#!/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
STATUS_ALERT_DELAY_SECONDS = 10 * 60
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 process_status_alert(previous, current_status):
"""Notify only after a new status persists for ten minutes across runs."""
# Older state files only have the latest observed status.
confirmed = previous.get("confirmed_status", previous.get("status"))
pending = previous.get("pending_status")
pending_since = previous.get("pending_status_since")
if not confirmed:
confirmed = current_status
pending = pending_since = None
elif current_status == confirmed:
pending = pending_since = None
elif current_status:
now = time.time()
if current_status != pending or pending_since is None:
pending = current_status
pending_since = now
elif now - pending_since >= STATUS_ALERT_DELAY_SECONDS:
notify(f"UPS status changed: {confirmed} -> {current_status}")
confirmed = current_status
pending = pending_since = None
return {
"confirmed_status": confirmed,
"pending_status": pending,
"pending_status_since": pending_since,
}
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)
status_alert_state = process_status_alert(previous, current_status)
# An empty state is the initial baseline, not an alert condition.
if previous:
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_alert_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"""
{label}
})
"""
for label, path in graphs
)
sections.append(
f""""""
)
document = f"""
{page_title}
{''.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()
current_values = update_rrd()
process_alerts(current_values)
create_graphs()
create_weekly_graphs()
create_dashboards()
if __name__ == "__main__":
main()