ups/ups2rrd.py

#!/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")

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 create_dashboard():
    """Create a responsive HTML dashboard for the generated graphs."""

    generated_at = time.strftime("%Y-%m-%d %H:%M:%S %Z")

    graph_groups = [
        ("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),
            ],
        ),
    ]

    sections = []

    for title, graphs in graph_groups:
        cards = "\n".join(
            f"""<article class="card">
          <h3>{label}</h3>
          <a href="{os.path.basename(path)}">
            <img src="{os.path.basename(path)}" alt="{title} — {label}" loading="lazy">
          </a>
        </article>"""
            for label, path in graphs
        )
        sections.append(
            f"""<section>
      <h2>{title}</h2>
      <div class="grid">
        {cards}
      </div>
    </section>"""
        )

    document = f"""<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <meta http-equiv="refresh" content="60">
  <title>APC BX950MI UPS Dashboard</title>
  <style>
    :root {{ color-scheme: light dark; font-family: system-ui, sans-serif; }}
    body {{ margin: 0; background: #10141c; color: #edf2f7; }}
    header, main {{ width: min(1500px, calc(100% - 2rem)); margin: auto; }}
    header {{ padding: 2rem 0 1rem; }}
    h1, h2, h3, p {{ margin-top: 0; }}
    h1 {{ margin-bottom: .35rem; }}
    header p {{ color: #aab5c5; }}
    section {{ margin: 1.5rem 0 2.5rem; }}
    h2 {{ border-bottom: 1px solid #344054; padding-bottom: .5rem; }}
    .grid {{ display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 1rem; }}
    .card {{ overflow: hidden; border: 1px solid #344054; border-radius: 12px; background: #19202c; box-shadow: 0 8px 24px #0004; }}
    .card h3 {{ padding: 1rem 1rem 0; color: #cbd5e1; font-size: 1rem; }}
    .card a {{ display: block; }}
    .card img {{ display: block; width: 100%; height: auto; background: white; }}
    footer {{ padding: 0 0 2rem; color: #8c99aa; text-align: center; }}
    @media (max-width: 760px) {{
      .grid {{ grid-template-columns: 1fr; }}
    }}
  </style>
</head>
<body>
  <header>
    <h1>APC BX950MI UPS Dashboard</h1>
    <p>Updated {generated_at} · This page refreshes every 60 seconds.</p>
  </header>
  <main>
    {''.join(sections)}
  </main>
  <footer>Powered by NUT, Python and RRDtool</footer>
</body>
</html>
"""

    with open(HTML_FILE, "w", encoding="utf-8") as dashboard:
        dashboard.write(document)

    print(f"Created dashboard: {HTML_FILE}")


def main():
    create_rrd()

    update_rrd()

    create_graphs()

    create_dashboard()


if __name__ == "__main__":
    main()