config_flow.py (3746B)
1 """Config flow for Hunter BTT.""" 2 3 from typing import Any 4 5 import voluptuous as vol 6 from homeassistant.components.bluetooth import ( 7 BluetoothServiceInfo, 8 async_discovered_service_info, 9 ) 10 from homeassistant.config_entries import ConfigFlow, ConfigFlowResult 11 from homeassistant.const import CONF_ADDRESS 12 13 from .const import BTT_SERVICE_UUID, CONF_PIN, DOMAIN 14 15 16 def _device_title(name: str | None, address: str) -> str: 17 """Return a distinguishable title from the advertisement or address. 18 19 Hunter exposes the same five-character suffix used as its serial number in 20 the app through the BLE address. Some platforms retain the advertised 21 ``Hunter BTT <suffix>`` name, while others report only ``Hunter BTT``. 22 """ 23 suffix = "".join(character for character in address if character.isalnum())[-5:] 24 advertised_name = (name or "Hunter BTT").strip() 25 if suffix.upper() not in advertised_name.upper(): 26 return f"{advertised_name} {suffix.upper()}" 27 return advertised_name 28 29 30 class HunterBTTConfigFlow(ConfigFlow, domain=DOMAIN): # type: ignore[call-arg] 31 """Configure a discovered Hunter BTT controller.""" 32 33 VERSION = 1 34 35 def __init__(self) -> None: 36 """Initialize the flow.""" 37 self._address: str | None = None 38 self._title: str | None = None 39 self._titles: dict[str, str] = {} 40 41 async def async_step_bluetooth( 42 self, discovery_info: BluetoothServiceInfo 43 ) -> ConfigFlowResult: 44 """Handle Bluetooth discovery.""" 45 self._address = discovery_info.address 46 self._title = _device_title(discovery_info.name, self._address) 47 self.context["title_placeholders"] = {"name": self._title} 48 await self.async_set_unique_id(self._address) 49 self._abort_if_unique_id_configured(updates={CONF_ADDRESS: self._address}) 50 return await self.async_step_confirm() 51 52 async def async_step_user( 53 self, user_input: dict[str, Any] | None = None 54 ) -> ConfigFlowResult: 55 """Allow selecting an already-discovered BTT service.""" 56 if user_input is not None: 57 self._address = user_input[CONF_ADDRESS] 58 self._title = self._titles[self._address] 59 self.context["title_placeholders"] = {"name": self._title} 60 await self.async_set_unique_id(self._address, raise_on_progress=False) 61 self._abort_if_unique_id_configured() 62 return await self.async_step_confirm() 63 devices = {} 64 for info in async_discovered_service_info(self.hass, connectable=True): 65 if BTT_SERVICE_UUID not in {uuid.lower() for uuid in info.service_uuids}: 66 continue 67 title = _device_title(info.name, info.address) 68 devices[info.address] = title 69 self._titles[info.address] = title 70 if not devices: 71 return self.async_abort(reason="no_devices_found") 72 return self.async_show_form( 73 step_id="user", 74 data_schema=vol.Schema({vol.Required(CONF_ADDRESS): vol.In(devices)}), 75 ) 76 77 async def async_step_confirm( 78 self, user_input: dict[str, Any] | None = None 79 ) -> ConfigFlowResult: 80 """Collect the passcode without attempting a second connection.""" 81 if self._address is None: 82 return self.async_abort(reason="no_devices_found") 83 if user_input is not None: 84 return self.async_create_entry( 85 title=self._title or _device_title(None, self._address), 86 data={CONF_ADDRESS: self._address, CONF_PIN: user_input[CONF_PIN]}, 87 ) 88 return self.async_show_form( 89 step_id="confirm", 90 data_schema=vol.Schema({vol.Required(CONF_PIN): str}), 91 )