757 lines
30 KiB
Python
757 lines
30 KiB
Python
"""Config flow for Nvidia Shield integration."""
|
||
|
||
import contextlib
|
||
import logging
|
||
|
||
import voluptuous as vol
|
||
from homeassistant import config_entries
|
||
from homeassistant.core import callback
|
||
from homeassistant.helpers import device_registry as dr
|
||
from homeassistant.helpers import entity_registry as er
|
||
from homeassistant.helpers import selector
|
||
|
||
from .adb_session import ensure_adb_key
|
||
from .connection import ShieldConnectionManager
|
||
from .const import (
|
||
ATTR_MEDIA_DAG_INPUT_MAP,
|
||
CONF_ADB_COMMAND_ENABLED,
|
||
CONF_ADB_ENABLED,
|
||
CONF_ADB_HOST,
|
||
CONF_ADB_PORT,
|
||
CONF_ADB_STORE_APP_INFO,
|
||
CONF_ANDROIDTV_ENTITY,
|
||
CONF_APP_ICON_MAP,
|
||
CONF_APP_NAME_MAP,
|
||
CONF_CHANNEL_DIGIT_DELAY,
|
||
CONF_CHANNEL_SEND_ENTER,
|
||
CONF_JELLYFIN_ENTITY,
|
||
CONF_LOGCAT_EVENTS,
|
||
CONF_PAIRING_CODE,
|
||
CONF_POWER_ENTITY,
|
||
CONF_PROTOCOL_ENABLED,
|
||
CONF_PROTOCOL_HOST,
|
||
CONF_RECEIVER_ENTITY,
|
||
CONF_RECEIVER_INPUT,
|
||
DEFAULT_ADB_PORT,
|
||
DEFAULT_NAME,
|
||
DOMAIN,
|
||
)
|
||
from .helpers import discover_adb_host, protocol_cert_paths
|
||
|
||
_LOGGER = logging.getLogger(__name__)
|
||
|
||
|
||
def _get_receiver_inputs(hass, receiver_entity_id: str | None) -> list[str]:
|
||
"""Read available inputs from a receiver's media_dag_input_map attribute."""
|
||
if not receiver_entity_id:
|
||
return []
|
||
state = hass.states.get(receiver_entity_id)
|
||
if not state:
|
||
return []
|
||
input_map = state.attributes.get(ATTR_MEDIA_DAG_INPUT_MAP, {})
|
||
return list(input_map.keys()) if isinstance(input_map, dict) else []
|
||
|
||
|
||
def _unique_id_from_atv_device(hass, androidtv_entity: str) -> str | None:
|
||
"""A stable unique_id from the atv entity's device MAC (t29 G2).
|
||
|
||
The device registry holds CONNECTION_NETWORK_MAC for Android TV
|
||
entities configured with a host — far stronger than the renameable
|
||
entity_id. None (→ no unique_id at all, not an entity-derived
|
||
placeholder) when the entity is unregistered or carries no MAC.
|
||
"""
|
||
if not androidtv_entity:
|
||
return None
|
||
registry = er.async_get(hass)
|
||
entity = registry.entities.get(androidtv_entity)
|
||
device_id = getattr(entity, "device_id", None) if entity else None
|
||
device = (
|
||
dr.async_get(hass).async_get(device_id) if device_id else None
|
||
)
|
||
if device is None:
|
||
return None
|
||
if isinstance(dr, dict): # stub shape guard
|
||
return None
|
||
for connection in device.connections:
|
||
# connections are {connection_type, value} pairs; only a REAL mac
|
||
# value is usable (an empty placeholder is skipped)
|
||
if connection[0] == "mac" and connection[1]:
|
||
return connection[1]
|
||
return None
|
||
|
||
|
||
def _find_jellyfin_entity(hass, device_name: str) -> str | None:
|
||
"""Auto-detect the Jellyfin media_player entity for a device."""
|
||
registry = er.async_get(hass)
|
||
device_name_lower = device_name.lower()
|
||
candidates = []
|
||
|
||
for entry in registry.entities.values():
|
||
if entry.domain != "media_player" or entry.platform != "jellyfin":
|
||
continue
|
||
if entry.disabled_by is not None:
|
||
continue
|
||
|
||
eid_lower = entry.entity_id.lower()
|
||
name_lower = (entry.original_name or "").lower()
|
||
|
||
if device_name_lower in eid_lower or device_name_lower in name_lower:
|
||
state = hass.states.get(entry.entity_id)
|
||
if state and state.state != "unavailable":
|
||
candidates.append(entry.entity_id)
|
||
|
||
if candidates:
|
||
if len(candidates) > 1:
|
||
_LOGGER.warning(
|
||
"Multiple Jellyfin entities found for %s: %s. Using first.",
|
||
device_name, candidates,
|
||
)
|
||
return candidates[0]
|
||
return None
|
||
|
||
|
||
def _app_map_to_text(app_map: dict, icon_map: dict | None = None) -> str:
|
||
"""Serialize the apps map to one ``package=name[|icon]`` per line."""
|
||
icon_map = icon_map or {}
|
||
lines = []
|
||
for pkg, name in app_map.items():
|
||
icon = icon_map.get(pkg)
|
||
lines.append(f"{pkg}={name}|{icon}" if icon else f"{pkg}={name}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _app_map_from_text(
|
||
text: str, icons: dict[str, str] | None = None
|
||
) -> dict[str, str] | None:
|
||
"""Parse one ``package=name[|icon]`` per line; None on malformed input.
|
||
|
||
The optional second column (a picture URL or entity id) feeds
|
||
entity_picture through CONF_APP_ICON_MAP; single-column maps stay
|
||
valid. Malformed icon text (spaces) invalidates the row the same way
|
||
a malformed name does.
|
||
"""
|
||
icons = icons if icons is not None else {}
|
||
app_map: dict[str, str] = {}
|
||
for line in text.split("\n"):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
if "=" not in line:
|
||
return None
|
||
package, value = line.split("=", 1)
|
||
parts = value.strip().split("|", 1)
|
||
name = parts[0].strip()
|
||
icon = parts[1].strip() if len(parts) > 1 else ""
|
||
package = package.strip()
|
||
name = name.strip()
|
||
if not package or not name or " " in package:
|
||
return None
|
||
if icon:
|
||
if " " in icon:
|
||
return None
|
||
icons[package] = icon
|
||
app_map[package] = name
|
||
return app_map
|
||
|
||
|
||
def _user_schema_dict(sel, defaults: dict | None = None) -> dict:
|
||
"""User-step schema — the androidtv entity is optional (standalone).
|
||
|
||
``defaults`` pre-fills name/hosts from a zeroconf discovery event.
|
||
"""
|
||
defaults = defaults or {}
|
||
schema_dict = {
|
||
vol.Required("name", default=defaults.get("name", DEFAULT_NAME)): str,
|
||
vol.Optional(CONF_ANDROIDTV_ENTITY): sel["androidtv"],
|
||
vol.Optional(CONF_JELLYFIN_ENTITY): sel["media_player"],
|
||
vol.Optional(CONF_RECEIVER_ENTITY): sel["receiver"],
|
||
}
|
||
|
||
# If there are any receivers with DAG input maps, show a text field
|
||
# for receiver input (will be a dropdown in options once receiver is set)
|
||
schema_dict[vol.Optional(CONF_RECEIVER_INPUT)] = str
|
||
|
||
schema_dict[vol.Optional(CONF_POWER_ENTITY)] = sel["power"]
|
||
|
||
schema_dict[vol.Optional(CONF_ADB_ENABLED)] = bool
|
||
schema_dict[vol.Optional(CONF_ADB_STORE_APP_INFO)] = bool
|
||
schema_dict[vol.Optional(CONF_ADB_HOST)] = str
|
||
schema_dict[vol.Optional(CONF_ADB_PORT)] = int
|
||
schema_dict[vol.Optional(CONF_PROTOCOL_ENABLED)] = bool
|
||
schema_dict[vol.Optional(CONF_PROTOCOL_HOST)] = str
|
||
if "host" in defaults:
|
||
schema_dict[vol.Optional(CONF_ADB_HOST)] = str
|
||
schema_dict[vol.Optional(CONF_PROTOCOL_HOST)] = str
|
||
return schema_dict
|
||
|
||
|
||
def _build_selectors():
|
||
"""Build reusable selector instances."""
|
||
return {
|
||
"androidtv": selector.EntitySelector(
|
||
selector.EntitySelectorConfig(domain="media_player", device_class="tv")
|
||
),
|
||
"media_player": selector.EntitySelector(
|
||
selector.EntitySelectorConfig(domain="media_player")
|
||
),
|
||
"receiver": selector.EntitySelector(
|
||
selector.EntitySelectorConfig(domain="media_player", device_class="receiver")
|
||
),
|
||
"power": selector.EntitySelector(
|
||
selector.EntitySelectorConfig(domain=["switch", "binary_sensor"])
|
||
),
|
||
}
|
||
|
||
|
||
class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||
"""Handle a config flow for Nvidia Shield."""
|
||
|
||
VERSION = 2
|
||
|
||
def __init__(self):
|
||
self._user_data = {}
|
||
self._protocol_client = None
|
||
self._reauth_entry = None
|
||
# Zeroconf discovery state (P4-1)
|
||
self._discovered_host: str | None = None
|
||
self._discovered_name: str | None = None
|
||
|
||
async def async_step_user(self, user_input=None):
|
||
"""Step 1: Basic setup.
|
||
|
||
Standalone setup (Phase 3): any one of the three state sources is
|
||
enough — the Android TV Remote entity, the Nvidia Remote Protocol
|
||
(pairing), or ADB media-session mode (CONF_ADB_ENABLED). The
|
||
``no_state_source`` error fires only when none of them is present.
|
||
"""
|
||
errors = {}
|
||
|
||
if user_input is not None:
|
||
androidtv_entity = user_input.get(CONF_ANDROIDTV_ENTITY)
|
||
|
||
has_source = (
|
||
bool(androidtv_entity)
|
||
or bool(user_input.get(CONF_PROTOCOL_ENABLED))
|
||
or bool(user_input.get(CONF_ADB_ENABLED))
|
||
)
|
||
if not has_source:
|
||
errors["base"] = "no_state_source"
|
||
sel = _build_selectors()
|
||
return self.async_show_form(
|
||
step_id="user",
|
||
data_schema=vol.Schema(_user_schema_dict(sel)),
|
||
errors=errors,
|
||
)
|
||
|
||
# t29 G2: the androidtv entity_id is a renameable reference —
|
||
# never a unique_id. A sqlite/registry MAC is preferred (the
|
||
# atv entity's device holds CONNECTION_NETWORK_MAC); without a
|
||
# resolvable MAC the entry gets NO unique_id (the same
|
||
# uniqueness-less semantics _async_handle_discovery_without_
|
||
# unique_id applies to MAC-less discovery) — renames can never
|
||
# break uniqueness.
|
||
fallback_unique_id = _unique_id_from_atv_device(
|
||
self.hass, androidtv_entity
|
||
)
|
||
if fallback_unique_id:
|
||
await self.async_set_unique_id(fallback_unique_id)
|
||
self._abort_if_unique_id_configured()
|
||
|
||
# Auto-detect Jellyfin if not manually specified
|
||
jellyfin = user_input.get(CONF_JELLYFIN_ENTITY)
|
||
if not jellyfin:
|
||
device_name = user_input.get("name", DEFAULT_NAME)
|
||
atv_state = (
|
||
self.hass.states.get(androidtv_entity)
|
||
if androidtv_entity
|
||
else None
|
||
)
|
||
atv_name = (atv_state.attributes.get("friendly_name", "") if atv_state else "")
|
||
|
||
for name in [device_name, atv_name]:
|
||
if not name:
|
||
continue
|
||
found = _find_jellyfin_entity(self.hass, name)
|
||
if found:
|
||
jellyfin = found
|
||
_LOGGER.info("Auto-detected Jellyfin entity: %s", found)
|
||
break
|
||
|
||
self._user_data = {
|
||
"name": user_input.get("name", DEFAULT_NAME),
|
||
CONF_ANDROIDTV_ENTITY: androidtv_entity or "",
|
||
CONF_JELLYFIN_ENTITY: jellyfin or "",
|
||
CONF_RECEIVER_ENTITY: user_input.get(CONF_RECEIVER_ENTITY, ""),
|
||
CONF_RECEIVER_INPUT: user_input.get(CONF_RECEIVER_INPUT, ""),
|
||
CONF_POWER_ENTITY: user_input.get(CONF_POWER_ENTITY, ""),
|
||
CONF_ADB_ENABLED: user_input.get(CONF_ADB_ENABLED, False),
|
||
CONF_ADB_STORE_APP_INFO: user_input.get(CONF_ADB_STORE_APP_INFO, True),
|
||
CONF_ADB_HOST: user_input.get(CONF_ADB_HOST, ""),
|
||
CONF_ADB_PORT: user_input.get(CONF_ADB_PORT, DEFAULT_ADB_PORT),
|
||
CONF_PROTOCOL_ENABLED: user_input.get(CONF_PROTOCOL_ENABLED, False),
|
||
CONF_PROTOCOL_HOST: user_input.get(CONF_PROTOCOL_HOST, ""),
|
||
}
|
||
|
||
# If ADB is enabled, go to pairing step
|
||
if self._user_data[CONF_ADB_ENABLED]:
|
||
return await self.async_step_adb_pair()
|
||
|
||
if self._user_data[CONF_PROTOCOL_ENABLED]:
|
||
return await self.async_step_protocol_pair()
|
||
|
||
return self._create_entry()
|
||
|
||
|
||
sel = _build_selectors()
|
||
|
||
defaults = {}
|
||
if self._discovered_host:
|
||
defaults = {
|
||
"name": self._discovered_name or DEFAULT_NAME,
|
||
"host": self._discovered_host,
|
||
}
|
||
schema_dict = _user_schema_dict(sel, defaults)
|
||
|
||
return self.async_show_form(
|
||
step_id="user",
|
||
data_schema=vol.Schema(schema_dict),
|
||
errors=errors,
|
||
)
|
||
|
||
async def async_step_zeroconf(self, discovery_info):
|
||
"""Zeroconf discovery of the Android TV remote service (P4-1).
|
||
|
||
Core androidtv_remote pattern, simplified: the discovery address
|
||
becomes the pre-filled protocol/ADB host on the user step; a known
|
||
MAC (from the service properties) aborts duplicate setup and
|
||
updates the existing entry's hosts on an IP change.
|
||
"""
|
||
host = str(
|
||
getattr(discovery_info, "ip_address", None)
|
||
or getattr(discovery_info, "host", None)
|
||
or ""
|
||
).strip()
|
||
if not host:
|
||
return self.async_abort(reason="zeroconf_no_host")
|
||
|
||
self._discovered_host = host
|
||
props = getattr(discovery_info, "properties", None) or {}
|
||
self._discovered_name = props.get("name")
|
||
|
||
mac = props.get("mac") or props.get("unique_id")
|
||
if mac:
|
||
await self.async_set_unique_id(mac)
|
||
for entry in self._async_current_entries():
|
||
if getattr(entry, "unique_id", None) != mac:
|
||
continue
|
||
# Known device — per-host updates (t21 F2): each host is
|
||
# tracked independently against the discovery address.
|
||
# - a blank (auto-detect) host self-heals from the address;
|
||
# - a stale protocol host is the IP change and is updated;
|
||
# - a deliberately-different ADB host (an explicit override
|
||
# that differs from both the address and the stored
|
||
# protocol host) is never clobbered.
|
||
protocol_host = entry.options.get(CONF_PROTOCOL_HOST, "")
|
||
adb_host = entry.options.get(CONF_ADB_HOST, "")
|
||
options = dict(entry.options)
|
||
if protocol_host != host:
|
||
options[CONF_PROTOCOL_HOST] = host
|
||
if not adb_host or adb_host == protocol_host:
|
||
# blank (self-heal) or auto-synced alongside the protocol
|
||
# host — safe to follow the device address on discovery
|
||
options[CONF_ADB_HOST] = host
|
||
if options != entry.options:
|
||
# t29 G3a: the deprecated update+schedule-reload combo is
|
||
# gone (2026.12 error) — the entry updates ONLY here; the
|
||
# self-heal rides on the t32 F1 poller factory, which
|
||
# re-reads entry.options at call time (no reload).
|
||
self.hass.config_entries.async_update_entry(
|
||
entry, options=options
|
||
)
|
||
return self.async_abort(reason="already_configured")
|
||
self._abort_if_unique_id_configured()
|
||
else:
|
||
# t32 F2 (review rider): MAC-less discovery cannot carry a
|
||
# unique_id — HA's helper records the uniqueness-less discovery
|
||
# (the same discovery context can't stack duplicates) and the
|
||
# host guard keeps a second discovery of an already-configured
|
||
# host from stacking another prefill.
|
||
self._async_handle_discovery_without_unique_id()
|
||
for entry in self._async_current_entries():
|
||
existing_host = entry.options.get(
|
||
CONF_PROTOCOL_HOST
|
||
) or entry.options.get(CONF_ADB_HOST)
|
||
if existing_host and existing_host == host:
|
||
return self.async_abort(reason="already_configured")
|
||
|
||
# First-time setup: pre-filled user step
|
||
return await self.async_step_user()
|
||
|
||
def _create_entry(self):
|
||
"""Create the config entry from stored user data."""
|
||
d = self._user_data
|
||
entry_data = {}
|
||
if d.get(CONF_ANDROIDTV_ENTITY):
|
||
entry_data[CONF_ANDROIDTV_ENTITY] = d[CONF_ANDROIDTV_ENTITY]
|
||
return self.async_create_entry(
|
||
title=d["name"],
|
||
data=entry_data,
|
||
options={
|
||
CONF_JELLYFIN_ENTITY: d.get(CONF_JELLYFIN_ENTITY, ""),
|
||
CONF_RECEIVER_ENTITY: d.get(CONF_RECEIVER_ENTITY, ""),
|
||
CONF_RECEIVER_INPUT: d.get(CONF_RECEIVER_INPUT, ""),
|
||
CONF_POWER_ENTITY: d.get(CONF_POWER_ENTITY, ""),
|
||
CONF_ADB_ENABLED: d.get(CONF_ADB_ENABLED, False),
|
||
CONF_ADB_STORE_APP_INFO: d.get(CONF_ADB_STORE_APP_INFO, True),
|
||
CONF_ADB_HOST: d.get(CONF_ADB_HOST, ""),
|
||
CONF_ADB_PORT: d.get(CONF_ADB_PORT, DEFAULT_ADB_PORT),
|
||
CONF_PROTOCOL_ENABLED: d.get(CONF_PROTOCOL_ENABLED, False),
|
||
CONF_PROTOCOL_HOST: d.get(CONF_PROTOCOL_HOST, ""),
|
||
},
|
||
)
|
||
|
||
async def _async_after_pairing(self):
|
||
"""Route after the pairing steps: protocol next, else create the entry."""
|
||
if self._user_data.get(CONF_PROTOCOL_ENABLED):
|
||
return await self.async_step_protocol_pair()
|
||
return self._create_entry()
|
||
|
||
def _protocol_host(self) -> str | None:
|
||
"""Protocol host: explicit override, else the ADB host fallback."""
|
||
return self._user_data.get(CONF_PROTOCOL_HOST) or self._resolve_adb_host()
|
||
|
||
async def _start_protocol_pairing(self):
|
||
"""Build a throwaway protocol client and make the Shield show its PIN.
|
||
|
||
Certificates live under ``.storage/<domain>/`` (core androidtv_remote
|
||
helpers.py pattern) and are shared between pairing and runtime.
|
||
"""
|
||
host = self._protocol_host()
|
||
if not host:
|
||
return None
|
||
certfile, keyfile = await self.hass.async_add_executor_job(
|
||
protocol_cert_paths, self.hass
|
||
)
|
||
client = ShieldConnectionManager.build_protocol_client(
|
||
host,
|
||
client_name=self._user_data.get("name", DEFAULT_NAME) or DEFAULT_NAME,
|
||
certfile=certfile,
|
||
keyfile=keyfile,
|
||
)
|
||
await client.async_generate_cert_if_missing()
|
||
await client.async_start_pairing()
|
||
return client
|
||
|
||
async def async_step_protocol_pair(self, user_input=None):
|
||
"""Pair with the remote protocol — the PIN is displayed on the Shield."""
|
||
errors = {}
|
||
if user_input is None:
|
||
# First visit: proactively start pairing so the PIN shows on the
|
||
# Shield while the form is open.
|
||
client = await self._start_protocol_pairing()
|
||
if client is None:
|
||
errors["base"] = "protocol_no_host"
|
||
else:
|
||
self._protocol_client = client
|
||
else:
|
||
client = self._protocol_client
|
||
code = user_input.get(CONF_PAIRING_CODE, "")
|
||
if client is None or not code:
|
||
errors["base"] = "protocol_invalid_code"
|
||
else:
|
||
try:
|
||
await client.async_finish_pairing(code)
|
||
except Exception:
|
||
errors["base"] = "protocol_pair_failed"
|
||
else:
|
||
try:
|
||
_name, mac = await client.async_get_name_and_mac()
|
||
except Exception:
|
||
mac = None
|
||
with contextlib.suppress(Exception):
|
||
client.disconnect()
|
||
self._protocol_client = None
|
||
if mac:
|
||
await self.async_set_unique_id(mac)
|
||
self._abort_if_unique_id_configured()
|
||
if self._reauth_entry is not None:
|
||
return self._finish_reauth()
|
||
return self._create_entry()
|
||
return self.async_show_form(
|
||
step_id="protocol_pair",
|
||
data_schema=vol.Schema({
|
||
vol.Required(CONF_PAIRING_CODE): str,
|
||
}),
|
||
errors=errors,
|
||
)
|
||
|
||
async def async_step_import(self, data):
|
||
"""Seed an interactive flow from YAML (t30 U1, seed-only ruling).
|
||
|
||
The import validates the block and routes into the standard user
|
||
flow — pairing stays interactive, exactly one visit. Duplicate
|
||
guards: the block's host/MAC-virtual unique_id aborts when an
|
||
entry already holds it, and a UI-configured entry with the same
|
||
protocol/adb host is never re-seeded.
|
||
"""
|
||
block = dict(data or {})
|
||
name = block.get("name") or DEFAULT_NAME
|
||
seed_unique_id = block.get("unique_id")
|
||
if not seed_unique_id:
|
||
host = block.get(CONF_ADB_HOST) or block.get(CONF_PROTOCOL_HOST)
|
||
seed_unique_id = f"yaml:{name}:{host or 'nohost'}"
|
||
|
||
# duplicate guards
|
||
yhost = block.get(CONF_ADB_HOST) or block.get(CONF_PROTOCOL_HOST)
|
||
for existing in self._async_current_entries():
|
||
ehost = existing.options.get(CONF_PROTOCOL_HOST) or existing.options.get(
|
||
CONF_ADB_HOST
|
||
)
|
||
if existing.unique_id == seed_unique_id or (
|
||
yhost and ehost == yhost
|
||
):
|
||
_LOGGER.info(
|
||
"nvidia_shield YAML seed for %r matches an existing "
|
||
"entry — not re-seeded",
|
||
name,
|
||
)
|
||
return self.async_abort(reason="already_configured")
|
||
await self.async_set_unique_id(seed_unique_id)
|
||
self._abort_if_unique_id_configured()
|
||
|
||
# route into the standard user flow (single interactive visit);
|
||
# t32 F4: the prefill honors the block shape — ADB only when the
|
||
# block actually carries an adb host.
|
||
self._user_data = {
|
||
"name": name,
|
||
CONF_ADB_ENABLED: bool(block.get(CONF_ADB_HOST)),
|
||
CONF_ADB_HOST: block.get(CONF_ADB_HOST, ""),
|
||
CONF_ADB_PORT: block.get(CONF_ADB_PORT, DEFAULT_ADB_PORT),
|
||
# protocol prefill hints ONLY (pairing stays interactive)
|
||
CONF_PROTOCOL_HOST: block.get(CONF_PROTOCOL_HOST, ""),
|
||
}
|
||
self._discovered_host = self._user_data[CONF_PROTOCOL_HOST] or self._user_data[
|
||
CONF_ADB_HOST
|
||
]
|
||
return await self.async_step_user(None)
|
||
|
||
async def async_step_reauth(self, entry_data):
|
||
"""Device hard reset invalidated our certificates (review trigger)."""
|
||
self._reauth_entry = self.hass.config_entries.async_get_entry(
|
||
self.context.get("entry_id")
|
||
)
|
||
if self._reauth_entry is None:
|
||
return self.async_abort(reason="reauth_entry_missing")
|
||
return self.async_show_form(
|
||
step_id="reauth_confirm", data_schema=vol.Schema({})
|
||
)
|
||
|
||
async def async_step_reauth_confirm(self, user_input=None):
|
||
"""Re-pair after a device hard reset, then reload the entry."""
|
||
if user_input is not None:
|
||
entry = self._reauth_entry
|
||
self._user_data = {
|
||
"name": entry.title,
|
||
CONF_PROTOCOL_ENABLED: True,
|
||
CONF_PROTOCOL_HOST: entry.options.get(CONF_PROTOCOL_HOST, ""),
|
||
}
|
||
return await self.async_step_protocol_pair() # first visit re-pairs
|
||
return self.async_show_form(
|
||
step_id="reauth_confirm", data_schema=vol.Schema({})
|
||
)
|
||
|
||
def _finish_reauth(self):
|
||
"""Reauth pairing done: update + reload + abort via the current API (G3a).
|
||
|
||
The deprecated update-entry + schedule-reload boomerang listener is
|
||
gone (2026.12 treats it as an error); the current config-entries
|
||
contract is async_update_reload_and_abort with _get_reauth_entry.
|
||
"""
|
||
entry = self._reauth_entry or self._get_reauth_entry()
|
||
self.async_update_reload_and_abort(
|
||
entry,
|
||
options_updates={
|
||
**entry.options,
|
||
CONF_PROTOCOL_ENABLED: True,
|
||
CONF_PROTOCOL_HOST: self._user_data.get(CONF_PROTOCOL_HOST, ""),
|
||
},
|
||
)
|
||
self._reauth_entry = None
|
||
return self.async_abort(reason="reauth_successful")
|
||
|
||
def _resolve_adb_host(self) -> str | None:
|
||
"""Resolve ADB host from explicit config or the Android TV entity."""
|
||
return self._user_data.get(CONF_ADB_HOST) or discover_adb_host(
|
||
self.hass, self._user_data.get(CONF_ANDROIDTV_ENTITY)
|
||
)
|
||
|
||
async def async_step_adb_pair(self, user_input=None):
|
||
"""Step 2: ADB pairing. User clicks Submit after approving on the Shield."""
|
||
errors = {}
|
||
|
||
if user_input is not None:
|
||
host = self._resolve_adb_host()
|
||
port = self._user_data.get(CONF_ADB_PORT, DEFAULT_ADB_PORT)
|
||
|
||
if not host:
|
||
errors["base"] = "adb_no_host"
|
||
else:
|
||
key_path = await self.hass.async_add_executor_job(
|
||
ensure_adb_key, self.hass.config.config_dir
|
||
)
|
||
|
||
connected = await ShieldConnectionManager.test_connect(
|
||
host, port, key_path
|
||
)
|
||
if connected:
|
||
return await self._async_after_pairing()
|
||
else:
|
||
errors["base"] = "adb_failed"
|
||
|
||
return self.async_show_form(
|
||
step_id="adb_pair",
|
||
data_schema=vol.Schema({}),
|
||
errors=errors,
|
||
)
|
||
|
||
@staticmethod
|
||
@callback
|
||
def async_get_options_flow(config_entry):
|
||
return NvidiaShieldOptionsFlow()
|
||
|
||
|
||
class NvidiaShieldOptionsFlow(config_entries.OptionsFlowWithReload):
|
||
"""Handle options for Nvidia Shield — single unified page (t29 G3b).
|
||
|
||
The constructor takes NO arguments: HA assigns ``self.config_entry``
|
||
on instantiation (the pre-2026 "OptionsFlow(config_entry)" pattern is
|
||
deprecated). Suggested values come through
|
||
``add_suggested_values_to_schema``, and saving reloads via the
|
||
OptionsFlowWithReload base — no listener boomerangs.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._options: dict | None = None
|
||
|
||
@property
|
||
def _options_dict(self) -> dict:
|
||
if self._options is None:
|
||
if self.config_entry is not None:
|
||
self._options = dict(self.config_entry.options)
|
||
else:
|
||
self._options = {}
|
||
return self._options
|
||
|
||
async def async_step_init(self, user_input=None):
|
||
"""Single-page options."""
|
||
errors = {}
|
||
if user_input is not None:
|
||
# P5-2: validate the digit delay (0.05–1.0 s, §7 window)
|
||
delay = user_input.pop(CONF_CHANNEL_DIGIT_DELAY, None)
|
||
if delay is not None:
|
||
try:
|
||
delay = float(delay)
|
||
except (TypeError, ValueError):
|
||
errors["base"] = "invalid_digit_delay"
|
||
delay = None
|
||
if delay is not None and not (0.05 <= delay <= 1.0):
|
||
errors["base"] = "invalid_digit_delay"
|
||
app_map_text = user_input.pop(CONF_APP_NAME_MAP, None)
|
||
icons: dict[str, str] = {}
|
||
if app_map_text is None:
|
||
app_map = self._options_dict.get(CONF_APP_NAME_MAP, {})
|
||
elif app_map_text.strip() == "":
|
||
app_map = {}
|
||
else:
|
||
app_map = _app_map_from_text(app_map_text, icons)
|
||
if app_map is None:
|
||
errors["base"] = "invalid_app_map"
|
||
if not errors:
|
||
if delay is not None:
|
||
self._options_dict[CONF_CHANNEL_DIGIT_DELAY] = delay
|
||
self._options_dict.update(user_input)
|
||
self._options_dict[CONF_APP_NAME_MAP] = app_map
|
||
# P5-2: the icon column — the editor is the source of
|
||
# truth, so a save without icons clears them (and a
|
||
# single-column map stays valid).
|
||
self._options_dict[CONF_APP_ICON_MAP] = icons
|
||
return self.async_create_entry(title="", data=self._options_dict)
|
||
|
||
sel = _build_selectors()
|
||
schema_dict = {}
|
||
|
||
# Apps name/icon map editor (P4-1, core androidtv_remote options
|
||
# pattern, simplified): one package=name[|icon] per line (P5-2 adds
|
||
# the optional icon column).
|
||
schema_dict[vol.Optional(CONF_APP_NAME_MAP)] = str
|
||
|
||
# Jellyfin
|
||
jf = self._options_dict.get(CONF_JELLYFIN_ENTITY)
|
||
jf_key = vol.Optional(CONF_JELLYFIN_ENTITY)
|
||
if jf:
|
||
jf_key = vol.Optional(CONF_JELLYFIN_ENTITY)
|
||
schema_dict[jf_key] = sel["media_player"]
|
||
|
||
# Receiver
|
||
recv = self._options_dict.get(CONF_RECEIVER_ENTITY)
|
||
recv_key = vol.Optional(CONF_RECEIVER_ENTITY)
|
||
if recv:
|
||
recv_key = vol.Optional(CONF_RECEIVER_ENTITY)
|
||
schema_dict[recv_key] = sel["receiver"]
|
||
|
||
# P5-2: channel digit behavior (the user-validated §7 flow)
|
||
schema_dict[vol.Optional(CONF_CHANNEL_DIGIT_DELAY)] = str
|
||
schema_dict[vol.Optional(CONF_CHANNEL_SEND_ENTER)] = bool
|
||
|
||
# P5-3: the live events push is opt-in (default off; the user
|
||
# tests stability on their device and reports)
|
||
schema_dict[vol.Optional(CONF_LOGCAT_EVENTS)] = bool
|
||
|
||
# P4-2a: the raw adb_command service is opt-in (default off)
|
||
schema_dict[vol.Optional(CONF_ADB_COMMAND_ENABLED)] = bool
|
||
|
||
# Receiver input — dropdown from DAG if receiver is set, text otherwise
|
||
current_input = self._options_dict.get(CONF_RECEIVER_INPUT, "")
|
||
receiver = self._options_dict.get(CONF_RECEIVER_ENTITY)
|
||
inputs = _get_receiver_inputs(self.hass, receiver)
|
||
|
||
if inputs:
|
||
input_sel = selector.SelectSelector(
|
||
selector.SelectSelectorConfig(
|
||
options=inputs,
|
||
mode=selector.SelectSelectorMode.DROPDOWN,
|
||
custom_value=True,
|
||
)
|
||
)
|
||
input_key = vol.Optional(CONF_RECEIVER_INPUT)
|
||
if current_input:
|
||
input_key = vol.Optional(CONF_RECEIVER_INPUT)
|
||
schema_dict[input_key] = input_sel
|
||
else:
|
||
schema_dict[vol.Optional(CONF_RECEIVER_INPUT)] = str
|
||
|
||
# Power
|
||
pwr = self._options_dict.get(CONF_POWER_ENTITY)
|
||
pwr_key = vol.Optional(CONF_POWER_ENTITY)
|
||
if pwr:
|
||
pwr_key = vol.Optional(CONF_POWER_ENTITY)
|
||
schema_dict[pwr_key] = sel["power"]
|
||
|
||
# ADB options
|
||
schema_dict[vol.Optional(CONF_ADB_ENABLED)] = bool
|
||
schema_dict[vol.Optional(CONF_ADB_STORE_APP_INFO)] = bool
|
||
|
||
# ADB host/port (optional overrides, shown when ADB is enabled)
|
||
schema_dict[vol.Optional(CONF_ADB_HOST)] = str
|
||
schema_dict[vol.Optional(CONF_ADB_PORT)] = int
|
||
|
||
# t29 G3b: suggested values come from the entry options via the
|
||
# documented helper — no per-key `default=` in the schema build.
|
||
return self.async_show_form(
|
||
step_id="init",
|
||
data_schema=self.add_suggested_values_to_schema(
|
||
vol.Schema(schema_dict), self._options_dict
|
||
),
|
||
errors=errors,
|
||
)
|