#!/usr/bin/env python3 """ APC Back-UPS BX950MI RRD monitor Collects UPS information from NUT every minute and stores it in RRDTool. Retention: UPS status: 3 months @ 1 minute Battery charge: 1 year @ 5 minutes 5 years @ 1 hour Battery runtime: 1 year @ 5 minutes 5 years @ 1 hour UPS load: 3 months @ 1 minute Nominal power: 1 year @ 5 minutes 5 years @ 1 hour Requirements: pip install -r requirements.txt NUT's server and USB driver must also be installed and running. NUT UPS name: BX950MI """ import os import sys import time import rrdtool from PyNUTClient import PyNUT # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- UPS_NAME = "BX950MI" RRD_DIR = "/var/lib/rrd" RRD_FILE = os.path.join(RRD_DIR, "bx950mi.rrd") GRAPH_DIR = "/var/www/html/ups" GRAPH_STATUS = os.path.join(GRAPH_DIR, "status-3m.png") GRAPH_BATTERY_1Y = os.path.join( GRAPH_DIR, "battery-1y.png" ) GRAPH_BATTERY_5Y = os.path.join( GRAPH_DIR, "battery-5y.png" ) GRAPH_RUNTIME_1Y = os.path.join( GRAPH_DIR, "runtime-1y.png" ) GRAPH_RUNTIME_5Y = os.path.join( GRAPH_DIR, "runtime-5y.png" ) GRAPH_LOAD = os.path.join( GRAPH_DIR, "load-3m.png" ) GRAPH_POWER_1Y = os.path.join( GRAPH_DIR, "power-1y.png" ) GRAPH_POWER_5Y = os.path.join( GRAPH_DIR, "power-5y.png" ) # UPS status -> numeric value # # Status is inherently categorical, so RRD stores a number. # # 0 = OL On Line # 1 = OB On Battery # 2 = LB Low Battery # 3 = RB Replace Battery # 4 = UNKNOWN STATUS_VALUES = { "OL": 0, "OB": 1, "LB": 2, "RB": 3, "UNKNOWN": 4, } # --------------------------------------------------------------------------- # Utility functions # --------------------------------------------------------------------------- def run_rrdtool(args): """Call an RRDtool Python binding function.""" try: operation, *operation_args = args function = getattr(rrdtool, operation) return function(*operation_args) except Exception as exc: print( f"RRDTool error: {exc}", file=sys.stderr, ) raise # --------------------------------------------------------------------------- # NUT # --------------------------------------------------------------------------- def get_ups_values(): """Read all values from the local NUT server.""" try: client = PyNUT.PyNUTClient( host="127.0.0.1", port=3493, timeout=10, ) values = client.GetUPSVars(UPS_NAME) # PyNUTClient currently returns byte strings on Python 3. return { key.decode("utf-8") if isinstance(key, bytes) else key: value.decode("utf-8") if isinstance(value, bytes) else value for key, value in values.items() } except Exception as exc: print(f"NUT error: {exc}", file=sys.stderr) return {} def get_status(values): """ Read UPS status. NUT may return multiple flags, for example: OB LB We prioritize the more serious states. """ value = values.get("ups.status") if not value: return "UNKNOWN" flags = value.split() if "LB" in flags: return "LB" if "RB" in flags: return "RB" if "OB" in flags: return "OB" if "OL" in flags: return "OL" return "UNKNOWN" def get_float(values, field): """Read a numeric value from NUT.""" value = values.get(field) if value is None: return None try: return float(value) except ValueError: print( f"Invalid value for {field}: {value}", file=sys.stderr, ) return None # --------------------------------------------------------------------------- # RRD creation # --------------------------------------------------------------------------- def create_rrd(): """ Create the RRD database. The database is created only once. IMPORTANT: RRD databases cannot easily have their DS/RRA layout changed. If you change this layout later, remove the RRD and recreate it. """ if os.path.exists(RRD_FILE): return os.makedirs(RRD_DIR, exist_ok=True) now = int(time.time()) # # Start two minutes in the past. # start = now - 120 command = [ "create", RRD_FILE, "--start", str(start), "--step", "60", # # Data sources # # Heartbeat = 2 minutes. # "DS:status:GAUGE:120:0:4", "DS:battery_charge:GAUGE:120:0:100", "DS:battery_runtime:GAUGE:120:0:U", "DS:ups_load:GAUGE:120:0:100", "DS:realpower_nominal:GAUGE:120:0:U", # # --------------------------------------------------------------- # STATUS # --------------------------------------------------------------- # # 3 months @ 1 minute # # 90 * 24 * 60 = 129600 # "RRA:LAST:0.5:1:129600", # # --------------------------------------------------------------- # BATTERY CHARGE / RUNTIME / POWER # --------------------------------------------------------------- # # # 1 year @ 5 minutes # # 365 * 24 * 12 = 105120 # "RRA:AVERAGE:0.5:5:105120", # # 5 years @ 1 hour # # 5 * 365 * 24 = 43800 # "RRA:AVERAGE:0.5:60:43800", # # --------------------------------------------------------------- # UPS LOAD # --------------------------------------------------------------- # # 3 months @ 1 minute # "RRA:AVERAGE:0.5:1:129600", ] run_rrdtool(command) print(f"Created RRD: {RRD_FILE}") # --------------------------------------------------------------------------- # RRD update # --------------------------------------------------------------------------- def update_rrd(): """Read UPS values and update RRD.""" ups_values = get_ups_values() status = get_status(ups_values) battery_charge = get_float( ups_values, "battery.charge" ) battery_runtime = get_float( ups_values, "battery.runtime" ) ups_load = get_float( ups_values, "ups.load" ) realpower_nominal = get_float( ups_values, "ups.realpower.nominal" ) status_value = STATUS_VALUES.get( status, STATUS_VALUES["UNKNOWN"], ) def value_or_unknown(value): if value is None: return "U" return str(value) values = [ str(status_value), value_or_unknown( battery_charge ), value_or_unknown( battery_runtime ), value_or_unknown( ups_load ), value_or_unknown( realpower_nominal ), ] timestamp = int(time.time()) update_string = ( f"{timestamp}:" + ":".join(values) ) run_rrdtool( [ "update", RRD_FILE, update_string, ] ) print( time.strftime( "%Y-%m-%d %H:%M:%S" ), f"status={status}", f"battery={battery_charge}", f"runtime={battery_runtime}", f"load={ups_load}", f"nominal={realpower_nominal}", ) # --------------------------------------------------------------------------- # Graph: UPS status # --------------------------------------------------------------------------- def graph_status(): """Create 3-month UPS status graph.""" now = int(time.time()) run_rrdtool( [ "graph", GRAPH_STATUS, "--start", str(now - 90 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", "APC BX950MI - UPS Status (3 Months)", "--vertical-label", "Status", "--lower-limit", "0", "--upper-limit", "4", "--rigid", # # Use the LAST RRA. # f"DEF:status={RRD_FILE}:status:LAST", "LINE2:status#0000FF:UPS status", # # Labels. # "COMMENT:\\n", "COMMENT:0 = On Line (OL)\\n", "COMMENT:1 = On Battery (OB)\\n", "COMMENT:2 = Low Battery (LB)\\n", "COMMENT:3 = Replace Battery (RB)\\n", "COMMENT:4 = Unknown\\n", ] ) # --------------------------------------------------------------------------- # Graph: battery charge # --------------------------------------------------------------------------- def graph_battery_1y(): """Battery charge, 1 year @ 5 minutes.""" now = int(time.time()) run_rrdtool( [ "graph", GRAPH_BATTERY_1Y, "--start", str(now - 365 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", "APC BX950MI - Battery Charge (1 Year)", "--vertical-label", "Charge (%)", "--lower-limit", "0", "--upper-limit", "100", "DEF:charge=" f"{RRD_FILE}:battery_charge:AVERAGE", "LINE2:charge#008000:Battery charge", "GPRINT:charge:MIN:" "Minimum %.1lf%%", "GPRINT:charge:AVERAGE:" "Average %.1lf%%", "GPRINT:charge:MAX:" "Maximum %.1lf%%", ] ) def graph_battery_5y(): """Battery charge, 5 years @ 1 hour.""" now = int(time.time()) run_rrdtool( [ "graph", GRAPH_BATTERY_5Y, "--start", str(now - 5 * 365 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", "APC BX950MI - Battery Charge (5 Years)", "--vertical-label", "Charge (%)", "--lower-limit", "0", "--upper-limit", "100", "DEF:charge=" f"{RRD_FILE}:battery_charge:AVERAGE", "LINE2:charge#008000:Battery charge", "GPRINT:charge:MIN:" "Minimum %.1lf%%", "GPRINT:charge:AVERAGE:" "Average %.1lf%%", "GPRINT:charge:MAX:" "Maximum %.1lf%%", ] ) # --------------------------------------------------------------------------- # Graph: battery runtime # --------------------------------------------------------------------------- def graph_runtime_1y(): """Battery runtime, 1 year @ 5 minutes.""" now = int(time.time()) run_rrdtool( [ "graph", GRAPH_RUNTIME_1Y, "--start", str(now - 365 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", "APC BX950MI - Battery Runtime (1 Year)", "--vertical-label", "Runtime (minutes)", "DEF:runtime=" f"{RRD_FILE}:battery_runtime:AVERAGE", # # Convert seconds to minutes. # "CDEF:minutes=runtime,60,/", "LINE2:minutes#800080:Battery runtime", "GPRINT:minutes:MIN:" "Minimum %.1lf min", "GPRINT:minutes:AVERAGE:" "Average %.1lf min", "GPRINT:minutes:MAX:" "Maximum %.1lf min", ] ) def graph_runtime_5y(): """Battery runtime, 5 years @ 1 hour.""" now = int(time.time()) run_rrdtool( [ "graph", GRAPH_RUNTIME_5Y, "--start", str(now - 5 * 365 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", "APC BX950MI - Battery Runtime (5 Years)", "--vertical-label", "Runtime (minutes)", "DEF:runtime=" f"{RRD_FILE}:battery_runtime:AVERAGE", "CDEF:minutes=runtime,60,/", "LINE2:minutes#800080:Battery runtime", "GPRINT:minutes:MIN:" "Minimum %.1lf min", "GPRINT:minutes:AVERAGE:" "Average %.1lf min", "GPRINT:minutes:MAX:" "Maximum %.1lf min", ] ) # --------------------------------------------------------------------------- # Graph: UPS load # --------------------------------------------------------------------------- def graph_load(): """UPS load, 3 months @ 1 minute.""" now = int(time.time()) run_rrdtool( [ "graph", GRAPH_LOAD, "--start", str(now - 90 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", "APC BX950MI - UPS Load (3 Months)", "--vertical-label", "Load (%)", "--lower-limit", "0", "--upper-limit", "100", f"DEF:load={RRD_FILE}:ups_load:AVERAGE", "LINE2:load#FF8000:UPS load", "GPRINT:load:MIN:" "Minimum %.1lf%%", "GPRINT:load:AVERAGE:" "Average %.1lf%%", "GPRINT:load:MAX:" "Maximum %.1lf%%", ] ) # --------------------------------------------------------------------------- # Graph: nominal power # --------------------------------------------------------------------------- def graph_power_1y(): """Nominal power, 1 year @ 5 minutes.""" now = int(time.time()) run_rrdtool( [ "graph", GRAPH_POWER_1Y, "--start", str(now - 365 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", "APC BX950MI - Nominal Power (1 Year)", "--vertical-label", "Power (W)", "DEF:power=" f"{RRD_FILE}:realpower_nominal:AVERAGE", "LINE2:power#CC0000:Nominal power", "GPRINT:power:MIN:" "Minimum %.1lf W", "GPRINT:power:AVERAGE:" "Average %.1lf W", "GPRINT:power:MAX:" "Maximum %.1lf W", ] ) def graph_power_5y(): """Nominal power, 5 years @ 1 hour.""" now = int(time.time()) run_rrdtool( [ "graph", GRAPH_POWER_5Y, "--start", str(now - 5 * 365 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", "APC BX950MI - Nominal Power (5 Years)", "--vertical-label", "Power (W)", "DEF:power=" f"{RRD_FILE}:realpower_nominal:AVERAGE", "LINE2:power#CC0000:Nominal power", "GPRINT:power:MIN:" "Minimum %.1lf W", "GPRINT:power:AVERAGE:" "Average %.1lf W", "GPRINT:power:MAX:" "Maximum %.1lf W", ] ) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def create_graphs(): """Generate all graphs.""" os.makedirs(GRAPH_DIR, exist_ok=True) graph_status() graph_battery_1y() graph_battery_5y() graph_runtime_1y() graph_runtime_5y() graph_load() graph_power_1y() graph_power_5y() def main(): create_rrd() update_rrd() create_graphs() if __name__ == "__main__": main()