New integration: Hue BLE

homeassistant/brands/philips.json

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

homeassistant/components/hue_ble/__init__.py

@@ -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.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
+
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setup(entry, "light")
+    )
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: HueBLEConfigEntry) -> bool:
+    """Unload a config entry."""
+
+    return await hass.config_entries.async_forward_entry_unload(entry, "light")

homeassistant/components/hue_ble/config_flow.py

@@ -0,0 +1,146 @@
+"""Config flow for Hue BLE integration."""
+
+from __future__ import annotations
+
+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
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def validate_input(hass: HomeAssistant, address: str) -> None:
+    """Validate that we can connect."""
+
+    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:
+            raise ScannerNotAvailable
+        raise NotFound
+
+    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:
+            raise InvalidAuth
+
+        if not light.connected:
+            raise CannotConnect
+
+        state_changed, errors = await light.poll_state()
+        if not len(errors) == 0:
+            raise CannotConnect
+    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:
+            try:
+                unique_id = dr.format_mac(self._discovery_info.address)
+                await self.async_set_unique_id(unique_id)
+                self._abort_if_unique_id_configured()
+                await validate_input(self.hass, unique_id)
+
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidAuth:
+                errors["base"] = "invalid_auth"
+            except ScannerNotAvailable:
+                errors["base"] = "no_scanners"
+            except NotFound:
+                errors["base"] = "not_found"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            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 CannotConnect(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""
+
+
+class ScannerNotAvailable(HomeAssistantError):
+    """Error to indicate no bluetooth scanners are available."""
+
+
+class NotFound(HomeAssistantError):
+    """Error to indicate the light could not be found."""

homeassistant/components/hue_ble/const.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/light.py

@@ -0,0 +1,168 @@
+"""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,
+    ATTR_XY_COLOR,
+    ColorMode,
+    LightEntity,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+_LOGGER = logging.getLogger(__name__)
+
+
+if TYPE_CHECKING:
+    from . import HueBLEConfigEntry
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: HueBLEConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Add light for passed config_entry in HA."""
+
+    light = config_entry.runtime_data
+    async_add_entities([HaHueBLE(light)])
+
+
+class HaHueBLE(LightEntity):
+    """Representation of a light."""
+
+    _attr_has_entity_name = True
+    _attr_name = None
+
+    def __init__(self, api: HueBleLight) -> None:
+        """Initialize the light object. Does not connect."""
+
+        self._light = api
+        self._name = self._light.name
+        self._address = self._light.address
+        self._attr_unique_id = self._light.address
+        self._attr_available = self._light.available
+        self._attr_is_on = self._light.power_state
+        self._attr_brightness = self._light.brightness
+        self._attr_color_temp = self._light.colour_temp
+        self._attr_xy_color = self._light.colour_xy
+        self._attr_min_mireds = self._light.minimum_mireds
+        self._attr_max_mireds = self._light.maximum_mireds
+        self._attr_device_info = DeviceInfo(
+            connections={(CONNECTION_BLUETOOTH, self._light.address)},
+            manufacturer=self._light.manufacturer,
+            model=self._light.model,
+            sw_version=self._light.firmware,
+        )
+
+    async def async_added_to_hass(self) -> None:
+        """Run when this Entity has been added to HA."""
+
+        self._light.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._light.remove_callback(self._state_change_callback)
+
+    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.async_write_ha_state()
+
+    async def async_update(self) -> None:
+        """Fetch latest state from light and make available via properties."""
+        await self._light.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._light.set_brightness(brightness)
+
+        if ATTR_COLOR_TEMP in kwargs:
+            mireds = kwargs[ATTR_COLOR_TEMP]
+            _LOGGER.debug("Setting color temp of %s to %s", self.name, mireds)
+            await self._light.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._light.set_colour_xy(xy_color[0], xy_color[1])
+
+        await self._light.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._light.set_power(False)
+
+        if ATTR_BRIGHTNESS in kwargs:
+            brightness = kwargs[ATTR_BRIGHTNESS]
+            _LOGGER.debug("Setting brightness of %s to %s", self.name, brightness)
+            await self._light.set_brightness(brightness)
+
+        if ATTR_COLOR_TEMP in kwargs:
+            mireds = kwargs[ATTR_COLOR_TEMP]
+            _LOGGER.debug("Setting color temp of %s to %s", self.name, mireds)
+            await self._light.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._light.set_colour_xy(xy_color[0], xy_color[1])
+
+    @property
+    def supported_color_modes(self) -> set[ColorMode] | None:
+        """Flag supported color modes."""
+
+        supported_modes = set()
+
+        if self._light.supports_colour_xy:
+            supported_modes.add(ColorMode.XY)
+
+        if self._light.supports_colour_temp:
+            supported_modes.add(ColorMode.COLOR_TEMP)
+
+        if self._light.supports_brightness and len(supported_modes) == 0:
+            supported_modes.add(ColorMode.BRIGHTNESS)
+
+        if self._light.supports_on_off and len(supported_modes) == 0:
+            supported_modes.add(ColorMode.ONOFF)
+
+        if len(supported_modes) == 0:
+            supported_modes.add(ColorMode.UNKNOWN)
+
+        return supported_modes
+
+    @property
+    def color_mode(self) -> ColorMode | None:
+        """Color mode of the light."""
+
+        if self._light.supports_colour_xy:
+            if self._light.supports_colour_temp and self._light.colour_temp_mode:
+                return ColorMode.COLOR_TEMP
+            return ColorMode.XY
+
+        if self._light.supports_brightness:
+            return ColorMode.BRIGHTNESS
+
+        if self._light.supports_on_off:
+            return ColorMode.ONOFF
+
+        return ColorMode.UNKNOWN