test: characterize contracts and make startup explicit

2026-08-06 17:24:20 -05:00
parent fc27feaea1
commit 9091f27283
9 changed files with 695 additions and 56 deletions
@@ -0,0 +1,98 @@
+"""Home Assistant discovery and legacy slot-publication helpers."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Any, Callable, Iterable
+
+from .config import HomeAssistantConfig
+from .contracts import LEGACY_STATE_QOS, LEGACY_STATE_RETAIN, SLOT_IDS, Topics
+
+
+Publish = Callable[..., Any]
+
+
+HA_DEVICE = {
+    "identifiers": ["ytplayer-controller"],
+    "name": "YouTube Controller",
+    "manufacturer": "River of Glory",
+    "model": "YT Controller",
+}
+
+
+@dataclass(frozen=True)
+class SlotAnnouncement:
+    slot: int
+    title: str
+    state: str
+    count: int
+    items: tuple[str, ...] = ()
+
+
+def discovery_messages(
+    topics: Topics,
+    config: HomeAssistantConfig,
+) -> Iterable[tuple[str, str, int, bool]]:
+    """Yield the existing thirty discovery documents unchanged."""
+
+    for slot in SLOT_IDS:
+        base = topics.slot_base(slot)
+        object_id = f"yt_slot_{slot}"
+        entities = (
+            (
+                "title",
+                {
+                    "name": f"YT Slot {slot} Title",
+                    "state_topic": f"{base}/title",
+                    "unique_id": f"{object_id}_title",
+                    "icon": "mdi:playlist-music",
+                    "device": HA_DEVICE,
+                },
+            ),
+            (
+                "state",
+                {
+                    "name": f"YT Slot {slot} State",
+                    "state_topic": f"{base}/state",
+                    "unique_id": f"{object_id}_state",
+                    "icon": "mdi:circle-small",
+                    "device": HA_DEVICE,
+                },
+            ),
+            (
+                "count",
+                {
+                    "name": f"YT Slot {slot} Count",
+                    "state_topic": f"{base}/count",
+                    "unique_id": f"{object_id}_count",
+                    "icon": "mdi:numeric",
+                    "device": HA_DEVICE,
+                },
+            ),
+        )
+        for suffix, payload in entities:
+            yield (
+                f"{config.discovery_prefix}/sensor/{object_id}_{suffix}/config",
+                json.dumps(payload),
+                LEGACY_STATE_QOS,
+                LEGACY_STATE_RETAIN,
+            )
+
+
+def slot_messages(
+    announcement: SlotAnnouncement,
+    topics: Topics,
+    *,
+    clear_items_through: int = 0,
+) -> Iterable[tuple[str, str, int, bool]]:
+    """Yield retained legacy slot topics, including stale item cleanup."""
+
+    base = topics.slot_base(announcement.slot)
+    yield f"{base}/title", announcement.title, LEGACY_STATE_QOS, LEGACY_STATE_RETAIN
+    yield f"{base}/state", announcement.state, LEGACY_STATE_QOS, LEGACY_STATE_RETAIN
+    yield f"{base}/count", str(announcement.count), LEGACY_STATE_QOS, LEGACY_STATE_RETAIN
+    for index, title in enumerate(announcement.items, start=1):
+        yield f"{base}/items/{index}", title, LEGACY_STATE_QOS, LEGACY_STATE_RETAIN
+    for index in range(len(announcement.items) + 1, clear_items_through + 1):
+        yield f"{base}/items/{index}", "", LEGACY_STATE_QOS, LEGACY_STATE_RETAIN