191 lines
7.1 KiB
Python
191 lines
7.1 KiB
Python
"""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:<key>"``; 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
|
||
|
||
_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:
|
||
"""Unavailable while the transport is active but unpaired/down.
|
||
|
||
A passive manager (no protocol transport) keeps the entity available:
|
||
nothing to transport is not a broken one.
|
||
"""
|
||
manager = self._manager_of()
|
||
if not isinstance(manager, ShieldConnectionManager):
|
||
return True
|
||
if not manager.protocol_active:
|
||
return True
|
||
return manager.is_available is not False
|
||
|
||
@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 with the POWER key when it reports off."""
|
||
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 with the POWER key when it reports on."""
|
||
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)
|