Initial release v0.1.0

Home Assistant custom integration for NVIDIA Shield TV that composes
the Android TV Remote, Jellyfin, smart plug, and receiver entities
into a single unified media player with ADB-based now-playing from
any app. Part of the Media Zone ecosystem.

Entities:
- media_player: power (with smart plug fallback), transport, input,
  volume routed to receiver, rich stream info overlay when Jellyfin
  is active
- sensor: DAG attributes and stream info
- button: Pair ADB

Config:
- Two-step flow: basic setup + optional ADB pairing on the Shield
- Configurable ADB port (default 5555)
- Pick an existing Jellyfin media_player entity — stream info is read
  from the HA Jellyfin integration's own DataUpdateCoordinator, so
  no server URL / API key / extra config needed here
- Privacy mode (adb_store_app_info=False) suppresses non-Jellyfin
  media details, showing only the app name

Ecosystem:
- Exposes media_dag_* attributes for ha-media-zone orchestration
- Exposes media_dag_running_apps and async_bring_to_foreground for
  the Media Zone card's app switcher

Shared ADB code lives in ltomes/ha-adb-common and is vendored at
custom_components/nvidia_shield/ha_adb_common/ for HACS installs.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

custom_components/nvidia_shield/ha_adb_common/__init__.py

@@ -0,0 +1,32 @@
+"""Shared ADB utilities for Home Assistant custom integrations."""
+
+from .adb_client import AdbClientBase
+from .const import (
+    ATTR_MEDIA_DAG_JELLYFIN_ENTITY,
+    ATTR_MEDIA_DAG_NOW_PLAYING,
+    ATTR_MEDIA_DAG_RECEIVER_ENTITY,
+    ATTR_MEDIA_DAG_RECEIVER_INPUT,
+    ATTR_MEDIA_DAG_ROLE,
+    ATTR_MEDIA_DAG_RUNNING_APPS,
+)
+from .intent import PlaybackIntent
+from .keys import ensure_adb_key
+from .media_session import MediaSessionInfo, parse_all_media_sessions, parse_media_session
+from .running_apps import running_apps_from_sessions, running_apps_from_tasks
+
+__all__ = [
+    "ATTR_MEDIA_DAG_JELLYFIN_ENTITY",
+    "ATTR_MEDIA_DAG_NOW_PLAYING",
+    "ATTR_MEDIA_DAG_RECEIVER_ENTITY",
+    "ATTR_MEDIA_DAG_RECEIVER_INPUT",
+    "ATTR_MEDIA_DAG_ROLE",
+    "ATTR_MEDIA_DAG_RUNNING_APPS",
+    "AdbClientBase",
+    "MediaSessionInfo",
+    "PlaybackIntent",
+    "ensure_adb_key",
+    "parse_all_media_sessions",
+    "parse_media_session",
+    "running_apps_from_sessions",
+    "running_apps_from_tasks",
+]

custom_components/nvidia_shield/ha_adb_common/adb_client.py

@@ -0,0 +1,213 @@
+"""Base ADB client with shared connection management and commands.
+
+Subclasses add device-specific polling (media sessions, wake state, etc.).
+"""
+
+import logging
+
+from .media_session import MediaSessionInfo, parse_all_media_sessions, parse_media_session
+
+_LOGGER = logging.getLogger(__name__)
+
+# Standard Android media key codes
+MEDIA_KEYCODES = {
+    "play": 126,
+    "pause": 127,
+    "play-pause": 85,
+    "stop": 86,
+    "next": 87,
+    "previous": 88,
+}
+
+DEVICE_KEYCODES = {
+    "power": 26,
+    "sleep": 223,
+    "wakeup": 224,
+    "volume_up": 24,
+    "volume_down": 25,
+    "mute": 164,
+}
+
+ALL_KEYCODES = {**MEDIA_KEYCODES, **DEVICE_KEYCODES}
+
+
+class AdbClientBase:
+    """Base ADB client with connection management and common commands."""
+
+    def __init__(self, host: str, port: int, key_path: str):
+        self._host = host
+        self._port = port
+        self._key_path = key_path
+        self._device = None
+        self._connected = False
+
+        # Media session state
+        self._last_info: MediaSessionInfo | None = None
+        self._all_sessions: list[MediaSessionInfo] = []
+
+    @property
+    def connected(self) -> bool:
+        return self._connected
+
+    @property
+    def last_info(self) -> MediaSessionInfo | None:
+        return self._last_info
+
+    @property
+    def all_sessions(self) -> list[MediaSessionInfo]:
+        return self._all_sessions
+
+    def clear_info(self) -> None:
+        """Clear the cached media session info."""
+        self._last_info = None
+        self._all_sessions = []
+
+    async def connect(self) -> bool:
+        """Connect to the device via ADB."""
+        try:
+            import asyncio
+
+            from adb_shell.adb_device_async import AdbDeviceTcpAsync
+            from adb_shell.auth.sign_pythonrsa import PythonRSASigner
+
+            # FromRSAKeyPath does blocking file I/O — run it in the default
+            # executor so we don't stall Home Assistant's event loop.
+            loop = asyncio.get_running_loop()
+            signer = await loop.run_in_executor(
+                None, PythonRSASigner.FromRSAKeyPath, self._key_path
+            )
+            self._device = AdbDeviceTcpAsync(self._host, self._port)
+            await self._device.connect(rsa_keys=[signer], auth_timeout_s=10)
+            self._connected = True
+            _LOGGER.info("ADB connected to %s:%d", self._host, self._port)
+            return True
+        except Exception as err:
+            _LOGGER.debug("ADB connection failed: %s", err)
+            self._connected = False
+            self._device = None
+            return False
+
+    async def disconnect(self) -> None:
+        """Disconnect from the device."""
+        if self._device:
+            try:
+                await self._device.close()
+            except Exception:
+                pass
+        self._device = None
+        self._connected = False
+
+    async def shell(self, cmd: str) -> str | None:
+        """Run a shell command, reconnecting if needed."""
+        if not self._connected or not self._device:
+            if not await self.connect():
+                return None
+        try:
+            return await self._device.shell(cmd)
+        except Exception as err:
+            _LOGGER.debug("ADB shell failed: %s", err)
+            self._connected = False
+            self._device = None
+            return None
+
+    # -- Common commands --
+
+    async def send_keyevent(self, command: str) -> bool:
+        """Send a key event via ADB input keyevent."""
+        keycode = ALL_KEYCODES.get(command)
+        if keycode is None:
+            _LOGGER.warning("Unknown keyevent command: %s", command)
+            return False
+        result = await self.shell(f"input keyevent {keycode}")
+        if result is not None:
+            _LOGGER.debug("ADB keyevent %d (%s)", keycode, command)
+        return result is not None
+
+    async def send_media_command(self, command: str) -> bool:
+        """Send a media command via ADB input keyevent."""
+        keycode = MEDIA_KEYCODES.get(command)
+        if keycode is None:
+            _LOGGER.warning("Unknown media command: %s", command)
+            return False
+        result = await self.shell(f"input keyevent {keycode}")
+        if result is not None:
+            _LOGGER.debug("ADB keyevent %d (%s)", keycode, command)
+            return True
+        return False
+
+    async def bring_to_foreground(self, package: str) -> bool:
+        """Bring an app to the foreground using monkey command."""
+        result = await self.shell(
+            f"monkey -p {package} -c android.intent.category.LAUNCHER 1 2>/dev/null"
+        )
+        if result is not None:
+            _LOGGER.debug("Brought %s to foreground via monkey", package)
+        return result is not None
+
+    async def is_awake(self) -> bool | None:
+        """Check if the device screen is on / awake via ADB."""
+        result = await self.shell("dumpsys power 2>/dev/null | grep mWakefulness")
+        if result:
+            return "Awake" in result
+        return None
+
+    # -- Media session polling --
+
+    async def poll_media_sessions(
+        self, foreground_package: str | None = None, head_lines: int = 200
+    ) -> MediaSessionInfo | None:
+        """Query media sessions. Stores all sessions and returns the best match."""
+        dump = await self.shell(
+            f"dumpsys media_session 2>/dev/null | head -{head_lines}"
+        )
+        if dump is None:
+            return None
+        self._all_sessions = parse_all_media_sessions(dump)
+        self._last_info = parse_media_session(dump, foreground_package)
+        return self._last_info
+
+    # -- Running apps (recent visible tasks) --
+
+    async def poll_running_apps(self) -> list[dict]:
+        """Get running apps from Android's recent tasks.
+
+        Returns a list of dicts with 'package' and 'foreground' keys,
+        ordered by recency (most recent first). Only includes visible
+        user apps, filtering out system UI, home, and invisible tasks.
+        """
+        result = await self.shell(
+            "dumpsys activity recents 2>/dev/null "
+            "| grep -E 'Recent #|realActivity|visible='"
+        )
+        if not result:
+            return []
+
+        import re
+        apps = []
+        seen = set()
+
+        # System packages to exclude
+        _EXCLUDED = {
+            "com.android.systemui",
+            "com.android.tv.settings",
+            "com.nvidia.diagtools",
+            "android",
+        }
+
+        for line in result.split("\n"):
+            stripped = line.strip()
+
+            # Match recent task lines with visibility and package info
+            if "Recent #" in stripped and "visible=true" in stripped:
+                # Extract package from A=uid:package pattern
+                match = re.search(r"A=\d+:(\S+)", stripped)
+                if not match:
+                    # Try I=package/activity pattern
+                    match = re.search(r"I=(\S+)/", stripped)
+                if match:
+                    pkg = match.group(1)
+                    if pkg not in seen and pkg not in _EXCLUDED and not pkg.startswith("."):
+                        seen.add(pkg)
+                        apps.append({"package": pkg})
+
+        return apps

custom_components/nvidia_shield/ha_adb_common/const.py

@@ -0,0 +1,25 @@
+"""Shared Media DAG attribute keys.
+
+These constants define the contract between source integrations (nvidia_shield,
+hisense_tv, etc.) and the media_zone aggregator. All source integrations MUST
+use these exact attribute keys so media_zone can read them uniformly.
+"""
+
+# Role of this entity in the media DAG
+ATTR_MEDIA_DAG_ROLE = "media_dag_role"
+
+# Receiver this source is wired to (entity_id)
+ATTR_MEDIA_DAG_RECEIVER_ENTITY = "media_dag_receiver_entity"
+
+# Which receiver input this source occupies
+ATTR_MEDIA_DAG_RECEIVER_INPUT = "media_dag_receiver_input"
+
+# Jellyfin media_player entity for rich metadata
+ATTR_MEDIA_DAG_JELLYFIN_ENTITY = "media_dag_jellyfin_entity"
+
+# Now-playing dict: {state, title, media_type, series?, season?, episode?,
+#                    duration?, position?, thumbnail?, artist?, album?, media_info?, source?}
+ATTR_MEDIA_DAG_NOW_PLAYING = "media_dag_now_playing"
+
+# List of running app dicts: [{package, state, title?, artist?}, ...]
+ATTR_MEDIA_DAG_RUNNING_APPS = "media_dag_running_apps"

custom_components/nvidia_shield/ha_adb_common/intent.py

@@ -0,0 +1,48 @@
+"""Playback intent tracking for optimistic UI updates.
+
+Bridges the gap between instant commands and slow state polling.
+"""
+
+import time
+from dataclasses import dataclass
+
+
+@dataclass
+class PlaybackIntent:
+    """A pending user intent for playback state.
+
+    Attributes:
+        target_state: The MediaPlayerState we expect after the command.
+        command: The service/command name (for retries and logging).
+        created_at: Monotonic timestamp when the intent was created.
+        prior_state: The source state before the intent (for detecting external overrides).
+        retry_count: Number of retries attempted so far.
+    """
+
+    target_state: str  # MediaPlayerState value (e.g. "playing", "paused")
+    command: str
+    created_at: float
+    prior_state: str | None = None
+    retry_count: int = 0
+
+    @classmethod
+    def create(cls, target_state: str, command: str, prior_state: str | None = None) -> "PlaybackIntent":
+        return cls(
+            target_state=target_state,
+            command=command,
+            created_at=time.monotonic(),
+            prior_state=prior_state,
+        )
+
+    @property
+    def age_seconds(self) -> float:
+        return round(time.monotonic() - self.created_at, 1)
+
+    def to_debug_dict(self) -> dict:
+        """Serialize for the _intent debug attribute."""
+        return {
+            "target": self.target_state,
+            "command": self.command,
+            "retries": self.retry_count,
+            "age_s": self.age_seconds,
+        }

custom_components/nvidia_shield/ha_adb_common/keys.py

@@ -0,0 +1,56 @@
+"""ADB key management — generation and migration."""
+
+import logging
+import os
+import shutil
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def ensure_adb_key(
+    hass_config_dir: str,
+    domain: str,
+    legacy_paths: list[str] | None = None,
+) -> str:
+    """Ensure an ADB key pair exists. Migrate from legacy paths or generate if missing.
+
+    Args:
+        hass_config_dir: Home Assistant config directory.
+        domain: Integration domain name (e.g. "nvidia_shield", "hisense_tv").
+            Used to namespace the key storage.
+        legacy_paths: Optional list of legacy key paths to check for migration,
+            in priority order. Avoids forcing device re-pairing.
+
+    Returns:
+        Path to the private key file.
+    """
+    key_dir = os.path.join(hass_config_dir, ".storage", domain)
+    os.makedirs(key_dir, exist_ok=True)
+    key_path = os.path.join(key_dir, "adbkey")
+
+    if os.path.exists(key_path):
+        return key_path
+
+    # Try migrating from legacy locations
+    all_legacy = list(legacy_paths or [])
+    # Always check the generic location as a last resort
+    generic = os.path.join(hass_config_dir, ".storage", "adbkey")
+    if generic not in all_legacy:
+        all_legacy.append(generic)
+
+    for legacy_key in all_legacy:
+        if os.path.exists(legacy_key):
+            shutil.copy2(legacy_key, key_path)
+            pub = legacy_key + ".pub"
+            if os.path.exists(pub):
+                shutil.copy2(pub, key_path + ".pub")
+            _LOGGER.info("Migrated ADB key from %s to %s", legacy_key, key_path)
+            return key_path
+
+    # Generate new key
+    from adb_shell.auth.keygen import keygen
+
+    keygen(key_path)
+    _LOGGER.info("Generated new ADB key at %s", key_path)
+
+    return key_path

custom_components/nvidia_shield/ha_adb_common/media_session.py

@@ -0,0 +1,122 @@
+"""Android media session parsing.
+
+Parses `dumpsys media_session` output from ADB to extract now-playing
+information from any app using Android's MediaSession API.
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class MediaSessionInfo:
+    """Parsed media session data from Android."""
+
+    title: str | None = None
+    artist: str | None = None
+    album: str | None = None
+    package: str | None = None
+    playing: bool = False
+    paused: bool = False
+
+
+def parse_all_media_sessions(dump: str) -> list[MediaSessionInfo]:
+    """Parse `dumpsys media_session` output for ALL active media sessions.
+
+    Returns a list of every active session that has metadata or a known
+    playback state, sorted: playing first, then paused, then others.
+    """
+    if not dump:
+        return []
+
+    sessions: list[MediaSessionInfo] = []
+    current_pkg: str | None = None
+    in_active = False
+    current_state = 0  # 0=unknown, 2=paused, 3=playing
+    current_meta: MediaSessionInfo | None = None
+
+    for line in dump.split("\n"):
+        stripped = line.strip()
+
+        if "package=" in stripped:
+            pkg = stripped.split("package=", 1)
+            if len(pkg) > 1:
+                # New session block — save the previous one if valid
+                if in_active and current_meta:
+                    current_meta.package = current_pkg
+                    sessions.append(current_meta)
+                current_pkg = pkg[1].strip()
+                in_active = False
+                current_state = 0
+                current_meta = None
+
+        if "active=true" in stripped:
+            in_active = True
+
+        if in_active and "state=PlaybackState" in stripped:
+            if "state=3" in stripped:
+                current_state = 3
+            elif "state=2" in stripped:
+                current_state = 2
+
+        if in_active and "metadata: size=" in stripped and "description=" in stripped:
+            desc_start = stripped.find("description=")
+            if desc_start < 0:
+                continue
+            desc = stripped[desc_start + 12:]
+            parts = [p.strip() for p in desc.split(", ")]
+
+            info = MediaSessionInfo()
+            info.package = current_pkg
+            info.playing = current_state == 3
+            info.paused = current_state == 2
+
+            if len(parts) >= 1 and parts[0] not in ("null", ""):
+                info.title = parts[0]
+            if len(parts) >= 2 and parts[1] not in ("null", ""):
+                info.artist = parts[1]
+            if len(parts) >= 3 and parts[2] not in ("null", ""):
+                info.album = parts[2]
+
+            if info.title:
+                current_meta = info
+
+    # Don't forget the last session
+    if in_active and current_meta:
+        current_meta.package = current_pkg
+        sessions.append(current_meta)
+
+    # Sort: playing first, then paused, then others
+    def sort_key(s: MediaSessionInfo) -> int:
+        if s.playing:
+            return 0
+        if s.paused:
+            return 1
+        return 2
+
+    sessions.sort(key=sort_key)
+    return sessions
+
+
+def parse_media_session(
+    dump: str, foreground_package: str | None = None
+) -> MediaSessionInfo | None:
+    """Parse `dumpsys media_session` output for the best active session.
+
+    Returns the best match from all active sessions:
+    1. The session whose package matches the foreground app (if provided)
+    2. The session that is currently playing (state=3)
+    3. The session that is paused (state=2)
+    4. Any active session with metadata
+    """
+    sessions = parse_all_media_sessions(dump)
+    if not sessions:
+        return None
+
+    # Prefer the session matching the foreground app
+    if foreground_package:
+        for s in sessions:
+            if s.package == foreground_package:
+                return s
+
+    # Already sorted: playing > paused > other
+    return sessions[0]

custom_components/nvidia_shield/ha_adb_common/running_apps.py

@@ -0,0 +1,72 @@
+"""Build the running apps list from recent tasks + media sessions.
+
+Produces the media_dag_running_apps attribute value that media_zone reads.
+Recent tasks (from dumpsys activity recents) provide the list of running apps.
+Media sessions provide playback state for apps that have active sessions.
+"""
+
+from .media_session import MediaSessionInfo
+
+
+def running_apps_from_tasks(
+    recent_tasks: list[dict],
+    media_sessions: list[MediaSessionInfo] | None = None,
+    foreground_package: str | None = None,
+) -> list[dict] | None:
+    """Build a list of running app dicts from recent tasks + media sessions.
+
+    Args:
+        recent_tasks: List of {"package": str} from ADB recent tasks.
+        media_sessions: Optional media sessions for playback state enrichment.
+        foreground_package: The current foreground app package.
+
+    Returns None if there are fewer than 2 apps (no point showing a switcher
+    with only one option).
+    """
+    if not recent_tasks or len(recent_tasks) < 2:
+        return None
+
+    # Index media sessions by package for quick lookup
+    session_by_pkg: dict[str, MediaSessionInfo] = {}
+    if media_sessions:
+        for s in media_sessions:
+            if s.package:
+                session_by_pkg[s.package] = s
+
+    apps = []
+    for task in recent_tasks:
+        pkg = task["package"]
+        app: dict = {"package": pkg}
+
+        # Mark foreground app
+        if pkg == foreground_package:
+            app["foreground"] = True
+
+        # Enrich with media session data if available
+        session = session_by_pkg.get(pkg)
+        if session:
+            if session.playing:
+                app["state"] = "playing"
+            elif session.paused:
+                app["state"] = "paused"
+            else:
+                app["state"] = "idle"
+            if session.title:
+                app["title"] = session.title
+            if session.artist:
+                app["artist"] = session.artist
+        else:
+            app["state"] = "running"
+
+        apps.append(app)
+
+    return apps if len(apps) >= 2 else None
+
+
+# Keep backward compat for any code still using this
+def running_apps_from_sessions(sessions: list[MediaSessionInfo]) -> list[dict] | None:
+    """Build running apps from media sessions only (legacy)."""
+    if not sessions:
+        return None
+    tasks = [{"package": s.package} for s in sessions if s.package]
+    return running_apps_from_tasks(tasks, sessions)