93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
#!/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()
|