"""Remote entities for NVIDIA Shield — protocol-backed (Phase 2, t10). Port of HA core androidtv_remote/remote.py semantics onto the absorbed remote-protocol transport (androidtvremote2) owned by ShieldConnectionManager: - ``send_command`` accepts SHORT/START_LONG/END_LONG direction prefixes (``"START_LONG:"``; no prefix = SHORT) and core-style service keys; the core port's hold_secs+direction validation quirks are not carried over — long presses are expressed as the START_LONG/END_LONG pair instead; - ``num_repeats`` / ``delay_secs`` are honored; long-press duration uses the ``hold_secs`` parameter; - ``RemoteEntityFeature.ACTIVITY`` with an ``activity_list`` built from OUR app directory — the library exposes NO app-list API (measured), which is exactly why core's list is user-maintained too — so the app/activity picker matches core; - ``current_activity`` resolves the foreground push (protocol first, ADB parse as the fallback) through the one ``resolve_app_name`` chain, and stays None while nothing is known: no fabricated activity, ever; - ``turn_on(activity=...)`` wakes when the device says it is off and then launches through the same path ``select_source`` uses; an activity name the directory does not carry is a ``ServiceValidationError``, never a silent no-op; - the transport's REAL state is published in ``extra_state_attributes`` so a never-connected transport cannot look like a paired one (no ``is_paired`` exists: a session or a push is the only proof); - a configured command script still owns turn_on/turn_off and never touches the protocol, so a press that worked through it says NOTHING about pairing — the attributes name which route each power command takes. """ import asyncio import os from homeassistant.components.remote import ( ATTR_ACTIVITY, RemoteEntity, RemoteEntityFeature, ) from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback from homeassistant.exceptions import ServiceValidationError from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback from .connection import ShieldConnectionManager from .const import ( CONF_APP_NAME_MAP, DOMAIN, PROTOCOL_REMOTE_KEYS, SCRIPT_VAR_COMMAND, SCRIPT_VAR_DELAY_SECS, SCRIPT_VAR_HOLD_SECS, SCRIPT_VAR_NUM_REPEATS, command_scripts, remote_command_script, ) from .helpers import protocol_cert_paths from .media_player import ( activity_names, app_directory, async_call_script, async_run_command_script, entity_available, resolve_app_name, ) _START_PREFIX = "START_LONG:" _END_PREFIX = "END_LONG:" _SHORT_PREFIX = "SHORT:" _POWER_KEY = "KEYCODE_POWER" def _split_command(command: str) -> tuple[str, str]: """Key + direction from a command string (default SHORT).""" lowered = command.strip() if lowered.startswith(_START_PREFIX): return lowered[len(_START_PREFIX):], "START_LONG" if lowered.startswith(_END_PREFIX): return lowered[len(_END_PREFIX):], "END_LONG" if lowered.startswith(_SHORT_PREFIX): return lowered[len(_SHORT_PREFIX):], "SHORT" return lowered, "SHORT" def _key_code_name(command: str) -> str: """Command kind → protocol keycode (raw KEYCODE_ names pass through).""" mapped = PROTOCOL_REMOTE_KEYS.get(command.lower()) if mapped is not None: return mapped upper = command.upper() if upper.startswith("KEYCODE_"): return upper return upper # unknown names: the library reports/validates them def _looks_launchable(value: str) -> bool: """True for a package id or a detail link — forms the launch path accepts (t15). The rule is the repo's own definition of a package (``_valid_app_row``: dotted, no spaces) plus URL-shaped deep links, which is exactly what ``send_launch_app_command`` handles. It is deliberately NOT a guess that an id exists: the device answers that. It exists so "Netflx" (a typo of a listed name) is refused while an app the directory does not carry can still be launched. """ if not value or " " in value: return False return "." in value or "://" in value async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: """Add the protocol-backed remote entity for this entry.""" async_add_entities([NvidiaShieldRemoteEntity(hass, entry)]) class NvidiaShieldRemoteEntity(RemoteEntity): """Remote entity over the manager's androidtvremote2 transport. t15: this is the entity the app/activity picker hangs off. HA renders the picker only when ``supported_features`` carries ``RemoteEntityFeature.ACTIVITY`` — the user's "the UI doesn't actually show any buttons like the android tv remote does" — and the list itself has to come from OUR directory, because ``AndroidTVRemote`` exposes no app-list API at all (measured on the installed library; core androidtv_remote keeps a user-maintained map for the same reason). """ _attr_has_entity_name = True _attr_name = None _attr_should_poll = False #: The feature flag that makes HA offer the activity/app picker at all. _attr_supported_features = RemoteEntityFeature.ACTIVITY def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: self.hass = hass self._entry = entry self._manager: ShieldConnectionManager | None = getattr( entry, "runtime_data", None ) self._attr_unique_id = f"{DOMAIN}_{entry.entry_id}_remote" # t15: the activity list is available from the FIRST render (a real-HA state # write needs no entity platform), and is refreshed with discovered apps once # ADB can answer (``async_added_to_hass``). self._attr_activity_list = activity_names(self._entry.options) self._attr_current_activity = self._resolve_activity_name() #: Which protocol pushes have EVER arrived — the only pairing evidence the #: library can offer (it has no is_paired). Sticky on purpose: the store #: expires every field after STATE_STORE_TTL, so "has it ever arrived?" #: cannot be read from the store. self._protocol_pushes_seen: set[str] = set() self._note_protocol_pushes() #: "present"/"missing"/"unknown" — the shared setup certificate, which is #: why enabling the transport on a paired entry asks for no new PIN. self._protocol_certificate = "unknown" self._activity_refresh_task = None def _manager_of(self) -> ShieldConnectionManager | None: return getattr(self._entry, "runtime_data", None) def _resolve_activity_name(self) -> str | None: """The foreground app as a friendly name (t15), or None when nothing is known. The same chain as the media player's ``app_name`` and the Current App sensor (``resolve_app_name``): the user's map, then the cached device label, then the curated table, then the raw id — so an activity and a source can never disagree about what is on screen. The foreground identity prefers the protocol push and falls back to the ADB parse, which keeps this entity useful on an ADB-only setup (its protocol surface alone degrades). """ manager = self._manager_of() if not isinstance(manager, ShieldConnectionManager): return None return resolve_app_name( manager.foreground_app, self._entry.options.get(CONF_APP_NAME_MAP) or {}, manager.app_label, ) def _note_protocol_pushes(self) -> None: """Remember which device pushes have arrived (t15) — sticky, see __init__.""" manager = self._manager_of() if not isinstance(manager, ShieldConnectionManager): return if manager.current_app is not None: self._protocol_pushes_seen.add("current_app") if manager.is_on is not None: self._protocol_pushes_seen.add("is_on") if manager.is_available is not None: self._protocol_pushes_seen.add("is_available") if manager.volume_info is not None: self._protocol_pushes_seen.add("volume") @property def device_info(self) -> DeviceInfo: return DeviceInfo( identifiers={(DOMAIN, self._entry.entry_id)}, name=self._entry.title, manufacturer="NVIDIA", model="Shield TV", ) @property def available(self) -> bool: """Available while ANY usable transport can serve a command (t40). The same ladder as the media player (``entity_available``): protocol wired+available, ADB connected, fresh wake data or a passive manager all count as usable, and a passive manager stays available — nothing to transport is not a broken one. """ return entity_available(self._manager_of()) @property def activity(self) -> str | None: """The raw foreground package (protocol push; ADB parse as the fallback). Kept alongside HA's ``current_activity``: this is the id (what ``turn_on(activity=...)`` accepts and what the Current App sensor stamps), while ``current_activity`` is the friendly name HA displays. """ manager = self._manager_of() if isinstance(manager, ShieldConnectionManager): return manager.foreground_app return None @property def extra_state_attributes(self) -> dict: """The transport's real state — the pairing answer, readable from HA (t15). There is no "is it paired?" API to call (the library exposes none either, measured), so everything here is a fact that can be PROVEN: * ``protocol_state`` — ``off`` / ``never_connected`` / ``connected`` / ``disconnected`` / ``auth_rejected``. ``never_connected`` is the state that used to be permanent (``keep_reconnecting()`` alone never connects), and it must never look like ``connected``. * ``protocol_connected`` / ``protocol_ever_authenticated`` — a session or a device push is the only proof of an authenticated (paired) client. * ``protocol_current_app_seen`` / ``protocol_is_on_seen`` / ``protocol_volume_seen`` — whether each device push has EVER arrived. * ``protocol_last_error`` / ``protocol_failures`` — a send that could not run is recorded here and logged, never swallowed into a debug line. * ``protocol_certificate`` — the setup pairing's shared ``client.pem`` is still on disk, which is exactly why enabling the transport asks for no new PIN (the other Android TV integration pairs its own client). * ``power_on_route`` / ``power_off_route`` — ``script:`` when a command script owns that command. That path never touches the protocol, so a success through it says NOTHING about pairing. """ return { "activity": self.activity, # t15: which tier produced it — a protocol push must never win SILENTLY # over the ADB parse the user's setup is built on (see the manager's # ``foreground_source``). "activity_source": self._protocol_read("foreground_source"), "power_on_route": self._power_route("turn_on"), "power_off_route": self._power_route("turn_off"), "protocol_active": self._protocol_active(), "protocol_state": self._protocol_state(), "protocol_connected": self._protocol_flag("protocol_connected"), "protocol_ever_authenticated": self._protocol_flag( "protocol_ever_authenticated" ), "protocol_last_error": self._protocol_read("protocol_last_error"), "protocol_failures": self._protocol_read("protocol_failures"), "protocol_certificate": self._protocol_certificate, "protocol_current_app_seen": "current_app" in self._protocol_pushes_seen, "protocol_is_on_seen": "is_on" in self._protocol_pushes_seen, "protocol_volume_seen": "volume" in self._protocol_pushes_seen, } def _protocol_read(self, name: str, default=None): """Read one transport fact off the manager, or the default (stub managers).""" manager = self._manager_of() if not isinstance(manager, ShieldConnectionManager): return default return getattr(manager, name, default) def _protocol_active(self) -> bool: return bool(self._protocol_read("protocol_active", False)) def _protocol_state(self) -> str: """The transport state, with ``off`` for a manager that carries no client.""" if not self._protocol_active(): return "off" return self._protocol_read("protocol_state", "never_connected") def _protocol_flag(self, name: str) -> bool: return bool(self._protocol_read(name, False)) def _power_route(self, command: str) -> str: """Which path ``turn_on``/``turn_off`` takes: a script, or the protocol (t15). A configured command script owns the sequence and never touches the protocol — so "the power press worked" through it is NOT evidence that the transport is paired, or even connected. That is the conclusion the user drew and then doubted, made readable instead of inferred. """ script = (command_scripts(self._entry.options) or {}).get(command) return f"script:{script}" if script else "protocol" @callback def _refresh_state(self) -> None: """Manager store changed — re-resolve activity, availability, transport.""" self._manager = self._manager_of() self._note_protocol_pushes() self._attr_current_activity = self._resolve_activity_name() self.async_write_ha_state() async def async_added_to_hass(self) -> None: manager = self._manager_of() if isinstance(manager, ShieldConnectionManager): self._manager_unregistered = manager.register_listener(self._refresh_state) self._note_protocol_pushes() await self._async_note_certificate() self._schedule_activity_refresh() async def _async_note_certificate(self) -> None: """Record whether the shared pairing certificate exists (t15). This is the "no new PIN is expected" evidence, and it is a design fact, not a guess: ``protocol_cert_paths`` is shared between the config flow's pairing step and this runtime, so an entry paired at setup reuses that ``client.pem`` when the transport is (re-)enabled. The OTHER Android TV integration pairs its own client and shares nothing with us. """ add_executor_job = getattr(self.hass, "async_add_executor_job", None) if add_executor_job is None: return # a hass stand-in: leave it "unknown" rather than guess try: certfile, keyfile = await add_executor_job(protocol_cert_paths, self.hass) except Exception: # pragma: no cover - a filesystem we cannot read return self._protocol_certificate = ( "present" if os.path.exists(certfile) and os.path.exists(keyfile) else "missing" ) def _schedule_activity_refresh(self) -> None: """Add the apps ADB discovery actually found to the activity list (t15). One bounded sweep, on demand — the t57 path the options editor uses — never a per-state-write shell. The list is already complete for a configured directory, and a sweep that cannot run changes nothing (no fabrication). """ manager = self._manager_of() create_task = getattr(self.hass, "async_create_task", None) if create_task is None or not isinstance(manager, ShieldConnectionManager): return if not callable(getattr(manager, "async_refresh_installed_apps", None)): return # a manager stand-in: the directory alone is the honest list self._activity_refresh_task = create_task( self._async_refresh_activity_list(manager), "nvidia_shield_remote_activities", ) async def _async_refresh_activity_list(self, manager) -> None: """Fold discovered apps into the activity list (t15) — one write if it grew.""" try: rows = await manager.async_refresh_installed_apps() except Exception: # pragma: no cover - discovery never breaks the entity return names = activity_names(self._entry.options, rows) if names != self._attr_activity_list: self._attr_activity_list = names self.async_write_ha_state() async def async_will_remove_from_hass(self) -> None: unregistered = getattr(self, "_manager_unregistered", None) if unregistered is not None: unregistered() self._manager_unregistered = None async def async_turn_on(self, **kwargs) -> None: """Wake the device, and launch a requested activity (t15). With NO activity — or an empty one — this behaves exactly as before: the configured script owns the wake sequence when there is one, otherwise POWER goes out only when the device says it is off (unknown counts as "maybe on", so nothing is toggled blindly). With ``activity``, the same wake happens first (a script that owns turn_on still owns it — it never touches the protocol), and then the app is launched over the protocol through the same path ``select_source`` uses. A name the activity list does not carry, and that is not a package/detail link either, raises ``ServiceValidationError``: an activity request that cannot launch must never look like it worked. """ activity = str(kwargs.get(ATTR_ACTIVITY) or "").strip() await self._async_wake() if activity: await self._async_launch_activity(activity) async def _async_wake(self) -> None: """Turn the device on: the configured script first, else the POWER key (t44). t44: `turn_on` is one configuration, and the remote entity is the other route to it; a script that owns the power sequence owns it here too. """ if await async_run_command_script(self.hass, self._entry.options, "turn_on"): return manager = self._manager_of() if not isinstance(manager, ShieldConnectionManager): return if manager.is_on is not False: # Unknown counts as "maybe on" — no superfluous toggle. return await manager.send_protocol_key(_POWER_KEY, "SHORT") async def _async_launch_activity(self, activity: str) -> None: """Launch an activity through the shared launch path (t15). The target comes from the same directory the activity list is built from, so anything the picker offers resolves. A package id or a deep link passes through: that is how an app the directory does not carry can still be launched, and what core does with an activity it does not know. Anything else is a request nothing can launch and raises instead of doing nothing. """ target = self._activity_target(activity) manager = self._manager_of() if not isinstance(manager, ShieldConnectionManager): raise ServiceValidationError( translation_domain=DOMAIN, translation_key="protocol_unavailable", translation_placeholders={"activity": activity}, ) if not await manager.launch_app(target): raise ServiceValidationError( translation_domain=DOMAIN, translation_key="activity_launch_failed", translation_placeholders={"activity": activity}, ) def _activity_target(self, activity: str) -> str: """Activity name (or id) → what the launch path takes (t15). A listed NAME wins, compared case-insensitively — the same rule the ``source_list``/``select_source`` pair uses, so the picker can never offer a name this does not resolve. A ``CONF_APP_NAME_MAP``-less entry still resolves the curated names, because both read the one ``app_directory``. """ directory = app_directory(self._entry.options) wanted = activity.casefold() for package, name in directory.items(): if name.casefold() == wanted: return package if activity in directory or _looks_launchable(activity): return activity raise ServiceValidationError( translation_domain=DOMAIN, translation_key="unknown_activity", translation_placeholders={"activity": activity}, ) async def async_turn_off(self, **kwargs) -> None: """Sleep the device — via the configured script, else the POWER key.""" if await async_run_command_script(self.hass, self._entry.options, "turn_off"): return manager = self._manager_of() if not isinstance(manager, ShieldConnectionManager): return if manager.is_on is not True: return await manager.send_protocol_key(_POWER_KEY, "SHORT") async def async_send_command( self, command: list[str], num_repeats: int = 1, delay_secs: float = 0.0, hold_secs: float = 0.0, ) -> None: """Send protocol keys — or hand the whole command to the configured script. t47: with a remote command script configured, it replaces the key path entirely and receives the service call's own parameters, named as they are on the service: ``command`` (a list of key commands), ``num_repeats``, ``delay_secs`` and ``hold_secs``. Unset, nothing changes: repeats, long presses and delays go out as protocol keys exactly as before. """ script = remote_command_script(self._entry.options) if script: await async_call_script( self.hass, script, "remote.send_command", { SCRIPT_VAR_COMMAND: command, SCRIPT_VAR_NUM_REPEATS: num_repeats, SCRIPT_VAR_DELAY_SECS: delay_secs, SCRIPT_VAR_HOLD_SECS: hold_secs, }, ) return manager = self._manager_of() if not isinstance(manager, ShieldConnectionManager): return for _ in range(max(1, num_repeats)): for cmd in command: key_part, direction = _split_command(cmd) key_name = _key_code_name(key_part) if hold_secs > 0 and direction == "SHORT": if await manager.send_protocol_key(key_name, "START_LONG"): await asyncio.sleep(max(0.0, hold_secs)) await manager.send_protocol_key(key_name, "END_LONG") else: await manager.send_protocol_key(key_name, direction) if delay_secs > 0: await asyncio.sleep(delay_secs) async def async_send_text(self, text: str) -> bool: """Text entry (P5-2) via the protocol send_text (remote surface). Falls back to ADB `input text` only when the protocol transport is down or the send failed (the Shield may swallow IME text while its on-screen keyboard is up, RESEARCH_IDEAS §2). """ manager = self._manager_of() if manager is None or not text: return False if await manager.send_text_ime(text): return True return await manager.send_adb_input_text(text)