"""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; - ``activity`` = the protocol current-app push (package; name resolution is deferred app-map work); - turn on/off send the POWER key and honor the is_on push (no superfluous toggling when the state already matches). """ import asyncio from homeassistant.components.remote import RemoteEntity from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback from .connection import ShieldConnectionManager from .const import DOMAIN, PROTOCOL_REMOTE_KEYS from .media_player import async_run_command_script, entity_available _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 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.""" _attr_has_entity_name = True _attr_name = None _attr_should_poll = False 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" def _manager_of(self) -> ShieldConnectionManager | None: return getattr(self._entry, "runtime_data", None) @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: """Current foreground app reported by the protocol push.""" manager = self._manager_of() if isinstance(manager, ShieldConnectionManager): return manager.current_app return None @callback def _refresh_state(self) -> None: """Manager store changed — re-evaluate state (activity/available).""" self._manager = self._manager_of() 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 ) 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 — via the configured script, else the POWER key.""" # 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_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: repeats × (key, or START_LONG/hold/END_LONG).""" 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)