New integration: Hue BLE (home-assistant#118635)

Co-authored-by: Mr. Bubbles <[email protected]>
Co-authored-by: Erik Montnemery <[email protected]>
Co-authored-by: Joostlek <[email protected]>

Author: 4 people

CODEOWNERS

@@ -1,5 +1,5 @@
 {
   "domain": "philips",
   "name": "Philips",
-  "integrations": ["dynalite", "hue", "philips_js"]
+  "integrations": ["dynalite", "hue", "hue_ble", "philips_js"]
 }

homeassistant/brands/philips.json

@@ -0,0 +1,54 @@
+"""Hue BLE integration."""
+
+import logging
+
+from HueBLE import HueBleLight
+
+from homeassistant.components.bluetooth import (
+    async_ble_device_from_address,
+    async_scanner_count,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+
+_LOGGER = logging.getLogger(__name__)
+
+type HueBLEConfigEntry = ConfigEntry[HueBleLight]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: HueBLEConfigEntry) -> bool:
+    """Set up the integration from a config entry."""
+
+    assert entry.unique_id is not None
+    address = entry.unique_id.upper()
+
+    ble_device = async_ble_device_from_address(hass, address, connectable=True)
+
+    if ble_device is None:
+        count_scanners = async_scanner_count(hass, connectable=True)
+        _LOGGER.debug("Count of BLE scanners: %i", count_scanners)
+
+        if count_scanners < 1:
+            raise ConfigEntryNotReady(
+                "No Bluetooth scanners are available to search for the light."
+            )
+        raise ConfigEntryNotReady("The light was not found.")
+
+    light = HueBleLight(ble_device)
+
+    if not await light.connect() or not await light.poll_state():
+        raise ConfigEntryNotReady("Device found but unable to connect.")
+
+    entry.runtime_data = light
+
+    await hass.config_entries.async_forward_entry_setups(entry, [Platform.LIGHT])
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: HueBLEConfigEntry) -> bool:
+    """Unload a config entry."""
+
+    return await hass.config_entries.async_unload_platforms(entry, [Platform.LIGHT])

homeassistant/components/hue_ble/__init__.py

@@ -0,0 +1,155 @@
+"""Config flow for Hue BLE integration."""
+
+from __future__ import annotations
+
+from enum import Enum
+import logging
+from typing import Any
+
+from HueBLE import HueBleLight
+import voluptuous as vol
+
+from homeassistant.components import bluetooth
+from homeassistant.components.bluetooth.api import (
+    async_ble_device_from_address,
+    async_scanner_count,
+)
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_MAC, CONF_NAME
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import device_registry as dr
+
+from .const import DOMAIN, URL_PAIRING_MODE
+from .light import get_available_color_modes
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def validate_input(hass: HomeAssistant, address: str) -> Error | None:
+    """Return error if cannot connect and validate."""
+
+    ble_device = async_ble_device_from_address(hass, address.upper(), connectable=True)
+
+    if ble_device is None:
+        count_scanners = async_scanner_count(hass, connectable=True)
+        _LOGGER.debug("Count of BLE scanners in HA bt: %i", count_scanners)
+
+        if count_scanners < 1:
+            return Error.NO_SCANNERS
+        return Error.NOT_FOUND
+
+    try:
+        light = HueBleLight(ble_device)
+
+        await light.connect()
+
+        if light.authenticated is None:
+            _LOGGER.warning(
+                "Unable to determine if light authenticated, proceeding anyway"
+            )
+        elif not light.authenticated:
+            return Error.INVALID_AUTH
+
+        if not light.connected:
+            return Error.CANNOT_CONNECT
+
+        try:
+            get_available_color_modes(light)
+        except HomeAssistantError:
+            return Error.NOT_SUPPORTED
+
+        _, errors = await light.poll_state()
+        if len(errors) != 0:
+            _LOGGER.warning("Errors raised when connecting to light: %s", errors)
+            return Error.CANNOT_CONNECT
+
+    except Exception:
+        _LOGGER.exception("Unexpected error validating light connection")
+        return Error.UNKNOWN
+    else:
+        return None
+    finally:
+        await light.disconnect()
+
+
+class HueBleConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Hue BLE."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._discovery_info: bluetooth.BluetoothServiceInfoBleak | None = None
+
+    async def async_step_bluetooth(
+        self, discovery_info: bluetooth.BluetoothServiceInfoBleak
+    ) -> ConfigFlowResult:
+        """Handle a flow initialized by the home assistant scanner."""
+
+        _LOGGER.debug(
+            "HA found light %s. Will show in UI but not auto connect",
+            discovery_info.name,
+        )
+
+        unique_id = dr.format_mac(discovery_info.address)
+        await self.async_set_unique_id(unique_id)
+        self._abort_if_unique_id_configured()
+
+        name = f"{discovery_info.name} ({discovery_info.address})"
+        self.context.update({"title_placeholders": {CONF_NAME: name}})
+
+        self._discovery_info = discovery_info
+
+        return await self.async_step_confirm()
+
+    async def async_step_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Confirm a single device."""
+
+        assert self._discovery_info is not None
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            unique_id = dr.format_mac(self._discovery_info.address)
+            await self.async_set_unique_id(unique_id)
+            self._abort_if_unique_id_configured()
+            error = await validate_input(self.hass, unique_id)
+            if error:
+                errors["base"] = error.value
+            else:
+                return self.async_create_entry(title=self._discovery_info.name, data={})
+
+        return self.async_show_form(
+            step_id="confirm",
+            data_schema=vol.Schema({}),
+            errors=errors,
+            description_placeholders={
+                CONF_NAME: self._discovery_info.name,
+                CONF_MAC: self._discovery_info.address,
+                "url_pairing_mode": URL_PAIRING_MODE,
+            },
+        )
+
+
+class Error(Enum):
+    """Potential validation errors when attempting to connect."""
+
+    CANNOT_CONNECT = "cannot_connect"
+    """Error to indicate we cannot connect."""
+
+    INVALID_AUTH = "invalid_auth"
+    """Error to indicate there is invalid auth."""
+
+    NO_SCANNERS = "no_scanners"
+    """Error to indicate no bluetooth scanners are available."""
+
+    NOT_FOUND = "not_found"
+    """Error to indicate the light could not be found."""
+
+    NOT_SUPPORTED = "not_supported"
+    """Error to indicate that the light is not a supported model."""
+
+    UNKNOWN = "unknown"
+    """Error to indicate that the issue is unknown."""

homeassistant/components/hue_ble/config_flow.py

@@ -0,0 +1,4 @@
+"""Constants for the Hue BLE integration."""
+
+DOMAIN = "hue_ble"
+URL_PAIRING_MODE = "https://www.home-assistant.io/integrations/hue_ble#initial-setup"

homeassistant/components/hue_ble/const.py

@@ -0,0 +1,160 @@
+"""Hue BLE light platform."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from HueBLE import HueBleLight
+
+from homeassistant.components.light import (
+    ATTR_BRIGHTNESS,
+    ATTR_COLOR_TEMP_KELVIN,
+    ATTR_XY_COLOR,
+    ColorMode,
+    LightEntity,
+    filter_supported_color_modes,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.util import color as color_util
+
+_LOGGER = logging.getLogger(__name__)
+
+
+if TYPE_CHECKING:
+    from . import HueBLEConfigEntry
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: HueBLEConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Add light for passed config_entry in HA."""
+
+    light = config_entry.runtime_data
+    async_add_entities([HueBLELight(light)])
+
+
+def get_available_color_modes(api: HueBleLight) -> set[ColorMode]:
+    """Return a set of available color modes."""
+    color_modes = set()
+    if api.supports_colour_xy:
+        color_modes.add(ColorMode.XY)
+    if api.supports_colour_temp:
+        color_modes.add(ColorMode.COLOR_TEMP)
+    if api.supports_brightness:
+        color_modes.add(ColorMode.BRIGHTNESS)
+    if api.supports_on_off:
+        color_modes.add(ColorMode.ONOFF)
+    return filter_supported_color_modes(color_modes)
+
+
+class HueBLELight(LightEntity):
+    """Representation of a light."""
+
+    _attr_has_entity_name = True
+    _attr_name = None
+
+    def __init__(self, light: HueBleLight) -> None:
+        """Initialize the light object. Does not connect."""
+
+        self._api = light
+        self._attr_unique_id = light.address
+        self._attr_min_color_temp_kelvin = (
+            color_util.color_temperature_mired_to_kelvin(light.maximum_mireds)
+            if light.maximum_mireds
+            else None
+        )
+        self._attr_max_color_temp_kelvin = (
+            color_util.color_temperature_mired_to_kelvin(light.minimum_mireds)
+            if light.minimum_mireds
+            else None
+        )
+        self._attr_device_info = DeviceInfo(
+            name=light.name,
+            connections={(CONNECTION_BLUETOOTH, light.address)},
+            manufacturer=light.manufacturer,
+            model_id=light.model,
+            sw_version=light.firmware,
+        )
+        self._attr_supported_color_modes = get_available_color_modes(self._api)
+        self._update_updatable_attributes()
+
+    async def async_added_to_hass(self) -> None:
+        """Run when this Entity has been added to HA."""
+
+        self._api.add_callback_on_state_changed(self._state_change_callback)
+
+    async def async_will_remove_from_hass(self) -> None:
+        """Run when entity will be removed from HA."""
+
+        self._api.remove_callback(self._state_change_callback)
+
+    def _update_updatable_attributes(self) -> None:
+        """Update this entities updatable attrs from the lights state."""
+        self._attr_available = self._api.available
+        self._attr_is_on = self._api.power_state
+        self._attr_brightness = self._api.brightness
+        self._attr_color_temp_kelvin = (
+            color_util.color_temperature_mired_to_kelvin(self._api.colour_temp)
+            if self._api.colour_temp is not None and self._api.colour_temp != 0
+            else None
+        )
+        self._attr_xy_color = self._api.colour_xy
+
+    def _state_change_callback(self) -> None:
+        """Run when light informs of state update. Updates local properties."""
+        _LOGGER.debug("Received state notification from light %s", self.name)
+        self._update_updatable_attributes()
+        self.async_write_ha_state()
+
+    async def async_update(self) -> None:
+        """Fetch latest state from light and make available via properties."""
+        await self._api.poll_state(run_callbacks=True)
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Set properties then turn the light on."""
+
+        _LOGGER.debug("Turning light %s on with args %s", self.name, kwargs)
+
+        if ATTR_BRIGHTNESS in kwargs:
+            brightness = kwargs[ATTR_BRIGHTNESS]
+            _LOGGER.debug("Setting brightness of %s to %s", self.name, brightness)
+            await self._api.set_brightness(brightness)
+
+        if ATTR_COLOR_TEMP_KELVIN in kwargs:
+            color_temp_kelvin = kwargs[ATTR_COLOR_TEMP_KELVIN]
+            mireds = color_util.color_temperature_kelvin_to_mired(color_temp_kelvin)
+            _LOGGER.debug("Setting color temp of %s to %s", self.name, mireds)
+            await self._api.set_colour_temp(mireds)
+
+        if ATTR_XY_COLOR in kwargs:
+            xy_color = kwargs[ATTR_XY_COLOR]
+            _LOGGER.debug("Setting XY color of %s to %s", self.name, xy_color)
+            await self._api.set_colour_xy(xy_color[0], xy_color[1])
+
+        await self._api.set_power(True)
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn light off then set properties."""
+
+        _LOGGER.debug("Turning light %s off with args %s", self.name, kwargs)
+        await self._api.set_power(False)
+
+    @property
+    def color_mode(self) -> ColorMode:
+        """Color mode of the light."""
+
+        if self._api.supports_colour_xy and not self._api.colour_temp_mode:
+            return ColorMode.XY
+
+        if self._api.colour_temp_mode:
+            return ColorMode.COLOR_TEMP
+
+        if self._api.supports_brightness:
+            return ColorMode.BRIGHTNESS
+
+        return ColorMode.ONOFF