feat(connection): protocol transport alongside ADB — AndroidTVRemote client, keep_reconnecting ownership, push callbacks into StateStore, InvalidAuth→reauth hook, cert/PIN client factory

custom_components/nvidia_shield/connection.py

@@ -47,7 +47,41 @@ 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)
+# 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"
+
+ALL_FIELDS = (
+    FIELD_AWAKE,
+    FIELD_SESSIONS,
+    FIELD_AUDIO_PLAYBACK,
+    FIELD_RUNNING_APPS,
+    FIELD_IS_ON,
+    FIELD_IS_AVAILABLE,
+    FIELD_CURRENT_APP,
+    FIELD_VOLUME_INFO,
+)
+
+
+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:
@@ -103,12 +137,25 @@ class ShieldConnectionManager:
         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,
     ):
         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
+        # 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_task: asyncio.Task | None = None
+        self._protocol_loop: asyncio.AbstractEventLoop | None = None
         self._listeners: list[Callable[[], None]] = []
         self._tasks: list[asyncio.Task] = []
         self._stopping = False
@@ -130,6 +177,118 @@ class ShieldConnectionManager:
         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.
+        """
+        import androidtvremote2
+
+        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"],
+        )
+
+    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
+        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)
+        self._protocol_task = self._create_task(
+            client.keep_reconnecting(self._on_invalid_auth),
+            "shield_protocol_keep_reconnecting",
+        )
+
+    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
@@ -161,6 +320,27 @@ class ShieldConnectionManager:
         """Visible recent-task apps, ordered by recency."""
         return self.store.get(FIELD_RUNNING_APPS, [])
 
+    @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 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:
@@ -217,6 +397,7 @@ class ShieldConnectionManager:
         """
         self._stopping = False
         self._tasks.append(self._create_task(self._watchdog(), "shield_adb_watchdog"))
+        self._start_protocol()
 
     async def async_stop(self) -> None:
         """Cancel all manager tasks and close the socket.
@@ -227,11 +408,18 @@ class ShieldConnectionManager:
         tasks, self._tasks = self._tasks, []
         for task in tasks:
             task.cancel()
+        if self._protocol_task is not None:
+            self._protocol_task.cancel()
+            self._protocol_task = None
         self._listeners.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: