feat(connection): ShieldConnectionManager + StateStore — persistent socket, watchdog backoff reconnect, TTL expiry, listener API (Phase 1, B1/B5)
This commit is contained in:
James Mackay committed
1 parent
2ade5091c4
commit
4779c20407
1 file changed
+321
@@ -0,0 +1,321 @@
|
||||
"""ShieldConnectionManager — entry-scoped owner of the ADB connection (Phase 1).
|
||||
|
||||
The ADB socket, its watchdog and all ADB-derived state live here instead of on
|
||||
the media-player entity (plan blockers B1/B2/B5):
|
||||
|
||||
- one persistent socket; a watchdog task reconnects with exponential backoff
|
||||
(5 s doubling to a 300 s cap, ±25 % jitter) when it drops, and pulls the
|
||||
heartbeat data (wake + media sessions + running apps) while connected;
|
||||
- ADB-derived values live in a
|
||||
:class:`StateStore` with per-field timestamps and TTL expiry — stale values
|
||||
read as expired (None/empty) instead of clinging on;
|
||||
- entities are thin callback-views: they ``register_listener()`` and write HA
|
||||
state when the store changes (the androidtv core ``entity.py`` pattern);
|
||||
- ADB is optional: when no poller factory is provided the manager is *passive*
|
||||
and the entry still sets up — a failed ADB connect is a background retry,
|
||||
never a ``ConfigEntryNotReady``.
|
||||
|
||||
This module deliberately stays Home-Assistant-free at import time so it is
|
||||
unit-testable with plain pytest; every hass capability is injected (the entry
|
||||
builds the concrete poller factory). Task creation additionally works without
|
||||
hass (plain ``asyncio.create_task``) for the unit tests.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import logging
|
||||
import random
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
from .adb_session import AdbMediaSessionPoller
|
||||
from .const import (
|
||||
ADB_POLL_INTERVAL,
|
||||
ADB_RECONNECT_BACKOFF_CAP,
|
||||
ADB_RECONNECT_BACKOFF_START,
|
||||
ADB_RECONNECT_JITTER,
|
||||
STATE_STORE_TTL,
|
||||
)
|
||||
from .ha_adb_common import MediaSessionInfo, parse_audio_playback
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
# StateStore field names
|
||||
FIELD_AWAKE = "awake"
|
||||
FIELD_SESSIONS = "sessions"
|
||||
FIELD_AUDIO_PLAYBACK = "audio_playback"
|
||||
FIELD_RUNNING_APPS = "running_apps"
|
||||
|
||||
ALL_FIELDS = (FIELD_AWAKE, FIELD_SESSIONS, FIELD_AUDIO_PLAYBACK, FIELD_RUNNING_APPS)
|
||||
|
||||
|
||||
class StateStore:
|
||||
"""Timed per-field state cache shared by all entities of one entry.
|
||||
|
||||
Reading an expired (or missing) field returns the caller-supplied
|
||||
default, so a stale "playing" session can never outlive the TTL (plan
|
||||
blocker B5). Values are written only via :meth:`set`, which re-stamps the
|
||||
field's freshness timestamp.
|
||||
"""
|
||||
|
||||
def __init__(self, ttl: float = STATE_STORE_TTL, clock: Callable[[], float] = time.monotonic):
|
||||
self._ttl = ttl
|
||||
self._clock = clock
|
||||
self._values: dict[str, tuple[float, Any]] = {}
|
||||
|
||||
def set(self, name: str, value: Any) -> None:
|
||||
"""Store a value, re-stamping its freshness timestamp."""
|
||||
self._values[name] = (self._clock(), value)
|
||||
|
||||
def get(self, name: str, default: Any = None) -> Any:
|
||||
"""Return the field's value, or ``default`` when it is missing/expired."""
|
||||
stamped = self._values.get(name)
|
||||
if stamped is None:
|
||||
return default
|
||||
at, value = stamped
|
||||
if self._clock() - at > self._ttl:
|
||||
return default
|
||||
return value
|
||||
|
||||
def clear(self, *names: str) -> None:
|
||||
"""Drop fields immediately so they read as expired on the next get."""
|
||||
for name in names:
|
||||
self._values.pop(name, None)
|
||||
|
||||
@property
|
||||
def ttl(self) -> float:
|
||||
return self._ttl
|
||||
|
||||
|
||||
class ShieldConnectionManager:
|
||||
"""Owner of one entry's persistent ADB connection and state.
|
||||
|
||||
:param poller_factory: builds the ADB poller when the watchdog starts;
|
||||
skipped entirely for entries without ADB (passive manager) or tests
|
||||
that inject a ready-to-use ``poller`` directly.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass=None,
|
||||
entry=None,
|
||||
poller_factory: Callable[[], AdbMediaSessionPoller] | None = None,
|
||||
poller: AdbMediaSessionPoller | None = None,
|
||||
store: StateStore | None = None,
|
||||
):
|
||||
self.hass = hass
|
||||
self.entry = entry
|
||||
self.store = store if store is not None else StateStore(ttl=STATE_STORE_TTL)
|
||||
self._poller_factory = poller_factory
|
||||
self._poller = poller
|
||||
self._listeners: list[Callable[[], None]] = []
|
||||
self._tasks: list[asyncio.Task] = []
|
||||
self._stopping = False
|
||||
self._failures = 0
|
||||
self._down_since = False # listeners already told the connection is down
|
||||
|
||||
# ── Introspection (thin views for entities) ──
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""True when ADB is active for this entry (a poller exists)."""
|
||||
return self._poller is not None
|
||||
|
||||
@property
|
||||
def connected(self) -> bool:
|
||||
return bool(self._poller and self._poller.connected)
|
||||
|
||||
@property
|
||||
def awake(self) -> bool | None:
|
||||
"""Device wake state (exact mWakefulness parse), None = unknown."""
|
||||
return self.store.get(FIELD_AWAKE)
|
||||
|
||||
@property
|
||||
def active_sessions(self) -> list[MediaSessionInfo]:
|
||||
"""Stored media sessions, sorted playing > paused > other."""
|
||||
return self.store.get(FIELD_SESSIONS, [])
|
||||
|
||||
@property
|
||||
def audio_playback(self) -> list[str]:
|
||||
"""Packages that recently played audio, ordered last-played-first."""
|
||||
return self.store.get(FIELD_AUDIO_PLAYBACK, [])
|
||||
|
||||
@property
|
||||
def running_apps(self) -> list[dict]:
|
||||
"""Visible recent-task apps, ordered by recency."""
|
||||
return self.store.get(FIELD_RUNNING_APPS, [])
|
||||
|
||||
def best_session(
|
||||
self, foreground_package: str | None = None
|
||||
) -> MediaSessionInfo | None:
|
||||
"""Best media session: foreground app first, then playing > paused."""
|
||||
sessions = self.active_sessions
|
||||
if not sessions:
|
||||
return None
|
||||
if foreground_package:
|
||||
for session in sessions:
|
||||
if session.package == foreground_package:
|
||||
return session
|
||||
return sessions[0]
|
||||
|
||||
# ── Listener API (entities are thin callback-views) ──
|
||||
|
||||
def register_listener(self, listener: Callable[[], None]) -> Callable[[], None]:
|
||||
"""Register a refresh callback; returns an unsubscribe callable."""
|
||||
self._listeners.append(listener)
|
||||
|
||||
def _unregister() -> None:
|
||||
with contextlib.suppress(ValueError):
|
||||
self._listeners.remove(listener)
|
||||
|
||||
return _unregister
|
||||
|
||||
def _notify(self) -> None:
|
||||
for listener in tuple(self._listeners):
|
||||
try:
|
||||
listener()
|
||||
except Exception:
|
||||
_LOGGER.exception("Shield state listener failed")
|
||||
|
||||
# ── Delegation (commands over the managed socket) ──
|
||||
|
||||
async def send_media_command(self, command: str) -> bool:
|
||||
"""Send a media key command over the persistent socket."""
|
||||
if not self.available:
|
||||
return False
|
||||
return await self._poller.send_media_command(command)
|
||||
|
||||
async def bring_to_foreground(self, package: str) -> bool:
|
||||
if not self.available:
|
||||
return False
|
||||
return await self._poller.bring_to_foreground(package)
|
||||
|
||||
# ── Lifecycle ──
|
||||
|
||||
async def async_start(self) -> None:
|
||||
"""Start the persistent connection loop.
|
||||
|
||||
Never validates the ADB connection here: setup must not fail because
|
||||
of ADB (it is optional), so all failure handling lives in the
|
||||
watchdog's background retry.
|
||||
"""
|
||||
self._stopping = False
|
||||
self._tasks.append(self._create_task(self._watchdog(), "shield_adb_watchdog"))
|
||||
|
||||
async def async_stop(self) -> None:
|
||||
"""Cancel all manager tasks and close the socket.
|
||||
|
||||
Called on entry unload and on EVENT_HOMEASSISTANT_STOP.
|
||||
"""
|
||||
self._stopping = True
|
||||
tasks, self._tasks = self._tasks, []
|
||||
for task in tasks:
|
||||
task.cancel()
|
||||
self._listeners.clear()
|
||||
if self._poller is not None:
|
||||
with contextlib.suppress(Exception):
|
||||
await self._poller.disconnect()
|
||||
self._poller = None
|
||||
|
||||
def _create_task(self, coro, name: str) -> asyncio.Task:
|
||||
if self.hass is not None:
|
||||
return self.hass.async_create_background_task(coro, name)
|
||||
return asyncio.create_task(coro, name=name)
|
||||
|
||||
async def _watchdog(self) -> None:
|
||||
"""Persistent loop: reconnect with backoff when down, refresh when up.
|
||||
|
||||
A passive manager (no poller possible) exits immediately — the entry
|
||||
recreates it when options change (full reload).
|
||||
"""
|
||||
try:
|
||||
if self._poller is None:
|
||||
if self._poller_factory is None:
|
||||
return
|
||||
self._poller = self._poller_factory()
|
||||
if self._poller is None:
|
||||
return
|
||||
|
||||
backoff = ADB_RECONNECT_BACKOFF_START
|
||||
while not self._stopping:
|
||||
delay = await self._watchdog_cycle(backoff)
|
||||
if self._stopping:
|
||||
break
|
||||
if self.connected:
|
||||
backoff = ADB_RECONNECT_BACKOFF_START
|
||||
else:
|
||||
backoff = min(backoff * 2, ADB_RECONNECT_BACKOFF_CAP)
|
||||
await asyncio.sleep(delay)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception:
|
||||
_LOGGER.exception("Shield connection watchdog crashed")
|
||||
|
||||
async def _watchdog_cycle(self, backoff: float) -> float:
|
||||
"""Run one watchdog cycle; return seconds to sleep before the next.
|
||||
|
||||
When the connection is up this is the heartbeat: wake check + (when
|
||||
awake) media sessions + running apps, all stored with timestamps.
|
||||
The current backoff sleeps after failures; success resets it."""
|
||||
|
||||
try:
|
||||
if not self.connected and not await self._poller.connect():
|
||||
self._note_failure(backoff)
|
||||
return self._jitter(backoff)
|
||||
if self._failures:
|
||||
self._failures = 0
|
||||
self._down_since = False
|
||||
_LOGGER.info("ADB connection recovered")
|
||||
await self._refresh()
|
||||
return ADB_POLL_INTERVAL
|
||||
except Exception as err:
|
||||
self._note_failure(backoff)
|
||||
_LOGGER.debug("Watchdog cycle failed: %s", err)
|
||||
return self._jitter(backoff)
|
||||
|
||||
def _note_failure(self, backoff: float) -> None:
|
||||
"""Bookkeep a failed cycle: log sparingly, drop stale store, notify."""
|
||||
self._failures += 1
|
||||
if self._failures == 1 or self._failures % 6 == 0:
|
||||
_LOGGER.warning(
|
||||
"ADB connection down (%s consecutive failures) — retrying in ~%.0fs",
|
||||
self._failures,
|
||||
backoff,
|
||||
)
|
||||
if not self._down_since:
|
||||
self._down_since = True
|
||||
self.store.clear(*ALL_FIELDS)
|
||||
self._notify()
|
||||
|
||||
@staticmethod
|
||||
def _jitter(delay: float) -> float:
|
||||
"""Backoff with ±jitter so many clients do not reconnect in lockstep."""
|
||||
return max(
|
||||
1.0,
|
||||
delay * (1.0 + random.uniform(-ADB_RECONNECT_JITTER, ADB_RECONNECT_JITTER)),
|
||||
)
|
||||
|
||||
async def _refresh(self) -> None:
|
||||
"""Heartbeat: pull all ADB-derived state into the store (with stamps)."""
|
||||
poller = self._poller
|
||||
awake = await poller.is_awake()
|
||||
if awake is None:
|
||||
# Failed poll (connection dropped / device unreachable) — surface
|
||||
# as a cycle failure instead of clinging to the last values.
|
||||
raise ConnectionError("ADB wake poll failed")
|
||||
self.store.set(FIELD_AWAKE, awake)
|
||||
if awake:
|
||||
# Full dump, no truncation (B4) — low sessions must not be lost.
|
||||
await poller.poll_media_sessions(head_lines=None)
|
||||
self.store.set(FIELD_SESSIONS, list(poller.all_sessions))
|
||||
self.store.set(
|
||||
FIELD_AUDIO_PLAYBACK, parse_audio_playback(poller.last_dump or "")
|
||||
)
|
||||
self.store.set(FIELD_RUNNING_APPS, await poller.poll_running_apps())
|
||||
else:
|
||||
# Standby: playback data is not current — expire it.
|
||||
self.store.set(FIELD_SESSIONS, [])
|
||||
self.store.set(FIELD_AUDIO_PLAYBACK, [])
|
||||
self.store.set(FIELD_RUNNING_APPS, [])
|
||||
self._notify()
|
||||
Reference in new issue
Block a user