#!/usr/bin/env python3 """Keep the dock's four knob rings showing what the room's lamps are doing. Polls Home Assistant's REST API for the configured lights, works out the four ring colours (`dock_leds.ring_colors`), and rewrites the akp05 device plugin's leds.toml when — and only when — they change. Then runs STREAM_DOCK_LED_APPLY_COMMAND, which is the hook that makes the plugin notice. THE HONEST LIMIT, and the reason that hook exists: the akp05 device plugin reads leds.toml when it starts and does not watch it. Nothing else can drive those LEDs either — the plugin holds the USB device open, so a second process cannot write HID reports to it. So this service is correct and complete on its own side, and the last few centimetres are somebody's decision about how to make the plugin re-read: a small upstream patch that watches the file (the right fix), or a command that restarts the plugin (available today, but it re-initialises the device, so it is only tolerable at the debounce intervals this service is designed around, not per detent). With the hook empty this still runs, still keeps the file right, and simply does not light anything new until the plugin next starts. See stream-dock/README.md §5. Why polling and not the websocket API: this is stdlib-only, like the rest of this repo's services, and the thing it feeds cannot react faster than its debounce anyway. A websocket subscription is a strict improvement the day the apply hook becomes cheap. Configuration is environment only (generated/led-sync.env, written by generate.py). """ from __future__ import annotations import json import logging import os import shlex import signal import subprocess import sys import time import urllib.error import urllib.request from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import dock_leds # noqa: E402 LOG = logging.getLogger("stream-dock-led-sync") HA_URL = os.environ.get("STREAM_DOCK_HA_URL", "").rstrip("/") HA_TOKEN = os.environ.get("STREAM_DOCK_HA_TOKEN", "") LIGHTS = [e.strip() for e in os.environ.get("STREAM_DOCK_LIGHTS", "").split(",") if e.strip()] LEDS_PATH = Path(os.path.expanduser( os.environ.get("STREAM_DOCK_LEDS_PATH", dock_leds.DEFAULT_LEDS_PATH))) LED_BRIGHTNESS = int(os.environ.get("STREAM_DOCK_LED_BRIGHTNESS", "100")) MIN_CHANNEL = int(os.environ.get("STREAM_DOCK_LED_MIN_CHANNEL", "0")) POLL_SECONDS = float(os.environ.get("STREAM_DOCK_LED_POLL_SECONDS", "2")) DEBOUNCE_MS = int(os.environ.get("STREAM_DOCK_LED_DEBOUNCE_MS", "400")) APPLY_COMMAND = os.environ.get("STREAM_DOCK_LED_APPLY_COMMAND", "").strip() # Rate limit for the apply command only — never for the file write, which is cheap and # always reflects the current state. Every strategy except "none" re-initialises the # device to some degree, so running one per detent of a spun dial would make the dock # blink rather than light. The last state is never dropped: a deferred apply runs as # soon as the interval is up. APPLY_MIN_INTERVAL = float(os.environ.get("STREAM_DOCK_LED_APPLY_MIN_INTERVAL", "0")) # The layer gate. The dock's lighting controls live on their own OpenDeck layer, and # the four rings are shared hardware: on any other layer those dials mean something # else, and painting a lamp's colour onto them there is worse than not lighting them # at all. Empty means no gate — drive the rings always. GATE_COMMAND = os.environ.get("STREAM_DOCK_LED_GATE_COMMAND", "").strip() IDLE_COLORS = [c.strip() for c in os.environ.get("STREAM_DOCK_LED_IDLE_COLORS", "").split(",") if c.strip()] IDLE_CYCLE_SECONDS = float(os.environ.get("STREAM_DOCK_LED_IDLE_CYCLE_SECONDS", "3")) # How long a failing Home Assistant is allowed to be quiet about it. Below this the # service just retries; above it, it says so once and then stays quiet again, because # a desktop service that logs a line every two seconds while HA reboots is a service # nobody keeps enabled. COMPLAIN_AFTER_SECONDS = 60 class Lights: """The lamps, and the last colour anybody saw them wearing. The remembered colour is the point of this class. Most integrations drop rgb_color to None when a lamp is off, and a ring set that forgets the colour every time the light is switched off would tell you nothing about what the colour dials are currently holding — see dock_leds' module docstring. """ def __init__(self, entity_ids: list[str]) -> None: self.entity_ids = entity_ids self.last_rgb: tuple[int, int, int] = (255, 167, 87) self.last_brightness = 255 def read(self) -> tuple[bool, bool, tuple[int, int, int], int]: """(HA answered at all, any lamp on, colour, brightness). The first flag is separate from the second on purpose: "every lamp is off" and "I could not ask" are different facts, and only one of them means the rings are lying. Same three-state honesty as the infra poller's ok/problem/unreachable. """ reachable = False any_on = False chosen = None for entity_id in self.entity_ids: state = fetch_state(entity_id) if state is None: continue reachable = True on = state.get("state") == "on" any_on = any_on or on if on and chosen is None: attrs = state.get("attributes") or {} rgb = attrs.get("rgb_color") if rgb and len(rgb) == 3: chosen = (tuple(int(c) for c in rgb), int(attrs.get("brightness") or 255)) if chosen is not None: self.last_rgb, self.last_brightness = chosen return reachable, any_on, self.last_rgb, self.last_brightness def fetch_state(entity_id: str) -> dict | None: request = urllib.request.Request( f"{HA_URL}/api/states/{entity_id}", headers={"Authorization": f"Bearer {HA_TOKEN}", "Accept": "application/json"}, ) try: with urllib.request.urlopen(request, timeout=5) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: # 404 is a configuration mistake, not a transient failure, and it deserves to # be loud every time: it means the entity in CoreSystemConfig.json does not # exist in this Home Assistant, which is exactly the failure the dials will # hit too. if exc.code == 404: LOG.error("no such entity in Home Assistant: %s", entity_id) else: LOG.debug("HTTP %s fetching %s", exc.code, entity_id) return None except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError) as exc: LOG.debug("fetching %s: %s", entity_id, exc) return None def write_leds(colors: list[list[int]]) -> None: """Atomic rewrite: a half-written leds.toml read by a starting plugin is a device that comes up with no LEDs and no explanation.""" LEDS_PATH.parent.mkdir(parents=True, exist_ok=True) temp = LEDS_PATH.with_suffix(".toml.tmp") temp.write_text(dock_leds.render_leds_toml(colors, LED_BRIGHTNESS)) os.replace(temp, LEDS_PATH) def layer_active() -> bool | None: """Is the dock currently showing the lighting layer? None means "cannot tell". Exit 0 yes, exit 1 no, anything else — including a command that will not run — is None, and the caller treats that as "not ours". A service that has lost track of which layer is up must not keep painting lamp colours onto rings that may now belong to something else; going idle is the recoverable mistake, hijacking is not. """ if not GATE_COMMAND: return True try: result = subprocess.run(shlex.split(GATE_COMMAND), capture_output=True, text=True, timeout=5) except (OSError, subprocess.SubprocessError) as exc: LOG.debug("layer gate could not run: %s", exc) return None if result.returncode == 0: return True if result.returncode == 1: return False LOG.debug("layer gate exited %s: %s", result.returncode, (result.stderr or "").strip()[:200]) return None def run_apply() -> None: if not APPLY_COMMAND: return try: result = subprocess.run(shlex.split(APPLY_COMMAND), capture_output=True, text=True, timeout=30) if result.returncode != 0: LOG.warning("apply command failed (%s): %s", result.returncode, (result.stderr or "").strip()[:200]) except (OSError, subprocess.SubprocessError) as exc: LOG.warning("apply command could not run: %s", exc) def main() -> int: logging.basicConfig(level=os.environ.get("STREAM_DOCK_LOG_LEVEL", "INFO"), format="%(levelname)s %(name)s: %(message)s") if not HA_URL or not HA_TOKEN: LOG.error("STREAM_DOCK_HA_URL and STREAM_DOCK_HA_TOKEN must be set " "(generate them with stream-dock/generate.py)") return 2 if not LIGHTS: LOG.error("STREAM_DOCK_LIGHTS is empty — nothing to follow") return 2 running = True def stop(signum, frame): # noqa: ARG001 nonlocal running running = False signal.signal(signal.SIGTERM, stop) signal.signal(signal.SIGINT, stop) lights = Lights(LIGHTS) LOG.info("following %s, writing %s", ", ".join(LIGHTS), LEDS_PATH) if GATE_COMMAND: LOG.info("gated on: %s", GATE_COMMAND) else: LOG.info("no layer gate configured — driving the rings on every layer") if APPLY_COMMAND: LOG.info("apply: %s (at most once every %ss)", APPLY_COMMAND, APPLY_MIN_INTERVAL) else: LOG.info("no apply command configured — leds.toml will be kept correct, but the " "device plugin only reads it at startup (README section 5)") written: list[list[int]] | None = None pending: list[list[int]] | None = None pending_since = 0.0 last_apply = 0.0 apply_due = False unreachable_since = 0.0 complained = False was_active: bool | None = None gate_complained = False while running: now = time.monotonic() active = layer_active() if active is None and not gate_complained: LOG.warning("the layer gate is not answering — treating the lighting layer as " "hidden and leaving the rings to their idle scheme (%s)", GATE_COMMAND) gate_complained = True elif active is not None: gate_complained = False if active is not True and was_active is True: LOG.debug("lighting layer hidden — rings released") if active is True and was_active is not True: LOG.debug("lighting layer showing — rings following %s", ", ".join(LIGHTS)) was_active = active if active is True: # Home Assistant is only polled while the layer is actually up. A dock # sitting on some other layer all day should not be asking about a lamp # nobody can see the colour of. reachable, any_on, rgb, brightness = lights.read() if not reachable: if unreachable_since == 0.0: unreachable_since = now elif not complained and now - unreachable_since > COMPLAIN_AFTER_SECONDS: LOG.warning("Home Assistant has been unreachable for %ds — the rings are " "showing the last state anybody saw, not the current one", int(now - unreachable_since)) complained = True else: if complained: LOG.info("Home Assistant is back") unreachable_since, complained = 0.0, False colors = dock_leds.ring_colors(rgb, brightness, any_on, floor=MIN_CHANNEL) elif IDLE_COLORS: unreachable_since, complained = 0.0, False step = int(now / IDLE_CYCLE_SECONDS) if IDLE_CYCLE_SECONDS > 0 else 0 colors = dock_leds.idle_ring_colors(IDLE_COLORS, step) else: # No idle scheme configured: hands entirely off. Whatever is on the rings # stays there, and the next time the layer comes up the colours are # rewritten from scratch — hence dropping `written`. written, pending = None, None time.sleep(POLL_SECONDS) continue # Debounce: a dial being spun changes the colour on every service call, and # each write is followed by whatever the apply command costs — which, if it is # a plugin restart, is far too expensive to do per detent. Collapse a burst # into one write once it settles. if colors != written and colors != pending: pending, pending_since = colors, now if pending is not None and (now - pending_since) * 1000 >= DEBOUNCE_MS: write_leds(pending) LOG.debug("rings -> %s", pending) written, pending = pending, None apply_due = True # The file is already current; this is only about when the (expensive) reload # runs. A deferred apply is never dropped — it fires on a later pass with the # newest state already on disk, which is exactly what should reach the device. if apply_due and APPLY_COMMAND and now - last_apply >= APPLY_MIN_INTERVAL: run_apply() last_apply, apply_due = now, False elif apply_due and not APPLY_COMMAND: apply_due = False interval = POLL_SECONDS if apply_due and APPLY_MIN_INTERVAL > 0: # Come back when the deferred reload is allowed to run, not a poll later. interval = min(interval, max(0.2, APPLY_MIN_INTERVAL - (now - last_apply))) if active is not True and IDLE_CYCLE_SECONDS > 0: # Keep the chase on time without polling faster than it steps. interval = min(interval, max(0.2, IDLE_CYCLE_SECONDS / 4)) if pending is not None: interval = min(interval, 0.2) time.sleep(interval) LOG.info("stopping") return 0 if __name__ == "__main__": sys.exit(main())