Add LeaOne integration (#108617)

CODEOWNERS

@@ -710,6 +710,8 @@ build.json @home-assistant/supervisor
 /tests/components/lcn/ @alengwenus
 /homeassistant/components/ld2410_ble/ @930913
 /tests/components/ld2410_ble/ @930913
+/homeassistant/components/leaone/ @bdraco
+/tests/components/leaone/ @bdraco
 /homeassistant/components/led_ble/ @bdraco
 /tests/components/led_ble/ @bdraco
 /homeassistant/components/lg_netcast/ @Drafteed

homeassistant/components/leaone/__init__.py

@@ -0,0 +1,49 @@
+"""The Leaone integration."""
+from __future__ import annotations
+
+import logging
+
+from leaone_ble import LeaoneBluetoothDeviceData
+
+from homeassistant.components.bluetooth import BluetoothScanningMode
+from homeassistant.components.bluetooth.passive_update_processor import (
+    PassiveBluetoothProcessorCoordinator,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+
+PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Leaone BLE device from a config entry."""
+    address = entry.unique_id
+    assert address is not None
+    data = LeaoneBluetoothDeviceData()
+    coordinator = hass.data.setdefault(DOMAIN, {})[
+        entry.entry_id
+    ] = PassiveBluetoothProcessorCoordinator(
+        hass,
+        _LOGGER,
+        address=address,
+        mode=BluetoothScanningMode.PASSIVE,
+        update_method=data.update,
+    )
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    entry.async_on_unload(
+        coordinator.async_start()
+    )  # only start after all platforms have had a chance to subscribe
+    return True
+
+
+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):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/leaone/config_flow.py

@@ -0,0 +1,57 @@
+"""Config flow for Leaone integration."""
+from __future__ import annotations
+
+from typing import Any
+
+from leaone_ble import LeaoneBluetoothDeviceData as DeviceData
+import voluptuous as vol
+
+from homeassistant.components.bluetooth import async_discovered_service_info
+from homeassistant.config_entries import ConfigFlow
+from homeassistant.const import CONF_ADDRESS
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DOMAIN
+
+
+class LeaoneConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for leaone."""
+
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+        self._discovered_devices: dict[str, str] = {}
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the user step to pick discovered device."""
+        if user_input is not None:
+            address = user_input[CONF_ADDRESS]
+            await self.async_set_unique_id(address, raise_on_progress=False)
+            self._abort_if_unique_id_configured()
+            return self.async_create_entry(
+                title=self._discovered_devices[address], data={}
+            )
+
+        current_addresses = self._async_current_ids()
+        for discovery_info in async_discovered_service_info(self.hass, False):
+            address = discovery_info.address
+            if address in current_addresses or address in self._discovered_devices:
+                continue
+            device = DeviceData()
+            if device.supported(discovery_info):
+                self._discovered_devices[address] = (
+                    device.title or device.get_device_name() or discovery_info.name
+                )
+
+        if not self._discovered_devices:
+            return self.async_abort(reason="no_devices_found")
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {vol.Required(CONF_ADDRESS): vol.In(self._discovered_devices)}
+            ),
+        )

homeassistant/components/leaone/const.py

@@ -0,0 +1,3 @@
+"""Constants for the Leaone integration."""
+
+DOMAIN = "leaone"

homeassistant/components/leaone/device.py

@@ -0,0 +1,15 @@
+"""Support for Leaone devices."""
+from __future__ import annotations
+
+from leaone_ble import DeviceKey
+
+from homeassistant.components.bluetooth.passive_update_processor import (
+    PassiveBluetoothEntityKey,
+)
+
+
+def device_key_to_bluetooth_entity_key(
+    device_key: DeviceKey,
+) -> PassiveBluetoothEntityKey:
+    """Convert a device key to an entity key."""
+    return PassiveBluetoothEntityKey(device_key.key, device_key.device_id)

homeassistant/components/leaone/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "leaone",
+  "name": "LeaOne",
+  "codeowners": ["@bdraco"],
+  "config_flow": true,
+  "dependencies": ["bluetooth_adapters"],
+  "documentation": "https://www.home-assistant.io/integrations/leaone",
+  "iot_class": "local_push",
+  "requirements": ["leaone-ble==0.1.0"]
+}

homeassistant/components/leaone/sensor.py

@@ -0,0 +1,152 @@
+"""Support for Leaone sensors."""
+from __future__ import annotations
+
+from leaone_ble import DeviceClass as LeaoneSensorDeviceClass, SensorUpdate, Units
+
+from homeassistant import config_entries
+from homeassistant.components.bluetooth.passive_update_processor import (
+    PassiveBluetoothDataProcessor,
+    PassiveBluetoothDataUpdate,
+    PassiveBluetoothProcessorCoordinator,
+    PassiveBluetoothProcessorEntity,
+)
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorEntityDescription,
+    SensorStateClass,
+)
+from homeassistant.const import (
+    SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
+    EntityCategory,
+    UnitOfMass,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.sensor import sensor_device_info_to_hass_device_info
+
+from .const import DOMAIN
+from .device import device_key_to_bluetooth_entity_key
+
+SENSOR_DESCRIPTIONS = {
+    (
+        LeaoneSensorDeviceClass.MASS_NON_STABILIZED,
+        Units.MASS_KILOGRAMS,
+    ): SensorEntityDescription(
+        key=f"{LeaoneSensorDeviceClass.MASS_NON_STABILIZED}_{Units.MASS_KILOGRAMS}",
+        device_class=SensorDeviceClass.WEIGHT,
+        native_unit_of_measurement=UnitOfMass.KILOGRAMS,
+        state_class=SensorStateClass.MEASUREMENT,
+        entity_category=EntityCategory.DIAGNOSTIC,
+    ),
+    (LeaoneSensorDeviceClass.MASS, Units.MASS_KILOGRAMS): SensorEntityDescription(
+        key=f"{LeaoneSensorDeviceClass.MASS}_{Units.MASS_KILOGRAMS}",
+        device_class=SensorDeviceClass.WEIGHT,
+        native_unit_of_measurement=UnitOfMass.KILOGRAMS,
+        state_class=SensorStateClass.MEASUREMENT,
+    ),
+    (LeaoneSensorDeviceClass.IMPEDANCE, Units.OHM): SensorEntityDescription(
+        key=f"{LeaoneSensorDeviceClass.IMPEDANCE}_{Units.OHM}",
+        icon="mdi:omega",
+        native_unit_of_measurement=Units.OHM,
+        state_class=SensorStateClass.MEASUREMENT,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        entity_registry_enabled_default=False,
+    ),
+    (
+        LeaoneSensorDeviceClass.SIGNAL_STRENGTH,
+        Units.SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
+    ): SensorEntityDescription(
+        key=f"{LeaoneSensorDeviceClass.SIGNAL_STRENGTH}_{Units.SIGNAL_STRENGTH_DECIBELS_MILLIWATT}",
+        device_class=SensorDeviceClass.SIGNAL_STRENGTH,
+        native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
+        state_class=SensorStateClass.MEASUREMENT,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        entity_registry_enabled_default=False,
+    ),
+    (
+        LeaoneSensorDeviceClass.PACKET_ID,
+        None,
+    ): SensorEntityDescription(
+        key=str(LeaoneSensorDeviceClass.PACKET_ID),
+        entity_category=EntityCategory.DIAGNOSTIC,
+        state_class=SensorStateClass.MEASUREMENT,
+        entity_registry_enabled_default=False,
+    ),
+}
+
+
+def sensor_update_to_bluetooth_data_update(
+    sensor_update: SensorUpdate,
+) -> PassiveBluetoothDataUpdate:
+    """Convert a sensor update to a bluetooth data update."""
+    return PassiveBluetoothDataUpdate(
+        devices={
+            device_id: sensor_device_info_to_hass_device_info(device_info)
+            for device_id, device_info in sensor_update.devices.items()
+        },
+        entity_descriptions={
+            device_key_to_bluetooth_entity_key(device_key): SENSOR_DESCRIPTIONS[
+                (description.device_class, description.native_unit_of_measurement)
+            ]
+            for device_key, description in sensor_update.entity_descriptions.items()
+            if description.device_class and description.native_unit_of_measurement
+        },
+        entity_data={
+            device_key_to_bluetooth_entity_key(device_key): sensor_values.native_value
+            for device_key, sensor_values in sensor_update.entity_values.items()
+        },
+        entity_names={
+            device_key_to_bluetooth_entity_key(device_key): sensor_values.name
+            for device_key, sensor_values in sensor_update.entity_values.items()
+        },
+    )
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: config_entries.ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the Leaone BLE sensors."""
+    coordinator: PassiveBluetoothProcessorCoordinator = hass.data[DOMAIN][
+        entry.entry_id
+    ]
+    processor = PassiveBluetoothDataProcessor(sensor_update_to_bluetooth_data_update)
+    entry.async_on_unload(
+        processor.async_add_entities_listener(
+            LeaoneBluetoothSensorEntity, async_add_entities
+        )
+    )
+    entry.async_on_unload(
+        coordinator.async_register_processor(processor, SensorEntityDescription)
+    )
+
+
+class LeaoneBluetoothSensorEntity(
+    PassiveBluetoothProcessorEntity[PassiveBluetoothDataProcessor[float | int | None]],
+    SensorEntity,
+):
+    """Representation of a Leaone sensor."""
+
+    @property
+    def native_value(self) -> int | float | None:
+        """Return the native value."""
+        return self.processor.entity_data.get(self.entity_key)
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available.
+
+        The sensor is only created when the device is seen.
+
+        Since these are sleepy devices which stop broadcasting
+        when not in use, we can't rely on the last update time
+        so once we have seen the device we always return True.
+        """
+        return True
+
+    @property
+    def assumed_state(self) -> bool:
+        """Return True if the device is no longer broadcasting."""
+        return not self.processor.available

homeassistant/components/leaone/strings.json

@@ -0,0 +1,21 @@
+{
+  "config": {
+    "flow_title": "[%key:component::bluetooth::config::flow_title%]",
+    "step": {
+      "user": {
+        "description": "[%key:component::bluetooth::config::step::user::description%]",
+        "data": {
+          "address": "[%key:common::config_flow::data::device%]"
+        }
+      },
+      "bluetooth_confirm": {
+        "description": "[%key:component::bluetooth::config::step::bluetooth_confirm::description%]"
+      }
+    },
+    "abort": {
+      "no_devices_found": "No supported LeaOne devices found in range; If the device is in range, ensure it has been activated in the last few minutes. If you need clarification on whether the device is in-range, download the diagnostics for the integration that provides your Bluetooth adapter or proxy and check if the MAC address of the LeaOne device is present.",
+      "already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -270,6 +270,7 @@ FLOWS = {
         "launch_library",
         "laundrify",
         "ld2410_ble",
+        "leaone",
         "led_ble",
         "lg_soundbar",
         "lidarr",

homeassistant/generated/integrations.json

@@ -3075,6 +3075,12 @@
       "config_flow": true,
       "iot_class": "local_push"
     },
+    "leaone": {
+      "name": "LeaOne",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "local_push"
+    },
     "led_ble": {
       "name": "LED BLE",
       "integration_type": "hub",

requirements_all.txt

@@ -1183,6 +1183,9 @@ laundrify-aio==1.1.2
 # homeassistant.components.ld2410_ble
 ld2410-ble==0.1.1
 
+# homeassistant.components.leaone
+leaone-ble==0.1.0
+
 # homeassistant.components.led_ble
 led-ble==1.0.1
 

requirements_test_all.txt

@@ -943,6 +943,9 @@ laundrify-aio==1.1.2
 # homeassistant.components.ld2410_ble
 ld2410-ble==0.1.1
 
+# homeassistant.components.leaone
+leaone-ble==0.1.0
+
 # homeassistant.components.led_ble
 led-ble==1.0.1
 

tests/components/leaone/__init__.py

@@ -0,0 +1,39 @@
+"""Tests for the Leaone integration."""
+
+
+from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo
+
+SCALE_SERVICE_INFO = BluetoothServiceInfo(
+    name="",
+    address="5F:5A:5C:52:D3:94",
+    rssi=-63,
+    manufacturer_data={57280: b"\x06\xa4\x00\x00\x00\x020_Z\\R\xd3\x94"},
+    service_uuids=[],
+    service_data={},
+    source="local",
+)
+SCALE_SERVICE_INFO_2 = BluetoothServiceInfo(
+    name="",
+    address="5F:5A:5C:52:D3:94",
+    rssi=-63,
+    manufacturer_data={
+        57280: b"\x06\xa4\x00\x00\x00\x020_Z\\R\xd3\x94",
+        63424: b"\x06\xa4\x13\x80\x00\x021_Z\\R\xd3\x94",
+    },
+    service_uuids=[],
+    service_data={},
+    source="local",
+)
+SCALE_SERVICE_INFO_3 = BluetoothServiceInfo(
+    name="",
+    address="5F:5A:5C:52:D3:94",
+    rssi=-63,
+    manufacturer_data={
+        57280: b"\x06\xa4\x00\x00\x00\x020_Z\\R\xd3\x94",
+        63424: b"\x06\xa4\x13\x80\x00\x021_Z\\R\xd3\x94",
+        6592: b"\x06\x8e\x00\x00\x00\x020_Z\\R\xd3\x94",
+    },
+    service_uuids=[],
+    service_data={},
+    source="local",
+)

tests/components/leaone/conftest.py

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

tests/components/leaone/test_config_flow.py

@@ -0,0 +1,94 @@
+"""Test the Leaone config flow."""
+from unittest.mock import patch
+
+from homeassistant import config_entries
+from homeassistant.components.leaone.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from . import SCALE_SERVICE_INFO
+
+from tests.common import MockConfigEntry
+
+
+async def test_async_step_user_no_devices_found(hass: HomeAssistant) -> None:
+    """Test setup from service info cache with no devices found."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": config_entries.SOURCE_USER},
+    )
+    assert result["type"] == FlowResultType.ABORT
+    assert result["reason"] == "no_devices_found"
+
+
+async def test_async_step_user_with_found_devices(hass: HomeAssistant) -> None:
+    """Test setup from service info cache with devices found."""
+    with patch(
+        "homeassistant.components.leaone.config_flow.async_discovered_service_info",
+        return_value=[SCALE_SERVICE_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"
+    with patch("homeassistant.components.leaone.async_setup_entry", return_value=True):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={"address": "5F:5A:5C:52:D3:94"},
+        )
+    assert result2["type"] == FlowResultType.CREATE_ENTRY
+    assert result2["title"] == "TZC4 D394"
+    assert result2["data"] == {}
+    assert result2["result"].unique_id == "5F:5A:5C:52:D3:94"
+
+
+async def test_async_step_user_device_added_between_steps(hass: HomeAssistant) -> None:
+    """Test the device gets added via another flow between steps."""
+    with patch(
+        "homeassistant.components.leaone.config_flow.async_discovered_service_info",
+        return_value=[SCALE_SERVICE_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"
+
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        unique_id="5F:5A:5C:52:D3:94",
+    )
+    entry.add_to_hass(hass)
+
+    with patch("homeassistant.components.leaone.async_setup_entry", return_value=True):
+        result2 = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={"address": "5F:5A:5C:52:D3:94"},
+        )
+    assert result2["type"] == FlowResultType.ABORT
+    assert result2["reason"] == "already_configured"
+
+
+async def test_async_step_user_with_found_devices_already_setup(
+    hass: HomeAssistant,
+) -> None:
+    """Test setup from service info cache with devices found."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        unique_id="5F:5A:5C:52:D3:94",
+    )
+    entry.add_to_hass(hass)
+
+    with patch(
+        "homeassistant.components.leaone.config_flow.async_discovered_service_info",
+        return_value=[SCALE_SERVICE_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_devices_found"

tests/components/leaone/test_sensor.py

@@ -0,0 +1,54 @@
+"""Test the Leaone sensors."""
+
+from homeassistant.components.leaone.const import DOMAIN
+from homeassistant.components.sensor import ATTR_STATE_CLASS
+from homeassistant.const import ATTR_FRIENDLY_NAME, ATTR_UNIT_OF_MEASUREMENT
+from homeassistant.core import HomeAssistant
+
+from . import SCALE_SERVICE_INFO, SCALE_SERVICE_INFO_2, SCALE_SERVICE_INFO_3
+
+from tests.common import MockConfigEntry
+from tests.components.bluetooth import inject_bluetooth_service_info
+
+
+async def test_sensors(hass: HomeAssistant) -> None:
+    """Test setting up creates the sensors."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        unique_id="5F:5A:5C:52:D3:94",
+    )
+    entry.add_to_hass(hass)
+
+    assert await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert len(hass.states.async_all("sensor")) == 0
+
+    inject_bluetooth_service_info(hass, SCALE_SERVICE_INFO)
+    await hass.async_block_till_done()
+    inject_bluetooth_service_info(hass, SCALE_SERVICE_INFO_2)
+    await hass.async_block_till_done()
+
+    assert len(hass.states.async_all("sensor")) == 2
+
+    mass_sensor = hass.states.get("sensor.tzc4_d394_mass")
+    mass_sensor_attrs = mass_sensor.attributes
+    assert mass_sensor.state == "77.11"
+    assert mass_sensor_attrs[ATTR_FRIENDLY_NAME] == "TZC4 D394 Mass"
+    assert mass_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "kg"
+    assert mass_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
+
+    mass_sensor = hass.states.get("sensor.tzc4_d394_non_stabilized_mass")
+    mass_sensor_attrs = mass_sensor.attributes
+    assert mass_sensor.state == "77.11"
+    assert mass_sensor_attrs[ATTR_FRIENDLY_NAME] == "TZC4 D394 Non Stabilized Mass"
+    assert mass_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "kg"
+    assert mass_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
+
+    inject_bluetooth_service_info(hass, SCALE_SERVICE_INFO_3)
+    await hass.async_block_till_done()
+
+    assert len(hass.states.async_all("sensor")) == 2
+
+    assert await hass.config_entries.async_unload(entry.entry_id)
+    await hass.async_block_till_done()