From e121ff421fb101f04f05c7b88e96cb756f18bda2 Mon Sep 17 00:00:00 2001 From: Holden Oullette Date: Mon, 13 Oct 2025 18:19:24 -0600 Subject: [PATCH 1/2] Migrated upsc over to native pynutclient --- README.md | 2 +- config.example.yaml | 14 ++- pyproject.toml | 1 + tests/test_config.py | 196 +++++++++++++++++++++++++++++- tests/test_monitor.py | 271 ++++++++++++++++++++++++++++++++++++++++++ wolnut/cli.py | 4 +- wolnut/config.py | 59 ++++++++- wolnut/monitor.py | 63 ++++++---- 8 files changed, 577 insertions(+), 33 deletions(-) create mode 100644 tests/test_monitor.py diff --git a/README.md b/README.md index 18537f7..185b4a1 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ wolnut... get it? When a UPS (connected to NUT) switches to battery power, WOLNUT: -1. Detects the power event via `upsc` +1. Detects the power event via PyNUTClient library 2. Tracks which clients were online before the outage 3. Waits for power to be restored and the battery to reach a safe threshold 4. Sends WOL packets to bring back any systems that powered down diff --git a/config.example.yaml b/config.example.yaml index be3ae2b..482f339 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1,10 +1,18 @@ -# copy this to config.yaml befor running wolnut +# copy this to config.yaml before running wolnut log_level: INFO # Options: DEBUG, INFO, WARNING, ERROR, CRITICAL nut: - ups: "ups@localhost" # Format: @ + # UPS identifier - supports multiple formats: + # - Simple: "myups" (connects to localhost:3493) + # - With host: "myups@nut-server" (connects to nut-server:3493) + # - Full: "myups@nut-server:9999" (custom port) + # OR use separate fields: + ups: "ups@localhost" # UPS device name (or full format above) + # host: "localhost" # Optional: override/specify NUT server hostname + # port: 3493 # Optional: override/specify NUT server port + # timeout: 5 # Optional: connection timeout in seconds username: "upsmon" # Optional: omit if NUT server doesn't require auth - password: "password" + password: "password" # Optional: password for NUT authentication poll_interval: 15 # Poll interval in seconds — should be shorter than the NUT shutdown delay on any client status_file: "/config/wolnut_state.json" # Path to status file, recommended you change this to be outside container if using Docker diff --git a/pyproject.toml b/pyproject.toml index 1947b60..204d944 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,6 +9,7 @@ dependencies = [ "click>=8.2.1", "pyyaml>=6.0.2", "wakeonlan>=3.1.0", + "PyNUTClient>=2.4.2", ] dynamic = ["version"] diff --git a/tests/test_config.py b/tests/test_config.py index 67d1446..f190a70 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -5,6 +5,7 @@ from pathlib import Path from wolnut import config +from wolnut.config import parse_ups_identifier @pytest.fixture @@ -56,7 +57,8 @@ def test_load_config_minimal(mocker, minimal_config_dict): cfg = config.load_config("dummy_path.yaml", None, False) - assert cfg.nut.ups == "ups@localhost" + assert cfg.nut.ups == "ups" # Parsed from "ups@localhost" + assert cfg.nut.host == "localhost" assert cfg.poll_interval == 10 # Default assert cfg.wake_on.min_battery_percent == 20 # Default assert len(cfg.clients) == 1 @@ -81,7 +83,9 @@ def test_load_config_full(mocker, full_config_dict): assert cfg.log_level == "DEBUG" assert cfg.poll_interval == 5 assert cfg.status_file == "/data/status.json" - assert cfg.nut.ups == "myups@nut-server" + assert cfg.nut.ups == "myups" # Parsed from "myups@nut-server" + assert cfg.nut.host == "nut-server" + assert cfg.nut.port == 1234 # From config assert cfg.nut.username == "monuser" assert cfg.wake_on.restore_delay_sec == 60 assert cfg.wake_on.min_battery_percent == 50 @@ -213,3 +217,191 @@ def test_find_state_file(tmp_path, caplog): m.setattr(Path, "mkdir", MagicMock(side_effect=OSError("Permission denied"))) config.find_state_file(str(unwritable_path)) assert "Could not create directory for state file" in caplog.text + + +class TestParseUpsIdentifier: + """Tests for the parse_ups_identifier function""" + + def test_parse_simple_ups_name(self): + """Test parsing a simple UPS name without host""" + ups_name, host, port = parse_ups_identifier("myups") + assert ups_name == "myups" + assert host == "localhost" + assert port == 3493 + + def test_parse_ups_with_host(self): + """Test parsing UPS name with host""" + ups_name, host, port = parse_ups_identifier("myups@nut-server") + assert ups_name == "myups" + assert host == "nut-server" + assert port == 3493 + + def test_parse_ups_with_host_and_port(self): + """Test parsing UPS name with host and port""" + ups_name, host, port = parse_ups_identifier("myups@nut-server:9999") + assert ups_name == "myups" + assert host == "nut-server" + assert port == 9999 + + def test_parse_ups_with_ip_address(self): + """Test parsing UPS with IP address""" + ups_name, host, port = parse_ups_identifier("ups1@192.168.1.100:3493") + assert ups_name == "ups1" + assert host == "192.168.1.100" + assert port == 3493 + + def test_parse_ups_explicit_host_override(self): + """Test that explicit host parameter overrides parsed host""" + ups_name, host, port = parse_ups_identifier("myups@oldhost", explicit_host="newhost") + assert ups_name == "myups" + assert host == "newhost" + assert port == 3493 + + def test_parse_ups_explicit_port_override(self): + """Test that explicit port parameter overrides parsed port""" + ups_name, host, port = parse_ups_identifier("myups@host:8888", explicit_port=7777) + assert ups_name == "myups" + assert host == "host" + assert port == 7777 + + def test_parse_ups_both_overrides(self): + """Test that both explicit parameters override parsed values""" + ups_name, host, port = parse_ups_identifier("myups@oldhost:8888", + explicit_host="newhost", + explicit_port=7777) + assert ups_name == "myups" + assert host == "newhost" + assert port == 7777 + + def test_parse_ups_invalid_port(self, caplog): + """Test handling of invalid port in UPS identifier""" + ups_name, host, port = parse_ups_identifier("myups@host:notaport") + assert ups_name == "myups" + assert host == "host" + assert port == 3493 # Falls back to default + assert "Invalid port" in caplog.text + + def test_parse_ups_with_explicit_host_only(self): + """Test parsing with only explicit host provided""" + ups_name, host, port = parse_ups_identifier("myups", explicit_host="custom-host") + assert ups_name == "myups" + assert host == "custom-host" + assert port == 3493 + + def test_parse_ups_with_explicit_port_only(self): + """Test parsing with only explicit port provided""" + ups_name, host, port = parse_ups_identifier("myups", explicit_port=5555) + assert ups_name == "myups" + assert host == "localhost" + assert port == 5555 + + +class TestLoadConfigWithParsing: + """Tests for load_config with UPS identifier parsing""" + + def test_load_config_simple_ups_name(self, mocker): + """Test loading config with simple UPS name (no host specified)""" + config_dict = { + "nut": {"ups": "myups"}, + "clients": [{"name": "client1", "host": "192.168.1.10", "mac": "AA:BB:CC:DD:EE:FF"}] + } + mocker.patch("builtins.open", mocker.mock_open(read_data=yaml.dump(config_dict))) + mocker.patch("wolnut.config.validate_config") + mocker.patch("wolnut.config.resolve_mac_from_host") + + cfg = config.load_config("dummy.yaml", None, False) + + assert cfg.nut.ups == "myups" + assert cfg.nut.host == "localhost" + assert cfg.nut.port == 3493 + + def test_load_config_ups_with_host(self, mocker): + """Test loading config with UPS@host format""" + config_dict = { + "nut": {"ups": "myups@nut-server"}, + "clients": [{"name": "client1", "host": "192.168.1.10", "mac": "AA:BB:CC:DD:EE:FF"}] + } + mocker.patch("builtins.open", mocker.mock_open(read_data=yaml.dump(config_dict))) + mocker.patch("wolnut.config.validate_config") + mocker.patch("wolnut.config.resolve_mac_from_host") + + cfg = config.load_config("dummy.yaml", None, False) + + assert cfg.nut.ups == "myups" + assert cfg.nut.host == "nut-server" + assert cfg.nut.port == 3493 + + def test_load_config_ups_with_host_and_port(self, mocker): + """Test loading config with UPS@host:port format""" + config_dict = { + "nut": {"ups": "myups@nut-server:9999"}, + "clients": [{"name": "client1", "host": "192.168.1.10", "mac": "AA:BB:CC:DD:EE:FF"}] + } + mocker.patch("builtins.open", mocker.mock_open(read_data=yaml.dump(config_dict))) + mocker.patch("wolnut.config.validate_config") + mocker.patch("wolnut.config.resolve_mac_from_host") + + cfg = config.load_config("dummy.yaml", None, False) + + assert cfg.nut.ups == "myups" + assert cfg.nut.host == "nut-server" + assert cfg.nut.port == 9999 + + def test_load_config_explicit_host_overrides_parsed(self, mocker): + """Test that explicit host field overrides parsed host from ups string""" + config_dict = { + "nut": {"ups": "myups@oldhost", "host": "newhost"}, + "clients": [{"name": "client1", "host": "192.168.1.10", "mac": "AA:BB:CC:DD:EE:FF"}] + } + mocker.patch("builtins.open", mocker.mock_open(read_data=yaml.dump(config_dict))) + mocker.patch("wolnut.config.validate_config") + mocker.patch("wolnut.config.resolve_mac_from_host") + + cfg = config.load_config("dummy.yaml", None, False) + + assert cfg.nut.ups == "myups" + assert cfg.nut.host == "newhost" + + def test_load_config_explicit_port_overrides_parsed(self, mocker): + """Test that explicit port field overrides parsed port from ups string""" + config_dict = { + "nut": {"ups": "myups@host:8888", "port": 7777}, + "clients": [{"name": "client1", "host": "192.168.1.10", "mac": "AA:BB:CC:DD:EE:FF"}] + } + mocker.patch("builtins.open", mocker.mock_open(read_data=yaml.dump(config_dict))) + mocker.patch("wolnut.config.validate_config") + mocker.patch("wolnut.config.resolve_mac_from_host") + + cfg = config.load_config("dummy.yaml", None, False) + + assert cfg.nut.ups == "myups" + assert cfg.nut.port == 7777 + + def test_load_config_with_username_password(self, mocker): + """Test loading config with authentication credentials""" + config_dict = { + "nut": {"ups": "myups@nut-server", "username": "monitor", "password": "secret"}, + "clients": [{"name": "client1", "host": "192.168.1.10", "mac": "AA:BB:CC:DD:EE:FF"}] + } + mocker.patch("builtins.open", mocker.mock_open(read_data=yaml.dump(config_dict))) + mocker.patch("wolnut.config.validate_config") + mocker.patch("wolnut.config.resolve_mac_from_host") + + cfg = config.load_config("dummy.yaml", None, False) + + assert cfg.nut.username == "monitor" + assert cfg.nut.password == "secret" + + def test_load_config_with_custom_timeout(self, mocker): + """Test loading config with custom timeout""" + config_dict = { + "nut": {"ups": "myups", "timeout": 10}, + "clients": [{"name": "client1", "host": "192.168.1.10", "mac": "AA:BB:CC:DD:EE:FF"}] + } + mocker.patch("builtins.open", mocker.mock_open(read_data=yaml.dump(config_dict))) + mocker.patch("wolnut.config.validate_config") + mocker.patch("wolnut.config.resolve_mac_from_host") + + cfg = config.load_config("dummy.yaml", None, False) + + assert cfg.nut.timeout == 10 diff --git a/tests/test_monitor.py b/tests/test_monitor.py new file mode 100644 index 0000000..7c0b418 --- /dev/null +++ b/tests/test_monitor.py @@ -0,0 +1,271 @@ +import pytest +from unittest.mock import MagicMock, patch + +from wolnut.monitor import get_ups_status, is_client_online +from wolnut.config import NutConfig + + +@pytest.fixture +def sample_nut_config(): + """Sample NutConfig for testing""" + return NutConfig( + ups="myups", + host="localhost", + port=3493, + timeout=5, + username=None, + password=None + ) + + +@pytest.fixture +def sample_nut_config_with_auth(): + """Sample NutConfig with authentication""" + return NutConfig( + ups="secureups", + host="nut-server", + port=3493, + timeout=5, + username="monitor", + password="secret" + ) + + +class TestGetUpsStatus: + """Tests for get_ups_status function""" + + def test_get_ups_status_success(self, mocker, sample_nut_config): + """Test successful UPS status retrieval via PyNUTClient""" + mock_client = MagicMock() + mock_client.GetUPSVars.return_value = { + b'battery.charge': b'95.5', + b'ups.status': b'OL', + b'input.voltage': b'120.0', + b'ups.model': b'Smart UPS 1000' + } + mock_pynut_class = mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(sample_nut_config) + + assert result['battery.charge'] == '95.5' + assert result['ups.status'] == 'OL' + assert result['input.voltage'] == '120.0' + assert result['ups.model'] == 'Smart UPS 1000' + + mock_pynut_class.assert_called_once_with( + host='localhost', + port=3493, + login=None, + password=None, + timeout=5 + ) + mock_client.GetUPSVars.assert_called_once_with('myups') + + def test_get_ups_status_with_authentication(self, mocker, sample_nut_config_with_auth): + """Test UPS status retrieval with username and password""" + mock_client = MagicMock() + mock_client.GetUPSVars.return_value = { + b'battery.charge': b'100', + b'ups.status': b'OL' + } + mock_pynut_class = mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(sample_nut_config_with_auth) + + assert result['battery.charge'] == '100' + mock_pynut_class.assert_called_once_with( + host='nut-server', + port=3493, + login='monitor', + password='secret', + timeout=5 + ) + + def test_get_ups_status_with_string_values(self, mocker, sample_nut_config): + """Test handling of non-byte string values (some versions may return strings)""" + mock_client = MagicMock() + mock_client.GetUPSVars.return_value = { + 'battery.charge': '85.0', + 'ups.status': 'OB' + } + mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(sample_nut_config) + + assert result['battery.charge'] == '85.0' + assert result['ups.status'] == 'OB' + + def test_get_ups_status_mixed_types(self, mocker, sample_nut_config): + """Test handling of mixed byte strings and regular strings""" + mock_client = MagicMock() + mock_client.GetUPSVars.return_value = { + b'battery.charge': b'75.0', + 'ups.status': 'OB', + b'input.voltage': '118.5' + } + mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(sample_nut_config) + + assert result['battery.charge'] == '75.0' + assert result['ups.status'] == 'OB' + assert result['input.voltage'] == '118.5' + + def test_get_ups_status_pynut_error(self, mocker, sample_nut_config): + """Test handling of PyNUTError exceptions""" + from wolnut.monitor import PyNUTError + + mock_client = MagicMock() + mock_client.GetUPSVars.side_effect = PyNUTError("Connection refused") + mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(sample_nut_config) + + assert result == {} + + def test_get_ups_status_connection_error(self, mocker, sample_nut_config): + """Test handling of connection errors""" + mocker.patch('wolnut.monitor.PyNUTClient', side_effect=ConnectionError("Network unreachable")) + + result = get_ups_status(sample_nut_config) + + assert result == {} + + def test_get_ups_status_generic_exception(self, mocker, sample_nut_config): + """Test handling of generic exceptions""" + mock_client = MagicMock() + mock_client.GetUPSVars.side_effect = Exception("Unexpected error") + mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(sample_nut_config) + + assert result == {} + + def test_get_ups_status_invalid_ups_name(self, mocker, sample_nut_config): + """Test handling of invalid UPS name""" + from wolnut.monitor import PyNUTError + + mock_client = MagicMock() + mock_client.GetUPSVars.side_effect = PyNUTError("UNKNOWN-UPS") + mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(sample_nut_config) + + assert result == {} + + def test_get_ups_status_pynutclient_not_available(self, mocker, sample_nut_config): + """Test handling when PyNUTClient is not installed""" + # Simulate PyNUTClient not being available + mocker.patch('wolnut.monitor.PyNUTClient', None) + + result = get_ups_status(sample_nut_config) + + assert result == {} + + def test_get_ups_status_empty_response(self, mocker, sample_nut_config): + """Test handling of empty UPS variables response""" + mock_client = MagicMock() + mock_client.GetUPSVars.return_value = {} + mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(sample_nut_config) + + assert result == {} + + def test_get_ups_status_custom_port(self, mocker): + """Test connection with custom port""" + config = NutConfig(ups="myups", host="custom-host", port=9999) + mock_client = MagicMock() + mock_client.GetUPSVars.return_value = {b'ups.status': b'OL'} + mock_pynut_class = mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(config) + + mock_pynut_class.assert_called_once_with( + host='custom-host', + port=9999, + login=None, + password=None, + timeout=5 + ) + assert result['ups.status'] == 'OL' + + def test_get_ups_status_custom_timeout(self, mocker): + """Test connection with custom timeout""" + config = NutConfig(ups="myups", timeout=10) + mock_client = MagicMock() + mock_client.GetUPSVars.return_value = {b'ups.status': b'OL'} + mock_pynut_class = mocker.patch('wolnut.monitor.PyNUTClient', return_value=mock_client) + + result = get_ups_status(config) + + mock_pynut_class.assert_called_once_with( + host='localhost', + port=3493, + login=None, + password=None, + timeout=10 + ) + assert result['ups.status'] == 'OL' + + +class TestIsClientOnline: + """Tests for is_client_online function""" + + def test_is_client_online_success(self, mocker): + """Test successful ping to online host""" + mock_result = MagicMock() + mock_result.returncode = 0 + mocker.patch('subprocess.run', return_value=mock_result) + + result = is_client_online("192.168.1.100") + + assert result is True + + def test_is_client_online_failure(self, mocker): + """Test ping to offline host""" + mock_result = MagicMock() + mock_result.returncode = 1 + mocker.patch('subprocess.run', return_value=mock_result) + + result = is_client_online("192.168.1.100") + + assert result is False + + def test_is_client_online_exception(self, mocker): + """Test handling of exceptions during ping""" + mocker.patch('subprocess.run', side_effect=Exception("Network error")) + + result = is_client_online("192.168.1.100") + + assert result is False + + @patch('wolnut.monitor.platform.system') + def test_is_client_online_windows(self, mock_platform, mocker): + """Test ping command on Windows""" + mock_platform.return_value = "Windows" + mock_result = MagicMock() + mock_result.returncode = 0 + mock_run = mocker.patch('subprocess.run', return_value=mock_result) + + result = is_client_online("192.168.1.100") + + assert result is True + # Verify Windows flag (-n) is used + args = mock_run.call_args[0][0] + assert "-n" in args + + @patch('wolnut.monitor.platform.system') + def test_is_client_online_linux(self, mock_platform, mocker): + """Test ping command on Linux/Unix""" + mock_platform.return_value = "Linux" + mock_result = MagicMock() + mock_result.returncode = 0 + mock_run = mocker.patch('subprocess.run', return_value=mock_result) + + result = is_client_online("192.168.1.100") + + assert result is True + # Verify Unix flag (-c) is used + args = mock_run.call_args[0][0] + assert "-c" in args diff --git a/wolnut/cli.py b/wolnut/cli.py index 53a0196..29e8c26 100644 --- a/wolnut/cli.py +++ b/wolnut/cli.py @@ -46,13 +46,13 @@ def main(config_file: str, status_file: str, verbose: bool = False) -> int: restoration_event = True state_tracker.reset() - ups_status = get_ups_status(config.nut.ups) + ups_status = get_ups_status(config.nut) battery_percent = get_battery_percent(ups_status) power_status = ups_status.get("ups.status", "OL") logger.info("UPS power status: %s, Battery: %s%%", power_status, battery_percent) while True: - ups_status = get_ups_status(config.nut.ups) + ups_status = get_ups_status(config.nut) battery_percent = get_battery_percent(ups_status) power_status = ups_status.get("ups.status", "OL") diff --git a/wolnut/config.py b/wolnut/config.py index bb5f604..ecf0923 100644 --- a/wolnut/config.py +++ b/wolnut/config.py @@ -16,7 +16,8 @@ @dataclass class NutConfig: - ups: str + ups: str # UPS device name (e.g., "myups" or "myups@host:port" for backward compatibility) + host: str = "localhost" # NUT server hostname/IP port: int = 3493 timeout: int = 5 username: str | None = None @@ -48,6 +49,42 @@ class WolnutConfig: log_level: str = "INFO" +def parse_ups_identifier(ups_str: str, explicit_host: str | None = None, explicit_port: int | None = None) -> tuple[str, str, int]: + """ + Parse UPS identifier string which may be in formats: + - "myups" -> ("myups", "localhost", 3493) + - "myups@host" -> ("myups", "host", 3493) + - "myups@host:port" -> ("myups", "host", port) + + Explicit host/port from config override any parsed values. + + Returns: (ups_name, host, port) + """ + ups_name = ups_str + host = "localhost" + port = 3493 + + # Parse ups@host:port format + if "@" in ups_str: + ups_name, host_part = ups_str.split("@", 1) + if ":" in host_part: + host, port_str = host_part.split(":", 1) + try: + port = int(port_str) + except ValueError: + logger.warning("Invalid port in UPS identifier '%s', using default 3493", ups_str) + else: + host = host_part + + # Explicit config values override parsed values + if explicit_host: + host = explicit_host + if explicit_port: + port = explicit_port + + return ups_name, host, port + + def find_state_file(state_file: Optional[str] = None) -> str: """Find an existing state file or return a writable default path.""" path = Path(state_file or DEFAULT_STATE_FILEPATH) @@ -78,8 +115,24 @@ def load_config( logger.exception("Failed to load or parse config file: '%s'.\n", config_path) return None - # LOGGING... - nut = NutConfig(**raw["nut"]) + # Parse NUT configuration with UPS identifier parsing + raw_nut = raw["nut"] + ups_str = raw_nut["ups"] + explicit_host = raw_nut.get("host") + explicit_port = raw_nut.get("port") + + # Parse the UPS identifier (supports ups@host:port format) + ups_name, host, port = parse_ups_identifier(ups_str, explicit_host, explicit_port) + + # Create NutConfig with parsed values + nut = NutConfig( + ups=ups_name, + host=host, + port=port, + timeout=raw_nut.get("timeout", 5), + username=raw_nut.get("username"), + password=raw_nut.get("password") + ) # get wake_on or use defaults wake_on = WakeOnConfig(**raw.get("wake_on", {})) diff --git a/wolnut/monitor.py b/wolnut/monitor.py index 9af8763..fbfcacd 100644 --- a/wolnut/monitor.py +++ b/wolnut/monitor.py @@ -1,41 +1,60 @@ -import subprocess import logging import platform -from typing import Optional +import subprocess +from typing import TYPE_CHECKING + +try: + from PyNUT import PyNUTClient, PyNUTError +except ImportError: + PyNUTClient = None + PyNUTError = Exception + +if TYPE_CHECKING: + from wolnut.config import NutConfig logger = logging.getLogger("wolnut") -def get_ups_status( - ups_name: str, username: Optional[str] = None, password: Optional[str] = None -) -> dict: - env = None +def get_ups_status(nut_config: "NutConfig") -> dict: + """ + Get UPS status using PyNUTClient. - if username and password: - env = {**subprocess.os.environ, "USERNAME": username, "PASSWORD": password} + Args: + nut_config: NutConfig object containing connection details + + Returns: + Dictionary of UPS variables (str -> str), or empty dict on error + """ + if PyNUTClient is None: + logger.error("PyNUTClient is not available. Please install: pip install PyNUTClient") + return {} try: - result = subprocess.run( - ["upsc", ups_name], - capture_output=True, - text=True, - env=env, - timeout=5, - check=False, + # Connect to NUT server + client = PyNUTClient( + host=nut_config.host, + port=nut_config.port, + login=nut_config.username, + password=nut_config.password, + timeout=nut_config.timeout ) - if result.returncode != 0: - logger.error("upsc returned error: %s", result.stderr.strip()) - return {} + # Get all variables for the UPS + raw_vars = client.GetUPSVars(nut_config.ups) + # Convert byte strings to regular strings (Python 3 compatibility) status = {} - for line in result.stdout.splitlines(): - if ":" in line: - key, value = line.split(":", 1) - status[key.strip()] = value.strip() + for key, value in raw_vars.items(): + # Decode bytes to strings + str_key = key.decode('utf-8') if isinstance(key, bytes) else key + str_value = value.decode('utf-8') if isinstance(value, bytes) else value + status[str_key] = str_value return status + except PyNUTError as e: + logger.error("PyNUT error: %s", e) + return {} except Exception as e: logger.error("Failed to get UPS status: %s", e) return {} From 5eeb669d8422e91cb1ca65f1f95ed803aed5468d Mon Sep 17 00:00:00 2001 From: Holden Oullette Date: Mon, 13 Oct 2025 18:20:29 -0600 Subject: [PATCH 2/2] Latest pynutclient --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 204d944..4a16e1c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ dependencies = [ "click>=8.2.1", "pyyaml>=6.0.2", "wakeonlan>=3.1.0", - "PyNUTClient>=2.4.2", + "PyNUTClient>=2.8.4", ] dynamic = ["version"]