config_flow.py (14626B)
1 """Local-only setup flow for new and existing in-lite meshes.""" 2 from __future__ import annotations 3 4 import logging 5 import secrets 6 from typing import Any 7 8 import voluptuous as vol 9 from bleak_retry_connector import BleakClientWithServiceCache, establish_connection 10 from homeassistant.components import bluetooth 11 from homeassistant.components.bluetooth import BluetoothServiceInfoBleak 12 from homeassistant.config_entries import ConfigFlow, ConfigFlowResult, OptionsFlow 13 from homeassistant.core import callback 14 from homeassistant.helpers import selector 15 16 from inlite_ble.hub import InliteHub 17 from inlite_ble.protocol import FACTORY_NETWORK_PASSPHRASE 18 19 from .const import ( 20 BLE_LOCAL_NAME, CONF_GARDEN_ID, CONF_GARDEN_NAME, CONF_IDLE_DISCONNECT, 21 CONF_NETWORK_KEY, CONF_PASSWORD, CONF_SCAN_INTERVAL, CONF_TRANSFORMERS, 22 DEFAULT_IDLE_DISCONNECT_SECONDS, DEFAULT_SCAN_INTERVAL, DOMAIN, 23 MAX_IDLE_DISCONNECT_SECONDS, MAX_SCAN_INTERVAL, MIN_IDLE_DISCONNECT_SECONDS, 24 MIN_SCAN_INTERVAL, 25 ) 26 27 _LOGGER = logging.getLogger(__name__) 28 29 SETUP_FACTORY = "factory" 30 SETUP_PASSPHRASE = "passphrase" 31 SETUP_NETWORK_KEY = "network_key" 32 SETUP_MODES = [SETUP_FACTORY, SETUP_PASSPHRASE, SETUP_NETWORK_KEY] 33 34 35 class InliteConfigFlow(ConfigFlow, domain=DOMAIN): 36 """Onboard one local in-lite mesh entirely over BLE.""" 37 38 VERSION = 2 39 40 def __init__(self) -> None: 41 self._service_info: BluetoothServiceInfoBleak | None = None 42 self._candidates: set[int] = set() 43 self._selected_id: int | None = None 44 self._garden_name = "in-lite" 45 self._setup_mode: str | None = None 46 self._passphrase: str | None = None 47 self._network_key: bytes | None = None 48 49 @staticmethod 50 @callback 51 def async_get_options_flow(config_entry: Any) -> InliteOptionsFlow: 52 return InliteOptionsFlow(config_entry) 53 54 async def async_step_bluetooth( 55 self, discovery_info: BluetoothServiceInfoBleak 56 ) -> ConfigFlowResult: 57 self._service_info = discovery_info 58 await self.async_set_unique_id(discovery_info.address) 59 self._abort_if_unique_id_configured() 60 return await self.async_step_setup() 61 62 async def async_step_user( 63 self, user_input: dict[str, Any] | None = None 64 ) -> ConfigFlowResult: 65 for info in bluetooth.async_discovered_service_info(self.hass, connectable=True): 66 if info.name and info.name.lower() == BLE_LOCAL_NAME: 67 self._service_info = info 68 break 69 if self._service_info is None: 70 return self.async_abort(reason="no_hub_found") 71 await self.async_set_unique_id(self._service_info.address) 72 self._abort_if_unique_id_configured() 73 return await self.async_step_setup() 74 75 async def async_step_setup( 76 self, user_input: dict[str, Any] | None = None 77 ) -> ConfigFlowResult: 78 """Choose whether to provision a new mesh or join an existing one.""" 79 if user_input is not None: 80 self._setup_mode = user_input["setup_mode"] 81 self._garden_name = user_input["garden_name"].strip() or "in-lite" 82 if self._setup_mode == SETUP_FACTORY: 83 return await self.async_step_bluetooth_confirm() 84 return await self.async_step_credentials() 85 return self.async_show_form( 86 step_id="setup", 87 data_schema=vol.Schema({ 88 vol.Required("setup_mode", default=SETUP_FACTORY): selector.SelectSelector( 89 selector.SelectSelectorConfig( 90 options=SETUP_MODES, translation_key="setup_mode" 91 ) 92 ), 93 vol.Required("garden_name", default=self._garden_name): str, 94 }), 95 ) 96 97 async def async_step_bluetooth_confirm( 98 self, user_input: dict[str, Any] | None = None 99 ) -> ConfigFlowResult: 100 if user_input is None: 101 self._set_confirm_only() 102 return self.async_show_form(step_id="bluetooth_confirm") 103 try: 104 self._candidates = await self._discover_hubs() 105 except Exception: 106 _LOGGER.exception("Factory-network discovery failed") 107 return self.async_show_form( 108 step_id="bluetooth_confirm", errors={"base": "cannot_connect"} 109 ) 110 if not self._candidates: 111 return self.async_show_form( 112 step_id="bluetooth_confirm", errors={"base": "no_factory_hubs"} 113 ) 114 return await self.async_step_identify() 115 116 async def async_step_credentials( 117 self, user_input: dict[str, Any] | None = None 118 ) -> ConfigFlowResult: 119 """Collect credentials for an existing mesh and discover its hubs.""" 120 errors: dict[str, str] = {} 121 if user_input is not None: 122 if self._setup_mode == SETUP_PASSPHRASE: 123 passphrase = user_input[CONF_PASSWORD] 124 if not passphrase: 125 errors["base"] = "invalid_credentials" 126 else: 127 self._passphrase = passphrase 128 elif self._setup_mode == SETUP_NETWORK_KEY: 129 network_key_text = user_input[CONF_NETWORK_KEY].strip() 130 if ( 131 len(network_key_text) != 32 132 or any(char not in "0123456789abcdefABCDEF" for char in network_key_text) 133 ): 134 network_key = b"" 135 else: 136 network_key = bytes.fromhex(network_key_text) 137 if len(network_key) != 16: 138 errors["base"] = "invalid_network_key" 139 else: 140 self._network_key = network_key 141 else: 142 return self.async_abort(reason="unknown") 143 144 if not errors: 145 try: 146 self._candidates = await self._discover_hubs() 147 except Exception: 148 _LOGGER.exception("Existing-mesh discovery failed") 149 errors["base"] = "cannot_connect" 150 else: 151 if self._candidates: 152 return await self.async_step_identify() 153 errors["base"] = "invalid_credentials" 154 155 credential_field = ( 156 CONF_PASSWORD if self._setup_mode == SETUP_PASSPHRASE else CONF_NETWORK_KEY 157 ) 158 return self.async_show_form( 159 step_id="credentials", 160 data_schema=vol.Schema({vol.Required(credential_field): str}), 161 errors=errors, 162 ) 163 164 async def async_step_identify( 165 self, user_input: dict[str, Any] | None = None 166 ) -> ConfigFlowResult: 167 choices = {f"0x{id_:04X}": id_ for id_ in sorted(self._candidates)} 168 errors: dict[str, str] = {} 169 if user_input is not None: 170 self._selected_id = choices[user_input["device_id"]] 171 try: 172 await self._identify(self._selected_id) 173 except Exception: 174 _LOGGER.exception("Identify command failed") 175 errors["base"] = "cannot_connect" 176 else: 177 return await self.async_step_confirm_blink() 178 return self.async_show_form( 179 step_id="identify", 180 data_schema=vol.Schema({ 181 vol.Required("device_id"): vol.In(choices), 182 }), 183 errors=errors, 184 ) 185 186 async def async_step_confirm_blink( 187 self, user_input: dict[str, Any] | None = None 188 ) -> ConfigFlowResult: 189 if user_input is None: 190 return self.async_show_form( 191 step_id="confirm_blink", 192 data_schema=vol.Schema({vol.Required("confirmed"): bool}), 193 ) 194 if not user_input["confirmed"]: 195 return await self.async_step_identify() 196 if self._selected_id is None: 197 return self.async_abort(reason="unknown") 198 try: 199 if self._setup_mode == SETUP_FACTORY: 200 password, zones, firmware = await self._associate_and_read(self._selected_id) 201 credential = {CONF_PASSWORD: password} 202 transformers = [ 203 self._transformer_data(self._selected_id, zones, firmware) 204 ] 205 else: 206 credential = self._credential_data() 207 mesh_hubs = await self._read_existing_mesh() 208 transformers = [ 209 self._transformer_data(device_id, zones, firmware) 210 for device_id, (zones, firmware) in mesh_hubs.items() 211 ] 212 except Exception: 213 _LOGGER.exception("Could not finish local mesh setup") 214 return self.async_show_form( 215 step_id="confirm_blink", errors={"base": "association_failed"} 216 ) 217 return self.async_create_entry( 218 title=f"in-lite {self._garden_name}", 219 data={ 220 CONF_GARDEN_ID: self._service_info.address, 221 CONF_GARDEN_NAME: self._garden_name, 222 **credential, 223 CONF_TRANSFORMERS: transformers, 224 }, 225 ) 226 227 @staticmethod 228 def _transformer_data( 229 device_id: int, zones: dict[int, Any], firmware: int | None 230 ) -> dict[str, Any]: 231 """Build config-entry data for one discovered mesh hub.""" 232 return { 233 "device_id": device_id, 234 "name": f"in-lite hub 0x{device_id:04X}", 235 "firmware_version": firmware if firmware is not None else "unknown", 236 "zones": [ 237 {"output_id": zone_id, "name": f"Zone {zone_id + 1}"} 238 for zone_id in sorted(zones) 239 ], 240 } 241 242 async def _connect(self) -> Any: 243 if self._service_info is None: 244 raise ConnectionError("No in-lite gateway available") 245 return await establish_connection( 246 BleakClientWithServiceCache, self._service_info.device, 247 self._service_info.address, max_attempts=3, 248 ) 249 250 def _new_hub(self, device_id: int) -> InliteHub: 251 """Build a hub using the credentials selected during setup.""" 252 if self._setup_mode == SETUP_FACTORY: 253 return InliteHub(device_id, FACTORY_NETWORK_PASSPHRASE) 254 if self._network_key is not None: 255 return InliteHub(device_id, network_key=self._network_key) 256 if self._passphrase is not None: 257 return InliteHub(device_id, self._passphrase) 258 raise ValueError("mesh credentials have not been selected") 259 260 def _credential_data(self) -> dict[str, str]: 261 """Return serializable credentials for an existing mesh entry.""" 262 if self._network_key is not None: 263 return {CONF_NETWORK_KEY: self._network_key.hex()} 264 if self._passphrase is not None: 265 return {CONF_PASSWORD: self._passphrase} 266 raise ValueError("mesh credentials have not been selected") 267 268 async def _with_hub(self, device_id: int) -> InliteHub: 269 hub = self._new_hub(device_id) 270 client = await self._connect() 271 if not await hub.connect(client=client): 272 await hub.disconnect() 273 raise ConnectionError("Could not subscribe to hub notifications") 274 return hub 275 276 async def _discover_hubs(self) -> set[int]: 277 hub = await self._with_hub(0) 278 try: 279 return await hub.discover_hubs() 280 finally: 281 await hub.disconnect() 282 283 async def _identify(self, device_id: int) -> None: 284 hub = await self._with_hub(device_id) 285 try: 286 if not await hub.identify(): 287 raise ConnectionError("Hub did not acknowledge identify") 288 finally: 289 await hub.disconnect() 290 291 async def _associate_and_read( 292 self, device_id: int 293 ) -> tuple[str, dict[int, Any], int | None]: 294 password = secrets.token_urlsafe(32) 295 hub = await self._with_hub(device_id) 296 try: 297 if not await hub.associate(password): 298 raise ConnectionError("Hub did not acknowledge association") 299 finally: 300 await hub.disconnect() 301 local_hub = InliteHub(device_id, password) 302 client = await self._connect() 303 try: 304 if not await local_hub.connect(client=client): 305 raise ConnectionError("Could not reconnect after association") 306 return password, await local_hub.query_zone_states(), local_hub.firmware_version 307 finally: 308 await local_hub.disconnect() 309 310 async def _read_existing( 311 self, device_id: int 312 ) -> tuple[dict[int, Any], int | None]: 313 """Verify credentials against a selected mesh hub and read its zones.""" 314 hub = await self._with_hub(device_id) 315 try: 316 return await hub.query_zone_states(), hub.firmware_version 317 finally: 318 await hub.disconnect() 319 320 async def _read_existing_mesh(self) -> dict[int, tuple[dict[int, Any], int | None]]: 321 """Read every hub that responded to authenticated mesh discovery.""" 322 result: dict[int, tuple[dict[int, Any], int | None]] = {} 323 for device_id in sorted(self._candidates): 324 result[device_id] = await self._read_existing(device_id) 325 return result 326 327 328 class InliteOptionsFlow(OptionsFlow): 329 """Options for local polling and connection lifetime.""" 330 331 def __init__(self, config_entry: Any) -> None: 332 self._config_entry = config_entry 333 334 async def async_step_init( 335 self, user_input: dict[str, Any] | None = None 336 ) -> ConfigFlowResult: 337 if user_input is not None: 338 return self.async_create_entry(title="", data=user_input) 339 return self.async_show_form( 340 step_id="init", 341 data_schema=vol.Schema({ 342 vol.Required( 343 CONF_SCAN_INTERVAL, 344 default=self._config_entry.options.get( 345 CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL 346 ), 347 ): vol.All( 348 vol.Coerce(int), 349 vol.Range(min=MIN_SCAN_INTERVAL, max=MAX_SCAN_INTERVAL), 350 ), 351 vol.Required( 352 CONF_IDLE_DISCONNECT, 353 default=self._config_entry.options.get( 354 CONF_IDLE_DISCONNECT, DEFAULT_IDLE_DISCONNECT_SECONDS 355 ), 356 ): vol.All( 357 vol.Coerce(int), 358 vol.Range( 359 min=MIN_IDLE_DISCONNECT_SECONDS, 360 max=MAX_IDLE_DISCONNECT_SECONDS, 361 ), 362 ), 363 }), 364 )