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 asyncio
import json
import os
import sys
import time

import paramiko
import rrdtool
from dbus_next.aio import MessageBus
from dbus_next.constants import BusType
from PyNUTClient import PyNUT
from ntfy_notify import send_notification


# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

UPS_NAME = "BX950MI"

RRD_DIR = "/var/lib/rrd"
RRD_FILE = os.path.join(RRD_DIR, "bx950mi.rrd")
ALERT_STATE_FILE = os.path.join(RRD_DIR, "bx950mi-alert-state.json")

SHUTDOWN_HOST = os.environ.get("UPS_SHUTDOWN_HOST", "plex.local")
SHUTDOWN_USER = os.environ.get("UPS_SHUTDOWN_USER", "ups")
SHUTDOWN_KEY_FILE = os.environ.get(
    "UPS_SHUTDOWN_KEY",
    "/opt/ups/.ssh/plex_shutdown",
)
SHUTDOWN_KNOWN_HOSTS = os.environ.get(
    "UPS_SHUTDOWN_KNOWN_HOSTS",
    "/opt/ups/.ssh/known_hosts",
)
SHUTDOWN_COMMAND = "sudo /usr/sbin/shutdown -h now"
REMOTE_SHUTDOWN_RUNTIME_SECONDS = 5 * 60
LOCAL_SHUTDOWN_RUNTIME_SECONDS = 200
BATTERY_CHARGE_MEAN_SAMPLES = 10

GRAPH_COLOR_ARGS = [
    "--color", "BACK#F7F5EFFF",
    "--color", "CANVAS#FFFFFFFF",
]

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}",
    )

    return {
        "status": status,
        "battery_charge": battery_charge,
        "battery_runtime": battery_runtime,
    }


# ---------------------------------------------------------------------------
# Notifications
# ---------------------------------------------------------------------------

def load_alert_state():
    """Load values saved by the previous collector run."""

    try:
        with open(ALERT_STATE_FILE, encoding="utf-8") as state_file:
            return json.load(state_file)
    except FileNotFoundError:
        return {}
    except (OSError, ValueError) as exc:
        print(f"Could not read alert state: {exc}", file=sys.stderr)
        return {}


def save_alert_state(state):
    """Atomically save values for the next collector run."""

    temporary_file = f"{ALERT_STATE_FILE}.tmp"

    try:
        with open(temporary_file, "w", encoding="utf-8") as state_file:
            json.dump(state, state_file, indent=2, sort_keys=True)
            state_file.write("\n")
        os.replace(temporary_file, ALERT_STATE_FILE)
    except OSError as exc:
        print(f"Could not save alert state: {exc}", file=sys.stderr)


def notify(message):
    """Send an alert without stopping UPS data collection on failure."""

    try:
        send_notification(message)
        print(f"Notification sent: {message}")
    except Exception as exc:
        print(f"Notification failed: {exc}", file=sys.stderr)


def crossed_below(previous, current, threshold):
    """Return whether a numeric value crossed below a threshold."""

    return (
        previous is not None
        and current is not None
        and previous >= threshold
        and current < threshold
    )


def battery_charge_history(previous, current_charge):
    """Return up to the last 10 valid battery-charge measurements."""

    history = previous.get("battery_charge_history", [])
    if not isinstance(history, list):
        history = []
    history = [
        value
        for value in history
        if isinstance(value, (int, float)) and not isinstance(value, bool)
    ]

    if current_charge is not None:
        history.append(current_charge)

    return history[-BATTERY_CHARGE_MEAN_SAMPLES:]


def complete_charge_mean(history):
    """Return the mean only when a complete charge window is available."""

    if len(history) < BATTERY_CHARGE_MEAN_SAMPLES:
        return None

    return sum(history) / len(history)


def shutdown_remote_server(
    host=SHUTDOWN_HOST,
    user=SHUTDOWN_USER,
    key_file=SHUTDOWN_KEY_FILE,
    known_hosts=SHUTDOWN_KNOWN_HOSTS,
    timeout=10,
):
    """Shut down a remote Linux server over restricted, verified SSH."""

    client = paramiko.SSHClient()
    client.load_host_keys(known_hosts)
    client.set_missing_host_key_policy(paramiko.RejectPolicy())

    try:
        client.connect(
            hostname=host,
            username=user,
            key_filename=key_file,
            timeout=timeout,
            banner_timeout=timeout,
            auth_timeout=timeout,
            look_for_keys=False,
            allow_agent=False,
        )
        client.exec_command(SHUTDOWN_COMMAND, timeout=timeout)
    finally:
        client.close()


async def request_local_poweroff():
    """Request a local power-off through systemd-logind over D-Bus."""

    bus = await MessageBus(bus_type=BusType.SYSTEM).connect()

    try:
        introspection = await bus.introspect(
            "org.freedesktop.login1",
            "/org/freedesktop/login1",
        )
        login = bus.get_proxy_object(
            "org.freedesktop.login1",
            "/org/freedesktop/login1",
            introspection,
        )
        manager = login.get_interface("org.freedesktop.login1.Manager")
        await manager.call_power_off(False)
    finally:
        bus.disconnect()


def shutdown_local_server():
    """Shut down the Linux server running this collector."""

    asyncio.run(request_local_poweroff())


def process_alerts(current):
    """Notify on status changes and configured threshold crossings."""

    previous = load_alert_state()
    previous_status = previous.get("status")
    current_status = current.get("status")
    previous_charge = previous.get("battery_charge")
    current_charge = current.get("battery_charge")
    previous_charge_history = battery_charge_history(previous, None)
    charge_history = battery_charge_history(previous, current_charge)
    previous_charge_mean = complete_charge_mean(previous_charge_history)
    current_charge_mean = complete_charge_mean(charge_history)
    previous_runtime = previous.get("battery_runtime")
    current_runtime = current.get("battery_runtime")
    shutdown_sent = previous.get("remote_shutdown_sent", False)
    local_shutdown_sent = previous.get("local_shutdown_sent", False)

    # An empty state is the initial baseline, not an alert condition.
    if previous:
        if (
            previous_status
            and current_status
            and current_status != previous_status
        ):
            notify(
                f"UPS status changed: {previous_status} -> {current_status}"
            )

        if current_charge_mean is not None and (
            previous_charge_mean is None
            or crossed_below(previous_charge_mean, current_charge_mean, 50)
        ) and current_charge_mean < 50:
            notify(
                "UPS mean battery charge over the last 10 measurements "
                f"is below 50%: {current_charge_mean:.1f}%"
            )

        if current_charge_mean is not None and (
            previous_charge_mean is None
            or crossed_below(previous_charge_mean, current_charge_mean, 5)
        ) and current_charge_mean < 5:
            notify(
                "UPS mean battery charge over the last 10 measurements "
                f"is critically low (below 5%): {current_charge_mean:.1f}%"
            )

        if crossed_below(previous_runtime, current_runtime, 5 * 60):
            notify(
                "UPS battery runtime is below 5 minutes: "
                f"{current_runtime / 60:.1f} minutes"
            )

        shutdown_required = (
            current_status in {"OB", "LB"}
            and current_runtime is not None
            and current_runtime < REMOTE_SHUTDOWN_RUNTIME_SECONDS
        )

        if shutdown_required and not shutdown_sent:
            try:
                shutdown_remote_server()
                shutdown_sent = True
                message = (
                    f"Remote shutdown requested for {SHUTDOWN_HOST}: "
                    f"UPS runtime is {current_runtime / 60:.1f} minutes"
                )
                print(message)
                notify(message)
            except Exception as exc:
                print(
                    f"Remote shutdown of {SHUTDOWN_HOST} failed: {exc}",
                    file=sys.stderr,
                )

        local_shutdown_required = (
            current_status in {"OB", "LB"}
            and current_runtime is not None
            and current_runtime < LOCAL_SHUTDOWN_RUNTIME_SECONDS
        )

        if local_shutdown_required and not local_shutdown_sent:
            message = (
                "Raspberry Pi shutdown requested: UPS runtime is "
                f"{current_runtime:.0f} seconds"
            )
            notify(message)

            try:
                shutdown_local_server()
                local_shutdown_sent = True
                print(message)
            except Exception as exc:
                print(
                    f"Local shutdown failed: {exc}",
                    file=sys.stderr,
                )

    # A return to utility power starts a new shutdown-alert cycle.
    if current_status == "OL":
        shutdown_sent = False
        local_shutdown_sent = False

    # Keep the last valid reading when NUT temporarily omits a value.
    state = {
        "status": current_status or previous_status,
        "battery_charge": (
            current_charge if current_charge is not None else previous_charge
        ),
        "battery_charge_history": charge_history,
        "battery_runtime": (
            current_runtime if current_runtime is not None else previous_runtime
        ),
        "remote_shutdown_sent": shutdown_sent,
        "local_shutdown_sent": local_shutdown_sent,
    }
    save_alert_state(state)


# ---------------------------------------------------------------------------
# Graph: UPS status
# ---------------------------------------------------------------------------

def graph_status():
    """Create 3-month UPS status graph."""

    now = int(time.time())

    run_rrdtool(
        [
            "graph",
            GRAPH_STATUS,
            *GRAPH_COLOR_ARGS,

            "--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,
            *GRAPH_COLOR_ARGS,

            "--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,
            *GRAPH_COLOR_ARGS,

            "--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,
            *GRAPH_COLOR_ARGS,

            "--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,
            *GRAPH_COLOR_ARGS,

            "--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,
            *GRAPH_COLOR_ARGS,

            "--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,
            *GRAPH_COLOR_ARGS,

            "--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,
            *GRAPH_COLOR_ARGS,

            "--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,
            *GRAPH_COLOR_ARGS,
            "--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,
        *GRAPH_COLOR_ARGS,
        "--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"""<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>{page_title}</title>
  <style>
    :root {{ color-scheme: light; font-family: system-ui, sans-serif; }}
    body {{ margin: 0; background: #f7f5ef; color: #28303d; }}
    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: #667085; }}
    nav {{ margin-top: 1rem; }}
    nav a {{ display: inline-block; padding: .6rem .9rem; border-radius: 8px; background: #e5e0d5; color: #28303d; text-decoration: none; }}
    nav a:hover {{ background: #d9d2c4; }}
    section {{ margin: 1.5rem 0 2.5rem; }}
    h2 {{ border-bottom: 1px solid #d6d0c4; padding-bottom: .5rem; }}
    .grid {{ display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 1rem; }}
    .card {{ overflow: hidden; border: 1px solid #d6d0c4; border-radius: 12px; background: #fffdf8; box-shadow: 0 8px 24px #28303d1a; }}
    .card h3 {{ padding: 1rem 1rem 0; color: #475467; font-size: 1rem; }}
    .card a {{ display: block; }}
    .card img {{ display: block; width: 100%; height: auto; background: white; }}
    footer {{ padding: 0 0 2rem; color: #667085; text-align: center; }}
    @media (max-width: 760px) {{
      .grid {{ grid-template-columns: 1fr; }}
    }}
  </style>
</head>
<body>
  <header>
    <h1>{page_title}</h1>
    <p>Updated {generated_at} · This page refreshes every 60 seconds.</p>
    <nav><a href="{other_page}">{other_page_label}</a></nav>
  </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 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()

    current_values = update_rrd()

    process_alerts(current_values)

    create_graphs()

    create_weekly_graphs()

    create_dashboards()


if __name__ == "__main__":
    main()