#!/usr/bin/env python3 """Simple device dashboard server. Serves HTML + proxies HA API + saves device names + TV control.""" import http.server import json import os import random import re import socketserver import time import threading import urllib.request import tv_control HA_URL = "http://supervisor/core/api" NAMES_FILE = "/share/device_names.json" PORT = 8888 SANDMAN_STATE = "/share/sandman_state.json" class Handler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory="/share", **kwargs) def do_GET(self): if self.path.startswith("/api/"): self._proxy_ha("GET") elif self.path == "/device_names.json": self._serve_names() elif self.path == "/tv/status": self._tv_status() elif self.path.startswith("/tv/screenshot"): self._tv_screenshot() elif self.path == "/tv/deficit": self._tv_deficit() elif self.path == "/sandman/status": self._sandman_status() elif self.path == "/sandman/history": self._sandman_history() elif self.path.startswith("/sandman/log"): self._sandman_log() elif self.path == "/sandman/config": self._sandman_config_get() elif self.path == "/tv/throttle_status": self._tv_throttle_status() else: super().do_GET() def do_POST(self): if self.path == "/save_names": self._save_names() elif self.path.startswith("/api/"): self._proxy_ha("POST") elif self.path == "/tv/key": self._tv_key() elif self.path == "/tv/launch": self._tv_launch() elif self.path == "/tv/ambilight": self._tv_ambilight() elif self.path == "/tv/volume": self._tv_volume() elif self.path == "/tv/mute": self._tv_mute() elif self.path == "/tv/throttle": self._tv_throttle() elif self.path == "/sandman/deficit": self._sandman_set_deficit() elif self.path == "/sandman/speed": self._sandman_set_speed() elif self.path == "/sandman/reset": self._sandman_reset() elif self.path == "/sandman/config": self._sandman_config_set() elif self.path == "/sandman/test_action": self._sandman_test_action() else: self.send_error(405) def _serve_names(self): try: with open(NAMES_FILE) as f: data = f.read() except FileNotFoundError: data = "{}" self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(data.encode()) def _save_names(self): length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) if length else b"{}" try: names = json.loads(body) with open(NAMES_FILE, "w") as f: json.dump(names, f, indent=2, ensure_ascii=False) self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(b'{"ok":true}') except Exception as e: self.send_response(500) self.end_headers() self.wfile.write(json.dumps({"error": str(e)}).encode()) def _proxy_ha(self, method): ha_path = self.path[4:] # strip /api url = f"{HA_URL}{ha_path}" try: body = None if method == "POST": length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) if length else None req = urllib.request.Request( url, data=body, headers={"Authorization": f"Bearer {tv_control.SUPERVISOR_TOKEN}", "Content-Type": "application/json"}, method=method, ) resp = urllib.request.urlopen(req, timeout=10) data = resp.read() self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(data) except Exception as e: self.send_response(502) self.end_headers() self.wfile.write(json.dumps({"error": str(e)}).encode()) # ── JSON helpers ──────────────────────────────────────────────── def _read_body(self): length = int(self.headers.get("Content-Length", 0)) return json.loads(self.rfile.read(length)) if length else {} def _json_ok(self, obj, status=200): data = json.dumps(obj).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(data) def _json_err(self, msg, status=500): self._json_ok({"error": msg}, status) # ── TV endpoints ───────────────────────────────────────────── def _tv_status(self): try: result = {} result["power"] = tv_control.get_power_state() audio = tv_control.get_audio_state() result["volume"] = audio["volume"] result["muted"] = audio["muted"] try: pkg, playback, extras = tv_control.adb_get_current_app() result["app"] = tv_control.APP_NAMES.get(pkg, pkg) if pkg else "Unknown" result["title"] = (extras or {}).get("title", "") result["artist"] = (extras or {}).get("artist", "") result["position_s"] = (extras or {}).get("position_s", 0) except Exception: result["app"] = "Unknown" result["title"] = "" result["artist"] = "" result["position_s"] = 0 # Sandman deficit + session tracking try: with open(SANDMAN_STATE) as f: sm = json.load(f) result["deficit"] = sm.get("deficit", 0) result["session_min"] = sm.get("session_min", 0) result["total_today_min"] = sm.get("total_today_min", 0) except Exception: result["deficit"] = 0 result["session_min"] = 0 result["total_today_min"] = 0 self._json_ok(result) except Exception as e: self._json_err(str(e)) def _tv_key(self): try: body = self._read_body() key = body.get("key", "") tv_control.js_key(key) self._json_ok({"ok": True, "key": key}) except Exception as e: self._json_err(str(e)) def _tv_launch(self): try: body = self._read_body() pkg = body.get("package", "") tv_control.adb_launch_app(pkg) self._json_ok({"ok": True, "package": pkg}) except Exception as e: self._json_err(str(e)) def _tv_ambilight(self): try: tv_control.ambilight_toggle() self._json_ok({"ok": True}) except Exception as e: self._json_err(str(e)) def _tv_volume(self): try: body = self._read_body() vol = int(body.get("volume", 0)) tv_control.js_post("audio/volume", {"current": vol, "muted": False}) self._json_ok({"ok": True, "volume": vol}) except Exception as e: self._json_err(str(e)) def _tv_mute(self): try: audio = tv_control.get_audio_state() currently_muted = audio["muted"] tv_control.set_mute(not currently_muted) self._json_ok({"ok": True, "muted": not currently_muted}) except Exception as e: self._json_err(str(e)) def _tv_screenshot(self): try: path = tv_control.adb_screenshot() if path and os.path.exists(path): with open(path, "rb") as f: data = f.read() self.send_response(200) self.send_header("Content-Type", "image/png") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) else: self._json_err("Screenshot failed") except Exception as e: self._json_err(str(e)) def _tv_throttle(self): try: body = self._read_body() bw = int(body.get("bandwidth", 0)) if bw > 0: tv_control.throttle_apply(bandwidth_kbps=bw) # Save throttle state import time as _time start_time = body.get("start_time") or int(_time.time() * 1000) with open("/share/throttle_state.json", "w") as f: json.dump({"active": True, "bandwidth": bw, "start_time": start_time}, f) else: tv_control.throttle_remove() try: os.remove("/share/throttle_state.json") except: pass self._json_ok({"ok": True, "bandwidth": bw}) except Exception as e: self._json_err(str(e)) def _tv_deficit(self): try: with open(SANDMAN_STATE) as f: data = json.load(f) self._json_ok(data) except FileNotFoundError: self._json_ok({"deficit": 0}) except Exception as e: self._json_err(str(e)) def _sandman_status(self): try: with open(SANDMAN_STATE) as f: state = json.load(f) # Check if speed file exists speed = 1.0 try: with open("/share/sandman_speed.json") as f: speed = json.load(f).get("speed", 1.0) except Exception: pass state["speed"] = speed # Derive tv_on from session_start state["tv_on"] = state.get("session_start") is not None self._json_response(state) except Exception as e: self._json_error(str(e)) def _sandman_history(self): """Return deficit history points from the sandman log.""" points = [] try: with open("/share/sandman.log") as f: for line in f.readlines()[-500:]: m = re.search(r"(\d{4}-\d{2}-\d{2}) (\d{2}:\d{2}):\d{2}.*deficit=([\d.]+)", line) if m: points.append({"date": m.group(1), "time": m.group(2), "deficit": float(m.group(3))}) # Thin to ~100 points max if len(points) > 100: step = len(points) // 100 points = points[::step] except Exception: pass self._json_response({"points": points}) def _sandman_log(self): """Return sandman actions. Accepts ?count=N (default 10).""" from urllib.parse import urlparse, parse_qs parsed = urlparse(self.path) params = parse_qs(parsed.query) count = int(params.get("count", [10])[0]) actions = [] try: with open("/share/sandman.log") as f: for line in f.readlines()[-(count * 20):]: m = re.search(r"(\d{4}-\d{2}-\d{2}) (\d{2}:\d{2}):\d{2}.*deficit=([\d.]+).*\| (\S+)", line) if m: actions.append({"date": m.group(1), "time": m.group(2), "deficit": float(m.group(3)), "action": m.group(4)}) except Exception: pass self._json_response(actions[-count:]) def _sandman_set_deficit(self): try: length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length)) if length else {} deficit = float(body.get("deficit", 0)) with open(SANDMAN_STATE) as f: state = json.load(f) state["deficit"] = max(0.0, min(1.0, deficit)) with open(SANDMAN_STATE, "w") as f: json.dump(state, f, indent=2) self._json_response({"ok": True, "deficit": state["deficit"]}) except Exception as e: self._json_error(str(e)) def _sandman_set_speed(self): try: length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length)) if length else {} speed = float(body.get("speed", 1.0)) with open("/share/sandman_speed.json", "w") as f: json.dump({"speed": speed}, f) self._json_response({"ok": True, "speed": speed}) except Exception as e: self._json_error(str(e)) def _tv_throttle_status(self): """Check if tc throttle is currently active.""" try: import subprocess r = subprocess.run(["tc", "qdisc", "show", "dev", "end0"], capture_output=True, text=True, timeout=5) active = "htb" in r.stdout start_time = None bandwidth = 1000 try: with open("/share/throttle_state.json") as f: ts = json.load(f) start_time = ts.get("start_time") bandwidth = ts.get("bandwidth", 1000) except Exception: pass self._json_response({"active": active, "start_time": start_time, "bandwidth": bandwidth}) except Exception as e: self._json_err(str(e)) def _sandman_config_get(self): try: with open("/share/sandman_config.json") as f: self._json_response(json.load(f)) except Exception as e: self._json_err(str(e)) def _sandman_config_set(self): """Update action config (min_deficit, weight).""" try: body = self._read_body() action_name = body.get("action") with open("/share/sandman_config.json") as f: config = json.load(f) # Golden hours update golden = body.get("golden_hours") if golden is not None: mults = config.get("attention_model", {}).get("time_multipliers", []) if golden.get("enabled"): # Find existing golden entry or create one found = False rate = golden.get("rate", 0.3) for m in mults: if m.get("no_actions"): m["start"] = golden["start"] m["end"] = golden["end"] m["rate"] = rate found = True break if not found: mults.append({"start": golden["start"], "end": golden["end"], "rate": rate, "no_actions": True, "comment": "Golden period"}) else: # Remove golden entry config["attention_model"]["time_multipliers"] = [m for m in mults if not m.get("no_actions")] with open("/share/sandman_config.json", "w") as f: json.dump(config, f, indent=4) self._json_response({"ok": True, "golden": golden}) return if action_name and action_name in config.get("actions", {}): if "min_deficit" in body: config["actions"][action_name]["min_deficit"] = float(body["min_deficit"]) if "weight" in body: config["actions"][action_name]["weight"] = int(body["weight"]) if "permissible_hours" in body: config["actions"][action_name]["permissible_hours"] = body["permissible_hours"] with open("/share/sandman_config.json", "w") as f: json.dump(config, f, indent=4) self._json_response({"ok": True, "action": action_name}) else: self._json_err("Unknown action", 400) except Exception as e: self._json_err(str(e)) def _sandman_reset(self): """Reset deficit to 0 and speed to 1x.""" try: with open(SANDMAN_STATE) as f: state = json.load(f) state["deficit"] = 0.0 with open(SANDMAN_STATE, "w") as f: json.dump(state, f, indent=2) with open("/share/sandman_speed.json", "w") as f: json.dump({"speed": 1.0}, f) self._json_response({"ok": True, "deficit": 0.0, "speed": 1.0}) except Exception as e: self._json_error(str(e)) def _sandman_test_action(self): """Execute a sandman action directly on the TV for testing.""" try: body = self._read_body() action = body.get("action", "") if not action: return self._json_error("Missing action", 400) def run_action(): try: if action == "volume_drop_big": for _ in range(random.randint(5, 15)): tv_control.js_key("VolumeDown") time.sleep(0.1) elif action == "audio_stutter_long": for _ in range(random.randint(5, 8)): tv_control.set_mute(True) time.sleep(0.3) tv_control.set_mute(False) time.sleep(0.2) elif action == "full_mute": tv_control.set_mute(True) elif action == "overlay_controls": tv_control.js_key("CursorDown") elif action == "pause_long": tv_control.js_key("Pause") elif action == "home": tv_control.js_key("Home") elif action == "back_spam": for _ in range(random.randint(3, 5)): tv_control.js_key("Back") time.sleep(0.2) elif action == "throttle": tv_control.throttle_apply(1500) elif action == "navigate_random": keys = ["CursorUp", "CursorDown", "CursorLeft", "CursorRight", "Confirm"] for _ in range(random.randint(3, 6)): tv_control.js_key(random.choice(keys)) time.sleep(0.3) elif action == "screensaver": tv_control.js_key("Home") time.sleep(1.5) tv_control.js_key("Back") elif action == "flicker_toilet_light": tv_control.ha_switch_flicker("switch.magic_switch_s1e_595a_kitchen_switch_2", random.uniform(0.1, 10.0)) elif action == "flicker_tv_light": tv_control.ha_switch_flicker("switch.magic_switch_s1e_bb55_switch_3", random.uniform(0.1, 10.0)) elif action == "flicker_laundry_light": tv_control.ha_switch_flicker("switch.magic_switch_s1e_595a_kitchen_switch_1", random.uniform(0.1, 10.0)) elif action == "flicker_kitchen_light": tv_control.ha_switch_flicker("switch.sonoff_10020b0c7c_2", random.uniform(0.1, 10.0)) elif action.startswith("sound_"): # Read sound file from config try: with open("/share/sandman_config.json") as f: cfg = json.load(f) sound_file = cfg["actions"][action]["params"].get("file", "knock_2.mp3") except Exception: sound_file = "knock_2.mp3" tv_control.speaker_play_sound(sound_file) elif action == "app_switch": apps = [ ("com.netflix.ninja", "com.netflix.ninja/.MainActivity"), ("com.google.android.youtube.tv", "com.google.android.youtube.tv/com.google.android.apps.youtube.tv.activity.ShellActivity"), ("com.apple.atve.androidtv.appletv", "com.apple.atve.androidtv.appletv/.MainActivity"), ] pkg, component = random.choice(apps) tv_control.adb_launch_app(pkg) elif action == "power_off": tv_control.js_key("Standby") except Exception: pass # Run in background thread so we can respond immediately threading.Thread(target=run_action, daemon=True).start() self._json_response({"ok": True, "action": action}) except Exception as e: self._json_error(str(e)) def _json_response(self, data, status=200): self.send_response(status) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps(data).encode()) def _json_error(self, msg, status=500): self.send_response(status) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps({"error": msg}).encode()) def log_message(self, format, *args): # Log to file for debugging import datetime try: with open("/share/devices_server.log", "a") as f: f.write(f"{datetime.datetime.now().strftime('%H:%M:%S')} {format % args}\n") except Exception: pass class ThreadedServer(socketserver.ThreadingMixIn, http.server.HTTPServer): daemon_threads = True if __name__ == "__main__": print(f"Devices dashboard on http://0.0.0.0:{PORT}/devices.html") ThreadedServer(("0.0.0.0", PORT), Handler).serve_forever()