engineer
c884c7e1ab
The "no real Shield was available, so this output shape is Android's documentation rather than a capture" mark lived only in the test module, where a reader of the discovery code never looks. It now sits in connection.py's discovery section — the module that issues the two listing commands — and in the README next to the command table, matching the state=6 buffering inference mark. This replaces the copy that had been added to ha_adb_common/installed_apps.py: t62's in-scope set is connection.py, sensor.py, README.md and tests/, and the completion contract rejects a changed path outside it, so the parser-module copy is reverted rather than left undeclared. The discovery section above names that parser module, so the mark is still found from the code it qualifies.
1413 lines
60 KiB
Python
1413 lines
60 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 dataclasses
|
|
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,
|
|
APP_ICON_FALLBACKS,
|
|
APP_ICON_GENERIC,
|
|
APP_LABEL_EXIT_MARKER,
|
|
APP_LABEL_TTL,
|
|
APP_LIST_TTL,
|
|
APP_SWEEP_CONCURRENCY,
|
|
APP_SWEEP_SPACING,
|
|
CONF_LOGCAT_EVENTS,
|
|
LOGCAT_SUPPRESS_WINDOW,
|
|
STATE_STORE_TTL,
|
|
SYSTEM_MEDIA_PACKAGES,
|
|
)
|
|
from .ha_adb_common import (
|
|
MediaSessionInfo,
|
|
foreground_from_logcat,
|
|
icon_for_package,
|
|
installed_packages,
|
|
is_launchable,
|
|
parse_application_label,
|
|
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"
|
|
|
|
# Protocol transport (androidtvremote2) push fields
|
|
FIELD_IS_ON = "is_on"
|
|
FIELD_IS_AVAILABLE = "is_available"
|
|
FIELD_CURRENT_APP = "current_app"
|
|
FIELD_VOLUME_INFO = "volume_info"
|
|
|
|
# ADB fallback foreground identity (P3-2): stamped only when the protocol
|
|
# push is silent; consumers prefer FIELD_CURRENT_APP over this.
|
|
FIELD_FOREGROUND = "foreground"
|
|
|
|
# ADB-derived fields (cleared when the ADB connection goes down — they are
|
|
# regenerated by the watchdog); protocol pushes (IS_ON/IS_AVAILABLE/
|
|
# CURRENT_APP/VOLUME_INFO) are device-reported truth and expire by TTL only,
|
|
# so an ADB outage never clobbers them.
|
|
ADB_FIELDS = (
|
|
FIELD_AWAKE,
|
|
FIELD_SESSIONS,
|
|
FIELD_AUDIO_PLAYBACK,
|
|
FIELD_RUNNING_APPS,
|
|
FIELD_FOREGROUND,
|
|
)
|
|
PROTOCOL_FIELDS = (
|
|
FIELD_IS_ON,
|
|
FIELD_IS_AVAILABLE,
|
|
FIELD_CURRENT_APP,
|
|
FIELD_VOLUME_INFO,
|
|
)
|
|
|
|
ALL_FIELDS = ADB_FIELDS + PROTOCOL_FIELDS
|
|
|
|
|
|
def volume_tuple(info) -> tuple | None:
|
|
"""Normalize the protocol's volume push to ``(level, max, muted)``.
|
|
|
|
The library delivers a VolumeInfo TypedDict (``level``/``max``/``muted``).
|
|
Readout only — there is no protocol-native VOLUME_SET (C-VOL), so nothing
|
|
here ever sends one.
|
|
"""
|
|
if info is None:
|
|
return None
|
|
try:
|
|
return (
|
|
float(info.get("level", 0.0)),
|
|
float(info.get("max", 1.0)),
|
|
bool(info.get("muted", False)),
|
|
)
|
|
except (AttributeError, TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
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
|
|
|
|
def values_stamped(self) -> dict[str, tuple[float, Any]]:
|
|
"""Public stamped-values snapshot (F6): ``{field: (stamp, value)}``."""
|
|
return dict(self._values)
|
|
|
|
|
|
class ShieldConnectionManager:
|
|
"""Owner of one entry's persistent connections (both transports) + state.
|
|
|
|
Transport A — ADB (adb_shell): the watchdog keeps a persistent socket,
|
|
pulls heartbeats and reconnects with its own backoff ladder.
|
|
Transport B — remote protocol (androidtvremote2, Phase 2): the library
|
|
owns reconnection (``keep_reconnecting`` with internal backoff), pushes
|
|
``is_on``/``is_available``/``current_app``/``volume_info`` into the state
|
|
store, and surfaces invalid certs (device hard reset) via a reauth hook.
|
|
|
|
: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.
|
|
:param protocol_factory: builds the AndroidTVRemote client; instead pass
|
|
``protocol_settings`` (host/client_name/certfile/keyfile) and the
|
|
manager builds it, or ``protocol`` directly (tests).
|
|
:param invalid_auth_callback: called when the device rejects our
|
|
certificates (hard reset) — the entry-owned reauth hook (t9).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
hass=None,
|
|
entry=None,
|
|
poller_factory: Callable[[], AdbMediaSessionPoller] | None = None,
|
|
poller: AdbMediaSessionPoller | None = None,
|
|
store: StateStore | None = None,
|
|
protocol_factory: Callable[[], Any] | None = None,
|
|
protocol: Any = None,
|
|
protocol_settings: dict | None = None,
|
|
invalid_auth_callback: Callable[[], None] | None = None,
|
|
clock: Callable[[], float] = time.monotonic,
|
|
):
|
|
self.hass = hass
|
|
self.entry = entry
|
|
self.store = (
|
|
store if store is not None else StateStore(ttl=STATE_STORE_TTL, clock=clock)
|
|
)
|
|
self._poller_factory = poller_factory
|
|
self._poller = poller
|
|
# Remote-protocol transport (androidtvremote2): entry passes either a
|
|
# settings dict (manager itself builds the real client) or a factory /
|
|
# ready-made client (tests, or the config flow when it pre-pairs).
|
|
self._protocol_factory = protocol_factory
|
|
self._protocol = protocol
|
|
self._protocol_settings = protocol_settings
|
|
self._invalid_auth_callback = invalid_auth_callback
|
|
self._protocol_loop: asyncio.AbstractEventLoop | None = None
|
|
self._listeners: list[Callable[[], None]] = []
|
|
# P3-2d last-known-state cache: package → (session info, stamp).
|
|
# Serves the foreground app's last known playback state when it
|
|
# publishes no MediaSession (TiviMate, R2). TTL-capped: 'playing'
|
|
# is NEVER resurrected past STATE_STORE_TTL.
|
|
self._last_known: dict[str, tuple[MediaSessionInfo, float]] = {}
|
|
# P5-1: device manifest labels — package → (label, stamp). Long-TTL
|
|
# (APP_LABEL_TTL) cache resolved via `pm dump` on demand, one
|
|
# package per watchdog cycle so a burst never blocks the watchdog.
|
|
# An entry means "a fetch RAN" (t57 UNIT 0): a tuple whose label is
|
|
# "" is a definitive negative, and a fetch that could not run is not
|
|
# in here at all — so the flag t55 carried to say so was dead weight
|
|
# (its only writer always passed True) and is gone.
|
|
self._app_labels: dict[str, tuple[str, float]] = {}
|
|
self._app_label_requests: set[str] = set()
|
|
# t55: packages whose label fetch is queued or in flight — a render
|
|
# during the fetch must not enqueue the same package twice.
|
|
self._app_label_inflight: set[str] = set()
|
|
# t57: the discovered installed-apps table — (rows, stamp), or None when
|
|
# no sweep has produced a fresh answer. Only a sweep that RAN writes
|
|
# here (t62 F1): a failed or unavailable sweep records itself as UNKNOWN
|
|
# instead, because stamping an empty answer would serve "0 apps" — and
|
|
# block every retry — for the whole APP_LIST_TTL. Refreshed on demand;
|
|
# the watchdog cycle never rebuilds it.
|
|
self._installed_apps: tuple[list[dict], float] | None = None
|
|
# t62 F1: True when the last sweep could not deliver an answer (no
|
|
# poller, ADB down, a shell that raised or answered nothing) — so an
|
|
# empty installed_apps() is "unknown", not "this device has no apps".
|
|
self._apps_unavailable = False
|
|
# ...and the clock stamp of an ATTEMPTED failure (shells really went
|
|
# out). A retry then costs two listing shells, so it is spaced by
|
|
# APP_SWEEP_RETRY_SECONDS instead of firing on every state poll; an
|
|
# unavailable transport costs nothing and retries immediately.
|
|
self._apps_failed_at: float | None = None
|
|
#: t57: the in-flight sweep, so concurrent callers share one sweep
|
|
#: instead of stacking a second burst on the same connection.
|
|
self._apps_sweep: asyncio.Task | None = None
|
|
# P5-3 (C-STREAM): the dedicated logcat line-stream consumer.
|
|
# NEVER shares the poller's ADB socket — its own device object and
|
|
# connection; CONF_LOGCAT_EVENTS gates it (default off) and the
|
|
# heartbeat cadence remains the floor when the stream is off/disabled.
|
|
self._logcat_task: asyncio.Task | None = None
|
|
self._logcat_disabled = False
|
|
self._logcat_failures: list[float] = []
|
|
self._logcat_last_pkg: str | None = None
|
|
# G10 (t30): protocol device identity captured once per session —
|
|
# (name, manufacturer, model, sw) for the registry's device card.
|
|
self.protocol_device_data: tuple[str, str, str, str] | None = None
|
|
self._logcat_last_stamp: float = -1e9
|
|
# Injectable ladder sleep (t26: deterministic tests / future alt-floors)
|
|
self._logcat_ladder_sleep = asyncio.sleep
|
|
# Injectable clock keeps the TTL expiry tests deterministic.
|
|
self._clock = clock
|
|
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()
|
|
|
|
# ── Remote-protocol transport (androidtvremote2, Phase 2) ──
|
|
|
|
@staticmethod
|
|
def build_protocol_client(
|
|
host: str,
|
|
client_name: str,
|
|
certfile: str,
|
|
keyfile: str,
|
|
api_port: int = 6466,
|
|
pair_port: int = 6467,
|
|
loop=None,
|
|
):
|
|
"""Build a real androidtvremote2 client (cert/PIN flow hook, t9).
|
|
|
|
Kept as a manager-level factory so the config flow never hand-rolls
|
|
the constructor; the import stays local so this module remains
|
|
importable where androidtvremote2 is not installed.
|
|
"""
|
|
try:
|
|
import androidtvremote2
|
|
except ImportError:
|
|
_LOGGER.warning(
|
|
"androidtvremote2 is not installed — the remote-protocol "
|
|
"transport is disabled and the entry degrades to ADB-only."
|
|
)
|
|
return None
|
|
|
|
return androidtvremote2.AndroidTVRemote(
|
|
client_name=client_name,
|
|
certfile=certfile,
|
|
keyfile=keyfile,
|
|
host=host,
|
|
api_port=api_port,
|
|
pair_port=pair_port,
|
|
loop=loop,
|
|
)
|
|
|
|
def _build_protocol(self):
|
|
"""Default factory from the entry settings (host + cert/key paths)."""
|
|
settings = self._protocol_settings or {}
|
|
return self.build_protocol_client(
|
|
host=settings["host"],
|
|
client_name=settings.get("client_name", "Home Assistant"),
|
|
certfile=settings["certfile"],
|
|
keyfile=settings["keyfile"],
|
|
)
|
|
|
|
async def _start_protocol(self) -> None:
|
|
"""Spin up the protocol transport alongside the ADB watchdog.
|
|
|
|
One owner per transport: the library's own ``keep_reconnecting()``
|
|
handles reconnection with its internal backoff, so no competing
|
|
watchdog is built here. Push callbacks write straight into the
|
|
StateStore and ping registered listeners.
|
|
"""
|
|
if self._protocol is None:
|
|
if self._protocol_factory is None:
|
|
if self._protocol_settings is None:
|
|
return # protocol transport disabled for this entry
|
|
self._protocol_factory = self._build_protocol
|
|
self._protocol = self._protocol_factory()
|
|
client = self._protocol
|
|
if client is None:
|
|
return
|
|
# Real client: make sure local certificate material exists before it
|
|
# connects (fails soft — pair flows can regenerate).
|
|
generate_cert = getattr(client, "async_generate_cert_if_missing", None)
|
|
if generate_cert is not None:
|
|
with contextlib.suppress(Exception):
|
|
await generate_cert()
|
|
self._protocol_loop = asyncio.get_running_loop()
|
|
client.add_is_on_updated_callback(self._on_is_on)
|
|
client.add_is_available_updated_callback(self._on_is_available)
|
|
client.add_current_app_updated_callback(self._on_current_app)
|
|
client.add_volume_info_updated_callback(self._on_volume_info)
|
|
# One owner per transport: the library's keep_reconnecting() creates
|
|
# and owns its reconnect task internally — we never wrap or duplicate
|
|
# it (the stop path closes the client, which tears the task down).
|
|
client.keep_reconnecting(self._on_invalid_auth)
|
|
|
|
def _on_is_on(self, on) -> None:
|
|
self._protocol_push(FIELD_IS_ON, bool(on))
|
|
|
|
def _on_is_available(self, is_available) -> None:
|
|
self._protocol_push(FIELD_IS_AVAILABLE, bool(is_available))
|
|
|
|
def _on_current_app(self, app_package) -> None:
|
|
self._protocol_push(FIELD_CURRENT_APP, app_package)
|
|
|
|
def _on_volume_info(self, info) -> None:
|
|
self._protocol_push(FIELD_VOLUME_INFO, volume_tuple(info))
|
|
|
|
def _on_invalid_auth(self) -> None:
|
|
"""Device hard reset — certificates rejected.
|
|
|
|
The library keeps its own backoff, but retrying can never succeed
|
|
since certificated auth is gone — the entry-level callback (reauth
|
|
flow) decides the user-facing response (t9).
|
|
"""
|
|
_LOGGER.warning(
|
|
"Shield remote-protocol auth is invalid (device hard reset?) — "
|
|
"notifying the entry's reauth hook"
|
|
)
|
|
if self._invalid_auth_callback is not None:
|
|
try:
|
|
self._invalid_auth_callback()
|
|
except Exception:
|
|
_LOGGER.exception("Shield invalid-auth callback failed")
|
|
|
|
def _protocol_push(self, field: str, value) -> None:
|
|
"""Stamp a push value into the store and ping listeners.
|
|
|
|
Push callbacks may arrive off the event loop (grpc worker threads);
|
|
route the listener notification back onto the captured loop.
|
|
"""
|
|
self.store.set(field, value)
|
|
loop = self._protocol_loop
|
|
if loop is not None and loop.is_running():
|
|
try:
|
|
loop.call_soon_threadsafe(self._notify)
|
|
return
|
|
except RuntimeError:
|
|
pass # loop closed during shutdown — skip the notify
|
|
self._notify()
|
|
|
|
# ── 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, [])
|
|
|
|
@property
|
|
def protocol_active(self) -> bool:
|
|
"""True when the remote-protocol transport is wired for this entry.
|
|
|
|
Entities use this instead of reaching into ``_protocol``: a passive
|
|
manager means "no transport", not "broken transport".
|
|
"""
|
|
return self._protocol is not None
|
|
|
|
async def async_protocol_device_info(self) -> tuple[str, str] | None:
|
|
"""(device name, MAC) via the protocol client (P5-1 migration).
|
|
|
|
None when the transport is absent or the lookup fails — callers
|
|
fall back to the config-entry id.
|
|
"""
|
|
if self._protocol is None:
|
|
return None
|
|
with contextlib.suppress(Exception):
|
|
name, mac = await asyncio.wait_for(
|
|
self._protocol.async_get_name_and_mac(), timeout=10
|
|
)
|
|
return name, mac
|
|
return None
|
|
|
|
@property
|
|
def adb_poller(self):
|
|
"""The ADB poller (or None) — public accessor replacing the
|
|
``_poller`` reach-in in the adb_command service handler (F6)."""
|
|
return self._poller
|
|
|
|
@property
|
|
def is_on(self) -> bool | None:
|
|
"""Protocol power push (device reports on/off state)."""
|
|
return self.store.get(FIELD_IS_ON)
|
|
|
|
@property
|
|
def is_available(self) -> bool | None:
|
|
"""Protocol availability push (device reachable at the protocol)."""
|
|
return self.store.get(FIELD_IS_AVAILABLE)
|
|
|
|
@property
|
|
def current_app(self) -> str | None:
|
|
"""Protocol current-app push (raw package; name resolution deferred)."""
|
|
return self.store.get(FIELD_CURRENT_APP)
|
|
|
|
@property
|
|
def foreground_app(self) -> str | None:
|
|
"""Best foreground identity: protocol push first, ADB parse fallback
|
|
(P3-2 store priority: FIELD_CURRENT_APP > FIELD_FOREGROUND)."""
|
|
return self.store.get(FIELD_CURRENT_APP) or self.store.get(FIELD_FOREGROUND)
|
|
|
|
@property
|
|
def volume_info(self) -> tuple | None:
|
|
"""Protocol volume push as ``(level, max, muted)`` — readout only
|
|
(no protocol VOLUME_SET exists, C-VOL)."""
|
|
return self.store.get(FIELD_VOLUME_INFO)
|
|
|
|
def best_session(
|
|
self, foreground_package: str | None = None
|
|
) -> MediaSessionInfo | None:
|
|
"""Best media session: foreground app first, then playing > paused.
|
|
|
|
P3-2d: when there is no live session at all, the foreground app's
|
|
last known state is served from the TTL-capped cache (TiviMate
|
|
publishes no MediaSession — RESEARCH_SHIELD R2).
|
|
"""
|
|
sessions = self.active_sessions
|
|
if foreground_package:
|
|
for session in sessions:
|
|
if session.package == foreground_package:
|
|
return session
|
|
if sessions:
|
|
return sessions[0]
|
|
if foreground_package:
|
|
return self.last_known_session(foreground_package)
|
|
return None
|
|
|
|
def _cache_last_known_sessions(self) -> None:
|
|
"""Stamp every live session's state as the package's last known."""
|
|
now = self._clock()
|
|
for session in self.active_sessions:
|
|
if session.package:
|
|
self._last_known[session.package] = (session, now)
|
|
|
|
def last_known_session(self, package: str) -> MediaSessionInfo | None:
|
|
"""Last known session state for a package, TTL-capped.
|
|
|
|
Returns None once STATE_STORE_TTL passes — 'playing' must never be
|
|
resurrected from a stale cache (P3-2d expiry contract).
|
|
"""
|
|
cached = self._last_known.get(package)
|
|
if cached is None:
|
|
return None
|
|
info, stamp = cached
|
|
if self._clock() - stamp > STATE_STORE_TTL:
|
|
self._last_known.pop(package, None)
|
|
return None
|
|
return info
|
|
|
|
def store_snapshot(self) -> dict[str, dict]:
|
|
"""Public StateStore snapshot for diagnostics (F6).
|
|
|
|
``{field: {"age": seconds, "value": ...}}`` with dataclass values
|
|
expanded to dicts so the payload stays JSON-serializable.
|
|
"""
|
|
snapshot: dict[str, dict] = {}
|
|
for name, (stamp, value) in self.store.values_stamped().items():
|
|
if dataclasses.is_dataclass(value) and not isinstance(value, type):
|
|
value = dataclasses.asdict(value)
|
|
snapshot[name] = {"age": round(time.monotonic() - stamp, 3), "value": value}
|
|
return snapshot
|
|
|
|
async def send_text_ime(self, text: str) -> bool:
|
|
"""Text entry via the protocol client's send_text (P5-2).
|
|
|
|
The Shield may swallow the text when its on-screen virtual keyboard
|
|
is up (RESEARCH_IDEAS §2 availability gate) — the call reports the
|
|
send outcome; callers fall back to the ADB input-text path only
|
|
when the protocol transport is down or the send failed.
|
|
"""
|
|
if self._protocol is None:
|
|
return False
|
|
try:
|
|
self._protocol.send_text(text)
|
|
return True
|
|
except Exception as err:
|
|
_LOGGER.debug("protocol send_text failed: %s", err)
|
|
return False
|
|
|
|
async def send_adb_input_text(self, text: str) -> bool:
|
|
"""ADB fallback for text entry: `input text` (P5-2).
|
|
|
|
Used only when the protocol send_text path is unavailable/failed.
|
|
Shell-quoted (shlex) so single quotes and spaces survive.
|
|
"""
|
|
import shlex
|
|
|
|
poller = self._poller
|
|
if poller is None or not self.connected:
|
|
return False
|
|
try:
|
|
result = await poller.shell(f"input text {shlex.quote(text)}")
|
|
return result is not None
|
|
except Exception as err:
|
|
_LOGGER.debug("adb input text failed: %s", err)
|
|
return False
|
|
|
|
def app_label(self, package: str | None) -> str | None:
|
|
"""Cached manifest label for a package (P5-1), or None (t55).
|
|
|
|
A cache miss enqueues an on-demand `pm dump` fetch — drained one
|
|
package per watchdog cycle via the manager task pool, so entity
|
|
refreshes and the watchdog are never blocked by a burst.
|
|
|
|
What is cached is the FETCH, not the wishful result (t55 root cause):
|
|
|
|
* a fetch that RAN caches its outcome for ``APP_LABEL_TTL`` — a label
|
|
when the manifest has one, and the empty string otherwise. It is
|
|
definitive: the device answered, the package simply has no default
|
|
label (several system/OEM packages do not). No re-enqueue, so a
|
|
permanently label-less package costs ZERO further round trips
|
|
(t24 F1 stays closed).
|
|
* a fetch that could NOT run (no poller, ADB down, shell error) caches
|
|
nothing, so the next render enqueues again — a miss is retried
|
|
visibly instead of being stamped as an eternal "no label". This was
|
|
the live defect: the old code stored ("", stamp) on every failure
|
|
path, which silenced the label tier for the whole 24 h TTL and left
|
|
app_name showing the raw package id.
|
|
* while a fetch for the package is queued or in flight, the next
|
|
render does not enqueue it a second time.
|
|
"""
|
|
if not package:
|
|
return None
|
|
cached = self._app_labels.get(package)
|
|
if cached is not None:
|
|
label, stamp = cached
|
|
if self._clock() - stamp <= APP_LABEL_TTL:
|
|
# empty label = negative cache (t24 F1): the pm dump RAN and
|
|
# found nothing — do not re-enqueue, zero per-heartbeat tax.
|
|
# A fetch that did not run is not cached at all (t57 UNIT 0),
|
|
# so there is no "unresolved" shape to fall through on.
|
|
return label or None
|
|
self._app_labels.pop(package, None) # stale — refetch on demand
|
|
if package not in self._app_label_requests and (
|
|
package not in self._app_label_inflight
|
|
):
|
|
self._app_label_requests.add(package)
|
|
return None
|
|
|
|
def _known_label(self, package: str) -> tuple[bool, str]:
|
|
"""``(known, label)`` from the label cache — "known" = a fetch RAN (t57).
|
|
|
|
The accessor the DISCOVERY sweep needs and ``app_label`` deliberately is
|
|
not: ``app_label`` answers "what should be displayed", so a definitive
|
|
negative (a package with no manifest label) comes back as ``None``,
|
|
exactly like a miss. A sweep that used it would re-dump every label-less
|
|
package on every run and throw away the 24 h cache it is supposed to
|
|
benefit from. Here the two are distinguishable.
|
|
"""
|
|
cached = self._app_labels.get(package)
|
|
if cached is None:
|
|
return False, ""
|
|
label, stamp = cached
|
|
if self._clock() - stamp > APP_LABEL_TTL:
|
|
self._app_labels.pop(package, None) # stale — refetch on demand
|
|
return False, ""
|
|
return True, label
|
|
|
|
def cache_app_label(self, package: str, label: str | None) -> None:
|
|
"""Record a label fetch that RAN, and what it found (t55, t57).
|
|
|
|
The cache's only writer. ``None``/"" is the negative result devices
|
|
genuinely report for label-less packages, and it is definitive: no
|
|
further ``pm dump`` for ``APP_LABEL_TTL``. A fetch that did NOT run
|
|
must not call this — that was the live defect (see
|
|
``_fetch_app_label``).
|
|
"""
|
|
self._app_labels[package] = (label or "", self._clock())
|
|
|
|
@property
|
|
def app_label_requests(self) -> set[str]:
|
|
"""Packages queued for a manifest-label fetch (public, for tests)."""
|
|
return set(self._app_label_requests)
|
|
|
|
@property
|
|
def app_label_inflight(self) -> set[str]:
|
|
"""Packages whose label fetch is running right now (public, tests)."""
|
|
return set(self._app_label_inflight)
|
|
|
|
async def async_capture_protocol_device_data(self) -> None:
|
|
"""Capture (name, manufacturer, model, sw) once (t30 G10)."""
|
|
if self._protocol is None or self.protocol_device_data is not None:
|
|
return
|
|
try:
|
|
info = self._protocol.device_info or {} # {manufacturer, model, sw_version}
|
|
name, _mac = await asyncio.wait_for(
|
|
self._protocol.async_get_name_and_mac(), timeout=10.0
|
|
)
|
|
self.protocol_device_data = (
|
|
name,
|
|
info.get("manufacturer", "Google"),
|
|
info.get("model", "Shield TV"),
|
|
info.get("sw_version", ""),
|
|
)
|
|
except Exception as err:
|
|
_LOGGER.debug("protocol device info unavailable: %s", err)
|
|
|
|
async def _dump_label(self, package: str) -> str | None:
|
|
"""One `pm dump` label read: the label, or None for a FAILED fetch.
|
|
|
|
The ONE place the dump is issued, shared by the on-demand path
|
|
(:meth:`_fetch_app_label`) and the discovery sweep, so the t57 UNIT 0
|
|
distinction — "the command ran and found no label" versus "the
|
|
transport delivered nothing" — cannot drift between them. A successful
|
|
fetch is also cached here, so both paths benefit from the 24 h cache and
|
|
neither double-stamps.
|
|
|
|
Returns the label (possibly "") when the dump RAN, and None when it did
|
|
not run or delivered no output at all.
|
|
|
|
The exit marker is a liveness tell, not a status code (t62 nit b): it sits
|
|
after a ``head`` pipeline, and the Android shell that runs this has no
|
|
PIPESTATUS, so ``$?`` is ``head``'s status rather than the dump's. Nothing
|
|
reads its VALUE — ``parse_application_label`` strips the marker line — and
|
|
the only property tested here is that a command which RAN leaves non-blank
|
|
output even when the device prints no label. That non-blankness is what
|
|
separates "ran, no label" from "the transport delivered nothing", which
|
|
this method turns into a cached negative versus a retry.
|
|
"""
|
|
poller = self._poller
|
|
if poller is None or not self.connected:
|
|
_LOGGER.debug(
|
|
"app label for %s not fetched (ADB not connected) — will "
|
|
"retry on the next render",
|
|
package,
|
|
)
|
|
return None
|
|
try:
|
|
import shlex
|
|
|
|
dump = await poller.shell(
|
|
f"pm dump {shlex.quote(package)} 2>/dev/null | head -300; "
|
|
f"echo {APP_LABEL_EXIT_MARKER}$?"
|
|
)
|
|
except Exception as err:
|
|
_LOGGER.debug(
|
|
"app label fetch failed for %s (%s) — will retry on the "
|
|
"next render",
|
|
package,
|
|
err,
|
|
)
|
|
return None
|
|
if dump is None or not dump.strip():
|
|
# the transport never delivered output: a FAILED fetch, not an
|
|
# answer. Caching here is the t55 defect this repairs.
|
|
_LOGGER.debug(
|
|
"app label fetch for %s returned no output — will retry on "
|
|
"the next render",
|
|
package,
|
|
)
|
|
return None
|
|
label = parse_application_label(dump)
|
|
# F1: label-less pm dumps negatively cache too (empty label) so a
|
|
# later cycle performs zero additional pm dumps for the package.
|
|
self.cache_app_label(package, label)
|
|
return label
|
|
|
|
async def _fetch_app_label(self, package: str) -> None:
|
|
"""One on-demand `pm dump` label fetch; stamps ONLY a fetch that RAN.
|
|
|
|
The on-demand half of the label path (one package per watchdog cycle,
|
|
driven by an entity render). The dump itself — including the t57 UNIT 0
|
|
distinction between a FAILED fetch and a genuinely label-less package —
|
|
lives in :meth:`_dump_label`, so the sweep and this path cannot drift.
|
|
"""
|
|
self._app_label_inflight.add(package)
|
|
try:
|
|
await self._dump_label(package)
|
|
finally:
|
|
self._app_label_inflight.discard(package)
|
|
|
|
def _on_logcat_app_start(self, package: str) -> None:
|
|
"""Stamp FIELD_FOREGROUND instantly from a parsed START (P5-3).
|
|
|
|
The next watchdog cycle confirms via poll_foreground/priority —
|
|
whatever dumpsys says wins, so logcat↔dumpsys disagreement never
|
|
becomes state churn. Duplicate STARTs for the same package inside
|
|
LOGCAT_SUPPRESS_WINDOW are not re-stamped.
|
|
"""
|
|
now = self._clock()
|
|
if (
|
|
package == self._logcat_last_pkg
|
|
and now - self._logcat_last_stamp < LOGCAT_SUPPRESS_WINDOW
|
|
):
|
|
return
|
|
self._logcat_last_pkg = package
|
|
self._logcat_last_stamp = now
|
|
self.store.set(FIELD_FOREGROUND, package)
|
|
self._notify()
|
|
|
|
def _open_logcat_stream(self):
|
|
"""Open the dedicated streaming logcat line source (P5-3).
|
|
|
|
Returns an async generator of decoded logcat lines on ITS OWN
|
|
connection — a fresh AdbDeviceTcpAsync sharing only the key signer
|
|
material with the poller, never the poller's socket.
|
|
"""
|
|
poller = self._poller
|
|
if poller is None:
|
|
raise RuntimeError("poller gone — logcat stream cannot start")
|
|
|
|
import asyncio
|
|
|
|
from adb_shell.adb_device_async import AdbDeviceTcpAsync
|
|
from adb_shell.auth.sign_pythonrsa import PythonRSASigner
|
|
|
|
config = poller.config # public accessor (t27 F-NEW3)
|
|
|
|
async def _lines():
|
|
loop = asyncio.get_running_loop()
|
|
signer = await loop.run_in_executor(
|
|
None, PythonRSASigner.FromRSAKeyPath, config["key_path"]
|
|
)
|
|
device = AdbDeviceTcpAsync(config["host"], config["port"])
|
|
connected = await device.connect(rsa_keys=[signer], auth_timeout_s=10.0)
|
|
if not connected:
|
|
raise RuntimeError("dedicated logcat connection failed")
|
|
# F-NEW1 (HIGH): quiet logcat is NORMAL on a bored device — the
|
|
# TCP transport timeout must NOT be the idle detector or every
|
|
# ≥10 s gap becomes a TcpTimeoutException streak that ends in
|
|
# the session auto-disable in the field. transport_timeout_s=None
|
|
# makes bulk_read block indefinitely; real loss (dead socket) is
|
|
# the ladder's job (reconnect with backoff), and async_stop is
|
|
# the cleanup path.
|
|
async for line in device.streaming_shell(
|
|
"logcat -v brief ActivityTaskManager:*",
|
|
transport_timeout_s=None,
|
|
):
|
|
yield line
|
|
|
|
return _lines()
|
|
|
|
def _note_logcat_failure(self, failure_time: float) -> None:
|
|
"""Record a stream failure; auto-disable after 5 within 10 minutes."""
|
|
self._logcat_failures = [
|
|
stamp
|
|
for stamp in self._logcat_failures
|
|
if failure_time - stamp <= 600.0
|
|
]
|
|
self._logcat_failures.append(failure_time)
|
|
if len(self._logcat_failures) >= 5:
|
|
self._logcat_disabled = True
|
|
_LOGGER.warning(
|
|
"Live events (logcat) stream failed 5 times in 10 minutes — "
|
|
"disabled for this session; the watchdog heartbeat remains "
|
|
"the state floor"
|
|
)
|
|
self._logcat_failures.clear()
|
|
|
|
async def _logcat_pass(self) -> None:
|
|
"""One full consumption of the logcat line-stream (P5-3).
|
|
|
|
Yields lines are buffered 3-deep (t27 F-NEW2) before parsing so a
|
|
START record whose cmp= continuation spans stream lines is not
|
|
missed. Returns when the stream ends (the dedicated connection
|
|
dropped); errors propagate to the caller's ladder.
|
|
"""
|
|
buffer: list[str] = []
|
|
async for line in self._open_logcat_stream():
|
|
buffer.append(str(line))
|
|
if len(buffer) < 3: # small wrap window (t27 F-NEW2)
|
|
continue
|
|
package = foreground_from_logcat(buffer)
|
|
if package:
|
|
self._on_logcat_app_start(package)
|
|
buffer = buffer[-2:] # keep the tail for a pending continuation
|
|
# flush the tail buffer once the stream ends
|
|
if buffer:
|
|
package = foreground_from_logcat(buffer)
|
|
if package:
|
|
self._on_logcat_app_start(package)
|
|
raise RuntimeError("stream ended (device closed)")
|
|
|
|
async def _run_logcat_stream(self) -> None:
|
|
"""Consume the logcat line-stream; backoff-reconnect on failure.
|
|
|
|
The stream owns its failure ladder (separate from the ADB socket
|
|
watchdog, reusing the same jitter): Stream failures reconnect with
|
|
5 s doubling to the 300 s cap ±jitter; five failures in ten minutes
|
|
auto-disable the option for the session (heartbeat floor). The
|
|
ladder sleep is injectable (_logcat_ladder_sleep) so tests stay
|
|
deterministic.
|
|
"""
|
|
delay = ADB_RECONNECT_BACKOFF_START
|
|
while not self._stopping and not self._logcat_disabled:
|
|
try:
|
|
await self._logcat_pass()
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as err:
|
|
_LOGGER.debug("logcat stream failure: %s", err)
|
|
self._note_logcat_failure(self._clock())
|
|
if self._stopping or self._logcat_disabled:
|
|
break
|
|
await self._logcat_ladder_sleep(self._jitter(delay))
|
|
delay = min(delay * 2, ADB_RECONNECT_BACKOFF_CAP)
|
|
|
|
def _ensure_logcat_stream(self) -> None:
|
|
"""Start the consumer when the option is on, connected, enabled."""
|
|
if self._logcat_disabled or not self._entry_options_logcat():
|
|
return
|
|
if self._poller is None or not self.connected:
|
|
return
|
|
if self._logcat_task is None or self._logcat_task.done():
|
|
self._logcat_task = self._create_task(
|
|
self._run_logcat_stream(), "shield_logcat_stream"
|
|
)
|
|
|
|
def _entry_options_logcat(self) -> bool:
|
|
"""CONF_LOGCAT_EVENTS from the entry's options (default off)."""
|
|
entry = getattr(self, "entry", None)
|
|
if entry is None:
|
|
return False
|
|
return bool((getattr(entry, "options", {}) or {}).get(CONF_LOGCAT_EVENTS, False))
|
|
|
|
def _stop_logcat_stream(self) -> None:
|
|
"""Cancel the consumer at unload/stop (the stream closes in its
|
|
own finally — the dedicated device never outlives the manager)."""
|
|
if self._logcat_task is not None:
|
|
self._logcat_task.cancel()
|
|
self._logcat_task = None
|
|
|
|
# ── t57: installed-app discovery ───────────────────────────────────────
|
|
#
|
|
# MARK — documented, not observed (t62 nit c): the output SHAPES the two
|
|
# listing commands rely on (`pm list packages -3` → `package:<id>` lines,
|
|
# `cmd package resolve-activity --brief -c …HOME` → `package/activity`) are
|
|
# Android's documentation, not a captured run: no real Shield was available
|
|
# when discovery was written (t57). Every fixture in
|
|
# tests/test_installed_apps.py is built from those shapes and the parsers in
|
|
# ``ha_adb_common.installed_apps`` are deliberately tolerant of whitespace
|
|
# and ordering. Same mark as the state=6 buffering inference in
|
|
# ``media_player.py``.
|
|
|
|
#: t62 F1: how long after an ATTEMPTED sweep failed before another one is
|
|
#: issued. A retry costs two listing shells, so it waits out one state poll
|
|
#: (30 s) rather than hammering a silent socket; ``force=True`` bypasses it.
|
|
#: An unavailable transport (no poller, ADB down) costs nothing and is
|
|
#: retried on the very next update.
|
|
APP_SWEEP_RETRY_SECONDS: float = 30.0
|
|
|
|
def installed_apps(self) -> list[dict]:
|
|
"""The discovered apps, ``[{package, name, icon}]``, cached (t57).
|
|
|
|
Returns the cached table whenever it is fresh, and an EMPTY list when
|
|
there is no fresh answer: ADB disabled, not connected, the first sweep
|
|
still running, or a sweep that could not run (t62 F1). Empty is
|
|
therefore "nothing to show", never by itself "this device has no apps"
|
|
— ask :meth:`installed_apps_status` which of the two it is. A caller
|
|
that wants the table filled calls
|
|
:meth:`async_refresh_installed_apps`; this accessor never blocks and
|
|
never issues a shell itself, so a state poll stays free of discovery
|
|
work.
|
|
"""
|
|
if self._installed_apps is None:
|
|
return []
|
|
rows, stamp = self._installed_apps
|
|
if self._clock() - stamp > APP_LIST_TTL:
|
|
return [] # stale — a refresh repopulates it, the poll does not
|
|
return [dict(row) for row in rows]
|
|
|
|
@property
|
|
def installed_apps_stale(self) -> bool:
|
|
"""True when no fresh table is available (including "never swept")."""
|
|
if self._installed_apps is None:
|
|
return True
|
|
_rows, stamp = self._installed_apps
|
|
return self._clock() - stamp > APP_LIST_TTL
|
|
|
|
@property
|
|
def installed_apps_status(self) -> str:
|
|
"""Why :meth:`installed_apps` may be empty, in one word (t62 F1).
|
|
|
|
* ``"ok"`` — a sweep RAN and its answer is fresh. An empty table with
|
|
this status is a real answer (the device reported no launchable
|
|
apps), which is the case the old stamped-empty cache could not say.
|
|
* ``"unknown"`` — the last sweep could not deliver an answer (no
|
|
poller, ADB down, a shell that raised or answered nothing). Nothing
|
|
is known about the device, and the next update retries.
|
|
* ``"pending"`` — no sweep has completed yet, or the last answer
|
|
expired and a refresh is due.
|
|
"""
|
|
if not self.installed_apps_stale:
|
|
return "ok"
|
|
return "unknown" if self._apps_unavailable else "pending"
|
|
|
|
async def async_refresh_installed_apps(self, force: bool = False) -> list[dict]:
|
|
"""(Re)build the installed-app table over ADB, bounded and shared (t57).
|
|
|
|
One sweep at a time: concurrent callers await the same task rather than
|
|
starting a second burst, and a fresh table is returned as-is unless
|
|
``force``. The sweep is deliberately NOT part of the watchdog: it is a
|
|
handful of shells on demand, not a per-cycle cost.
|
|
|
|
Failure is not an answer (t62 F1). No poller, a disconnected transport,
|
|
a shell that raises, or a sweep that delivered no output records
|
|
``unknown`` and stores NO table, so the next non-forced call — an
|
|
entity poll or the standard ``update_entity`` action — retries instead
|
|
of serving a stamped empty answer for the whole APP_LIST_TTL. A sweep
|
|
that reached the device and failed is retried at most once per
|
|
``APP_SWEEP_RETRY_SECONDS`` so a silent socket is not hammered; a sweep
|
|
that RAN stores its answer, including a genuinely empty list.
|
|
"""
|
|
if not force and not self.installed_apps_stale:
|
|
return self.installed_apps()
|
|
if (
|
|
not force
|
|
and self._apps_failed_at is not None
|
|
and self._clock() - self._apps_failed_at < self.APP_SWEEP_RETRY_SECONDS
|
|
):
|
|
# The previous attempt really issued shells and failed: space the
|
|
# retries by one poll interval instead of re-issuing them every
|
|
# 30 s on a socket that is not answering (t62 F1).
|
|
return self.installed_apps()
|
|
if self._apps_sweep is not None and not self._apps_sweep.done():
|
|
await asyncio.shield(self._apps_sweep)
|
|
return self.installed_apps()
|
|
task = self._create_task(self._sweep_installed_apps(), "apps_sweep")
|
|
self._apps_sweep = task
|
|
try:
|
|
rows = await task
|
|
finally:
|
|
if self._apps_sweep is task:
|
|
self._apps_sweep = None
|
|
return [dict(row) for row in rows]
|
|
|
|
async def _sweep_installed_apps(self) -> list[dict]:
|
|
"""The sweep itself: list, label, icon — then stamp the cache (t57).
|
|
|
|
Only a sweep that RAN stamps the cache (t62 F1): every failure path
|
|
records ``unknown`` and leaves ``_installed_apps`` alone, so the table
|
|
the entity serves is the last one that was real — never a fabricated
|
|
empty answer that also blocks the retry for APP_LIST_TTL.
|
|
"""
|
|
poller = self._poller
|
|
if poller is None or not self.connected:
|
|
_LOGGER.debug(
|
|
"installed-app discovery skipped (ADB unavailable) — the "
|
|
"options page and the app list stay on what is already known"
|
|
)
|
|
self._note_apps_unavailable(attempted=False)
|
|
return []
|
|
try:
|
|
user_output = await poller.shell("pm list packages -3")
|
|
launcher_output = await poller.shell(
|
|
"cmd package resolve-activity --brief -c "
|
|
"android.intent.category.HOME"
|
|
)
|
|
except Exception as err: # pragma: no cover - shell swallows its own
|
|
_LOGGER.debug("installed-app discovery failed: %s", err)
|
|
self._note_apps_unavailable(attempted=True)
|
|
return []
|
|
if user_output is None and launcher_output is None:
|
|
# the transport delivered nothing at all: no list is NOT an answer
|
|
_LOGGER.debug(
|
|
"installed-app discovery got no output — will retry on the "
|
|
"next update"
|
|
)
|
|
self._note_apps_unavailable(attempted=True)
|
|
return []
|
|
|
|
packages = [p for p in installed_packages(user_output, launcher_output)
|
|
if is_launchable(p)]
|
|
labels = await self._sweep_labels(packages)
|
|
rows = [
|
|
{
|
|
"package": package,
|
|
"name": labels.get(package) or package,
|
|
"icon": icon_for_package(package, APP_ICON_FALLBACKS, APP_ICON_GENERIC),
|
|
}
|
|
for package in packages
|
|
]
|
|
self._installed_apps = (rows, self._clock())
|
|
self._apps_unavailable = False
|
|
self._apps_failed_at = None
|
|
_LOGGER.debug("installed-app discovery found %d apps", len(rows))
|
|
return rows
|
|
|
|
def _note_apps_unavailable(self, *, attempted: bool) -> None:
|
|
"""Record a sweep that could not deliver an answer (t62 F1).
|
|
|
|
``attempted`` says whether shells were actually issued: only then does
|
|
the retry window apply, because only then does a retry cost anything.
|
|
"""
|
|
self._apps_unavailable = True
|
|
self._apps_failed_at = self._clock() if attempted else None
|
|
|
|
async def _sweep_labels(self, packages: list[str]) -> dict[str, str]:
|
|
"""Labels for a whole sweep, in a bounded burst (t57).
|
|
|
|
Deliberately NOT through ``_app_label_requests``: that queue is drained
|
|
ONE package per watchdog cycle, which is right for the live entity path
|
|
(a render must never fan out into shells) but would take a couple of
|
|
minutes for a Shield with ~40 apps. Here the fetches run
|
|
``APP_SWEEP_CONCURRENCY`` at a time with a short spacing, so a typical
|
|
Shield finishes in seconds while the socket still sees a trickle rather
|
|
than a flood.
|
|
|
|
Cached labels are reused, so a second sweep costs almost nothing; a
|
|
failed fetch is NOT cached (t57 UNIT 0) and the app falls back to its
|
|
package id for this table while the next sweep retries it.
|
|
|
|
The check goes through :meth:`_known_label` rather than
|
|
:meth:`app_label`, so a package the device has definitively answered
|
|
"no label" for is not re-dumped on every sweep.
|
|
"""
|
|
cached: dict[str, str] = {}
|
|
missing: list[str] = []
|
|
for package in packages:
|
|
known, label = self._known_label(package)
|
|
if not known:
|
|
missing.append(package)
|
|
elif label:
|
|
cached[package] = label
|
|
# a known NEGATIVE needs no entry: the row falls back to its
|
|
# package id, and the cache already holds the definitive answer
|
|
if not missing:
|
|
return cached
|
|
|
|
semaphore = asyncio.Semaphore(APP_SWEEP_CONCURRENCY)
|
|
|
|
async def _one(package: str) -> tuple[str, str | None]:
|
|
async with semaphore:
|
|
# a small spacing keeps the burst polite on a single socket
|
|
await asyncio.sleep(APP_SWEEP_SPACING)
|
|
return package, await self._dump_label(package)
|
|
|
|
results = await asyncio.gather(
|
|
*(_one(package) for package in missing), return_exceptions=True
|
|
)
|
|
for result in results:
|
|
if isinstance(result, BaseException):
|
|
continue # one bad package never fails the sweep
|
|
package, label = result
|
|
if label:
|
|
cached[package] = label
|
|
else:
|
|
# None here means the fetch FAILED (nothing cached); the empty
|
|
# string means the device answered with no label. Both leave
|
|
# the row on its package id, and both keep the app in the list.
|
|
cached[package] = ""
|
|
return cached
|
|
|
|
|
|
def _drain_app_label_requests(self) -> None:
|
|
"""Start at most one label fetch per watchdog cycle."""
|
|
if not self._app_label_requests or self._poller is None:
|
|
return
|
|
package = min(self._app_label_requests)
|
|
self._app_label_requests.discard(package)
|
|
self._tasks.append(
|
|
self._create_task(self._fetch_app_label(package), f"app_label_{package}")
|
|
)
|
|
|
|
def last_known_map(self) -> dict[str, MediaSessionInfo]:
|
|
"""TTL-filtered snapshot of the per-package last-known cache."""
|
|
result: dict[str, MediaSessionInfo] = {}
|
|
for package in tuple(self._last_known):
|
|
info = self.last_known_session(package)
|
|
if info is not None:
|
|
result[package] = info
|
|
return result
|
|
|
|
def last_known_state(
|
|
self, package: str
|
|
) -> tuple[MediaSessionInfo | None, float | None]:
|
|
"""(last known session, freshness stamp) for a package — the stamp
|
|
lets callers age the served state in their UI."""
|
|
cached = self._last_known.get(package)
|
|
if cached is None:
|
|
return None, None
|
|
info, stamp = cached
|
|
if self._clock() - stamp > STATE_STORE_TTL:
|
|
self._last_known.pop(package, None)
|
|
return None, None
|
|
return info, stamp
|
|
|
|
# ── 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)
|
|
|
|
# -- Protocol command delegation (Phase 2) --
|
|
|
|
async def send_protocol_key(self, key: str, direction: str = "SHORT") -> bool:
|
|
"""Send a remote-protocol key push (fast, non-blocking).
|
|
|
|
Returns False when the transport is down or the send raises — entities
|
|
then fall back to the ADB keyevent path.
|
|
"""
|
|
if self._protocol is None:
|
|
return False
|
|
try:
|
|
self._protocol.send_key_command(key, direction)
|
|
return True
|
|
except Exception as err:
|
|
_LOGGER.debug("Protocol key %s failed: %s", key, err)
|
|
return False
|
|
|
|
async def launch_app(self, app_link_or_app_id: str) -> bool:
|
|
"""Launch an app or URL over the protocol transport.
|
|
|
|
``send_launch_app_command`` auto-prefixes ``market://launch?id=`` for
|
|
bare package ids (C-LAUNCH); deep links pass through unchanged.
|
|
"""
|
|
if self._protocol is None:
|
|
return False
|
|
try:
|
|
self._protocol.send_launch_app_command(app_link_or_app_id)
|
|
return True
|
|
except Exception as err:
|
|
_LOGGER.debug("Protocol launch %s failed: %s", app_link_or_app_id, err)
|
|
return False
|
|
|
|
async def send_digit_key(self, digit: str) -> bool:
|
|
"""Channel digit via the ADB keyevent path (fallback for §7)."""
|
|
if not self.available:
|
|
return False
|
|
return await self._poller.send_digit_keyevent(digit)
|
|
|
|
async def send_enter_key(self) -> bool:
|
|
"""ENTER via the ADB keyevent path (KEYCODE_ENTER = 66, §7 fallback)."""
|
|
if not self.available:
|
|
return False
|
|
return await self._poller.send_enter_keyevent()
|
|
|
|
# ── 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"))
|
|
await self._start_protocol()
|
|
|
|
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
|
|
self._stop_logcat_stream()
|
|
tasks, self._tasks = self._tasks, []
|
|
for task in tasks:
|
|
task.cancel()
|
|
self._listeners.clear()
|
|
self._last_known.clear()
|
|
self._app_labels.clear()
|
|
self._app_label_requests.clear()
|
|
self._app_label_inflight.clear()
|
|
if self._poller is not None:
|
|
with contextlib.suppress(Exception):
|
|
await self._poller.disconnect()
|
|
self._poller = None
|
|
if self._protocol is not None:
|
|
with contextlib.suppress(Exception):
|
|
self._protocol.disconnect()
|
|
self._protocol = 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)
|
|
# Only ADB-derived fields expire here: protocol pushes are
|
|
# device-reported truth and must survive an ADB outage (t11 F2).
|
|
self.store.clear(*ADB_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._cache_last_known_sessions()
|
|
# Zero-cost hint from the same dump; system processes filtered
|
|
# (mediashell/systemui/tv settings/diagtools report "audio").
|
|
playback = [
|
|
pkg
|
|
for pkg in parse_audio_playback(poller.last_dump or "")
|
|
if pkg not in SYSTEM_MEDIA_PACKAGES
|
|
]
|
|
self.store.set(FIELD_AUDIO_PLAYBACK, playback)
|
|
self.store.set(FIELD_RUNNING_APPS, await poller.poll_running_apps())
|
|
# Foreground fallback (P3-2): ADB parse only when the protocol
|
|
# push is silent — store priority is protocol > ADB, so this
|
|
# never overwrites a live push.
|
|
if self.store.get(FIELD_CURRENT_APP) is None:
|
|
foreground = await poller.poll_foreground()
|
|
if foreground:
|
|
self.store.set(FIELD_FOREGROUND, foreground)
|
|
# G10: capture the protocol device identity for the registry card
|
|
if self._protocol is not None:
|
|
self._tasks.append(
|
|
self._create_task(
|
|
self.async_capture_protocol_device_data(),
|
|
"protocol_device_capture",
|
|
)
|
|
)
|
|
# P5-1: manifest labels resolve on demand, one package per cycle
|
|
self._drain_app_label_requests()
|
|
# P5-3: ensure the (optional, default-off) logcat fast-path runs
|
|
self._ensure_logcat_stream()
|
|
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()
|