fix(media_player): the adb_command participant registry key is computed once at registration and stored on the entity — add/remove always match (F4)
This commit is contained in:
1 parent
c44c2c2170
commit
b37a14b920
5 files changed
+147
-60
No files matched your search
@@ -23,6 +23,7 @@ hass (plain ``asyncio.create_task``) for the unit tests.
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import dataclasses
|
||||
import logging
|
||||
import random
|
||||
import time
|
||||
@@ -135,6 +136,10 @@ class StateStore:
|
||||
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.
|
||||
@@ -377,6 +382,12 @@ class ShieldConnectionManager:
|
||||
"""
|
||||
return self._protocol is not 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)."""
|
||||
@@ -446,6 +457,19 @@ class ShieldConnectionManager:
|
||||
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
|
||||
|
||||
def last_known_map(self) -> dict[str, MediaSessionInfo]:
|
||||
"""TTL-filtered snapshot of the per-package last-known cache."""
|
||||
result: dict[str, MediaSessionInfo] = {}
|
||||
|
||||
Reference in new issue
Block a user