#!/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" HTML_FILE = os.path.join(GRAPH_DIR, "index.html") WEEKLY_HTML_FILE = os.path.join(GRAPH_DIR, "week.html") GRAPH_STATUS_1W = os.path.join(GRAPH_DIR, "status-1w.png") GRAPH_BATTERY_1W = os.path.join(GRAPH_DIR, "battery-1w.png") GRAPH_RUNTIME_1W = os.path.join(GRAPH_DIR, "runtime-1w.png") GRAPH_LOAD_1W = os.path.join(GRAPH_DIR, "load-1w.png") GRAPH_POWER_1W = os.path.join(GRAPH_DIR, "power-1w.png") 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", ] ) # --------------------------------------------------------------------------- # Graphs: one-week history # --------------------------------------------------------------------------- def graph_weekly_status(): """UPS status for the last week.""" now = int(time.time()) run_rrdtool( [ "graph", GRAPH_STATUS_1W, "--start", str(now - 7 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", "APC BX950MI - UPS Status (1 Week)", "--vertical-label", "Status", "--lower-limit", "0", "--upper-limit", "4", "--rigid", f"DEF:status={RRD_FILE}:status:LAST", "LINE2:status#0000FF:UPS status", "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", ] ) def graph_weekly_metric( output_file, title, vertical_label, definition, line, value_name, unit, limits=None, calculation=None, ): """Create a one-week graph for a numeric RRD data source.""" now = int(time.time()) command = [ "graph", output_file, "--start", str(now - 7 * 86400), "--end", str(now), "--width", "1000", "--height", "350", "--title", f"APC BX950MI - {title} (1 Week)", "--vertical-label", vertical_label, ] if limits: command.extend( [ "--lower-limit", str(limits[0]), "--upper-limit", str(limits[1]), ] ) command.append(definition) if calculation: command.append(calculation) command.extend( [ line, f"GPRINT:{value_name}:MIN:Minimum %.1lf{unit}", f"GPRINT:{value_name}:AVERAGE:Average %.1lf{unit}", f"GPRINT:{value_name}:MAX:Maximum %.1lf{unit}", ] ) run_rrdtool(command) def create_weekly_graphs(): """Generate one-week graphs for every collected value.""" graph_weekly_status() graph_weekly_metric( GRAPH_BATTERY_1W, "Battery Charge", "Charge (%)", f"DEF:charge={RRD_FILE}:battery_charge:AVERAGE", "LINE2:charge#008000:Battery charge", "charge", "%%", limits=(0, 100), ) graph_weekly_metric( GRAPH_RUNTIME_1W, "Battery Runtime", "Runtime (minutes)", f"DEF:runtime={RRD_FILE}:battery_runtime:AVERAGE", "LINE2:minutes#800080:Battery runtime", "minutes", " min", calculation="CDEF:minutes=runtime,60,/", ) graph_weekly_metric( GRAPH_LOAD_1W, "UPS Load", "Load (%)", f"DEF:load={RRD_FILE}:ups_load:AVERAGE", "LINE2:load#FF8000:UPS load", "load", "%%", limits=(0, 100), ) graph_weekly_metric( GRAPH_POWER_1W, "Nominal Power", "Power (W)", f"DEF:power={RRD_FILE}:realpower_nominal:AVERAGE", "LINE2:power#CC0000:Nominal power", "power", " 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 write_dashboard( html_file, page_title, graph_groups, other_page, other_page_label, ): """Write a responsive HTML dashboard for a set of graphs.""" generated_at = time.strftime("%Y-%m-%d %H:%M:%S %Z") sections = [] for title, graphs in graph_groups: cards = "\n".join( f"""

{label}

{title} — {label}
""" for label, path in graphs ) sections.append( f"""

{title}

{cards}
""" ) document = f""" {page_title}

{page_title}

Updated {generated_at} · This page refreshes every 60 seconds.

{''.join(sections)}
""" with open(html_file, "w", encoding="utf-8") as dashboard: dashboard.write(document) print(f"Created dashboard: {html_file}") def create_dashboards(): """Create the long-term and one-week HTML dashboards.""" write_dashboard( HTML_FILE, "APC BX950MI UPS Dashboard", [ ("UPS status", [("Last 3 months", GRAPH_STATUS)]), ( "Battery charge", [ ("Last year", GRAPH_BATTERY_1Y), ("Last 5 years", GRAPH_BATTERY_5Y), ], ), ( "Battery runtime", [ ("Last year", GRAPH_RUNTIME_1Y), ("Last 5 years", GRAPH_RUNTIME_5Y), ], ), ("UPS load", [("Last 3 months", GRAPH_LOAD)]), ( "Nominal power", [ ("Last year", GRAPH_POWER_1Y), ("Last 5 years", GRAPH_POWER_5Y), ], ), ], "week.html", "View one-week history", ) write_dashboard( WEEKLY_HTML_FILE, "APC BX950MI — One-Week History", [ ("UPS status", [("Last week", GRAPH_STATUS_1W)]), ("Battery charge", [("Last week", GRAPH_BATTERY_1W)]), ("Battery runtime", [("Last week", GRAPH_RUNTIME_1W)]), ("UPS load", [("Last week", GRAPH_LOAD_1W)]), ("Nominal power", [("Last week", GRAPH_POWER_1W)]), ], "index.html", "View long-term history", ) def main(): create_rrd() update_rrd() create_graphs() create_weekly_graphs() create_dashboards() if __name__ == "__main__": main()