import json import threading import time import urllib.error import urllib.request from collections import deque from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from . import paths from .profiles import list_profiles, load_yaml HISTORY_SIZE = 2880 POLL_TIMEOUT = 3 class Sampler: def __init__(self, interval=5.0, history=HISTORY_SIZE): self.interval = interval self.history = deque(maxlen=history) self.snapshot = {"anker": [], "shelly": [], "updated": None, "engine": False} self.lock = threading.Lock() self.stop_event = threading.Event() self.thread = None def start(self): self.thread = threading.Thread(target=self._loop, daemon=True) self.thread.start() def stop(self): self.stop_event.set() def _loop(self): while not self.stop_event.is_set(): try: self._sample() except Exception: pass self.stop_event.wait(self.interval) def _read_live(self, serial): path = paths.STATE_DIR / f"live-{serial}.json" if not path.exists(): return None try: data = json.loads(path.read_text(encoding="utf-8")) except Exception: return None if time.time() - float(data.get("epoch", 0)) > 120: data["fresh"] = False else: data["fresh"] = True return data def _anker_devices(self): devices = [] for path in list_profiles(paths.ANKER_PROFILE_DIR): try: profile = load_yaml(path) except Exception: continue identity = profile.get("identity", {}) serial = identity.get("serial") live = self._read_live(serial) if serial else None values = (live or {}).get("values", {}) devices.append( { "name": identity.get("name") or identity.get("model") or path.stem, "model": identity.get("model") or identity.get("part_number") or "", "serial": serial, "live": bool(live and live.get("fresh")), "updated": (live or {}).get("updated"), "age": (live or {}).get("age_seconds"), "profile": (live or {}).get("profile"), "floor_latched": (live or {}).get("floor_latched", False), "battery_soc": values.get("battery_soc"), "output_watts": values.get("output_power_total"), "ac_in_watts": values.get("ac_input_power"), "pv_watts": values.get("pv_total"), "pv_surplus": values.get("pv_surplus"), "temperature": values.get("temperature"), } ) return devices def _poll_shelly(self, host, generation, channel): path = "/rpc/Shelly.GetStatus" if generation >= 2 else "/status" try: with urllib.request.urlopen( f"http://{host}{path}", timeout=POLL_TIMEOUT ) as response: data = json.loads(response.read().decode("utf-8")) except Exception: return None, None if generation >= 2: entry = data.get(f"switch:{channel}") or {} return entry.get("output"), entry.get("apower") relays = data.get("relays") or [] meters = data.get("meters") or [] state = relays[channel].get("ison") if channel < len(relays) else None power = meters[channel].get("power") if channel < len(meters) else None return state, power def _shelly_devices(self): devices = [] for path in list_profiles(paths.SHELLY_PROFILE_DIR): try: profile = load_yaml(path) except Exception: continue identity = profile.get("identity", {}) access = profile.get("access", {}) host = access.get("host") generation = int(identity.get("generation", 1) or 1) channels = profile.get("channels") or {"0": {}} channel = int(sorted(channels)[0]) state, power = self._poll_shelly(host, generation, channel) if host else (None, None) devices.append( { "name": identity.get("name") or path.stem, "model": identity.get("model") or "", "host": host, "channel": channel, "state": state, "watts": power, "reachable": state is not None, } ) return devices def _sample(self): anker = self._anker_devices() shelly = self._shelly_devices() point = {"t": time.time()} for device in anker: if not device["live"]: continue serial = device["serial"] point[f"{serial}:soc"] = device["battery_soc"] point[f"{serial}:out"] = device["output_watts"] point[f"{serial}:ac"] = device["ac_in_watts"] point[f"{serial}:pv"] = device["pv_watts"] with self.lock: if len(point) > 1: self.history.append(point) self.snapshot = { "anker": anker, "shelly": shelly, "updated": time.strftime("%H:%M:%S"), "engine": any(d["live"] for d in anker), "interval": self.interval, } def payload(self): with self.lock: return { **self.snapshot, "history": list(self.history), } PAGE = """ Power monitor

Power monitor

updated --:--:--

Anker SOLIX

Shelly switches

Recent history

solar W grid in W load W battery %
""" class Handler(BaseHTTPRequestHandler): sampler = None def log_message(self, *args): pass def _send(self, code, body, content_type): payload = body.encode("utf-8") self.send_response(code) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(payload))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(payload) def do_GET(self): if self.path.startswith("/api/data"): self._send( 200, json.dumps(Handler.sampler.payload()), "application/json" ) return if self.path in ("/", "/index.html"): self._send(200, PAGE, "text/html; charset=utf-8") return self._send(404, "not found", "text/plain") def serve(port=8765, interval=5.0, host="127.0.0.1"): paths.ensure_dirs() sampler = Sampler(interval=interval) sampler.start() Handler.sampler = sampler server = ThreadingHTTPServer((host, port), Handler) return server, sampler