"""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 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_PORT, CONF_ADB_STORE_APP_INFO, CONF_APP_ICON_MAP, CONF_APP_NAME_MAP, CONF_CHANNEL_DIGIT_DELAY, CONF_CHANNEL_SEND_ENTER, CONF_COMMAND_SCRIPTS, CONF_HOST, CONF_JELLYFIN_ENTITY, CONF_LOGCAT_EVENTS, CONF_PAIRING_CODE, CONF_POWER_ENTITY, CONF_PROTOCOL_ENABLED, CONF_RECEIVER_ENTITY, CONF_RECEIVER_INPUT, CONF_REMOTE_COMMAND_SCRIPT, CONF_SKIP_PAIRING, DEFAULT_ADB_PORT, DEFAULT_NAME, DOMAIN, ENTRY_VERSION, SCRIPT_COMMANDS, SCRIPT_FIELD_PREFIX, command_scripts, ) from .helpers import entry_host, protocol_cert_paths, seed_adb_enabled, seed_host _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 _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 _apps_table_text( stored_map: dict, stored_icons: dict, discovered: list[dict] ) -> str: """The Editable apps table: stored rows first, then discovered apps (t57). Every app the Shield reports gets a row, so the page opens on the REAL app list instead of the four curated entries ("it doesn't find all the apps"). The stored rows keep their position, name and icon — the user's edit is never overwritten by a discovery — and a discovered app the map does not mention is appended with the device label (or its package id when the label fetch did not answer) and its heuristic icon. The icon column is emitted only for a stored icon, because the discovered icon is an MDI NAME (``mdi:netflix``) and the icon column's documented values are a picture URL or a ``camera.*`` entity that feeds ``entity_picture``. Writing an ``mdi:`` string there would put a value the media player cannot use into the user's own map — the icon is offered in the table's description instead, and in the installed-apps sensor's attributes. """ lines: list[str] = [] seen: set[str] = set() for package, name in (stored_map or {}).items(): seen.add(package) icon = (stored_icons or {}).get(package) lines.append(f"{package}={name}|{icon}" if icon else f"{package}={name}") for app in discovered or []: package = app.get("package") if not package or package in seen: continue seen.add(package) name = app.get("name") or package lines.append(f"{package}={name}") return "\n".join(lines) def _options_field_values(options: dict | None) -> dict: """Entry options as FRONTEND-SERIALIZABLE field values (t56). The bug this exists to kill: the apps map is stored as a DICT, and the options page handed that dict straight to ``add_suggested_values_to_schema`` as the suggested value of a TEXT field. Real Home Assistant serializes the suggested value verbatim into the form the frontend receives (verified on a real install: ``{"suggested_value": {"ar.tvplayer.tv": "TiviMate"}}`` for a ``type: string`` field), and the browser renders a non-string as ``[object Object]``. So the suggestion layer gets only what a text field can display: the apps map (plus its icon column) serialized to the documented ``package=name[|icon]`` lines, and every other option value unchanged. """ options = dict(options or {}) app_map = options.pop(CONF_APP_NAME_MAP, None) icon_map = options.pop(CONF_APP_ICON_MAP, None) if isinstance(app_map, dict): options[CONF_APP_NAME_MAP] = _app_map_to_text( app_map, icon_map if isinstance(icon_map, dict) else None ) return options def _app_map_submission(value) -> tuple[dict, dict] | None: """Parse the apps-table field into ``(app_map, icon_map)``; None if invalid. A blank field is the documented way to clear the map; anything else must parse as ``package=name[|icon]`` lines or the whole submission is rejected (never half-stored). """ if value is None: return {}, {} if not str(value).strip(): return {}, {} icons: dict[str, str] = {} app_map = _app_map_from_text(str(value), icons) if app_map is None: return None return app_map, icons def _host_field_key(field: str, host: str): """The host-field KEY, carrying the known address as its default. t33 F-A: the default MUST ride on this key. Re-declaring a bare ``vol.Optional(field)`` afterwards is a no-op that silently drops it — voluptuous keys compare by field name, so the first (defaultless) key object survives and the form renders blank. """ if host: return vol.Optional(field, default=host) return vol.Optional(field) def _user_schema_dict(sel, defaults: dict | None = None) -> dict: """User-step schema — every field belongs to THIS integration (t41). ``defaults`` pre-fills the name and the ONE address (t41) from a zeroconf discovery event or a YAML seed hint; the ADB port is pre-filled with the standard 5555 so it reads as an advanced setting rather than required knowledge (t33 F-D). """ defaults = defaults or {} host = str(defaults.get("host") or "") port = defaults.get("port") if not isinstance(port, int): port = DEFAULT_ADB_PORT schema_dict = { vol.Required("name", default=defaults.get("name", DEFAULT_NAME)): str, 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"] # t41: both transports are OURS and both are ON by default — the user opts # out (or skips pairing) rather than opting in. The rendered checkboxes must # agree with the code's fallbacks in async_step_user / _create_entry, so all # three carry True (t33 F-A: a default the schema cannot see is a bug). schema_dict[vol.Optional(CONF_ADB_ENABLED, default=True)] = bool schema_dict[vol.Optional(CONF_ADB_STORE_APP_INFO, default=True)] = bool schema_dict[_host_field_key(CONF_HOST, host)] = str schema_dict[vol.Optional(CONF_ADB_PORT, default=port)] = int schema_dict[vol.Optional(CONF_PROTOCOL_ENABLED, default=True)] = bool return schema_dict class _ClearingEntitySelector(selector.EntitySelector): """An entity selector that accepts a CLEARED field (t47 F1). Clearing an optional entity field used to depend on the frontend OMITTING the key. It does not have to: a submission can carry ``''`` or ``None``, and real Home Assistant's EntitySelector rejects both — the reviewer's real-HA run produced "Schema validation failed at script_turn_on" from an explicit clear. That made clearing look like a broken form rather than a choice, on every entity field of this page (the script selectors are the newest, not the only ones). Returning the empty value unchanged lets the flow read it as "no selection", while a real entity id still goes through the parent's validation. The class stays an EntitySelector, so the frontend still renders an entity picker. """ def __call__(self, data): if data is None or (isinstance(data, str) and not data.strip()): return "" return super().__call__(data) def _build_selectors(): """Build reusable selector instances.""" return { "media_player": _ClearingEntitySelector( selector.EntitySelectorConfig(domain="media_player") ), # t56: a receiver is a media_player (an AVR/soundbar) OR a plain switch # (a soundbar/speaker that only knows on and off). The device_class # constraint belongs to the media_player half only, so a switch is not # filtered out by it — and the switch behaviour itself (turning it on # during playback) lives in the media_player task that follows. "receiver": _ClearingEntitySelector( selector.EntitySelectorConfig(domain=["media_player", "switch"]) ), "power": _ClearingEntitySelector( selector.EntitySelectorConfig(domain=["switch", "binary_sensor"]) ), # t44: a script entity per command override "script": _ClearingEntitySelector( selector.EntitySelectorConfig(domain="script") ), } class NvidiaShieldConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): """Handle a config flow for Nvidia Shield.""" VERSION = ENTRY_VERSION def __init__(self): self._user_data = {} self._protocol_client = None self._reauth_entry = None # audit G5 / t35 F1: the entry a "Reconfigure" visit targets self._reconfigure_entry = None # Zeroconf discovery state (P4-1) self._discovered_host: str | None = None self._discovered_name: str | None = None # YAML-seed prefill hint (t33 F-A): feeds the same user-step prefill as # a discovery, without claiming the address was discovered. self._seed_host: str | None = None async def async_step_user(self, user_input=None): """Step 1: Basic setup. We own the whole stack (t41), so the two state sources are ours: the Nvidia Remote Protocol (pairing) and ADB media-session mode (CONF_ADB_ENABLED). The ``no_state_source`` error fires only when neither is enabled. """ errors = {} if user_input is not None: has_source = bool(user_input.get(CONF_PROTOCOL_ENABLED)) or bool( user_input.get(CONF_ADB_ENABLED) ) if not has_source: errors["base"] = "no_state_source" return self._show_user_form(errors) # t41: nothing here derives a unique_id from another integration. # The entry's identity is the protocol MAC learned during pairing # (`async_set_unique_id(mac)` in the pairing step) or the config # entry itself — never a renameable entity id (t29 G2's # atv-device derivation went with the entity path). # Auto-detect Jellyfin from the names setup already knows: the # entry name the user typed, or the device name from discovery. jellyfin = user_input.get(CONF_JELLYFIN_ENTITY) if not jellyfin: for name in (user_input.get("name", DEFAULT_NAME), self._discovered_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_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, True), CONF_ADB_STORE_APP_INFO: user_input.get(CONF_ADB_STORE_APP_INFO, True), CONF_HOST: str(user_input.get(CONF_HOST) or "").strip(), CONF_ADB_PORT: user_input.get(CONF_ADB_PORT, DEFAULT_ADB_PORT), CONF_PROTOCOL_ENABLED: user_input.get(CONF_PROTOCOL_ENABLED, True), } # t56: the first form stays exactly one screen (so nothing it used # to accept moves or disappears), and the flow then becomes a MENU # of sub-screens — Connection / Receiver / the pairing steps / # Finish — instead of a single long page or a hidden sequence. return await self.async_step_menu() return self._show_user_form(errors) def _user_defaults(self) -> dict: """Defaults for the user step: whatever setup already knows. A zeroconf discovery (or a YAML seed hint) supplies the address; the port keeps a seed's hint instead of being flattened to the standard value. This is the ONLY place the known address enters the form — the schema keys carry these values as their defaults (t33 F-A). """ defaults = { "name": self._discovered_name or self._user_data.get("name") or DEFAULT_NAME, } host = self._discovered_host or self._seed_host if host: defaults["host"] = host port = self._user_data.get(CONF_ADB_PORT) if isinstance(port, int): defaults["port"] = port return defaults def _show_user_form(self, errors=None): """Render the user step with the known address actually applied. t33 F-A2: a discovery-driven form says so in its description, so a pre-filled host can never be misread as "unknown". """ note = "" if self._discovered_host: note = ( f"Discovered Shield at {self._discovered_host} — its address " "is already filled in below. " ) return self.async_show_form( step_id="user", data_schema=vol.Schema( _user_schema_dict(_build_selectors(), self._user_defaults()) ), errors=errors or {}, description_placeholders={"discovery_note": note}, ) 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 — IP change (t21 F2, collapsed by t41): there is # ONE address field now, so the discovery address simply IS the # entry's address. The per-host dance (blank self-heal / stale # protocol host / deliberately different ADB host) only existed # because the address was stored twice; it went with the # duplication. options = dict(entry.options) if entry_host(entry) != host: options[CONF_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 / t37 D3: MAC-less discovery cannot carry a unique_id, so # HA's own helper owns the dedup — and it MUST be awaited. It is a # coroutine that aborts `already_configured` when the handler # already has entries, claims DEFAULT_DISCOVERY_UNIQUE_ID, and # aborts `already_in_progress` when another flow for this handler # is running. t32 shipped the call without the await, so none of # that ran ("coroutine was never awaited") and the host-equality # guard that used to sit here was the accidental dedup; the real # helper is stricter (any entry for the domain, any in-flight # flow), so nothing is left to do here. await self._async_handle_discovery_without_unique_id() # 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 + setup sub-screens. Everything the integration needs lives in the OPTIONS (t41: one host field among them); ``data`` stays empty because no value has to be immutable. The setup sub-screens (t56) contribute ``_setup_extras`` — disjoint keys merged over the first form, so a value configured on a sub-screen wins over the same key left blank on the first form. """ d = {**self._user_data, **self._setup_extras} return self.async_create_entry( # the name is the entry title; a flow driven programmatically (and # the first form always carries it) may omit it, so fall back title=d.get("name") or DEFAULT_NAME, 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, True), CONF_ADB_STORE_APP_INFO: d.get(CONF_ADB_STORE_APP_INFO, True), CONF_HOST: d.get(CONF_HOST, ""), CONF_ADB_PORT: d.get(CONF_ADB_PORT, DEFAULT_ADB_PORT), CONF_PROTOCOL_ENABLED: d.get(CONF_PROTOCOL_ENABLED, True), }, ) async def _async_after_pairing(self): """Route after a pairing step: the protocol next, else the setup menu (t56). The "ADB paired, now the protocol" sequence is deliberate and stays automatic — a user who asked for both transports should not have to find the second one in a menu — and the setup menu afterwards is where the remaining sub-screens (Connection / Receiver) and Finish live. """ if self._user_data.get(CONF_PROTOCOL_ENABLED) and not getattr( self, "_paired_protocol", False ): return await self.async_step_protocol_pair() return self._show_setup_menu() def _discard_protocol_client(self) -> None: """Drop a pairing client whose session became void (host corrected).""" client = self._protocol_client self._protocol_client = None if client is not None: with contextlib.suppress(Exception): client.disconnect() async def _start_protocol_pairing(self): """Build a throwaway protocol client and make the Shield show its PIN. Certificates live under ``.storage//`` (core androidtv_remote helpers.py pattern) and are shared between pairing and runtime. Returns ``(client, None)`` or ``(None, error_key)`` — no host, a missing ``androidtvremote2`` and an unreachable Shield are all keys the form can show, never an exception out of the step. t44 F1: androidtvremote2 raises ``CannotConnect`` from both calls below, and the protocol transport is ON by default, so an asleep, rebooting or mis-addressed Shield is the ordinary case rather than an edge one. Left uncaught it escaped the config flow entirely, and the protocol_pair form — the address field AND the "continue without pairing" escape — never rendered. """ host = self._host() if not host: return None, "protocol_no_host" 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, ) if client is None: # build_protocol_client already logged the missing library; # pairing cannot start, and the form still offers the skip path return None, "protocol_unavailable" try: await client.async_generate_cert_if_missing() await client.async_start_pairing() except Exception as err: # noqa: BLE001 - the library's error IS the finding _LOGGER.warning( "protocol pairing could not start at %s (%s: %s) — showing the " "form so the address can be corrected or pairing skipped", host, type(err).__name__, err, ) with contextlib.suppress(Exception): client.disconnect() return None, "protocol_unreachable" return client, None async def _async_begin_protocol_pairing(self): """Start a PIN pairing session: ``(client, None)`` or ``(None, error)``.""" return await self._start_protocol_pairing() def _show_protocol_pair_form(self, errors=None): """protocol_pair form: the PIN, the address, and the skip escape. t33 F-B/F-C: the host field lives on THIS form, so an unresolvable address is fixable where the error is shown (the user step is gone), and the skip option keeps the step from being a dead end. """ return self.async_show_form( step_id="protocol_pair", data_schema=vol.Schema( { vol.Required(CONF_PAIRING_CODE): str, _host_field_key(CONF_HOST, self._host()): str, vol.Optional(CONF_SKIP_PAIRING, default=False): bool, } ), errors=errors or {}, ) async def _async_skip_protocol_pairing(self): """Continue without the protocol transport (t33 F-B, passive mode). The transport is switched OFF rather than left half-paired: an unpaired protocol client fails auth on its first connect and would only bounce the user into the reauth flow — a fresh dead end. """ self._discard_protocol_client() self._user_data[CONF_PROTOCOL_ENABLED] = False if self._reauth_entry is not None: # an existing entry cannot be re-created: leave it untouched and # say so instead of pretending the re-pair worked self._reauth_entry = None return self.async_abort(reason="reauth_skipped") if self._reconfigure_entry is not None: # same for a Reconfigure visit: nothing changed, the entry keeps # its options and can be reconfigured again at any time self._reconfigure_entry = None return self.async_abort(reason="reconfigure_skipped") return self._create_entry() async def _async_restart_protocol_pairing(self, errors: dict): """Start a fresh pairing session and render the form for its code. Called on the first visit, after a host correction, and when the user submits without a code: each is a request for the Shield to show a new code, and none of them may leave the step stuck. """ self._discard_protocol_client() client, error = await self._async_begin_protocol_pairing() if error: errors["base"] = error else: self._protocol_client = client return self._show_protocol_pair_form(errors) async def async_step_protocol_pair(self, user_input=None): """Pair with the remote protocol — the PIN is displayed on the Shield. Never a dead end (t33 F-B): a corrected address restarts pairing right here, submitting without a code asks for a fresh one, and ``skip_pairing`` finishes the entry in passive mode. """ errors = {} if user_input is None: # First visit: proactively start pairing so the PIN shows on the # Shield while the form is open. return await self._async_restart_protocol_pairing(errors) if user_input.get(CONF_SKIP_PAIRING): return await self._async_skip_protocol_pairing() typed_host = str(user_input.get(CONF_HOST) or "").strip() if typed_host and typed_host != self._host(): # A host typed HERE is authoritative — the user step is gone, so # the session started against the old address is dropped and the # Shield is asked for a NEW code at the corrected address (t35 F1). self._user_data[CONF_HOST] = typed_host self._discard_protocol_client() code = str(user_input.get(CONF_PAIRING_CODE, "") or "").strip() if self._protocol_client is None or not code: # No live session, or no code typed: (re)start pairing so the # Shield shows a code for the address currently on the form. return await self._async_restart_protocol_pairing(errors) client = self._protocol_client try: await client.async_finish_pairing(code) except Exception: errors["base"] = "protocol_pair_failed" return self._show_protocol_pair_form(errors) try: _name, mac = await client.async_get_name_and_mac() except Exception: mac = None with contextlib.suppress(Exception): client.disconnect() self._protocol_client = None target = self._reauth_entry or self._reconfigure_entry if mac and mac != getattr(target, "unique_id", None): # A MAC that is already configured must never be adopted silently. # A reauth/reconfigure flow already OWNS an entry, and # _abort_if_unique_id_configured looks the domain up by unique_id — # it would abort on that very entry when the MAC is unchanged # (t35 F1: re-pairing an entry used to abort as "already_configured"). await self.async_set_unique_id(mac) self._abort_if_unique_id_configured() if self._reauth_entry is not None: return self._finish_reauth() if self._reconfigure_entry is not None: return self._finish_reconfigure(mac) # t56: the protocol is paired now, so the setup menu must not offer to # pair it again — it returns to the menu (Receiver / Connection / # Finish) instead of creating the entry here. self._paired_protocol = True return self._show_setup_menu() 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 # the block's address comes from `host`, or from the legacy keys block_host = seed_host(block) seed_unique_id = block.get("unique_id") if not seed_unique_id: seed_unique_id = f"yaml:{name}:{block_host or 'nohost'}" # duplicate guards for existing in self._async_current_entries(): ehost = entry_host(existing) if existing.unique_id == seed_unique_id or ( block_host and ehost == block_host ): _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, # t32 F4's shape rule on the t41 keys: an address names the Shield, # and an explicit adb_enabled says otherwise CONF_ADB_ENABLED: seed_adb_enabled(block), CONF_HOST: block_host, CONF_ADB_PORT: int(block.get(CONF_ADB_PORT) or DEFAULT_ADB_PORT), } # t33 F-A: the seed host feeds the same user-step prefill as a # zeroconf discovery, but WITHOUT the discovery announcement — the # address came from configuration.yaml, not from the network. self._seed_host = block_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_HOST: entry_host(entry), } return await self.async_step_protocol_pair() # first visit re-pairs return self.async_show_form( step_id="reauth_confirm", data_schema=vol.Schema({}) ) async def async_step_reconfigure(self, user_input=None): """Re-run protocol pairing for an existing entry (audit G5, t35 F1). Home Assistant offers this as the entry's "Reconfigure" action — the discoverable way back for an entry whose protocol transport is off (typically the "continue without protocol pairing" outcome): the address is editable, the Shield shows a PIN, and the entry's options are updated + reloaded. """ entry = self._reconfigure_entry or self._get_reconfigure_entry() if entry is None: # HA always links the entry; a missing one gets a plain abort # instead of an exception return self.async_abort(reason="reconfigure_entry_missing") self._reconfigure_entry = entry # Seed the flow from the entry so the pairing step offers the address # the entry already knows (and _create_entry is never reached here). self._user_data = { "name": entry.title, CONF_ADB_ENABLED: entry.options.get(CONF_ADB_ENABLED, False), CONF_HOST: entry_host(entry), CONF_ADB_PORT: entry.options.get(CONF_ADB_PORT, DEFAULT_ADB_PORT), CONF_PROTOCOL_ENABLED: True, } return await self.async_step_protocol_pair() def _finish_reconfigure(self, mac: str | None): """Reconfigure pairing done: update + reload + abort (G3a contract). The protocol host persists as the address that just paired, so the entry no longer depends on the ADB-host fallback for its transport. HA derives the abort reason ("reconfigure_successful") from the flow source, so this needs no integration string. """ entry = self._reconfigure_entry or self._get_reconfigure_entry() updates: dict = { "options": { **entry.options, CONF_PROTOCOL_ENABLED: True, CONF_HOST: self._host(), }, } if mac: # a swapped device keeps the entry's identity honest updates["unique_id"] = mac self._reconfigure_entry = None return self.async_update_reload_and_abort(entry, **updates) 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. Core's kwarg is `options` (REPLACE) — there is no `options_updates`. """ entry = self._reauth_entry or self._get_reauth_entry() self.async_update_reload_and_abort( entry, options={ **entry.options, CONF_PROTOCOL_ENABLED: True, CONF_HOST: self._host(), }, ) self._reauth_entry = None return self.async_abort(reason="reauth_successful") def _host(self) -> str: """The Shield's address as this flow knows it (t41: ONE field). What the user typed (or a seed carried) lives in ``_user_data``; the discovery/seed hints stay as fallbacks for a form that was rendered with a pre-filled field and submitted unchanged. """ return str( self._user_data.get(CONF_HOST) or self._discovered_host or self._seed_host or "" ).strip() # ── t56: the setup menu and its sub-screens ── @property def _setup_extras(self) -> dict: """Options collected on the setup sub-screens (t56), merged, not last-writer-wins. Lazily created so the flow's ``__init__`` stays untouched; the menu's sub-screens each own a disjoint set of keys and merge into this dict, so configuring the receiver and then the connection (in either order) cannot drop the other's edits. """ if getattr(self, "_setup_extra_options", None) is None: self._setup_extra_options = {} return self._setup_extra_options def _show_setup_menu(self, errors=None): """The setup menu: connection, receiver, and finishing the entry (t56). Home Assistant's standard menu step: it renders every option as a row and dispatches the choice to ``async_step_