diff --git a/README.md b/README.md index 64ef327..0cbee13 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ You need a SOLIX device that connects to WiFi, a Shelly plug on the same network, and your Anker account login. ```bash -git clone https://github.com/JustinOros/anker-shelly-bridge.git +git clone https://github.com/YOURNAME/anker-shelly-bridge.git cd anker-shelly-bridge ./start.sh ``` @@ -121,6 +121,29 @@ The same document is written into your data directory during setup. --- +## Live monitor + +```bash +solixauto monitor +``` + +Opens a dashboard at `http://127.0.0.1:8765` showing every device you have +saved: battery level, solar in, grid in, load out, and each Shelly's switch +state and draw. It refreshes every five seconds and keeps a rolling chart. + +Each device gets a **power bus** — solar and grid drawn from the left, load +from the right, all to one scale. If the left outweighs the right, the battery +is filling. Light and dark themes, remembered between visits. + +Add `--host 0.0.0.0` to reach it from a phone on the same network. + +Anker readings come from a running automation rather than a second cloud +connection, so start a service first. Two MQTT sessions on one Anker account +invite rate limiting. Shelly devices are polled directly over local HTTP and +always show live. + +--- + ## Safety Controlling the power supply to a battery has a specific failure mode: turn @@ -303,6 +326,7 @@ solixauto fields field names usable in rules solixauto status live telemetry solixauto switch on|off manual control solixauto conflicts automation set on the plug itself +solixauto monitor live dashboard in your browser solixauto new-profile scaffold a power profile solixauto profiles list power profiles diff --git a/solixauto/cli.py b/solixauto/cli.py index aef5819..a13bc44 100644 --- a/solixauto/cli.py +++ b/solixauto/cli.py @@ -364,6 +364,47 @@ def cmd_service(args): print("machine set System Settings > Energy Saver to prevent sleep.") +def cmd_monitor(args): + import webbrowser + from . import monitor as monitor_module + + try: + server, sampler = monitor_module.serve( + port=args.port, interval=args.interval, host=args.host + ) + except OSError as err: + fail( + f"could not start on {args.host}:{args.port}: {err}\n" + "Another monitor may already be running. Try --port 8766." + ) + + url = f"http://{args.host}:{args.port}" + print() + print(f"Power monitor running at {url}") + print() + print(" Readings come from a running automation, so start one first if the") + print(" Anker box says it has no live data:") + print(f" {paths.command('service ')}") + print() + print(" Press Ctrl-C to stop.") + + if not args.no_open: + try: + webbrowser.open(url) + except Exception: + pass + + try: + server.serve_forever() + except KeyboardInterrupt: + print() + print("stopped") + finally: + sampler.stop() + server.shutdown() + server.server_close() + + def cmd_conflicts(args): from .shelly import ( ShellyTarget, @@ -1321,6 +1362,23 @@ def build_parser(): status_parser.add_argument("--settle", type=int, default=45) status_parser.set_defaults(func=cmd_status) + monitor_parser = subparsers.add_parser( + "monitor", help="live dashboard in your browser" + ) + monitor_parser.add_argument("--port", type=int, default=8765) + monitor_parser.add_argument( + "--host", + default="127.0.0.1", + help="use 0.0.0.0 to reach it from other devices on your network", + ) + monitor_parser.add_argument( + "--interval", type=float, default=5.0, help="seconds between samples" + ) + monitor_parser.add_argument( + "--no-open", action="store_true", help="do not open a browser" + ) + monitor_parser.set_defaults(func=cmd_monitor) + service_parser = subparsers.add_parser( "service", help="run a power profile in the background at login" ) diff --git a/solixauto/engine.py b/solixauto/engine.py index 117d33a..0a9eeb7 100644 --- a/solixauto/engine.py +++ b/solixauto/engine.py @@ -445,6 +445,34 @@ class Engine: force=True, ) + def publish_live(self, variables, age): + identity = self.anker_profile.get("identity", {}) + serial = identity.get("serial") or "unknown" + + payload = { + "serial": serial, + "name": identity.get("name") or identity.get("model") or serial, + "model": identity.get("model") or identity.get("part_number") or "", + "updated": stamp(), + "epoch": time.time(), + "age_seconds": round(age) if age is not None else None, + "profile": self.profile.name, + "target_state": self.last_commanded, + "floor_latched": self.floor_latched, + "values": { + key: value + for key, value in variables.items() + if isinstance(value, (int, float, bool)) or value is None + }, + } + + try: + paths.STATE_DIR.mkdir(parents=True, exist_ok=True) + destination = paths.STATE_DIR / f"live-{serial}.json" + destination.write_text(json.dumps(payload), encoding="utf-8") + except Exception: + pass + def save_state(self, desired, reason): record = { "profile": self.profile.name, @@ -581,6 +609,7 @@ class Engine: variables = derived_values(self.anker_profile, status) self.last_variables = dict(variables) self.last_seen_at = stamp() + self.publish_live(variables, age) missing = sorted( name diff --git a/solixauto/monitor.py b/solixauto/monitor.py new file mode 100644 index 0000000..fbeaf6a --- /dev/null +++ b/solixauto/monitor.py @@ -0,0 +1,754 @@ +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