init

server.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Expose `upsc <ups>` as JSON: GET /jsonstatus?=nutdev1@localhost"""
+import json
+import os
+import re
+import subprocess
+from urllib.parse import unquote
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+# upsname[@hostname[:port]] -- no leading '-', so it can't be parsed as a upsc flag
+UPS_RE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.-]*(@[A-Za-z0-9_.\[\]:-]+)?$")
+HOST = os.environ.get("LISTEN_HOST", "0.0.0.0")
+PORT = int(os.environ.get("LISTEN_PORT", "8000"))
+
+# keys that look numeric but must stay strings (leading zeros / identifiers)
+KEEP_STRING = ("firmware", "productid", "vendorid", "model", "version")
+
+
+def convert(key, value):
+    if any(k in key for k in KEEP_STRING):
+        return value
+    try:
+        f = float(value)
+        return int(f) if f.is_integer() and "." not in value else f
+    except ValueError:
+        return value
+
+
+def read_ups(ups):
+    proc = subprocess.run(
+        ["upsc", ups], capture_output=True, text=True, timeout=10
+    )
+    if proc.returncode != 0:
+        raise RuntimeError(proc.stderr.strip() or f"upsc exited {proc.returncode}")
+
+    data = {}
+    for line in proc.stdout.splitlines():
+        if ": " not in line:
+            continue
+        key, value = line.split(": ", 1)
+        key = key.strip()
+        if key.startswith("Init SSL"):
+            continue
+        data[key] = convert(key, value.strip())
+
+    if "ups.status" not in data:
+        raise RuntimeError("no ups.status in upsc output")
+
+    status = str(data["ups.status"])
+    return {
+        "ups": ups,
+        "online": "OL" in status.split(),
+        "on_battery": "OB" in status.split(),
+        "low_battery": "LB" in status.split(),
+        "status": status,
+        "charge": data.get("battery.charge"),
+        "load": data.get("ups.load"),
+        "input_voltage": data.get("input.voltage"),
+        "raw": data,
+    }
+
+
+class Handler(BaseHTTPRequestHandler):
+    def _send(self, code, payload):
+        body = json.dumps(payload, indent=2).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        path, _, query = self.path.partition("?")
+        if path != "/jsonstatus":
+            return self._send(404, {"error": "not found"})
+        ups = unquote(query).lstrip("=").strip()
+        if not ups:
+            return self._send(400, {"error": "usage: /jsonstatus?=upsname@host"})
+        if not UPS_RE.match(ups):
+            return self._send(400, {"error": f"invalid ups name: {ups!r}"})
+        try:
+            self._send(200, read_ups(ups))
+        except Exception as e:  # upsc missing, timeout, NUT down
+            self._send(503, {"ups": ups, "online": False, "error": str(e)})
+
+    def log_message(self, fmt, *args):
+        pass  # keep journal quiet; remove to log every request
+
+
+if __name__ == "__main__":
+    print(f"listening on http://{HOST}:{PORT}/jsonstatus?=ups@host", flush=True)
+    ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()