Add LED BLE integration (#77489)

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>

.coveragerc

@@ -659,6 +659,9 @@ omit =
     homeassistant/components/lcn/helpers.py
     homeassistant/components/lcn/scene.py
     homeassistant/components/lcn/services.py
+    homeassistant/components/led_ble/__init__.py
+    homeassistant/components/led_ble/light.py
+    homeassistant/components/led_ble/util.py
     homeassistant/components/lg_netcast/media_player.py
     homeassistant/components/lg_soundbar/media_player.py
     homeassistant/components/life360/__init__.py

CODEOWNERS

@@ -601,6 +601,8 @@ build.json @home-assistant/supervisor
 /tests/components/laundrify/ @xLarry
 /homeassistant/components/lcn/ @alengwenus
 /tests/components/lcn/ @alengwenus
+/homeassistant/components/led_ble/ @bdraco
+/tests/components/led_ble/ @bdraco
 /homeassistant/components/lg_netcast/ @Drafteed
 /homeassistant/components/life360/ @pnbruckner
 /tests/components/life360/ @pnbruckner

homeassistant/components/led_ble/__init__.py

@@ -0,0 +1,119 @@
+"""The LED BLE integration."""
+from __future__ import annotations
+
+import asyncio
+from datetime import timedelta
+import logging
+
+import async_timeout
+from led_ble import BLEAK_EXCEPTIONS, LEDBLE
+
+from homeassistant.components import bluetooth
+from homeassistant.components.bluetooth.match import ADDRESS, BluetoothCallbackMatcher
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_ADDRESS, EVENT_HOMEASSISTANT_STOP, Platform
+from homeassistant.core import Event, HomeAssistant, callback
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DEVICE_TIMEOUT, DOMAIN, UPDATE_SECONDS
+from .models import LEDBLEData
+
+PLATFORMS: list[Platform] = [Platform.LIGHT]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up LED BLE from a config entry."""
+    address: str = entry.data[CONF_ADDRESS]
+    ble_device = bluetooth.async_ble_device_from_address(hass, address.upper(), True)
+    if not ble_device:
+        raise ConfigEntryNotReady(
+            f"Could not find LED BLE device with address {address}"
+        )
+
+    led_ble = LEDBLE(ble_device)
+
+    @callback
+    def _async_update_ble(
+        service_info: bluetooth.BluetoothServiceInfoBleak,
+        change: bluetooth.BluetoothChange,
+    ) -> None:
+        """Update from a ble callback."""
+        led_ble.set_ble_device(service_info.device)
+
+    entry.async_on_unload(
+        bluetooth.async_register_callback(
+            hass,
+            _async_update_ble,
+            BluetoothCallbackMatcher({ADDRESS: address}),
+            bluetooth.BluetoothScanningMode.PASSIVE,
+        )
+    )
+
+    async def _async_update():
+        """Update the device state."""
+        try:
+            await led_ble.update()
+        except BLEAK_EXCEPTIONS as ex:
+            raise UpdateFailed(str(ex)) from ex
+
+    startup_event = asyncio.Event()
+    cancel_first_update = led_ble.register_callback(lambda *_: startup_event.set())
+    coordinator = DataUpdateCoordinator(
+        hass,
+        _LOGGER,
+        name=led_ble.name,
+        update_method=_async_update,
+        update_interval=timedelta(seconds=UPDATE_SECONDS),
+    )
+
+    try:
+        await coordinator.async_config_entry_first_refresh()
+    except ConfigEntryNotReady:
+        cancel_first_update()
+        raise
+
+    try:
+        async with async_timeout.timeout(DEVICE_TIMEOUT):
+            await startup_event.wait()
+    except asyncio.TimeoutError as ex:
+        raise ConfigEntryNotReady(
+            "Unable to communicate with the device; "
+            f"Try moving the Bluetooth adapter closer to {led_ble.name}"
+        ) from ex
+    finally:
+        cancel_first_update()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = LEDBLEData(
+        entry.title, led_ble, coordinator
+    )
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    entry.async_on_unload(entry.add_update_listener(_async_update_listener))
+
+    async def _async_stop(event: Event) -> None:
+        """Close the connection."""
+        await led_ble.stop()
+
+    entry.async_on_unload(
+        hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_stop)
+    )
+    return True
+
+
+async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Handle options update."""
+    data: LEDBLEData = hass.data[DOMAIN][entry.entry_id]
+    if entry.title != data.title:
+        await hass.config_entries.async_reload(entry.entry_id)
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        data: LEDBLEData = hass.data[DOMAIN].pop(entry.entry_id)
+        await data.device.stop()
+
+    return unload_ok

homeassistant/components/led_ble/config_flow.py

@@ -0,0 +1,117 @@
+"""Config flow for LEDBLE integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from led_ble import BLEAK_EXCEPTIONS, LEDBLE
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.components.bluetooth import (
+    BluetoothServiceInfoBleak,
+    async_discovered_service_info,
+)
+from homeassistant.const import CONF_ADDRESS
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DOMAIN, LOCAL_NAMES, UNSUPPORTED_SUB_MODEL
+from .util import human_readable_name
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Yale Access Bluetooth."""
+
+    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
+    ) -> FlowResult:
+        """Handle the bluetooth discovery step."""
+        if discovery_info.name.startswith(UNSUPPORTED_SUB_MODEL):
+            # These versions speak a different protocol
+            # that we do not support yet.
+            return self.async_abort(reason="not_supported")
+        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": human_readable_name(
+                None, discovery_info.name, discovery_info.address
+            )
+        }
+        return await self.async_step_user()
+
+    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] = {}
+
+        if user_input is not None:
+            address = user_input[CONF_ADDRESS]
+            discovery_info = self._discovered_devices[address]
+            local_name = discovery_info.name
+            await self.async_set_unique_id(
+                discovery_info.address, raise_on_progress=False
+            )
+            self._abort_if_unique_id_configured()
+            led_ble = LEDBLE(discovery_info.device)
+            try:
+                await led_ble.update()
+            except BLEAK_EXCEPTIONS:
+                errors["base"] = "cannot_connect"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected error")
+                errors["base"] = "unknown"
+            else:
+                await led_ble.stop()
+                return self.async_create_entry(
+                    title=local_name,
+                    data={
+                        CONF_ADDRESS: discovery_info.address,
+                    },
+                )
+
+        if discovery := self._discovery_info:
+            self._discovered_devices[discovery.address] = discovery
+        else:
+            current_addresses = self._async_current_ids()
+            for discovery in async_discovered_service_info(self.hass):
+                if (
+                    discovery.address in current_addresses
+                    or discovery.address in self._discovered_devices
+                    or not any(
+                        discovery.name.startswith(local_name)
+                        and not discovery.name.startswith(UNSUPPORTED_SUB_MODEL)
+                        for local_name in LOCAL_NAMES
+                    )
+                ):
+                    continue
+                self._discovered_devices[discovery.address] = discovery
+
+        if not self._discovered_devices:
+            return self.async_abort(reason="no_unconfigured_devices")
+
+        data_schema = vol.Schema(
+            {
+                vol.Required(CONF_ADDRESS): vol.In(
+                    {
+                        service_info.address: f"{service_info.name} ({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/led_ble/const.py

@@ -0,0 +1,10 @@
+"""Constants for the LED BLE integration."""
+
+DOMAIN = "led_ble"
+
+DEVICE_TIMEOUT = 30
+LOCAL_NAMES = {"LEDnet", "BLE-LED", "LEDBLE", "Triones", "LEDBlue"}
+
+UNSUPPORTED_SUB_MODEL = "LEDnetWF"
+
+UPDATE_SECONDS = 15

homeassistant/components/led_ble/light.py

@@ -0,0 +1,99 @@
+"""LED BLE integration light platform."""
+from __future__ import annotations
+
+from typing import Any
+
+from led_ble import LEDBLE
+
+from homeassistant.components.light import (
+    ATTR_BRIGHTNESS,
+    ATTR_RGB_COLOR,
+    ATTR_WHITE,
+    ColorMode,
+    LightEntity,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+    DataUpdateCoordinator,
+)
+
+from .const import DOMAIN
+from .models import LEDBLEData
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the light platform for LEDBLE."""
+    data: LEDBLEData = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities([LEDBLEEntity(data.coordinator, data.device, entry.title)])
+
+
+class LEDBLEEntity(CoordinatorEntity, LightEntity):
+    """Representation of LEDBLE device."""
+
+    _attr_supported_color_modes = {ColorMode.RGB, ColorMode.WHITE}
+    _attr_has_entity_name = True
+
+    def __init__(
+        self, coordinator: DataUpdateCoordinator, device: LEDBLE, name: str
+    ) -> None:
+        """Initialize an ledble light."""
+        super().__init__(coordinator)
+        self._device = device
+        self._attr_unique_id = device._address
+        self._attr_device_info = DeviceInfo(
+            name=name,
+            model=hex(device.model_num),
+            sw_version=hex(device.version_num),
+            connections={(dr.CONNECTION_BLUETOOTH, device._address)},
+        )
+        self._async_update_attrs()
+
+    @callback
+    def _async_update_attrs(self) -> None:
+        """Handle updating _attr values."""
+        device = self._device
+        self._attr_color_mode = ColorMode.WHITE if device.w else ColorMode.RGB
+        self._attr_brightness = device.brightness
+        self._attr_rgb_color = device.rgb_unscaled
+        self._attr_is_on = device.on
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Instruct the light to turn on."""
+        brightness = kwargs.get(ATTR_BRIGHTNESS, self.brightness)
+        if ATTR_RGB_COLOR in kwargs:
+            rgb = kwargs[ATTR_RGB_COLOR]
+            await self._device.set_rgb(rgb, brightness)
+            return
+        if ATTR_BRIGHTNESS in kwargs:
+            await self._device.set_brightness(brightness)
+            return
+        if ATTR_WHITE in kwargs:
+            await self._device.set_white(kwargs[ATTR_WHITE])
+            return
+        await self._device.turn_on()
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Instruct the light to turn off."""
+        await self._device.turn_off()
+
+    @callback
+    def _handle_coordinator_update(self, *args: Any) -> None:
+        """Handle data update."""
+        self._async_update_attrs()
+        self.async_write_ha_state()
+
+    async def async_added_to_hass(self) -> None:
+        """Register callbacks."""
+        self.async_on_remove(
+            self._device.register_callback(self._handle_coordinator_update)
+        )
+        return await super().async_added_to_hass()

homeassistant/components/led_ble/manifest.json

@@ -0,0 +1,17 @@
+{
+  "domain": "led_ble",
+  "name": "LED BLE",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/ble_ble",
+  "requirements": ["led-ble==0.5.4"],
+  "dependencies": ["bluetooth"],
+  "codeowners": ["@bdraco"],
+  "bluetooth": [
+    { "local_name": "LEDnet*" },
+    { "local_name": "BLE-LED*" },
+    { "local_name": "LEDBLE*" },
+    { "local_name": "Triones*" },
+    { "local_name": "LEDBlue*" }
+  ],
+  "iot_class": "local_polling"
+}

homeassistant/components/led_ble/models.py

@@ -0,0 +1,17 @@
+"""The led ble integration models."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from led_ble import LEDBLE
+
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+
+@dataclass
+class LEDBLEData:
+    """Data for the led ble integration."""
+
+    title: str
+    device: LEDBLE
+    coordinator: DataUpdateCoordinator

homeassistant/components/led_ble/strings.json

@@ -0,0 +1,23 @@
+{
+  "config": {
+    "flow_title": "{name}",
+    "step": {
+      "user": {
+        "data": {
+          "address": "Bluetooth address"
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "abort": {
+      "not_supported": "Device not supported",
+      "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+      "no_unconfigured_devices": "No unconfigured devices found.",
+      "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]"
+    }
+  }
+}

homeassistant/components/led_ble/translations/en.json

@@ -0,0 +1,23 @@
+{
+    "config": {
+        "abort": {
+            "already_configured": "Device is already configured",
+            "already_in_progress": "Configuration flow is already in progress",
+            "no_devices_found": "No devices found on the network",
+            "no_unconfigured_devices": "No unconfigured devices found.",
+            "not_supported": "Device not supported"
+        },
+        "error": {
+            "cannot_connect": "Failed to connect",
+            "unknown": "Unexpected error"
+        },
+        "flow_title": "{name}",
+        "step": {
+            "user": {
+                "data": {
+                    "address": "Bluetooth address"
+                }
+            }
+        }
+    }
+}

homeassistant/components/led_ble/util.py

@@ -0,0 +1,51 @@
+"""The yalexs_ble integration models."""
+from __future__ import annotations
+
+from homeassistant.components.bluetooth import (
+    BluetoothScanningMode,
+    BluetoothServiceInfoBleak,
+    async_discovered_service_info,
+    async_process_advertisements,
+)
+from homeassistant.components.bluetooth.match import ADDRESS, BluetoothCallbackMatcher
+from homeassistant.core import HomeAssistant, callback
+
+from .const import DEVICE_TIMEOUT
+
+
+@callback
+def async_find_existing_service_info(
+    hass: HomeAssistant, local_name: str, address: str
+) -> BluetoothServiceInfoBleak | None:
+    """Return the service info for the given local_name and address."""
+    for service_info in async_discovered_service_info(hass):
+        device = service_info.device
+        if device.address == address:
+            return service_info
+    return None
+
+
+async def async_get_service_info(
+    hass: HomeAssistant, local_name: str, address: str
+) -> BluetoothServiceInfoBleak:
+    """Wait for the service info for the given local_name and address."""
+    if service_info := async_find_existing_service_info(hass, local_name, address):
+        return service_info
+    return await async_process_advertisements(
+        hass,
+        lambda service_info: True,
+        BluetoothCallbackMatcher({ADDRESS: address}),
+        BluetoothScanningMode.ACTIVE,
+        DEVICE_TIMEOUT,
+    )
+
+
+def short_address(address: str) -> str:
+    """Convert a Bluetooth address to a short address."""
+    split_address = address.replace("-", ":").split(":")
+    return f"{split_address[-2].upper()}{split_address[-1].upper()}"[-4:]
+
+
+def human_readable_name(name: str | None, local_name: str, address: str) -> str:
+    """Return a human readable name for the given name, local_name, and address."""
+    return f"{name or local_name} ({short_address(address)})"

homeassistant/generated/bluetooth.py

@@ -118,6 +118,26 @@ BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [
         "local_name": "tps",
         "connectable": False
     },
+    {
+        "domain": "led_ble",
+        "local_name": "LEDnet*"
+    },
+    {
+        "domain": "led_ble",
+        "local_name": "BLE-LED*"
+    },
+    {
+        "domain": "led_ble",
+        "local_name": "LEDBLE*"
+    },
+    {
+        "domain": "led_ble",
+        "local_name": "Triones*"
+    },
+    {
+        "domain": "led_ble",
+        "local_name": "LEDBlue*"
+    },
     {
         "domain": "moat",
         "local_name": "Moat_S*",

homeassistant/generated/config_flows.py

@@ -201,6 +201,7 @@ FLOWS = {
         "landisgyr_heat_meter",
         "launch_library",
         "laundrify",
+        "led_ble",
         "lg_soundbar",
         "life360",
         "lifx",

requirements_all.txt

@@ -961,6 +961,9 @@ lakeside==0.12
 # homeassistant.components.laundrify
 laundrify_aio==1.1.2
 
+# homeassistant.components.led_ble
+led-ble==0.5.4
+
 # homeassistant.components.foscam
 libpyfoscam==1.0
 

requirements_test_all.txt

@@ -699,6 +699,9 @@ lacrosse-view==0.0.9
 # homeassistant.components.laundrify
 laundrify_aio==1.1.2
 
+# homeassistant.components.led_ble
+led-ble==0.5.4
+
 # homeassistant.components.foscam
 libpyfoscam==1.0
 

tests/components/led_ble/__init__.py

@@ -0,0 +1,51 @@
+"""Tests for the LED BLE Bluetooth integration."""
+from bleak.backends.device import BLEDevice
+from bleak.backends.scanner import AdvertisementData
+
+from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
+
+LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
+    name="Triones:F30200000152C",
+    address="AA:BB:CC:DD:EE:FF",
+    rssi=-60,
+    manufacturer_data={},
+    service_uuids=[],
+    service_data={},
+    source="local",
+    device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Triones:F30200000152C"),
+    advertisement=AdvertisementData(),
+    time=0,
+    connectable=True,
+)
+
+UNSUPPORTED_LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
+    name="LEDnetWFF30200000152C",
+    address="AA:BB:CC:DD:EE:FF",
+    rssi=-60,
+    manufacturer_data={},
+    service_uuids=[],
+    service_data={},
+    source="local",
+    device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="LEDnetWFF30200000152C"),
+    advertisement=AdvertisementData(),
+    time=0,
+    connectable=True,
+)
+
+
+NOT_LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
+    name="Not",
+    address="AA:BB:CC:DD:EE:FF",
+    rssi=-60,
+    manufacturer_data={
+        33: b"\x00\x00\xd1\xf0b;\xd8\x1dE\xd6\xba\xeeL\xdd]\xf5\xb2\xe9",
+        21: b"\x061\x00Z\x8f\x93\xb2\xec\x85\x06\x00i\x00\x02\x02Q\xed\x1d\xf0",
+    },
+    service_uuids=[],
+    service_data={},
+    source="local",
+    device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Aug"),
+    advertisement=AdvertisementData(),
+    time=0,
+    connectable=True,
+)

tests/components/led_ble/conftest.py

@@ -0,0 +1,8 @@
+"""led_ble session fixtures."""
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def mock_bluetooth(enable_bluetooth):
+    """Auto mock bluetooth."""

tests/components/led_ble/test_config_flow.py

@@ -0,0 +1,229 @@
+"""Test the LED BLE Bluetooth config flow."""
+from unittest.mock import patch
+
+from bleak import BleakError
+
+from homeassistant import config_entries
+from homeassistant.components.led_ble.const import DOMAIN
+from homeassistant.const import CONF_ADDRESS
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from . import (
+    LED_BLE_DISCOVERY_INFO,
+    NOT_LED_BLE_DISCOVERY_INFO,
+    UNSUPPORTED_LED_BLE_DISCOVERY_INFO,
+)
+
+from tests.common import MockConfigEntry
+
+
+async def test_user_step_success(hass: HomeAssistant) -> None:
+    """Test user step success path."""
+    with patch(
+        "homeassistant.components.led_ble.config_flow.async_discovered_service_info",
+        return_value=[NOT_LED_BLE_DISCOVERY_INFO, LED_BLE_DISCOVERY_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch("homeassistant.components.led_ble.config_flow.LEDBLE.update",), patch(
+        "homeassistant.components.led_ble.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.CREATE_ENTRY
+    assert result2["title"] == LED_BLE_DISCOVERY_INFO.name
+    assert result2["data"] == {
+        CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+    }
+    assert result2["result"].unique_id == LED_BLE_DISCOVERY_INFO.address
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_user_step_no_devices_found(hass: HomeAssistant) -> None:
+    """Test user step with no devices found."""
+    with patch(
+        "homeassistant.components.led_ble.config_flow.async_discovered_service_info",
+        return_value=[NOT_LED_BLE_DISCOVERY_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "no_unconfigured_devices"
+
+
+async def test_user_step_no_new_devices_found(hass: HomeAssistant) -> None:
+    """Test user step with only existing devices found."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+        },
+        unique_id=LED_BLE_DISCOVERY_INFO.address,
+    )
+    entry.add_to_hass(hass)
+    with patch(
+        "homeassistant.components.led_ble.config_flow.async_discovered_service_info",
+        return_value=[LED_BLE_DISCOVERY_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "no_unconfigured_devices"
+
+
+async def test_user_step_cannot_connect(hass: HomeAssistant) -> None:
+    """Test user step and we cannot connect."""
+    with patch(
+        "homeassistant.components.led_ble.config_flow.async_discovered_service_info",
+        return_value=[LED_BLE_DISCOVERY_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch(
+        "homeassistant.components.led_ble.config_flow.LEDBLE.update",
+        side_effect=BleakError,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["step_id"] == "user"
+    assert result2["errors"] == {"base": "cannot_connect"}
+
+    with patch("homeassistant.components.led_ble.config_flow.LEDBLE.update",), patch(
+        "homeassistant.components.led_ble.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {
+                CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == FlowResultType.CREATE_ENTRY
+    assert result3["title"] == LED_BLE_DISCOVERY_INFO.name
+    assert result3["data"] == {
+        CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+    }
+    assert result3["result"].unique_id == LED_BLE_DISCOVERY_INFO.address
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_user_step_unknown_exception(hass: HomeAssistant) -> None:
+    """Test user step with an unknown exception."""
+    with patch(
+        "homeassistant.components.led_ble.config_flow.async_discovered_service_info",
+        return_value=[NOT_LED_BLE_DISCOVERY_INFO, LED_BLE_DISCOVERY_INFO],
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch(
+        "homeassistant.components.led_ble.config_flow.LEDBLE.update",
+        side_effect=RuntimeError,
+    ):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.FORM
+    assert result2["step_id"] == "user"
+    assert result2["errors"] == {"base": "unknown"}
+
+    with patch("homeassistant.components.led_ble.config_flow.LEDBLE.update",), patch(
+        "homeassistant.components.led_ble.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result3 = await hass.config_entries.flow.async_configure(
+            result2["flow_id"],
+            {
+                CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result3["type"] == FlowResultType.CREATE_ENTRY
+    assert result3["title"] == LED_BLE_DISCOVERY_INFO.name
+    assert result3["data"] == {
+        CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+    }
+    assert result3["result"].unique_id == LED_BLE_DISCOVERY_INFO.address
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_bluetooth_step_success(hass: HomeAssistant) -> None:
+    """Test bluetooth step success path."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=LED_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch("homeassistant.components.led_ble.config_flow.LEDBLE.update",), patch(
+        "homeassistant.components.led_ble.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {
+                CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+            },
+        )
+        await hass.async_block_till_done()
+
+    assert result2["type"] == FlowResultType.CREATE_ENTRY
+    assert result2["title"] == LED_BLE_DISCOVERY_INFO.name
+    assert result2["data"] == {
+        CONF_ADDRESS: LED_BLE_DISCOVERY_INFO.address,
+    }
+    assert result2["result"].unique_id == LED_BLE_DISCOVERY_INFO.address
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_bluetooth_unsupported_model(hass: HomeAssistant) -> None:
+    """Test bluetooth step with an unsupported model path."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=UNSUPPORTED_LED_BLE_DISCOVERY_INFO,
+    )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "not_supported"