diff --git a/README.md b/README.md index 8ddf3a7..4ec2196 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,10 @@ RelayTV integrates with Home Assistant as a local `media_player` plus RelayTV-sp ## Current Feature Set - Creates a `media_player` entity for each RelayTV config entry. -- Polls RelayTV `GET /status` every 3 seconds. +- Uses a hybrid RelayTV state model: + - `GET /status` for bootstrap, reconnect, and full refresh fallback + - `GET /ui/events` SSE for immediate UI-state updates + - authoritative `status` events plus fast-path `playback` / refresh-hint `queue` and `jellyfin` events - Supports media controls from Home Assistant: - Play, pause, stop - Next and previous @@ -101,7 +104,7 @@ data: - No dedicated `enqueue` or `clear_queue` Home Assistant service is currently registered by this integration. - Overlay calls must include at least `text` or `image_url`. - Snapshots require active playback on the RelayTV server. -- Integration uses local polling; it does not currently use WebSocket push updates. +- `/ui/events` is treated as a live push stream, not a replay log; `/status` remains the reconnect/bootstrap fallback. ## Compatibility diff --git a/custom_components/relaytv/README.md b/custom_components/relaytv/README.md index 446b146..e680488 100644 --- a/custom_components/relaytv/README.md +++ b/custom_components/relaytv/README.md @@ -5,7 +5,11 @@ This integration provides a RelayTV `media_player` entity, RelayTV service actio ## Implemented Behavior - `media_player` platform is enabled (`custom_components/relaytv/media_player.py`). -- Polling coordinator refreshes RelayTV `GET /status` every 3 seconds. +- Hybrid state updates: + - bootstrap/reconnect uses RelayTV `GET /status` + - RelayTV `GET /ui/events` SSE provides hot-state updates + - `status` events are treated as authoritative full snapshots + - `playback` / `queue` / `jellyfin` events trigger fast updates or targeted refreshes - Sidebar panel is registered via Home Assistant frontend iframe panel APIs. - RelayTV services are registered from `services.yaml`: - `smart_url` @@ -41,5 +45,6 @@ This integration provides a RelayTV `media_player` entity, RelayTV service actio - Overlay requires `text` or `image_url`. - Snapshot requires active playback on the RelayTV server. - Snapshot responses are normalized to absolute URLs for Home Assistant entity attributes. +- The integration keeps `/status` as bootstrap/fallback and does not treat `/ui/events` as a replay log. For fuller documentation and examples, see the repository root README. diff --git a/custom_components/relaytv/__init__.py b/custom_components/relaytv/__init__.py index a71a709..d45494a 100644 --- a/custom_components/relaytv/__init__.py +++ b/custom_components/relaytv/__init__.py @@ -321,6 +321,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: coordinator.async_add_listener(_save_resume_position) await coordinator.async_config_entry_first_refresh() + await coordinator.async_start() await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) hass.data[DOMAIN][entry.entry_id]["mapping_unsubs"] = _setup_mapping_listeners(hass, entry) @@ -449,6 +450,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) entry_data = hass.data.get(DOMAIN, {}).pop(entry.entry_id, None) if entry_data: + await entry_data[DATA_COORDINATOR].async_stop() for unsub in entry_data.get("mapping_unsubs", []): unsub() @@ -483,6 +485,7 @@ async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> Non store = _get_entry_data(hass, entry.entry_id) if store and base_url: store[DATA_API].base_url = base_url + await store[DATA_COORDINATOR].async_restart() for unsub in store.get("mapping_unsubs", []): unsub() store["mapping_unsubs"] = _setup_mapping_listeners(hass, entry) diff --git a/custom_components/relaytv/coordinator.py b/custom_components/relaytv/coordinator.py index e2e8c1d..af92c08 100644 --- a/custom_components/relaytv/coordinator.py +++ b/custom_components/relaytv/coordinator.py @@ -1,9 +1,14 @@ -"""Coordinator for RelayTV polling.""" +"""Coordinator for RelayTV status refresh and UI event streaming.""" from __future__ import annotations -from datetime import timedelta +import asyncio +import json import logging +from datetime import timedelta +from typing import Any + +import aiohttp from homeassistant.core import HomeAssistant from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed @@ -12,21 +17,196 @@ from .relaytv_api import RelayTVApi _LOGGER = logging.getLogger(__name__) +_POLL_INTERVAL_FALLBACK = timedelta(seconds=3) +_POLL_INTERVAL_SSE = timedelta(seconds=30) +_SSE_CONNECT_TIMEOUT = 10 +_SSE_READ_TIMEOUT = 90 +_SSE_REFRESH_DEBOUNCE_SEC = 0.25 -class RelayTVCoordinator(DataUpdateCoordinator[dict]): - """Poll RelayTV for its current status.""" + +def _merge_playback_snapshot(current: dict[str, Any] | None, payload: dict[str, Any]) -> dict[str, Any] | None: + """Overlay compact playback-state fields onto the last full status payload.""" + if not isinstance(current, dict): + return None + + merged = dict(current) + merged.update(payload) + return merged + + +class RelayTVCoordinator(DataUpdateCoordinator[dict[str, Any]]): + """Hybrid RelayTV coordinator using /status plus /ui/events.""" def __init__(self, hass: HomeAssistant, api: RelayTVApi) -> None: super().__init__( hass, _LOGGER, name="RelayTV status", - update_interval=timedelta(seconds=3), + update_interval=_POLL_INTERVAL_FALLBACK, ) self.api = api + self._sse_task: asyncio.Task[None] | None = None + self._refresh_task: asyncio.Task[None] | None = None + self._sse_enabled = False - async def _async_update_data(self) -> dict: + async def _async_update_data(self) -> dict[str, Any]: data = await self.api.get_status() if data is None: raise UpdateFailed("Unable to fetch RelayTV status") return data + + async def async_start(self) -> None: + """Start the background SSE listener.""" + if self._sse_task and not self._sse_task.done(): + return + self._sse_task = asyncio.create_task(self._async_sse_loop()) + + async def async_stop(self) -> None: + """Stop background tasks owned by the coordinator.""" + tasks = [task for task in (self._refresh_task, self._sse_task) if task is not None] + self._refresh_task = None + self._sse_task = None + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + self._set_sse_enabled(False) + + async def async_restart(self) -> None: + """Reconnect the SSE stream, used after base URL changes.""" + await self.async_stop() + await self.async_start() + + def _set_sse_enabled(self, enabled: bool) -> None: + if self._sse_enabled == enabled: + return + self._sse_enabled = enabled + self.update_interval = _POLL_INTERVAL_SSE if enabled else _POLL_INTERVAL_FALLBACK + _LOGGER.debug("RelayTV SSE %s for %s", "enabled" if enabled else "disabled", self.api.base_url) + + def _schedule_refresh(self) -> None: + if self._refresh_task and not self._refresh_task.done(): + return + + async def _delayed_refresh() -> None: + try: + await asyncio.sleep(_SSE_REFRESH_DEBOUNCE_SEC) + await self.async_request_refresh() + except asyncio.CancelledError: + raise + except Exception: + _LOGGER.debug("RelayTV SSE-triggered refresh failed", exc_info=True) + + self._refresh_task = asyncio.create_task(_delayed_refresh()) + + async def _async_dispatch_event(self, event_name: str | None, data_lines: list[str]) -> None: + if not data_lines and not event_name: + return + + raw = "\n".join(data_lines).strip() + payload: Any + if raw: + try: + payload = json.loads(raw) + except Exception: + _LOGGER.debug("Ignoring non-JSON RelayTV SSE payload for %s: %r", event_name, raw) + return + else: + payload = {} + + if not event_name and isinstance(payload, dict): + event_name = str(payload.get("type") or "").strip() or None + if not event_name: + return + + if event_name == "status": + if isinstance(payload, dict): + self.async_set_updated_data(payload) + else: + self._schedule_refresh() + return + + if event_name == "playback": + if isinstance(payload, dict): + merged = _merge_playback_snapshot(self.data, payload) + if merged is not None: + self.async_set_updated_data(merged) + else: + self._schedule_refresh() + else: + self._schedule_refresh() + return + + if event_name in ("queue", "jellyfin"): + self._schedule_refresh() + return + + if event_name == "hello": + if not isinstance(self.data, dict) or not self.last_update_success: + self._schedule_refresh() + return + + if event_name == "ping": + return + + _LOGGER.debug("Ignoring unsupported RelayTV SSE event %s", event_name) + + async def _async_sse_loop(self) -> None: + backoff = 1.0 + headers = { + "Accept": "text/event-stream", + "Cache-Control": "no-cache", + } + + while True: + try: + timeout = aiohttp.ClientTimeout( + total=None, + connect=_SSE_CONNECT_TIMEOUT, + sock_read=_SSE_READ_TIMEOUT, + ) + async with self.api.session.get(self.api.url_for("ui/events"), headers=headers, timeout=timeout) as resp: + if resp.status >= 400: + raise aiohttp.ClientResponseError( + resp.request_info, + resp.history, + status=resp.status, + message=f"Unexpected RelayTV SSE response: {resp.status}", + headers=resp.headers, + ) + + self._set_sse_enabled(True) + backoff = 1.0 + event_name: str | None = None + data_lines: list[str] = [] + + async for raw_line in resp.content: + line = raw_line.decode("utf-8", "ignore").rstrip("\r\n") + if line == "": + await self._async_dispatch_event(event_name, data_lines) + event_name = None + data_lines = [] + continue + if line.startswith(":"): + continue + + field, _, value = line.partition(":") + if value.startswith(" "): + value = value[1:] + + if field == "event": + event_name = value.strip() or None + elif field == "data": + data_lines.append(value) + + if event_name or data_lines: + await self._async_dispatch_event(event_name, data_lines) + except asyncio.CancelledError: + raise + except Exception: + _LOGGER.debug("RelayTV SSE loop disconnected for %s", self.api.base_url, exc_info=True) + finally: + self._set_sse_enabled(False) + + await asyncio.sleep(backoff) + backoff = min(backoff * 2.0, 30.0) diff --git a/custom_components/relaytv/relaytv_api.py b/custom_components/relaytv/relaytv_api.py index 221aa25..099f02a 100644 --- a/custom_components/relaytv/relaytv_api.py +++ b/custom_components/relaytv/relaytv_api.py @@ -4,6 +4,7 @@ This integration targets the RelayTV server API documented in relaytv/docs/API.m We intentionally prefer the canonical endpoints: - GET /status +- GET /ui/events - POST /play - POST /smart - POST /enqueue @@ -43,6 +44,10 @@ class RelayTVApi: base_url: str timeout_s: float = 8.0 + def url_for(self, path: str) -> str: + """Build an absolute RelayTV URL for a relative API path.""" + return _join(self.base_url, path) + async def _request_json( self, method: str, @@ -50,7 +55,7 @@ class RelayTVApi: *, json: Optional[dict[str, Any]] = None, ) -> Optional[dict[str, Any]]: - url = _join(self.base_url, path) + url = self.url_for(path) try: async with asyncio.timeout(self.timeout_s): async with self.session.request(method, url, json=json) as resp: