ha-nvidia-shield/custom_components/nvidia_shield/remote.py

"""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;
- ``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. The
          ERROR is the failure the CURRENT state has not recovered from (t20/O1):
          it is cleared as soon as a session is proven again — the next connect or
          the transport's own availability push — so it can never read as a
          problem beside ``protocol_connected: true``, and a failure that happened
          AFTER the session came up is kept because that one is still current.
          ``protocol_failures`` keeps counting, so a recovery stays visible as
          history instead of as a stale message.
        * ``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:<id>`` 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)