"""Media player entity for Nvidia Shield. Phase 2 architecture (docs/MVP_PLAN.md §2): the entry's ShieldConnectionManager (connection.py) owns both transports — ADB (adb_shell, state source) and the absorbed remote protocol (androidtvremote2, local push) — plus their StateStore; this entity is a thin callback-view over it. State architecture: - Power ladder: protocol is_on push → ADB wake (exact mWakefulness) → plug off - Playback state: Jellyfin (polling, slow 10-30s) or ADB media session - Metadata: Jellyfin attributes - Commands: a configured per-command script (t44) → protocol key push → ADB keyevent. The external androidtv_remote entity tier that used to back this is gone — t41 removed it with the entity — so a command has exactly one owner. Intent tracking bridges the gap between instant commands and slow Jellyfin polling: 1. User presses pause → intent recorded → UI shows paused immediately 2. Command sent over the serving transport → Shield pauses on screen 3. 10-30s later Jellyfin confirms paused → intent cleared 4. If Jellyfin doesn't confirm in 30s → retry once on the serving transport (protocol/ADB stamped on the intent) → surrender to truth """ import asyncio import logging import time import voluptuous as vol from homeassistant.components.media_player import ( BrowseMedia, MediaClass, MediaPlayerDeviceClass, MediaPlayerEntity, MediaPlayerEntityFeature, MediaPlayerState, MediaType, ) from homeassistant.config_entries import ConfigEntry from homeassistant.const import STATE_PAUSED, STATE_PLAYING from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback from homeassistant.exceptions import HomeAssistantError, ServiceValidationError from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.event import async_call_later, async_track_state_change_event from .connection import ShieldConnectionManager 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, CHANNEL_DIGIT_DELAY, CONF_ADB_COMMAND_ENABLED, CONF_ADB_STORE_APP_INFO, CONF_APP_ICON_MAP, CONF_APP_NAME_MAP, CONF_CHANNEL_DIGIT_DELAY, CONF_CHANNEL_SEND_ENTER, CONF_JELLYFIN_ENTITY, CONF_POWER_ENTITY, CONF_RECEIVER_ENTITY, CONF_RECEIVER_INPUT, DEFAULT_APP_DIRECTORY, DOMAIN, PROTOCOL_MEDIA_KEYS, PROTOCOL_REMOTE_KEYS, SCRIPT_VAR_APP_NAME, SCRIPT_VAR_LEVEL, SCRIPT_VAR_MUTED, TIVIMATE_PACKAGE, command_scripts, ) from .ha_adb_common import PlaybackIntent, running_apps_from_tasks from .jellyfin import fetch_session_media_info from .power import discover_power_siblings _LOGGER = logging.getLogger(__name__) INTENT_TIMEOUT = 10 # seconds before self-heal fires MAX_RETRIES = 1 # Map service names to the MediaPlayerState we expect after the command SERVICE_TO_STATE = { "media_pause": MediaPlayerState.PAUSED, "media_play": MediaPlayerState.PLAYING, "media_stop": MediaPlayerState.IDLE, } _BASE_FEATURES = ( MediaPlayerEntityFeature.TURN_ON | MediaPlayerEntityFeature.TURN_OFF | MediaPlayerEntityFeature.VOLUME_STEP | MediaPlayerEntityFeature.VOLUME_MUTE | MediaPlayerEntityFeature.PLAY_MEDIA | MediaPlayerEntityFeature.BROWSE_MEDIA ) _PLAYBACK_FEATURES = ( MediaPlayerEntityFeature.PLAY | MediaPlayerEntityFeature.PAUSE | MediaPlayerEntityFeature.STOP | MediaPlayerEntityFeature.NEXT_TRACK | MediaPlayerEntityFeature.PREVIOUS_TRACK ) async def async_run_command_script( hass: HomeAssistant, options: dict, command: str, variables: dict | None = None ) -> bool: """Run the script configured for ``command``; True when one handled it (t44). ``blocking=True`` so a script that fails surfaces HERE: a mapped command that cannot run must be a visible error, never a silent no-op. Module-level so the remote entity shares it — one configuration governs both routes to POWER. """ script = command_scripts(options).get(command) if not script: return False data: dict = {"entity_id": script} if variables: data["variables"] = variables try: await hass.services.async_call("script", "turn_on", data, blocking=True) except Exception as err: # noqa: BLE001 - re-raised as a user-facing error _LOGGER.error("the script for %s (%s) failed: %s", command, script, err) raise HomeAssistantError( f"the script configured for {command} ({script}) failed: {err}" ) from err return True async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: async_add_entities([NvidiaShieldMediaPlayer(hass, entry)]) # P5-1: one-time legacy unique_id migration (t20 F3 closure) hass.async_create_task(_migrate_legacy_unique_id(hass, entry)) async def _migrate_legacy_unique_id(hass, entry) -> None: """Migrate legacy androidtv-entity-id unique_ids once (t20 F3). Legacy entries carry the androidtv entity id as unique_id; the one-time migration moves it to the protocol MAC when the transport can resolve it, else to the config-entry id. After the single update_entry the legacy pattern no longer matches, so this never runs twice for an entry. No behavior change: the entry's data/options are untouched. """ # t41: the entity path is gone, so the SHAPE of the stale id is the marker. # Legacy entries (pre-t29 G2) carry the androidtv entity id as unique_id; # every other shape (a MAC, the config-entry id, a yaml: seed id) is left # exactly as it is. legacy = entry.unique_id or "" if not legacy.startswith("media_player."): return manager = getattr(entry, "runtime_data", None) new_id = None if manager is not None: info = await manager.async_protocol_device_info() if info is not None: _name, new_id = info if not new_id: new_id = entry.entry_id # t26 F4: another entry may already own the target MAC — the collision # pre-check falls back to the config-entry id instead of re-binding the # unique_id to a MAC that could collide with a real second instance. other_entries = getattr( getattr(hass, "config_entries", None), "entries", [] ) or [] if any( getattr(other, "unique_id", None) == new_id for other in other_entries if getattr(other, "entry_id", None) != entry.entry_id ): _LOGGER.warning( "P5-1: unique_id migration target %r already in use — falling " "back to the config-entry id", new_id, ) new_id = entry.entry_id hass.config_entries.async_update_entry(entry, unique_id=new_id) _LOGGER.info("P5-1: legacy unique_id %r migrated to %r", legacy, new_id) def entity_available(manager: ShieldConnectionManager | None) -> bool: """Is ANY usable transport healthy for this entry? (t40 live install) The single ladder behind ``media_player.available``, ``remote.available`` and the diagnostics payload, so the three cannot disagree: * the protocol transport is wired and has not reported the device unreachable — it serves power/app/volume even when ADB is down; * or ADB is connected (live state data); * or fresh wake data exists (the poller has spoken at least once); * or the manager is passive (no ADB configured at all): nothing to transport is not a broken transport. Unavailable only when a CONFIGURED transport is actively down and no usable data exists. The t5-F2 rule ("an ADB manager that is down means unreliable state") took the whole entity down while a healthy paired protocol transport was still serving commands — the reported defect. """ if not isinstance(manager, ShieldConnectionManager): return True if manager.protocol_active and manager.is_available is not False: return True if manager.connected: return True if manager.awake is not None: return True return not manager.available def resolve_app_name( package: str | None, app_name_map, app_label, ) -> str | None: """Resolve a package id to a friendly name (t40; entity tier removed t41). The single chain behind ``media_player.app_name`` and the Current App sensor: 1. the user's explicit apps map (``CONF_APP_NAME_MAP``); 2. the cached device-manifest label (``manager.app_label`` — a cache miss enqueues the on-demand fetch and falls through); 3. the raw package text — a bare or unknown package falls back to itself, never to None. """ if package: mapped = (app_name_map or {}).get(package) if mapped: return mapped label = app_label(package) if app_label is not None else None if label: return label return package or None class NvidiaShieldMediaPlayer(MediaPlayerEntity): _attr_device_class = MediaPlayerDeviceClass.TV _attr_has_entity_name = True _attr_name = None _attr_should_poll = False # Overridden as a property below: ADB-health aware def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: self.hass = hass self._entry = entry self._attr_unique_id = f"{DOMAIN}_{entry.entry_id}_media_player" self._unsub_listeners: list = [] # Intent tracking self._intent: PlaybackIntent | None = None self._intent_timer_unsub: CALLBACK_TYPE | None = None # Power monitoring — discovered from power switch entity self._power_sensors: dict[str, str] = {} # Jellyfin media stream info (resolution, codec, etc.) self._media_info: dict | None = None # ADB-derived state comes from the entry's ShieldConnectionManager # (connection.py) — this entity is a thin callback-view over its store. self._manager: ShieldConnectionManager | None = None self._manager_unsub: CALLBACK_TYPE | None = None # In-flight channel digit entry (play_media(channel) — §7); a new # play_media cancels the previous sequence (core task-cancel guard). self._channel_task = None # Last channel entered via play_media(channel) — the foreground # tier's title fallback when an app publishes no session (P3-3a). self._last_channel: str | None = None # Entity-owned background tasks (intent retries, media-info fetch); # cancelled on unload. The manager owns the socket — never touch it # here directly. self._entity_tasks: set = set() @property def available(self) -> bool: """Available while ANY usable transport is healthy (t40 live install). ADB enabled but unreachable no longer takes the entity down: a paired, available protocol transport still serves power/app/volume. The ladder is shared with the remote and diagnostics (``entity_available``). """ return entity_available(getattr(self._entry, "runtime_data", None)) @property def _jellyfin_entity(self) -> str | None: return self._entry.options.get(CONF_JELLYFIN_ENTITY) or None @property def _receiver_entity(self) -> str | None: return self._entry.options.get(CONF_RECEIVER_ENTITY) or None @property def _receiver_input(self) -> str | None: return self._entry.options.get(CONF_RECEIVER_INPUT) or None @property def device_info(self) -> DeviceInfo: # G10 (t30): when the protocol transport is live and captured, the # registry card carries the protocol-pushed identity (name from the # device itself, manufacturer 'Google' per the audit ruling); the # pre-capture shape stays the entry-title fallback. captured = ( self._manager.protocol_device_data if self._manager is not None else None ) if captured: name, manufacturer, model, sw = captured return DeviceInfo( identifiers={(DOMAIN, self._entry.entry_id)}, name=name or self._entry.title, manufacturer=manufacturer, model=model, sw_version=sw or None, ) return DeviceInfo( identifiers={(DOMAIN, self._entry.entry_id)}, name=self._entry.title, manufacturer="NVIDIA", model="Shield TV", ) @property def _power_entity(self) -> str | None: return self._entry.options.get(CONF_POWER_ENTITY) or None @property def _adb_store_info(self) -> bool: return self._entry.options.get(CONF_ADB_STORE_APP_INFO, True) async def async_added_to_hass(self) -> None: # P4-2a: opt-in as an adb_command target (option default off). # Registered at setup; unregistered at unload when the last # participating entity goes away. if self._entry.options.get(CONF_ADB_COMMAND_ENABLED, False): # F4: the registry key is computed once and stored on the # entity, so add/remove always agree even when entity_id is # assigned by HA core after the first read. self._adb_command_key = ( getattr(self, "entity_id", None) or self._attr_unique_id ) _ADB_COMMAND_TARGETS[self._adb_command_key] = self._manager # Discover power sensor siblings from the power switch if self._power_entity: self._power_sensors = discover_power_siblings(self.hass, self._power_entity) # Consume the entry's ShieldConnectionManager (thin-view pattern) manager = getattr(self._entry, "runtime_data", None) if manager is not None: self._manager = manager self._manager_unsub = manager.register_listener( self._on_manager_update ) entities = [ entity_id for entity_id in (self._jellyfin_entity, self._power_entity) if entity_id ] self._unsub_listeners.append( async_track_state_change_event( self.hass, entities, self._on_source_change ) ) async def async_will_remove_from_hass(self) -> None: for unsub in self._unsub_listeners: unsub() self._unsub_listeners.clear() self._clear_intent() if self._manager_unsub: self._manager_unsub() self._manager_unsub = None # Cancel in-flight entity tasks so they cannot outlive the entity. for task in self._entity_tasks: task.cancel() self._entity_tasks.clear() # P4-2a/t29 G8: drop this entity's service participation (same key # as add). The service itself stays registered (always-on setup). # ── Intent management ── def _set_intent(self, service: str, target: MediaPlayerState) -> None: """Record a new playback intent. Replaces any existing intent.""" # Cancel existing timer if self._intent_timer_unsub: self._intent_timer_unsub() # Capture current Jellyfin state before the intent jf = self._read(self._jellyfin_entity) prior = jf.state if jf else None self._intent = PlaybackIntent.create( target_state=target.value, command=service, prior_state=prior, ) # Schedule self-heal timeout self._intent_timer_unsub = async_call_later( self.hass, INTENT_TIMEOUT, self._on_intent_timeout ) _LOGGER.debug("Intent set: %s → %s (prior JF: %s)", service, target, prior) def _clear_intent(self) -> None: """Clear the pending intent and its timer.""" if self._intent_timer_unsub: self._intent_timer_unsub() self._intent_timer_unsub = None self._intent = None @callback def _on_intent_timeout(self, _now) -> None: """Self-heal: intent wasn't confirmed by Jellyfin in time.""" intent = self._intent if intent is None: return jf = self._read(self._jellyfin_entity) jf_state = jf.state if jf else None # Check if Jellyfin caught up during the timeout window if jf_state == intent.target_state: _LOGGER.debug("Intent %s confirmed at timeout check", intent.command) self._clear_intent() self.async_write_ha_state() return # Retry once if intent.retry_count < MAX_RETRIES: intent.retry_count += 1 intent.created_at = time.monotonic() _LOGGER.warning( "Intent %s not confirmed (JF=%s, want=%s), retrying (attempt %d)", intent.command, jf_state, intent.target_state, intent.retry_count, ) if ( intent.via_protocol and intent.protocol_key and self._manager is not None ): # The command was served by the protocol key push — retry there. self._track_task( self._manager.send_protocol_key(intent.protocol_key) ) elif ( intent.via_adb and intent.adb_command and self._manager is not None and self._manager.connected ): # The command was originally served by ADB — retry over ADB. self._track_task( self._manager.send_media_command(intent.adb_command) ) else: # t41: neither of OUR transports served this command (both were # down when it was sent), so there is nothing to re-send. Retrying # by delegating to another integration is exactly what this change # removes — surrender to the truth instead. _LOGGER.warning( "Intent %s had no serving transport to retry — falling back " "to truth", intent.command, ) self._clear_intent() self.async_write_ha_state() return self._intent_timer_unsub = async_call_later( self.hass, INTENT_TIMEOUT, self._on_intent_timeout ) return # Exhausted retries — surrender to Jellyfin truth _LOGGER.error( "Intent %s failed after %d retries (JF=%s), falling back to truth", intent.command, MAX_RETRIES, jf_state, ) self._clear_intent() self.async_write_ha_state() # ── State change listener ── @callback def _on_source_change(self, event: Event) -> None: """React to state changes in Jellyfin or the power switch.""" entity_id = event.data.get("entity_id") new_state_obj = event.data.get("new_state") # Power off clears any intent — device is off, intent is moot if ( entity_id == self._power_entity and new_state_obj and not self._is_power_on() and self._intent ): _LOGGER.debug("Intent cleared: device powered off") self._clear_intent() # Jellyfin state change — check for intent confirmation and fetch media info if entity_id == self._jellyfin_entity and new_state_obj: new_jf = new_state_obj.state # Intent confirmation if self._intent: intent = self._intent if new_jf == intent.target_state: _LOGGER.debug("Intent %s confirmed by Jellyfin", intent.command) self._clear_intent() elif new_jf not in ( intent.prior_state, "unavailable", "unknown", None ): _LOGGER.debug( "Intent %s overridden by external action (JF=%s)", intent.command, new_jf, ) self._clear_intent() # Fetch media stream info when playing if new_jf in (STATE_PLAYING, STATE_PAUSED): entity_picture = new_state_obj.attributes.get("entity_picture", "") self._track_task( self._fetch_media_info(entity_picture) ) else: self._media_info = None # Always re-evaluate state self.async_write_ha_state() # ── Helpers ── def _read(self, entity_id: str | None): if not entity_id: return None return self.hass.states.get(entity_id) @callback def _on_manager_update(self) -> None: """Connection manager's store changed — re-evaluate our state.""" self.async_write_ha_state() def _track_task(self, coro) -> None: """Dispatch an entity background task; cancelled on unload. Entity-owned tasks (intent retries, media-info fetch) must never go through a bare hass.async_create_task: they would survive removal (reviewer finding F2). The connection manager owns its own tasks and the socket — nothing here touches it directly. """ task = self.hass.async_create_task(coro) self._entity_tasks.add(task) task.add_done_callback(self._entity_tasks.discard) return task async def _fetch_media_info(self, entity_picture: str) -> None: """Read stream info from the HA Jellyfin integration's coordinator.""" info = await fetch_session_media_info( self.hass, self._jellyfin_entity, self._entry.title ) if info != self._media_info: self._media_info = info self.async_write_ha_state() def _jf_active(self) -> bool: """Is Jellyfin reporting active playback?""" jf = self._read(self._jellyfin_entity) if jf and jf.state in (STATE_PLAYING, STATE_PAUSED): return True # If we have an active intent for play/pause, media is conceptually active return bool( self._intent and self._intent.target_state in (STATE_PLAYING, STATE_PAUSED) ) # ── State ── def _is_power_on(self) -> bool | None: """Determine if the Shield is powered on. Priority: ADB wake state > power entity plug off > unknown. ADB is the authority — it knows if the device is awake or asleep. The plug being "on" just means the outlet has power (standby). """ # 1. Protocol is_on push — fastest truth when the transport is live protocol_is_on = self._manager.is_on if self._manager else None if protocol_is_on is not None: return protocol_is_on # 2. ADB wake state — exact mWakefulness parse adb_awake = self._manager.awake if self._manager else None if adb_awake is not None: return adb_awake # 3. Power entity — only trust "off" (plug pulled = definitely off) if self._power_entity: power_state = self._read(self._power_entity) if power_state and power_state.state == "off": return False return None @property def state(self) -> MediaPlayerState: # Check if plug is off first — truly off if self._power_entity: power_state = self._read(self._power_entity) if power_state and power_state.state == "off": return MediaPlayerState.OFF power = self._is_power_on() if power is False: if self._manager is not None and self._manager.is_on is False: return MediaPlayerState.OFF # protocol says powered off # ADB says asleep — standby (yellow dot) return MediaPlayerState.IDLE if power is None: return MediaPlayerState.IDLE # Tier 1: Active intent overrides Jellyfin if self._intent is not None: return MediaPlayerState(self._intent.target_state) # Tier 2: Jellyfin is authoritative jf = self._read(self._jellyfin_entity) if jf: if jf.state == STATE_PLAYING: return MediaPlayerState.PLAYING if jf.state == STATE_PAUSED: return MediaPlayerState.PAUSED # Tier 3: ADB media session (non-Jellyfin apps) adb_info = self._manager.best_session(self.app_id) if self._manager else None if adb_info: if adb_info.playing: return MediaPlayerState.PLAYING if adb_info.paused: return MediaPlayerState.PAUSED # Tier 3.5 (P3-3): foreground fallback — an app is open (protocol # current_app push, else the ADB mResumedActivity parse) but publishes # no media session. The device is still ON; the now-playing tier # carries the last-known/channel/audio-hint identity. if self._manager is not None and self._manager.foreground_app: return MediaPlayerState.ON # Tier 4: On but no media return MediaPlayerState.ON @property def supported_features(self) -> MediaPlayerEntityFeature: if self._jf_active(): features = _BASE_FEATURES | _PLAYBACK_FEATURES else: # ADB media session also supports playback controls adb_info = self._manager.best_session(self.app_id) if self._manager else None if adb_info and (adb_info.playing or adb_info.paused): features = _BASE_FEATURES | _PLAYBACK_FEATURES else: features = _BASE_FEATURES # t44: a command that exists ONLY as a script still has to be reachable, and # the feature flag is what makes HA offer it. volume_set and select_source # have no native path at all, so they appear exactly when configured — # without a script the feature set is what it always was. if self._script_for("volume_set"): features |= MediaPlayerEntityFeature.VOLUME_SET if self._script_for("select_source"): features |= MediaPlayerEntityFeature.SELECT_SOURCE return features # ── Metadata from Jellyfin ── def _jf_attr(self, key: str): jf = self._read(self._jellyfin_entity) return jf.attributes.get(key) if jf else None @property def app_id(self): """The raw foreground package from the manager's store (t41). Both native transports stamp it: the protocol current-app push and the ADB mResumedActivity parse. None until one of them has reported. """ return self._manager.current_app if self._manager is not None else None @property def _app_name_map(self) -> dict: """The user's apps name/icon map (P4-1 options editor).""" return self._entry.options.get(CONF_APP_NAME_MAP) or {} @property def app_name(self): """P5-1 resolution order — explicit override first: 1. CONF_APP_NAME_MAP (the user's explicit override) 2. device manifest label (`pm dump`, cached ≥24 h, resolved on demand — a miss enqueues the fetch and falls through) 3. raw package (protocol push / ADB foreground parse) """ # t40: one implementation, shared with the Current App sensor # (``resolve_app_name``) — the two can no longer disagree about what # is on screen. package = self._manager.foreground_app if self._manager is not None else None return resolve_app_name( package, self._app_name_map, self._manager.app_label if self._manager is not None else None, ) @property def media_content_type(self): return self._jf_attr("media_content_type") if self._jf_active() else None @property def media_title(self): return self._jf_attr("media_title") if self._jf_active() else None @property def media_series_title(self): return self._jf_attr("media_series_title") if self._jf_active() else None @property def media_season(self): return self._jf_attr("media_season") if self._jf_active() else None @property def media_episode(self): return self._jf_attr("media_episode") if self._jf_active() else None @property def media_duration(self): return self._jf_attr("media_duration") if self._jf_active() else None @property def media_position(self): return self._jf_attr("media_position") if self._jf_active() else None @property def media_position_updated_at(self): return self._jf_attr("media_position_updated_at") if self._jf_active() else None @property def media_image_url(self): """Jellyfin session picture, else the P5-2 apps-map icon override.""" if self._jf_active(): jf = self._jf_attr("entity_picture") if jf: return jf package = self._manager.foreground_app if self._manager else None if package: icon_map = self._entry.options.get(CONF_APP_ICON_MAP, {}) return icon_map.get(package) return None @property def volume_level(self) -> float | None: """Protocol volume push readout (C-VOL — level only, no VOLUME_SET).""" if self._manager is None: return None info = self._manager.volume_info if info is None: return None level, volume_max, _muted = info try: if volume_max and volume_max > 0: return level / volume_max except (TypeError, ValueError): pass return None @property def is_volume_muted(self) -> bool | None: """Real mute state from the protocol push (was unknowable pre-Phase 2).""" if self._manager is None: return None info = self._manager.volume_info return info[2] if info is not None else None def _get_now_playing(self) -> dict | None: if not self._jf_active(): return None jf = self._read(self._jellyfin_entity) if not jf: return None attrs = jf.attributes np = {"state": self.state.value, "title": attrs.get("media_title"), "media_type": attrs.get("media_content_type")} for src, dst in [("media_series_title", "series"), ("media_season", "season"), ("media_episode", "episode"), ("media_duration", "duration"), ("media_position", "position"), ("media_position_updated_at", "position_updated_at")]: v = attrs.get(src) if v is not None: np[dst] = v if attrs.get("entity_picture"): np["thumbnail"] = attrs["entity_picture"] if self._media_info: np["media_info"] = self._media_info return np def _get_adb_now_playing(self) -> dict | None: """Build now-playing from ADB media session (non-Jellyfin apps).""" info = self._manager.best_session(self.app_id) if self._manager else None if not info: return None if not self._adb_store_info: # Privacy mode: show that something is playing but not what if info.playing or info.paused: return { "state": "playing" if info.playing else "paused", "title": self.app_name or self.app_id or "Playing", "source": "media_session", } return None if not info.title: return None np = { "state": "playing" if info.playing else ("paused" if info.paused else "idle"), "title": info.title, "source": "media_session", } if info.artist: np["artist"] = info.artist if info.album: np["album"] = info.album return np def _get_app_now_playing(self) -> dict | None: """Foreground-tier now-playing (P3-3a): an app is open but publishes no media session (TiviMate, R2). Title = last known state, else the last channel entered, else the zero-cost audio-playback hint. """ manager = self._manager if manager is None: return None foreground = manager.foreground_app if not foreground: return None state = "idle" title = None info = manager.last_known_session(foreground) if info is not None and (info.playing or info.paused): state = "playing" if info.playing else "paused" # Privacy mode: identity stays the app, never the media title title = None if not self._adb_store_info else info.title if title is None and not self._adb_store_info: # F3: privacy mode first — identity stays the app, so neither the # channel number nor the media title can leak as the identity. title = self.app_name or self.app_id or foreground if title is None and self._last_channel and ( # F1: the channel number belongs to the channel app only — never # leak it to the launcher or another foreground app. foreground == TIVIMATE_PACKAGE or foreground in manager.audio_playback ): title = self._last_channel if title is None and foreground in manager.audio_playback: # E hint (zero round-trip cost): audio was recently played title = "Media recently active" if title is None: return None # bare foreground app — the state stays ON regardless return {"state": state, "title": title, "source": "app"} def _get_power_data(self) -> dict | None: """Read current values from discovered power sensors.""" if not self._power_sensors: return None data = {} for key, entity_id in self._power_sensors.items(): if key == "switch": continue state = self._read(entity_id) if state and state.state not in ("unavailable", "unknown"): try: data[key] = float(state.state) except (ValueError, TypeError): data[key] = state.state return data if data else None def _get_running_apps(self) -> list[dict] | None: """Build a list of running apps from ADB recent tasks + media sessions.""" if not self._manager: return None apps = self._manager.running_apps if not apps: return None if not self._adb_store_info: return None return running_apps_from_tasks( apps, self._manager.active_sessions, foreground_package=self.app_id, last_known=self._manager.last_known_map(), ) # G12 (t30): attributes that change with every poll must not bloat the # database (entity doc) — dag internals are declared unrecorded. _unrecorded_attributes = frozenset( { ATTR_MEDIA_DAG_NOW_PLAYING, ATTR_MEDIA_DAG_RUNNING_APPS, ATTR_MEDIA_DAG_RECEIVER_ENTITY, ATTR_MEDIA_DAG_RECEIVER_INPUT, ATTR_MEDIA_DAG_ROLE, } ) @property def extra_state_attributes(self) -> dict: """Extra state attributes — a PROPERTY, as HA requires (t40 blocker). ``Entity.__async_calculate_state`` merges them with ``attr |= self.extra_state_attributes``; declaring this as a plain method made every state write raise ``TypeError: 'method' object is not iterable``, so the entity could never render on a live install. """ # Now playing: prefer Jellyfin, fall back to ADB media session, # then the foreground tier (P3-3a). now_playing = self._get_now_playing() if not now_playing: now_playing = self._get_adb_now_playing() if not now_playing: now_playing = self._get_app_now_playing() attrs = {"app_id": self.app_id, "app_name": self.app_name, ATTR_MEDIA_DAG_ROLE: "source", ATTR_MEDIA_DAG_NOW_PLAYING: now_playing} if self._receiver_entity: attrs[ATTR_MEDIA_DAG_RECEIVER_ENTITY] = self._receiver_entity if self._receiver_input: attrs[ATTR_MEDIA_DAG_RECEIVER_INPUT] = self._receiver_input if self._jellyfin_entity: attrs[ATTR_MEDIA_DAG_JELLYFIN_ENTITY] = self._jellyfin_entity # Running apps from ADB media sessions running_apps = self._get_running_apps() if running_apps: attrs[ATTR_MEDIA_DAG_RUNNING_APPS] = running_apps # Power monitoring if self._power_entity: attrs["power_switch"] = self._power_entity power_data = self._get_power_data() if power_data: attrs["power_monitoring"] = power_data # Intent state for debugging if self._intent: attrs["_intent"] = self._intent.to_debug_dict() return attrs # ── Commands — our own transports only (t41: the external entity is gone) ── async def _power_key(self, *, send_when_on: bool) -> None: """Send the protocol POWER key when the known state disagrees (t41). Both native transports are ours now, so this no longer delegates to an androidtv_remote entity: the protocol POWER key is the path we own. An unknown state sends nothing (no superfluous toggle — the same rule the remote entity applies), and without the protocol transport there is nothing to transport the key over. """ manager = self._manager if manager is None or manager.is_on is not send_when_on: return await manager.send_protocol_key(PROTOCOL_REMOTE_KEYS["power"]) def _script_for(self, command: str) -> str | None: """The script configured to own ``command``, if any (t44).""" return command_scripts(self._entry.options).get(command) async def _script_served(self, command: str, variables: dict | None = None) -> bool: """True when a configured script handled ``command`` instead of the transports.""" return await async_run_command_script( self.hass, self._entry.options, command, variables ) async def async_turn_on(self): """Wake the Shield (only when it reports OFF) — or run the turn_on script.""" if await self._script_served("turn_on"): return await self._power_key(send_when_on=False) async def async_turn_off(self): """Sleep the Shield (only when it reports ON) — or run the turn_off script.""" if await self._script_served("turn_off"): return await self._power_key(send_when_on=True) async def async_volume_up(self): if await self._script_served("volume_up"): return await self._remote_key_cmd("volume_up", "volume_up") async def async_volume_down(self): if await self._script_served("volume_down"): return await self._remote_key_cmd("volume_down", "volume_down") async def async_mute_volume(self, mute): """Mute — via the configured script, else the protocol/ADB key (C-VOL).""" if await self._script_served("volume_mute", {SCRIPT_VAR_MUTED: mute}): return await self._remote_key_cmd("volume_mute", "mute") async def async_set_volume_level(self, volume): """Set absolute volume — reachable only with a script (t44). There is no native path to replace: the protocol carries UP/DOWN/MUTE keys only (C-VOL) and ADB has no absolute volume either. Configured, this is the hook for a device whose volume is owned elsewhere (a speaker switch); unconfigured it stays the no-op it has always been. """ await self._script_served("volume_set", {SCRIPT_VAR_LEVEL: volume}) async def async_select_source(self, source): """Select a source — reachable only with a script (t44). The integration has no native source selection (app launching goes through play_media with the app map). A script for this command is the hook for an AVR input or anything else that has to move with the Shield. """ await self._script_served("select_source", {SCRIPT_VAR_APP_NAME: source}) async def _remote_key_cmd(self, remote_key: str, adb_key: str) -> None: """Simple key command without intent bookkeeping: protocol → ADB keyevent.""" protocol_key = PROTOCOL_REMOTE_KEYS.get(remote_key) if ( protocol_key and self._manager is not None and await self._manager.send_protocol_key(protocol_key) ): return if self._manager is not None and await self._manager.send_media_command(adb_key): return async def _media_cmd( self, ha_service: str, adb_command: str, intent_state: MediaPlayerState | None = None ) -> None: """Send a media command over the fastest available command path. Order: protocol key push → ADB keyevent (t41: our own transports only — the transitional external-entity tier is gone). With neither transport up the command is a no-op, never a call into another integration. """ if intent_state: self._set_intent(ha_service, intent_state) self.async_write_ha_state() # Use ADB media dispatch when ADB is active and Jellyfin isn't playing jf = self._read(self._jellyfin_entity) jf_active = jf and jf.state in (STATE_PLAYING, STATE_PAUSED) # 1. Protocol key push (fast; works whenever the transport is up) protocol_key = PROTOCOL_MEDIA_KEYS.get(ha_service) if ( protocol_key and self._manager is not None and await self._manager.send_protocol_key(protocol_key) ): if self._intent is not None: self._intent.via_protocol = True self._intent.protocol_key = protocol_key return # 2. ADB keyevent (socket-independent; exact same commands as before) if self._manager and self._manager.connected and not jf_active: success = await self._manager.send_media_command(adb_command) if success: # The ADB path served this command — the self-heal retry must # also go over ADB, not back to the remote. if self._intent is not None: self._intent.via_adb = True self._intent.adb_command = adb_command return async def async_media_play(self): await self._media_cmd("media_play", "play", MediaPlayerState.PLAYING) async def async_media_pause(self): await self._media_cmd("media_pause", "pause", MediaPlayerState.PAUSED) async def async_media_stop(self): await self._media_cmd("media_stop", "stop", MediaPlayerState.IDLE) async def async_media_next_track(self): await self._media_cmd("media_next_track", "next") async def async_media_previous_track(self): await self._media_cmd("media_previous_track", "previous") async def async_bring_to_foreground(self, package: str) -> bool: """Bring a specific app to the foreground via ADB.""" if self._manager is None or not self._manager.connected: _LOGGER.warning("ADB not available for bring_to_foreground") return False return await self._manager.bring_to_foreground(package) # ── BrowseMedia app directory (P4-2c) ── async def async_browse_media(self, media_content_type=None, media_content_id=None): """Root lists the apps map (or the curated default) → play_media(app). t37 D1: the standard media_player tree, exactly as core's own androidtv_remote/media_player.py builds it — ``BrowseMedia`` from homeassistant.components.media_player, every node carrying an explicit media_content_id/media_content_type, and the directory naming the class of its children. ``BrowseMediaSource`` only ever existed in homeassistant.components.media_source.models (t30 G18's domain parameter is ITS contract: it builds ``media_source://`` ids), so an app-directory node must never claim one — and the import above removed a name media_player has not exported for a long time, which made every real browse raise. """ app_map = self._entry.options.get(CONF_APP_NAME_MAP) or DEFAULT_APP_DIRECTORY children = [ BrowseMedia( title=name, media_class=MediaClass.APP, media_content_type=MediaType.APP, media_content_id=pkg, can_play=True, can_expand=False, ) for pkg, name in app_map.items() ] return BrowseMedia( title="Apps", media_class=MediaClass.DIRECTORY, media_content_type=MediaType.APPS, media_content_id="apps", children_media_class=MediaClass.APP, can_play=False, can_expand=True, children=children, ) # ── play_media (§7): channel digits, app/url launch ── async def async_play_media(self, media_type: str, media_id: str) -> None: """Channel selection by digit number; app/url launches (C-LAUNCH). Channel path replicates HA core androidtv_remote exactly: digits only, configurable inter-digit delay (default 0.1 s), no ENTER — and the padded prior in-flight channel sequence is cancelled first. TEXT entry (P5-2): play_media("text", "...") — protocol IME send_text first; the ADB `input text` fallback tier runs only when the protocol send is unavailable or failed (the Shield may swallow IME text while its on-screen keyboard is up, RESEARCH_IDEAS §2). """ if media_type == "text": if not media_id: _LOGGER.warning("play_media(text): empty text ignored") return if self._manager is not None and await self._manager.send_text_ime(media_id): return if self._manager is not None and await self._manager.send_adb_input_text(media_id): return _LOGGER.warning( "play_media(text): no transport available (protocol down, ADB" " unavailable) — text not sent" ) return if media_type == "channel": digits = [c for c in media_id if c.isdigit()] if not digits: _LOGGER.warning("play_media(channel): no digits in %r", media_id) return self._last_channel = "".join(digits) self._cancel_inflight_channel() self._channel_task = self._track_task( self._channel_entry_sequence(digits) ) return if media_type in ("app", "url"): if self._manager is not None and await self._manager.launch_app(media_id): return if ( media_type == "app" and self._manager is not None and await self._manager.bring_to_foreground(media_id) ): # Fallback: monkey launch of the bare package (no deep links). return _LOGGER.warning( "play_media(%s): could not launch %s — no available transport", media_type, media_id, ) return _LOGGER.warning( "play_media: unsupported media type %r (supported: channel/app/url)" " — nothing sent", media_type, ) def _cancel_inflight_channel(self) -> None: task = self._channel_task if task is not None and not task.done(): task.cancel() self._channel_task = None async def _channel_entry_sequence(self, digits: list[str]) -> None: delay = self._entry.options.get(CONF_CHANNEL_DIGIT_DELAY, CHANNEL_DIGIT_DELAY) send_enter = self._entry.options.get(CONF_CHANNEL_SEND_ENTER, False) try: for digit in digits: # 1. Protocol digit key; 2. ADB keyevent 7..16 (t41: our own # transports only — an unserved digit is dropped, never handed # to another integration) served = False if self._manager is not None: served = await self._manager.send_protocol_key(f"KEYCODE_{digit}") if not served: served = await self._manager.send_digit_key(digit) if not served: _LOGGER.warning( "channel digit %s not sent — no available transport", digit ) await asyncio.sleep(delay) if ( send_enter and self._manager is not None and not await self._manager.send_protocol_key("KEYCODE_ENTER") ): # ENTER fallback chain: protocol key → ADB keyevent 66. # (No legacy-entity tier: the external androidtv_remote has # no ENTER service to delegate to.) await self._manager.send_enter_key() except asyncio.CancelledError: raise # ── P4-2a: the option-gated adb_command service ── # Entities register their manager as a target when their entry enables the # option; the domain service unregisters once the last participant unloads # (core androidtv service posture: documented, opt-in, default off). _ADB_COMMAND_TARGETS: dict[str, ShieldConnectionManager] = {} _ADB_COMMAND_SCHEMA = vol.Schema( { vol.Required("command"): str, vol.Required("entity_id"): str, } ) async def _async_adb_command_service(call) -> None: """Run a raw ADB shell command on the target connection (t29 G8). The integration's async_setup registers this ALWAYS; the entry's adb_command_enabled option gates USE (participants register from async_added_to_hass). Failures raise ServiceValidationError with translated keys (action-exceptions) — never warning logs. """ command = str(call.data.get("command") or "").strip() if not command: raise ServiceValidationError( translation_domain=DOMAIN, translation_key="invalid_command" ) entity_id = call.data.get("entity_id") manager = _ADB_COMMAND_TARGETS.get(entity_id) if manager is None: raise ServiceValidationError( translation_domain=DOMAIN, translation_key="unsupported_target", ) poller = manager.adb_poller # public accessor (t21 F6) if poller is None or not manager.connected: raise ServiceValidationError( translation_domain=DOMAIN, translation_key="adb_unavailable" ) result = await poller.shell(command) _LOGGER.debug("adb_command %r → output=%s", command, result is not None)