"""Config flow for Hunter Douglas PowerView BLE integration.""" import hashlib import struct from typing import Any import aiohttp import voluptuous as vol from homeassistant import config_entries from homeassistant.components.bluetooth import BluetoothServiceInfoBleak from homeassistant.config_entries import ConfigFlowResult from homeassistant.core import HomeAssistant from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.selector import ( SelectOptionDict, SelectSelector, SelectSelectorConfig, TextSelector, TextSelectorConfig, TextSelectorType, ) from .const import CONF_HOME_KEY, CONF_HUB_URL, DOMAIN, LOGGER def _hub_unique_id(home_key: str) -> str: """Derive a stable unique ID for a hub entry from the home key.""" if home_key: digest = hashlib.sha256(home_key.encode()).hexdigest()[:16] return f"pvhome_{digest}" return "pvhome_unencrypted" def _parse_key_response(ble_name: str, result: dict) -> bytes | None: # noqa: PLR0911 """Parse a shade exec response and return the 16-byte key, or None.""" if result.get("err"): err_msg = (result.get("responses") or [{}])[0].get("errMsg", "unknown") LOGGER.warning( "Shade %s: hub BLE command failed (err=%s: %s)", ble_name, result["err"], err_msg, ) return None responses = result.get("responses", []) if len(responses) != 1 or "hex" not in responses[0]: LOGGER.warning( "Shade %s returned unexpected response structure: %s", ble_name, result, ) return None response_bytes = bytes.fromhex(responses[0]["hex"]) if len(response_bytes) < 5: LOGGER.warning( "Shade %s response too short (%d bytes)", ble_name, len(response_bytes) ) return None _s, _c, _q, length = struct.unpack(" bytes: """Fetch 16-byte homekey from a PowerView G3 hub. Tries each shade on the hub until one returns a valid key. The key is network-wide so any reachable shade returns the same value. Raises ValueError on protocol/key errors. Raises aiohttp.ClientError on network errors. Raises asyncio.TimeoutError on timeout. """ session = async_get_clientsession(hass) timeout = aiohttp.ClientTimeout(total=10) async with session.get(f"{hub_url}/home/shades", timeout=timeout) as resp: resp.raise_for_status() shades = await resp.json(content_type=None) if not shades: raise ValueError("No shades found on the hub") ble_names = [s.get("bleName", "") for s in shades if s.get("bleName")] if not ble_names: raise ValueError("No BLE-capable shades found on the hub") # GetShadeKey BLE request: sid=251, cid=18, seqId=1, data_len=0 request_frame = struct.pack(" None: """Initialize the config flow.""" self._home_key: str = "" self._hub_url: str = "" def _create_entry(self) -> ConfigFlowResult: """Create the hub config entry.""" data: dict[str, str] = {CONF_HOME_KEY: self._home_key} if self._hub_url: data[CONF_HUB_URL] = self._hub_url return self.async_create_entry(title="PowerView Home", data=data) def _validate_manual_key( self, user_input: dict[str, Any], errors: dict[str, str] ) -> bool: """Validate a manually entered hex key. Returns True on success, False on validation error. """ raw = user_input.get("home_key", "").strip() if "\\x" in raw: raw = raw.replace("\\x", "") if len(raw) != 32: errors["home_key"] = "invalid_key_length" return False try: bytes.fromhex(raw) except ValueError: errors["home_key"] = "invalid_key_format" return False self._home_key = raw.lower() return True async def _validate_homekey_input( self, user_input: dict[str, Any], errors: dict[str, str] ) -> bool: """Parse and validate homekey user_input, populating self state. Returns True on success, False on validation error (errors dict is populated). """ method = user_input.get("key_method", "skip") if method == "skip": self._home_key = "" return True if method == "manual": return self._validate_manual_key(user_input, errors) if method != "hub": return False hub_url = user_input.get(CONF_HUB_URL, "").rstrip("/") _HUB_ERROR_MAP: dict[type[Exception], str] = { aiohttp.ClientResponseError: "hub_http_error", aiohttp.ClientConnectionError: "hub_connection_error", TimeoutError: "hub_timeout", ValueError: "hub_protocol_error", } try: key = await _fetch_key_from_hub(self.hass, hub_url) except tuple(_HUB_ERROR_MAP) as ex: LOGGER.warning("Hub key fetch failed: %s", ex) errors[CONF_HUB_URL] = _HUB_ERROR_MAP[type(ex)] return False self._home_key = key.hex() self._hub_url = hub_url return True async def async_step_bluetooth( self, discovery_info: BluetoothServiceInfoBleak ) -> ConfigFlowResult: """Handle a flow initialized by Bluetooth discovery.""" LOGGER.debug("Bluetooth device detected: %s", discovery_info) # Tag the flow with this address so HA deduplicates future # discovery flows for the same device await self.async_set_unique_id(discovery_info.address) # If a hub entry already exists, shades are auto-discovered for entry in self._async_current_entries(): if entry.version >= 2: return self.async_abort(reason="already_configured") # No hub entry yet — redirect to user setup return await self.async_step_user() async def async_step_user( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Handle the user step — create a hub entry.""" LOGGER.debug("user step") # Only one hub entry allowed (per key, but for simplicity one total) for entry in self._async_current_entries(): if entry.version >= 2: return self.async_abort(reason="single_instance_allowed") errors: dict[str, str] = {} if user_input is not None and await self._validate_homekey_input( user_input, errors ): unique_id = _hub_unique_id(self._home_key) await self.async_set_unique_id(unique_id, raise_on_progress=False) self._abort_if_unique_id_configured() return self._create_entry() return self.async_show_form( step_id="user", data_schema=_HOMEKEY_SCHEMA, errors=errors, description_placeholders={ "hub_url_example": "http://powerview-g3.local", }, )