cheap_system_monitor/generate_graphs.py

#!/usr/bin/env python3
"""Generate system-monitor PNG graphs from its RRD database."""

from __future__ import annotations

import argparse
import configparser
import html
from pathlib import Path
import subprocess


PERIODS = {"3hours": "3h", "1day": "1d", "2weeks": "2w", "1year": "1y"}
COLORS = ["#2563EB", "#DC2626", "#16A34A", "#9333EA"]
LOAD_COLORS = ["#FACC15", "#F59E0B", "#DC2626"]  # bottom to top
DEFAULT_GRAPHS = (
    "load", "cpu", "thermal", "trouble", "network", "disk",
    "storage-health", "uptime",
)


def graph(output: Path, rrd: Path, start: str, title: str, unit: str,
          series: list[tuple[str, str, str]], logarithmic: bool = False,
          stacked: bool = False, disk_memory: bool = False,
          network_mixed: bool = False, thermal_mixed: bool = False,
          trouble_mixed: bool = False) -> None:
    args = ["rrdtool", "graph", str(output), "--start", f"end-{start}",
            "--end", "now", "--width", "900", "--height", "260",
            "--title", title, "--vertical-label", unit,
            "--watermark", "cheap-system-monitor"]
    if not trouble_mixed:
        args.append("--slope-mode")
    if logarithmic:
        args += ["--logarithmic", "--lower-limit", "1"]
    if title.startswith("Processor load"):
        # RRD's `m` axis suffix means milli-load: 1000m is a load of 1.0.
        args += ["--lower-limit", "0", "--upper-limit", "1", "--rigid",
                 "--units-exponent", "-3"]
    if title.startswith("CPU usage"):
        args += ["--lower-limit", "0", "--upper-limit", "80", "--rigid"]
    if trouble_mixed:
        args += ["--lower-limit", "0", "--upper-limit", "20", "--rigid"]
    if disk_memory:
        args += ["--lower-limit", "0", "--upper-limit", "100", "--rigid"]
    if thermal_mixed:
        args += ["--lower-limit", "0", "--upper-limit", "100", "--rigid"]
    health_graph = any(ds.startswith("health_") for ds, _label, _cf in series)
    if health_graph:
        args += ["--lower-limit", "0", "--upper-limit", "10", "--rigid",
                 "HRULE:5#9CA3AF:action threshold",
                 "HRULE:3#DC2626:urgent threshold",
                 "COMMENT:\\n"]
    value_suffix = "%%" if unit == "%" else ""
    for index, (ds, label, cf) in enumerate(series):
        var = f"v{index}"
        if thermal_mixed and ds == "throttled":
            args += [f"DEF:{var}_raw={rrd}:{ds}:{cf}",
                     f"CDEF:{var}={var}_raw,100,*",
                     f"AREA:{var}#DC262640:{label}",
                     f"GPRINT:{var}_raw:LAST:Current\\:%8.0lf",
                     f"GPRINT:{var}_raw:AVERAGE:Average\\:%8.2lf",
                     f"GPRINT:{var}_raw:MAX:Maximum\\:%8.0lf\\n"]
            continue
        if trouble_mixed and ds == "oom_kills":
            args += [f"DEF:{var}_raw={rrd}:{ds}:{cf}",
                     f"CDEF:{var}={var}_raw,60,*",
                     f"AREA:{var}#DC262680:{label}",
                     f"GPRINT:{var}:LAST:Current\\:%8.2lf",
                     f"GPRINT:{var}:AVERAGE:Average\\:%8.2lf",
                     f"GPRINT:{var}:MAX:Maximum\\:%8.2lf\\n"]
            continue
        if trouble_mixed and ds == "failed_services":
            drawing = f"AREA:{var}#2563EB80:{label}"
        elif network_mixed and ds == "net_out":
            drawing = f"AREA:{var}#86EFAC:{label}"
        elif network_mixed and ds == "net_in":
            drawing = f"LINE3:{var}#2563EB:{label}"
        elif disk_memory and ds == "memory":
            drawing = f"LINE3:{var}#DC2626:{label}"
        elif disk_memory:
            # Transparency keeps multiple mountpoints visible when their
            # percentage areas overlap.
            drawing = f"AREA:{var}{COLORS[index % len(COLORS)]}80:{label}"
        elif stacked:
            stack = ":STACK" if index else ""
            stacked_colors = LOAD_COLORS if title.startswith("Processor load") else COLORS
            drawing = f"AREA:{var}{stacked_colors[index]}:{label}{stack}"
        else:
            drawing = f"LINE{2 if index < 3 else 1}:{var}{COLORS[index % len(COLORS)]}:{label}"
        args += [f"DEF:{var}={rrd}:{ds}:{cf}",
                 drawing,
                 f"GPRINT:{var}:LAST:Current\\:%8.2lf%s{value_suffix}",
                 f"GPRINT:{var}:AVERAGE:Average\\:%8.2lf%s{value_suffix}",
                 f"GPRINT:{var}:MAX:Maximum\\:%8.2lf%s{value_suffix}\\n"]
    subprocess.run(args, check=True)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", default="/opt/cheap_system_monitor/monitor.conf")
    args = parser.parse_args()
    config_parser = configparser.ConfigParser()
    if not config_parser.read(args.config):
        raise SystemExit(f"Configuration file not found: {args.config}")
    config = config_parser["monitor"]
    rrd = Path(config["rrd_file"])
    if not rrd.exists():
        raise SystemExit(f"RRD file does not exist: {rrd}")
    output_dir = Path(config.get("graph_dir", "/var/www/html/system-monitor"))
    output_dir.mkdir(parents=True, exist_ok=True)
    mounts = [item.strip() for item in config["mountpoints"].split(",") if item.strip()]
    if not mounts or mounts[0] != "/":
        mounts.insert(0, "/")
    all_definitions = [
        ("load", "Processor load", "load", [("load15", "15 minutes", "AVERAGE"), ("load5", "5 minutes", "AVERAGE"), ("load1", "1 minute", "AVERAGE")], False, True, False, False, False, False),
        ("cpu", "CPU usage", "%", [("cpu_user", "user", "AVERAGE"), ("cpu_system", "system", "AVERAGE"), ("cpu_nice", "nice", "AVERAGE"), ("io_wait", "I/O wait", "AVERAGE")], False, True, False, False, False, False),
        ("thermal", "CPU temperature and throttling", "°C", [("temperature", "temperature", "AVERAGE"), ("throttled", "throttled", "MAX")], False, False, False, False, True, False),
        ("trouble", "Failed services and OOM kills", "count", [("failed_services", "failed services", "MAX"), ("oom_kills", "OOM kills/min", "MAX")], False, False, False, False, False, True),
        ("network", "Network throughput", "bytes/s", [("net_out", "sent", "AVERAGE"), ("net_in", "received", "AVERAGE")], False, False, False, True, False, False),
        ("disk", "Disk and memory usage", "%", [(f"disk_{'root' if i == 0 else i}", mount, "AVERAGE") for i, mount in enumerate(mounts)] + [("memory", "memory", "AVERAGE")], False, False, True, False, False, False),
        ("storage-health", "Storage operational health", "0–10", [(f"health_{'root' if i == 0 else i}", mount, "MIN") for i, mount in enumerate(mounts)], False, False, False, False, False, False),
        ("uptime", "System uptime", "days", [("uptime", "uptime", "AVERAGE")], False, False, False, False, False, False),
    ]
    configured_graphs = [
        item.strip()
        for item in config.get("graphs", ",".join(DEFAULT_GRAPHS)).split(",")
        if item.strip()
    ]
    if not configured_graphs:
        raise SystemExit("At least one graph must be configured in 'graphs'")
    definitions_by_name = {definition[0]: definition for definition in all_definitions}
    unknown = [name for name in configured_graphs if name not in definitions_by_name]
    if unknown:
        available = ", ".join(DEFAULT_GRAPHS)
        raise SystemExit(
            f"Unknown graph name(s): {', '.join(unknown)}. Available: {available}"
        )
    if len(configured_graphs) != len(set(configured_graphs)):
        raise SystemExit("Graph names in 'graphs' must not be repeated")
    definitions = [definitions_by_name[name] for name in configured_graphs]
    for period_name, start in PERIODS.items():
        for filename, title, unit, series, logarithmic, stacked, disk_memory, network_mixed, thermal_mixed, trouble_mixed in definitions:
            graph(output_dir / f"{filename}-{period_name}.png", rrd, start,
                  f"{title} — {period_name}", unit, series, logarithmic,
                  stacked, disk_memory, network_mixed, thermal_mixed,
                  trouble_mixed)
    page_files = {
        "3hours": "index.html",
        "1day": "1day.html",
        "2weeks": "2weeks.html",
        "1year": "1year.html",
    }
    period_labels = {
        "3hours": "3 hours",
        "1day": "1 day",
        "2weeks": "2 weeks",
        "1year": "1 year",
    }
    for period in PERIODS:
        cards = "\n".join(
            f'<section><h2>{html.escape(title)}</h2>'
            f'<img src="{filename}-{period}.png" '
            f'alt="{html.escape(title)}, {period_labels[period]}"></section>'
            for filename, title, *_rest in definitions
        )
        navigation = " ".join(
            f'<a href="{page_files[item]}"'
            f'{" class=active" if item == period else ""}>'
            f'{period_labels[item]}</a>'
            for item in PERIODS
        )
        page = f"""<!doctype html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width">
<meta http-equiv="refresh" content="300"><title>System monitor — {period_labels[period]}</title>
<style>body{{font:16px system-ui;margin:auto;max-width:1100px;padding:1rem;background:#f8fafc;color:#172033}}
nav{{display:flex;gap:.5rem;flex-wrap:wrap}} nav a{{padding:.5rem .75rem;background:#e2e8f0;color:#172033;text-decoration:none;border-radius:.35rem}}
nav a.active{{background:#2563eb;color:white}} section{{background:white;padding:1rem;margin:1rem 0;border-radius:.5rem;box-shadow:0 1px 4px #0002}}
img{{width:100%;height:auto}}</style>
</head><body><h1>System monitor — {period_labels[period]}</h1><nav>{navigation}</nav>{cards}</body></html>"""
        (output_dir / page_files[period]).write_text(page)


if __name__ == "__main__":
    main()