Simplify switchbot config flow

homeassistant/components/switchbot/__init__.py

@@ -9,6 +9,7 @@
 from homeassistant.const import (
     CONF_ADDRESS,
     CONF_MAC,
+    CONF_NAME,
     CONF_PASSWORD,
     CONF_SENSOR_TYPE,
     Platform,
@@ -17,38 +18,34 @@
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr
 
-from .const import (
-    ATTR_BOT,
-    ATTR_CONTACT,
-    ATTR_CURTAIN,
-    ATTR_HYGROMETER,
-    ATTR_MOTION,
-    ATTR_PLUG,
-    CONF_RETRY_COUNT,
-    DEFAULT_RETRY_COUNT,
-    DOMAIN,
-)
+from .const import CONF_RETRY_COUNT, DEFAULT_RETRY_COUNT, DOMAIN, SupportedModels
 from .coordinator import SwitchbotDataUpdateCoordinator
 
 PLATFORMS_BY_TYPE = {
-    ATTR_BOT: [Platform.SWITCH, Platform.SENSOR],
-    ATTR_PLUG: [Platform.SWITCH, Platform.SENSOR],
-    ATTR_CURTAIN: [Platform.COVER, Platform.BINARY_SENSOR, Platform.SENSOR],
-    ATTR_HYGROMETER: [Platform.SENSOR],
-    ATTR_CONTACT: [Platform.BINARY_SENSOR, Platform.SENSOR],
-    ATTR_MOTION: [Platform.BINARY_SENSOR, Platform.SENSOR],
+    SupportedModels.BULB.value: [Platform.SENSOR],
+    SupportedModels.BOT.value: [Platform.SWITCH, Platform.SENSOR],
+    SupportedModels.PLUG.value: [Platform.SWITCH, Platform.SENSOR],
+    SupportedModels.CURTAIN.value: [
+        Platform.COVER,
+        Platform.BINARY_SENSOR,
+        Platform.SENSOR,
+    ],
+    SupportedModels.HYGROMETER.value: [Platform.SENSOR],
+    SupportedModels.CONTACT.value: [Platform.BINARY_SENSOR, Platform.SENSOR],
+    SupportedModels.MOTION.value: [Platform.BINARY_SENSOR, Platform.SENSOR],
 }
 CLASS_BY_DEVICE = {
-    ATTR_CURTAIN: switchbot.SwitchbotCurtain,
-    ATTR_BOT: switchbot.Switchbot,
-    ATTR_PLUG: switchbot.SwitchbotPlugMini,
+    SupportedModels.CURTAIN.value: switchbot.SwitchbotCurtain,
+    SupportedModels.BOT.value: switchbot.Switchbot,
+    SupportedModels.PLUG.value: switchbot.SwitchbotPlugMini,
 }
 
 _LOGGER = logging.getLogger(__name__)
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up Switchbot from a config entry."""
+    assert entry.unique_id is not None
     hass.data.setdefault(DOMAIN, {})
     if CONF_ADDRESS not in entry.data and CONF_MAC in entry.data:
         # Bleak uses addresses not mac addresses which are are actually
@@ -81,7 +78,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         retry_count=entry.options[CONF_RETRY_COUNT],
     )
     coordinator = hass.data[DOMAIN][entry.entry_id] = SwitchbotDataUpdateCoordinator(
-        hass, _LOGGER, ble_device, device
+        hass,
+        _LOGGER,
+        ble_device,
+        device,
+        entry.unique_id,
+        entry.data.get(CONF_NAME, entry.title),
     )
     entry.async_on_unload(coordinator.async_start())
     if not await coordinator.async_wait_ready():

homeassistant/components/switchbot/binary_sensor.py

@@ -7,7 +7,6 @@
     BinarySensorEntityDescription,
 )
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import CONF_ADDRESS, CONF_NAME
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity import EntityCategory
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
@@ -53,20 +52,10 @@ async def async_setup_entry(
 ) -> None:
     """Set up Switchbot curtain based on a config entry."""
     coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
-    unique_id = entry.unique_id
-    assert unique_id is not None
     async_add_entities(
-        [
-            SwitchBotBinarySensor(
-                coordinator,
-                unique_id,
-                binary_sensor,
-                entry.data[CONF_ADDRESS],
-                entry.data[CONF_NAME],
-            )
-            for binary_sensor in coordinator.data["data"]
-            if binary_sensor in BINARY_SENSOR_TYPES
-        ]
+        SwitchBotBinarySensor(coordinator, binary_sensor)
+        for binary_sensor in coordinator.data["data"]
+        if binary_sensor in BINARY_SENSOR_TYPES
     )
 
 
@@ -78,15 +67,12 @@ class SwitchBotBinarySensor(SwitchbotEntity, BinarySensorEntity):
     def __init__(
         self,
         coordinator: SwitchbotDataUpdateCoordinator,
-        unique_id: str,
         binary_sensor: str,
-        mac: str,
-        switchbot_name: str,
     ) -> None:
         """Initialize the Switchbot sensor."""
-        super().__init__(coordinator, unique_id, mac, name=switchbot_name)
+        super().__init__(coordinator)
         self._sensor = binary_sensor
-        self._attr_unique_id = f"{unique_id}-{binary_sensor}"
+        self._attr_unique_id = f"{coordinator.base_unique_id}-{binary_sensor}"
         self.entity_description = BINARY_SENSOR_TYPES[binary_sensor]
         self._attr_name = self.entity_description.name
 

homeassistant/components/switchbot/config_flow.py

@@ -12,9 +12,9 @@
     async_discovered_service_info,
 )
 from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
-from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_PASSWORD, CONF_SENSOR_TYPE
+from homeassistant.const import CONF_ADDRESS, CONF_PASSWORD, CONF_SENSOR_TYPE
 from homeassistant.core import callback
-from homeassistant.data_entry_flow import FlowResult
+from homeassistant.data_entry_flow import AbortFlow, FlowResult
 
 from .const import CONF_RETRY_COUNT, DEFAULT_RETRY_COUNT, DOMAIN, SUPPORTED_MODEL_TYPES
 
@@ -26,6 +26,17 @@ def format_unique_id(address: str) -> str:
     return address.replace(":", "").lower()
 
 
+def short_address(address: str) -> str:
+    """Convert a Bluetooth address to a short address."""
+    results = address.replace("-", ":").split(":")
+    return f"{results[-2].upper()}{results[-1].upper()}"[-4:]
+
+
+def name_from_discovery(discovery: SwitchBotAdvertisement) -> str:
+    """Get the name from a discovery."""
+    return f'{discovery.data["modelFriendlyName"]} {short_address(discovery.address)}'
+
+
 class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
     """Handle a config flow for Switchbot."""
 
@@ -59,62 +70,128 @@ async def async_step_bluetooth(
         self._discovered_adv = parsed
         data = parsed.data
         self.context["title_placeholders"] = {
-            "name": data["modelName"],
-            "address": discovery_info.address,
+            "name": data["modelFriendlyName"],
+            "address": short_address(discovery_info.address),
         }
-        return await self.async_step_user()
+        if self._discovered_adv.data["isEncrypted"]:
+            return await self.async_step_password()
+        return await self.async_step_confirm()
 
-    async def async_step_user(
-        self, user_input: dict[str, Any] | None = None
+    async def _async_create_entry_from_discovery(
+        self, user_input: dict[str, Any]
     ) -> FlowResult:
-        """Handle the user step to pick discovered device."""
-        errors: dict[str, str] = {}
+        """Create an entry from a discovery."""
+        assert self._discovered_adv is not None
+        discovery = self._discovered_adv
+        name = name_from_discovery(discovery)
+        model_name = discovery.data["modelName"]
+        return self.async_create_entry(
+            title=name,
+            data={
+                **user_input,
+                CONF_ADDRESS: discovery.address,
+                CONF_SENSOR_TYPE: str(SUPPORTED_MODEL_TYPES[model_name]),
+            },
+        )
+
+    async def async_step_confirm(self, user_input: dict[str, Any] = None) -> FlowResult:
+        """Confirm a single device."""
+        assert self._discovered_adv is not None
+        if user_input is not None:
+            return await self._async_create_entry_from_discovery(user_input)
+
+        self._set_confirm_only()
+        return self.async_show_form(
+            step_id="confirm",
+            data_schema=vol.Schema({}),
+            description_placeholders={
+                "name": name_from_discovery(self._discovered_adv)
+            },
+        )
 
+    async def async_step_password(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the password step."""
+        assert self._discovered_adv is not None
         if user_input is not None:
-            address = user_input[CONF_ADDRESS]
-            await self.async_set_unique_id(
-                format_unique_id(address), raise_on_progress=False
+            # There is currently no api to validate the password
+            # that does not operate the device so we have
+            # to accept it as-is
+            return await self._async_create_entry_from_discovery(user_input)
+
+        return self.async_show_form(
+            step_id="password",
+            data_schema=vol.Schema({vol.Required(CONF_PASSWORD): str}),
+            description_placeholders={
+                "name": name_from_discovery(self._discovered_adv)
+            },
+        )
+
+    @callback
+    def _async_discover_devices(self) -> None:
+        current_addresses = self._async_current_ids()
+        for discovery_info in async_discovered_service_info(self.hass):
+            address = discovery_info.address
+            if (
+                format_unique_id(address) in current_addresses
+                or address in self._discovered_advs
+            ):
+                continue
+            parsed = parse_advertisement_data(
+                discovery_info.device, discovery_info.advertisement
             )
-            self._abort_if_unique_id_configured()
-            user_input[CONF_SENSOR_TYPE] = SUPPORTED_MODEL_TYPES[
-                self._discovered_advs[address].data["modelName"]
-            ]
-            return self.async_create_entry(title=user_input[CONF_NAME], data=user_input)
-
-        if discovery := self._discovered_adv:
-            self._discovered_advs[discovery.address] = discovery
-        else:
-            current_addresses = self._async_current_ids()
-            for discovery_info in async_discovered_service_info(self.hass):
-                address = discovery_info.address
-                if (
-                    format_unique_id(address) in current_addresses
-                    or address in self._discovered_advs
-                ):
-                    continue
-                parsed = parse_advertisement_data(
-                    discovery_info.device, discovery_info.advertisement
-                )
-                if parsed and parsed.data.get("modelName") in SUPPORTED_MODEL_TYPES:
-                    self._discovered_advs[address] = parsed
+            if parsed and parsed.data.get("modelName") in SUPPORTED_MODEL_TYPES:
+                self._discovered_advs[address] = parsed
 
         if not self._discovered_advs:
-            return self.async_abort(reason="no_unconfigured_devices")
-
-        data_schema = vol.Schema(
-            {
-                vol.Required(CONF_ADDRESS): vol.In(
-                    {
-                        address: f"{parsed.data['modelName']} ({address})"
-                        for address, parsed in self._discovered_advs.items()
-                    }
-                ),
-                vol.Required(CONF_NAME): str,
-                vol.Optional(CONF_PASSWORD): str,
-            }
+            raise AbortFlow("no_unconfigured_devices")
+
+    async def _async_set_device(self, discovery: SwitchBotAdvertisement) -> None:
+        """Set the device to work with."""
+        self._discovered_adv = discovery
+        address = discovery.address
+        await self.async_set_unique_id(
+            format_unique_id(address), raise_on_progress=False
         )
+        self._abort_if_unique_id_configured()
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the user step to pick discovered device."""
+        errors: dict[str, str] = {}
+        device_adv: SwitchBotAdvertisement | None = None
+        if user_input is not None:
+            device_adv = self._discovered_advs[user_input[CONF_ADDRESS]]
+            await self._async_set_device(device_adv)
+            if device_adv.data["isEncrypted"]:
+                return await self.async_step_password()
+            return await self._async_create_entry_from_discovery(user_input)
+
+        self._async_discover_devices()
+        if len(self._discovered_advs) == 1:
+            # If there is only one device we can ask for a password
+            # or simply confirm it
+            device_adv = list(self._discovered_advs.values())[0]
+            await self._async_set_device(device_adv)
+            if device_adv.data["isEncrypted"]:
+                return await self.async_step_password()
+            return await self.async_step_confirm()
+
         return self.async_show_form(
-            step_id="user", data_schema=data_schema, errors=errors
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_ADDRESS): vol.In(
+                        {
+                            address: name_from_discovery(parsed)
+                            for address, parsed in self._discovered_advs.items()
+                        }
+                    ),
+                }
+            ),
+            errors=errors,
         )
 
 

homeassistant/components/switchbot/const.py

@@ -1,25 +1,39 @@
 """Constants for the switchbot integration."""
+from switchbot import SwitchbotModel
+
+from homeassistant.backports.enum import StrEnum
+
 DOMAIN = "switchbot"
 MANUFACTURER = "switchbot"
 
 # Config Attributes
-ATTR_BOT = "bot"
-ATTR_CURTAIN = "curtain"
-ATTR_HYGROMETER = "hygrometer"
-ATTR_CONTACT = "contact"
-ATTR_PLUG = "plug"
-ATTR_MOTION = "motion"
+
 DEFAULT_NAME = "Switchbot"
 
+
+class SupportedModels(StrEnum):
+    """Supported Switchbot models."""
+
+    BOT = "bot"
+    BULB = "bulb"
+    CURTAIN = "curtain"
+    HYGROMETER = "hygrometer"
+    CONTACT = "contact"
+    PLUG = "plug"
+    MOTION = "motion"
+
+
 SUPPORTED_MODEL_TYPES = {
-    "WoHand": ATTR_BOT,
-    "WoCurtain": ATTR_CURTAIN,
-    "WoSensorTH": ATTR_HYGROMETER,
-    "WoContact": ATTR_CONTACT,
-    "WoPlug": ATTR_PLUG,
-    "WoPresence": ATTR_MOTION,
+    SwitchbotModel.BOT: SupportedModels.BOT,
+    SwitchbotModel.CURTAIN: SupportedModels.CURTAIN,
+    SwitchbotModel.METER: SupportedModels.HYGROMETER,
+    SwitchbotModel.CONTACT_SENSOR: SupportedModels.CONTACT,
+    SwitchbotModel.PLUG_MINI: SupportedModels.PLUG,
+    SwitchbotModel.MOTION_SENSOR: SupportedModels.MOTION,
+    SwitchbotModel.COLOR_BULB: SupportedModels.BULB,
 }
 
+
 # Config Defaults
 DEFAULT_RETRY_COUNT = 3