From 47f077a461e96640fedfba001f68d1b42e77cb73 Mon Sep 17 00:00:00 2001 From: steven Date: Sat, 30 May 2026 14:29:33 +1000 Subject: [PATCH 1/2] Added APC UPS SNMP Monitor --- docs/monitors/apcupssnmp.rst | 55 +++++ poetry.lock | 18 +- pyproject.toml | 1 + simplemonitor/Monitors/__init__.py | 2 + simplemonitor/Monitors/apcupssnmp.py | 345 +++++++++++++++++++++++++++ 5 files changed, 419 insertions(+), 2 deletions(-) create mode 100644 docs/monitors/apcupssnmp.rst create mode 100644 simplemonitor/Monitors/apcupssnmp.py diff --git a/docs/monitors/apcupssnmp.rst b/docs/monitors/apcupssnmp.rst new file mode 100644 index 00000000..b7297f25 --- /dev/null +++ b/docs/monitors/apcupssnmp.rst @@ -0,0 +1,55 @@ +apcupssnmp - APC UPS status via NMC +^^^^^^^^^^^^^^^^^^^^^^^^ + +Connects directly to an APC UPS with an NMC (Network Management Card) installed, using SNMP. + +Currently only supports SNMP v1 + + +.. confval:: host + + :type: string + :required: true + + The hostname or IP address of the UPS + +.. confval:: community + + :type: string + :required: false + :default: ``public`` + + The SNMP Community name to use when connecting to the UPS. + +.. conval:: maxloadpct + + :type: int + :required: false + :default: ``80`` + + The Load Percentage of the UPS to be OK. A value over 100 will never trigger this warning. + +.. conval:: battpctwarn + + :type: int + :required: false + :default: ``20`` + + The minimum Battery Percentage warning, above this number the battery is deemed OK. Monitor will fail if battery is under 20%. Useful to ensure the monitor stays in a failed state until the battery has recovered enough after a power outage. + +.. conval:: runtimemin + + :type: int + :required: false + :default: ``10`` + + The minimum allowed runtime of the UPS, defaults to 10 (10 minutes). This default is used to allow enough time to trigger shutdown of systems. + +.. conval:: textdelimeter + + :type: string + :required: false + :default: ``,`` + + The text to use to seperate each item in the Details when returned from the test. Defaults to using a comma ','. + Useful if you want to use this text in another system, or make it easier to read diff --git a/poetry.lock b/poetry.lock index c3ce3e48..dd718c17 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1179,11 +1179,11 @@ description = "Cross-platform colored terminal text." optional = false python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7" groups = ["main", "dev"] -markers = "sys_platform == \"win32\" or platform_system == \"Windows\"" files = [ {file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"}, {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"}, ] +markers = {main = "sys_platform == \"win32\" or platform_system == \"Windows\"", dev = "platform_system == \"Windows\" or sys_platform == \"win32\""} [[package]] name = "colorlog" @@ -3015,6 +3015,20 @@ files = [ {file = "sniffio-1.3.1.tar.gz", hash = "sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc"}, ] +[[package]] +name = "snmp" +version = "1.2.1" +description = "A user-friendly SNMP library" +optional = false +python-versions = ">=3.6" +groups = ["main"] +files = [ + {file = "snmp-1.2.1.tar.gz", hash = "sha256:b958d73d92ff003346ce3419447eb7413d8927a4862f25eae6b909cbec2d6dd3"}, +] + +[package.dependencies] +pycryptodome = ">=3.4,<4.0" + [[package]] name = "snowballstemmer" version = "2.2.0" @@ -3719,4 +3733,4 @@ propcache = ">=0.2.0" [metadata] lock-version = "2.1" python-versions = "^3.10.0" -content-hash = "4ec306478a471b8c6e80282f1d715d5c1f578eade33d57fd54c94fff3d4e8e53" +content-hash = "c14a63cde05fcbf70ef435ba4c3d8d4b83ec19b8ad68d826afae51cf608aa637" diff --git a/pyproject.toml b/pyproject.toml index c1637f5d..104fd2af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -55,6 +55,7 @@ paramiko = ">=2.7.2,<4.0.0" pyaarlo = ">=0.8.0.19,<0.9.0.0" icmplib = "^3.0.3" packaging = "^26.2" +snmp = "^1.2.1" [tool.poetry.group.dev.dependencies] boto3-stubs = {extras = ["sns"], version = "^1.43.10"} diff --git a/simplemonitor/Monitors/__init__.py b/simplemonitor/Monitors/__init__.py index 94a6c543..0bb78bb7 100644 --- a/simplemonitor/Monitors/__init__.py +++ b/simplemonitor/Monitors/__init__.py @@ -2,6 +2,7 @@ Monitors for SimpleMonitor """ +from .apcupssnmp import MonitorAPCUPSSNMP from .arlo import MonitorArloCamera from .compound import CompoundMonitor, RemoteHostsMonitor from .file import MonitorBackup @@ -44,6 +45,7 @@ from .unifi import MonitorUnifiFailover, MonitorUnifiFailoverWatchdog __all__ = [ + "MonitorAPCUPSNMP", "MonitorMQTT", "CompoundMonitor", "MonitorApcupsd", diff --git a/simplemonitor/Monitors/apcupssnmp.py b/simplemonitor/Monitors/apcupssnmp.py new file mode 100644 index 00000000..d2472df1 --- /dev/null +++ b/simplemonitor/Monitors/apcupssnmp.py @@ -0,0 +1,345 @@ +""" +Checks an APC UPS via SNMP using an APC NMC (Network Management card) +""" + +from typing import Tuple, cast + +import snmp + +from .monitor import Monitor, register + +# General SNMP OIDs +apcupssnmp_oid_upstype = ".1.3.6.1.4.1.318.1.1.1.1.1.1.0" # UPS Model Information +# Status +apcupssnmp_oid_battery_capacity = ( + ".1.3.6.1.4.1.318.1.1.1.2.2.1.0" # current Capacity in % +) +apcupssnmp_oid_battery_runtimeremain = ( + ".1.3.6.1.4.1.318.1.1.1.2.2.3.0" # runtime remain in timeticks +) +apcupssnmp_oid_output_load = ( + ".1.3.6.1.4.1.318.1.1.1.4.2.3.0" # Output load in percentage +) +apcupssnmp_oid_upsBasicStateOutputState = ( + ".1.3.6.1.4.1.318.1.1.1.11.1.1.0" # 64-bit binary String +) +apcupssnmp_oid_upsOutputStatus = ".1.3.6.1.4.1.318.1.1.1.4.1.1.0" # Output Status +# Health +apcupssnmp_oid_test_result = ".1.3.6.1.4.1.318.1.1.1.7.2.3.0" # UPS Self Test Result + + +@register +class MonitorAPCUPSSNMP(Monitor): + """ + Monitor APC UPS via Direct SNMP queries to management card + """ + + monitor_type = "apcupssnmp" + + def __init__(self, name: str, config_options: dict) -> None: + super().__init__(name, config_options) + self.host = cast(str, self.get_config_option("host", required=True)) + self.community = cast( + str, self.get_config_option("community", required=False, default="public") + ) + self.maxloadpct = cast( + int, self.get_config_option("maxloadpct", required=False, default=80) + ) + self.battpctwarn = cast( + int, self.get_config_option("battpctwarn", required=False, default=20) + ) + self.runtimemin = cast( + int, self.get_config_option("runtimemin", required=False, default=10) + ) + self.textdelimeter = cast( + str, self.get_config_option("textdelimeter", required=False, default=",") + ) + self.snmpengine = snmp.Engine(snmp.SNMPv1) + self.snmphost = self.snmpengine.Manager( + self.host, community=self.community.encode("utf-8") + ) + + def DecodeBasicStateOutput(self, error_state, state): + """decode the 64-bit string encoded state to text""" + BasicStateOutputTableText = [ + "Abnormal Condition", + "Running On Battery", + "LowBattery", + "OnLine", + "Replace Battery", + "Comm:OK", + "AVR Boost (Low Input V)", + "AVR Trim (High Input V)", + "Overload", + "Runtime Calibration", + "Batteries Discharged", + "Manual Bypass", + "Software Bypass", + "Bypass - Internal Fault", + "Bypass - Supply Failure", + "Bypass - Fan Failure", + "Sleeping on a Timer", + "Sleeping until Utility Power Returns", + "Out:On", + "Rebooting", + "Batt Comm Lost", + "Graceful Shutdown Initiated", + "Smart Boost/Trim Fault", + "Bad Output Voltage", + "Batt Charger Failure", + "High Batt Temperature", + "Warning Batt Temperature", + "Critical Batt Temperature", + "Self Test In Progress", + "Low Batt / On Batt", + "Graceful Shutdown Issued by Upstream Device", + "Graceful Shutdown Issued by Downstream Device", + "No Batteries Attached", + "Synchronized Command is in Progress", + "Synchronized Sleeping Command is in Progress", + "Synchronized Rebooting Command is in Progress", + "Inverter DC Imbalance", + "Transfer Relay Failure", + "Shutdown or Unable to Transfer", + "Low Batt Shutdown", + "Electronic Unit Fan Failure", + "Main Relay Failure", + "Bypass Relay Failure", + "Temporary Bypass", + "High Internal Temp", + "Batt Temp Sensor Fault", + "Input Out of Range for Bypass", + "DC Bus Overvoltage", + "PFC Failure", + "Critical Hardware Fault", + "Green/ECO Mode", + "Hot Standby", + "EPO Activated", # Emergency Power Off + "Load Alarm Violation", + "Bypass Phase Fault", + "UPS Internal Comm Failure", + "Efficiency Booster Mode", + "Off", + "Standby", + ] + BasicStateOutputTableError = [ + 1, + 1, + 1, + 0, + 0, + 0, + 0, + 0, + 1, + 0, + 1, + 1, + 1, + 1, + 1, + 1, + 0, + 0, + 0, + 0, + 0, + 0, + 1, + 1, + 1, + 1, + 1, + 2, + 0, + 1, + 0, + 0, + 1, + 0, + 0, + 0, + 0, + 1, + 0, + 2, + 1, + 2, + 1, + 0, + 1, + 1, + 1, + 1, + 1, + 2, + 0, + 0, + 2, + 0, + 0, + 1, + 0, + 1, + 1, + ] + bitlocation = 0 + text = "" + for bit in state: + if state[bitlocation] == "1": + state_text = BasicStateOutputTableText[bitlocation] + state_status = BasicStateOutputTableError[bitlocation] + if len(text): + text = text + self.textdelimeter + state_text + else: + text = state_text + # Increase error level as needed - 0 is OK, 1 = Warning, 2 = Error + error_state = max(error_state, state_status) + bitlocation += 1 + return error_state, text + + def decodeoutputstatus(self, error_state, val): + OutputStatusTableText = [ + "unknown", + "OnLine", + "OnBattery", + "OnSmartBoost", + "TimedSleeping", + "SoftwareBypass", + "OFF", + "Rebooting", + "SwitchedBypass", + "HardwareFailureBypass", + "SleepingUntilPowerReturn", + "OnSmartTrim", + "EcoMode", + "HotStandby", + "OnBatteryTest", + "EmergencyStaticBypass", + "StaticBypassStandby", + "PowerSavingMode", + "SpotMode", + "eConversion", + ] + OutputStatusTableError = [ + 1, + 0, + 1, + 0, + 1, + 1, + 2, + 1, + 1, + 2, + 1, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + ] + return OutputStatusTableError[val - 1] + error_state, OutputStatusTableText[ + val - 1 + ] + + def decodeselftest(self, error_state, val): + """Decode the Last Calibration value to text""" + tabletext = ["Pass", "Never", "InProg", "Never"] + tableerror = [0, 1, 0, 1] + return tableerror[val - 1] + error_state, tabletext[val - 1] + + def snmpresulttext(self, result): + for name, val in result: + # print(f"X:{dir(val)}:X") + if isinstance(val, snmp.smi.Gauge32): + print(f"State: {name} @ {int(val.value)}") + elif isinstance(val, snmp.smi.OctetString): + print(f"State: {name} # {val.data.decode('utf-8')}") + elif isinstance(val, snmp.smi.Integer32): + print(f"State: {name} @ {int(val.value)}") + elif isinstance(val, snmp.smi.TimeTicks): + print(f"State: {name} % {int(val.value / 6000)}m") + else: + print(f"State: {name} $ {str(val)}") + + def processbattpct(self, error_state, batt_pct): + if batt_pct < self.battpctwarn: + error_state += 1 + return error_state, batt_pct + + def processloadpct(self, error_state, load_pct): + if load_pct > self.maxloadpct: + error_state += 1 + return error_state, load_pct + + def processruntime(self, error_state, runtimeval): + runtime = int(runtimeval / 6000) # Convert to minutes + if runtime < self.runtimemin: + error_state += 1 + return error_state, str(runtime) + "m" + + def geterrorstatetext(self, error_state): + if error_state > 1: + error_text = f"CRIT{error_state}" + elif error_state == 1: + error_text = "WARN" + else: + error_text = "OK" + return error_text + " - " + + def run_test(self) -> bool: + state = self.snmphost.get( + apcupssnmp_oid_upstype, + apcupssnmp_oid_upsBasicStateOutputState, + apcupssnmp_oid_battery_capacity, + apcupssnmp_oid_output_load, + apcupssnmp_oid_upsOutputStatus, + apcupssnmp_oid_test_result, + apcupssnmp_oid_battery_runtimeremain, + ) + error_state = 0 + ups_type = state[0].value.data.decode("utf-8") + error_state, status_state = self.DecodeBasicStateOutput( + error_state, state[1].value.data.decode("utf-8") + ) + error_state, batt_pct = self.processbattpct(error_state, state[2].value.value) + error_state, load_pct = self.processloadpct(error_state, state[3].value.value) + error_state, outputstate = self.decodeoutputstatus( + error_state, state[4].value.value + ) + error_state, selftest = self.decodeselftest(error_state, state[5].value.value) + error_state, runtime = self.processruntime(error_state, state[6].value.value) + error_text = self.geterrorstatetext(error_state) + output_text = error_text + self.textdelimeter.join( + [ + f"{ups_type}:{status_state}", + f"Batt:{batt_pct}%", + f"Load:{load_pct}%", + f"Runtime:{runtime}", + f"OutStatus:{outputstate}", + f"Test:{selftest}", + ] + ) + # print(output_text) + if error_state: + return self.record_fail(output_text) + return self.record_success(output_text) + + def get_params(self) -> Tuple: + return ( + self.host, + self.community, + self.maxloadpct, + self.battpctwarn, + self.runtimemin, + self.textdelimeter, + self.snmpengine, + ) + + def describe(self) -> str: + return "Checking Status & Health of APC UPS {} via SNMP".format(self.host) From ad51f2c7af831b5b52c2f67d1871c281132998dc Mon Sep 17 00:00:00 2001 From: steven Date: Sun, 31 May 2026 10:50:22 +1000 Subject: [PATCH 2/2] Fixed Monitors module import list name --- simplemonitor/Monitors/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/simplemonitor/Monitors/__init__.py b/simplemonitor/Monitors/__init__.py index 0bb78bb7..48c9b2ab 100644 --- a/simplemonitor/Monitors/__init__.py +++ b/simplemonitor/Monitors/__init__.py @@ -45,7 +45,7 @@ from .unifi import MonitorUnifiFailover, MonitorUnifiFailoverWatchdog __all__ = [ - "MonitorAPCUPSNMP", + "MonitorAPCUPSSNMP", "MonitorMQTT", "CompoundMonitor", "MonitorApcupsd",