Add config flow to Avea

homeassistant/components/avea/__init__.py

@@ -1 +1,39 @@
-"""The avea component."""
+"""The Avea integration."""
+
+from __future__ import annotations
+
+import avea
+
+from homeassistant.components.bluetooth import async_ble_device_from_address
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_ADDRESS, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+
+type AveaConfigEntry = ConfigEntry[avea.Bulb]
+
+PLATFORMS: list[Platform] = [Platform.LIGHT]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: AveaConfigEntry) -> bool:
+    """Set up Avea from a config entry."""
+    if not (
+        ble_device := async_ble_device_from_address(
+            hass, entry.data[CONF_ADDRESS], connectable=True
+        )
+    ):
+        raise ConfigEntryNotReady(
+            f"Could not find Avea device with address {entry.data[CONF_ADDRESS]}"
+        )
+
+    entry.runtime_data = avea.Bulb(ble_device)
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: AveaConfigEntry) -> bool:
+    """Unload an Avea config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        await hass.async_add_executor_job(entry.runtime_data.close)
+
+    return unload_ok

homeassistant/components/avea/config_flow.py

@@ -0,0 +1,137 @@
+"""Config flow for Avea."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import avea
+import voluptuous as vol
+
+from homeassistant.components.bluetooth import (
+    BluetoothServiceInfoBleak,
+    async_discovered_service_info,
+)
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_ADDRESS
+
+from .const import AVEA_SERVICE_UUID, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class CannotConnect(Exception):
+    """Error to indicate an Avea device cannot be connected to."""
+
+
+def _validate_device(discovery_info: BluetoothServiceInfoBleak) -> str:
+    """Validate the device is reachable and return a title for it."""
+    bulb = avea.Bulb(discovery_info.device)
+
+    try:
+        if not bulb.connect():
+            raise CannotConnect
+        name = bulb.get_name()
+        bulb.get_brightness()
+    finally:
+        bulb.close()
+
+    return name or discovery_info.name or discovery_info.address
+
+
+def _is_avea_discovery(discovery_info: BluetoothServiceInfoBleak) -> bool:
+    """Return if the bluetooth discovery matches an Avea bulb."""
+    return AVEA_SERVICE_UUID in discovery_info.service_uuids or bool(
+        discovery_info.name and "avea" in discovery_info.name.lower()
+    )
+
+
+class AveaConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Avea."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._discovery_info: BluetoothServiceInfoBleak | None = None
+        self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {}
+
+    async def async_step_bluetooth(
+        self, discovery_info: BluetoothServiceInfoBleak
+    ) -> ConfigFlowResult:
+        """Handle the bluetooth discovery step."""
+        await self.async_set_unique_id(discovery_info.address)
+        self._abort_if_unique_id_configured()
+        self._discovery_info = discovery_info
+        self.context["title_placeholders"] = {
+            "name": discovery_info.name or discovery_info.address
+        }
+        return await self.async_step_user()
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the user step to pick a discovered device."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            address = user_input[CONF_ADDRESS]
+            discovery_info = self._discovered_devices[address]
+            await self.async_set_unique_id(address, raise_on_progress=False)
+            self._abort_if_unique_id_configured()
+
+            try:
+                title = await self.hass.async_add_executor_job(
+                    _validate_device, discovery_info
+                )
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected error while validating Avea device")
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(
+                    title=title,
+                    data={CONF_ADDRESS: address},
+                )
+
+        if discovery := self._discovery_info:
+            self._discovered_devices[discovery.address] = discovery
+        else:
+            current_addresses = self._async_current_ids(include_ignore=False)
+            for discovery in async_discovered_service_info(self.hass):
+                if (
+                    discovery.address in current_addresses
+                    or discovery.address in self._discovered_devices
+                    or not _is_avea_discovery(discovery)
+                ):
+                    continue
+                self._discovered_devices[discovery.address] = discovery
+
+        if not self._discovered_devices:
+            return self.async_abort(reason="no_devices_found")
+
+        if self._discovery_info:
+            data_schema = vol.Schema(
+                {vol.Required(CONF_ADDRESS): self._discovery_info.address}
+            )
+        else:
+            data_schema = vol.Schema(
+                {
+                    vol.Required(CONF_ADDRESS): vol.In(
+                        {
+                            service_info.address: (
+                                f"{service_info.name or service_info.address}"
+                                f" ({service_info.address})"
+                            )
+                            for service_info in self._discovered_devices.values()
+                        }
+                    ),
+                }
+            )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=data_schema,
+            errors=errors,
+        )

homeassistant/components/avea/const.py

@@ -0,0 +1,6 @@
+"""Constants for the Avea integration."""
+
+DOMAIN = "avea"
+MANUFACTURER = "Elgato"
+MODEL = "Avea"
+AVEA_SERVICE_UUID = "f815e810-456c-6761-746f-4d756e696368"

homeassistant/components/avea/light.py

@@ -1,75 +1,166 @@
-"""Support for the Elgato Avea lights."""
+"""Light platform for Avea."""
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 
-import avea
+from bleak.exc import BleakError
 
+from homeassistant.components import bluetooth
 from homeassistant.components.light import (
     ATTR_BRIGHTNESS,
     ATTR_HS_COLOR,
     ColorMode,
     LightEntity,
 )
+from homeassistant.const import CONF_ADDRESS
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import PlatformNotReady
-from homeassistant.helpers.entity_platform import AddEntitiesCallback
-from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 from homeassistant.util import color as color_util
 
+from . import AveaConfigEntry
+from .const import DOMAIN, MANUFACTURER, MODEL
 
-def setup_platform(
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
     hass: HomeAssistant,
-    config: ConfigType,
-    add_entities: AddEntitiesCallback,
-    discovery_info: DiscoveryInfoType | None = None,
+    entry: AveaConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
 ) -> None:
-    """Set up the Avea platform."""
-    try:
-        nearby_bulbs = avea.discover_avea_bulbs()
-        for bulb in nearby_bulbs:
-            bulb.get_name()
-            bulb.get_brightness()
-    except OSError as err:
-        raise PlatformNotReady from err
-
-    add_entities(AveaLight(bulb) for bulb in nearby_bulbs)
+    """Set up the Avea light platform."""
+    async_add_entities([AveaLight(hass, entry)], update_before_add=True)
 
 
 class AveaLight(LightEntity):
     """Representation of an Avea."""
 
     _attr_color_mode = ColorMode.HS
+    _attr_has_entity_name = True
+    _attr_name = None
     _attr_supported_color_modes = {ColorMode.HS}
 
-    def __init__(self, light):
+    def __init__(self, hass: HomeAssistant, entry: AveaConfigEntry) -> None:
         """Initialize an AveaLight."""
-        self._light = light
-        self._attr_name = light.name
-        self._attr_brightness = light.brightness
+        self.hass = hass
+        self._light = entry.runtime_data
+        self._address: str = entry.data[CONF_ADDRESS]
+        self._attr_unique_id = self._address
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, self._address)},
+            connections={(CONNECTION_BLUETOOTH, self._address)},
+            manufacturer=MANUFACTURER,
+            model=MODEL,
+            name=entry.title,
+        )
+
+    def _update_ble_device(self) -> None:
+        """Update the library with the latest BLE device if available."""
+        if ble_device := bluetooth.async_ble_device_from_address(
+            self.hass, self._address, connectable=True
+        ):
+            self._light.addr = ble_device
+
+    def _sync_update(self) -> tuple[int, tuple[int, int, int]]:
+        """Fetch the latest state from the device."""
+        if not self._light.connect():
+            raise ConnectionError(f"Could not connect to Avea device {self._address}")
+
+        try:
+            self._light.get_name()
+            brightness = self._light.get_brightness()
+            rgb = self._light.get_rgb()
+        finally:
+            self._light.disconnect()
 
-    def turn_on(self, **kwargs: Any) -> None:
+        return brightness, rgb
+
+    def _sync_turn_on(
+        self, brightness: int | None, hs_color: tuple[float, float] | None
+    ) -> None:
         """Instruct the light to turn on."""
-        if not kwargs:
-            self._light.set_brightness(4095)
-        else:
-            if ATTR_BRIGHTNESS in kwargs:
-                bright = round((kwargs[ATTR_BRIGHTNESS] / 255) * 4095)
-                self._light.set_brightness(bright)
-            if ATTR_HS_COLOR in kwargs:
-                rgb = color_util.color_hs_to_RGB(*kwargs[ATTR_HS_COLOR])
+        if not self._light.connect():
+            raise ConnectionError(f"Could not connect to Avea device {self._address}")
+
+        try:
+            if brightness is None and hs_color is None:
+                self._light.set_brightness(4095)
+                return
+
+            if brightness is not None:
+                self._light.set_brightness(round((brightness / 255) * 4095))
+
+            if hs_color is not None:
+                rgb = color_util.color_hs_to_RGB(*hs_color)
                 self._light.set_rgb(rgb[0], rgb[1], rgb[2])
+                if brightness is None and not self._attr_is_on:
+                    self._light.set_brightness(4095)
+        finally:
+            self._light.disconnect()
+
+    def _sync_turn_off(self) -> None:
+        """Instruct the light to turn off."""
+        if not self._light.connect():
+            raise ConnectionError(f"Could not connect to Avea device {self._address}")
+
+        try:
+            self._light.set_brightness(0)
+        finally:
+            self._light.disconnect()
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Instruct the light to turn on."""
+        brightness: int | None = kwargs.get(ATTR_BRIGHTNESS)
+        hs_color: tuple[float, float] | None = kwargs.get(ATTR_HS_COLOR)
+        self._update_ble_device()
+
+        await self.hass.async_add_executor_job(
+            self._sync_turn_on, brightness, hs_color
+        )
+
+        if not kwargs:
+            self._attr_brightness = 255
+            self._attr_is_on = True
+            return
+
+        if hs_color is not None:
+            self._attr_hs_color = hs_color
+
+        if brightness is not None:
+            self._attr_brightness = brightness
+            self._attr_is_on = brightness > 0
+        elif hs_color is not None and not self._attr_is_on:
+            self._attr_brightness = 255
+            self._attr_is_on = True
 
-    def turn_off(self, **kwargs: Any) -> None:
+    async def async_turn_off(self, **kwargs: Any) -> None:
         """Instruct the light to turn off."""
-        self._light.set_brightness(0)
+        self._update_ble_device()
+        await self.hass.async_add_executor_job(self._sync_turn_off)
+        self._attr_is_on = False
+        self._attr_brightness = 0
 
-    def update(self) -> None:
-        """Fetch new state data for this light.
+    async def async_update(self) -> None:
+        """Fetch new state data for this light."""
+        self._update_ble_device()
+        try:
+            brightness, rgb = await self.hass.async_add_executor_job(self._sync_update)
+        except ConnectionError:
+            self._attr_available = False
+            return
+        except (BleakError, OSError, RuntimeError):
+            _LOGGER.warning(
+                "Unexpected error while updating Avea device %s",
+                self._address,
+                exc_info=True,
+            )
+            self._attr_available = False
+            return
 
-        This is the only method that should fetch new data for Home Assistant.
-        """
-        if (brightness := self._light.get_brightness()) is not None:
-            self._attr_is_on = brightness != 0
-            self._attr_brightness = round(255 * (brightness / 4095))
+        self._attr_available = True
+        self._attr_is_on = brightness != 0
+        self._attr_brightness = round(255 * (brightness / 4095))
+        self._attr_hs_color = color_util.color_RGB_to_hs(*rgb)