344 lines
13 KiB
Python
344 lines
13 KiB
Python
"""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
|
|
|
|
# ── One-shot connection test (pairing UX) ──
|
|
|
|
@staticmethod
|
|
async def test_connect(host: str, port: int, key_path: str) -> bool:
|
|
"""Single-shot ADB connect used by the config-flow pairing step.
|
|
|
|
Throws away the poller immediately — the config flow runs before any
|
|
entry exists, so it must never touch an entry's persistent socket.
|
|
"""
|
|
poller = AdbMediaSessionPoller(host, port, key_path)
|
|
try:
|
|
return await poller.connect()
|
|
finally:
|
|
await poller.disconnect()
|
|
|
|
# ── 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():
|
|
await 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:
|
|
await self._note_failure(backoff)
|
|
_LOGGER.debug("Watchdog cycle failed: %s", err)
|
|
return self._jitter(backoff)
|
|
|
|
async 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
|
|
# Reconnect decision (health_check's documented role): close the
|
|
# socket so the next cycle rebuilds it from scratch instead of
|
|
# trusting shell()'s auto-reconnect on a broken transport.
|
|
try:
|
|
await self._poller.disconnect()
|
|
except Exception:
|
|
_LOGGER.debug("disconnect during reconnect decision failed", exc_info=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()
|