99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
"""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
|