Add Tibber binary sensors

Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>

homeassistant/components/tibber/__init__.py

@@ -33,7 +33,7 @@ from .const import (
 from .coordinator import TibberDataAPICoordinator
 from .services import async_setup_services
 
-PLATFORMS = [Platform.NOTIFY, Platform.SENSOR]
+PLATFORMS = [Platform.BINARY_SENSOR, Platform.NOTIFY, Platform.SENSOR]
 
 CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
 

homeassistant/components/tibber/binary_sensor.py

@@ -0,0 +1,138 @@
+"""Support for Tibber binary sensors."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+import logging
+
+from tibber.data_api import TibberDevice
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorDeviceClass,
+    BinarySensorEntity,
+    BinarySensorEntityDescription,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN, TibberConfigEntry
+from .coordinator import TibberDataAPICoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, kw_only=True)
+class TibberBinarySensorEntityDescription(BinarySensorEntityDescription):
+    """Describes Tibber binary sensor entity."""
+
+    is_on_fn: Callable[[str | None], bool | None]
+
+
+def _connector_status_is_on(value: str | None) -> bool | None:
+    """Map connector status value to binary sensor state."""
+    if value == "connected":
+        return True
+    if value == "disconnected":
+        return False
+    return None
+
+
+def _charging_status_is_on(value: str | None) -> bool | None:
+    """Map charging status value to binary sensor state."""
+    if value == "charging":
+        return True
+    if value == "idle":
+        return False
+    return None
+
+
+def _device_status_is_on(value: str | None) -> bool | None:
+    """Map device status value to binary sensor state."""
+    if value == "on":
+        return True
+    if value == "off":
+        return False
+    return None
+
+
+DATA_API_BINARY_SENSORS: tuple[TibberBinarySensorEntityDescription, ...] = (
+    TibberBinarySensorEntityDescription(
+        key="connector.status",
+        device_class=BinarySensorDeviceClass.PLUG,
+        is_on_fn=_connector_status_is_on,
+    ),
+    TibberBinarySensorEntityDescription(
+        key="charging.status",
+        device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
+        is_on_fn=_charging_status_is_on,
+    ),
+    TibberBinarySensorEntityDescription(
+        key="onOff",
+        device_class=BinarySensorDeviceClass.POWER,
+        is_on_fn=_device_status_is_on,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: TibberConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the Tibber binary sensors."""
+    coordinator = entry.runtime_data.data_api_coordinator
+    assert coordinator is not None
+
+    entities: list[TibberDataAPIBinarySensor] = []
+    api_binary_sensors = {sensor.key: sensor for sensor in DATA_API_BINARY_SENSORS}
+
+    for device in coordinator.data.values():
+        for sensor in device.sensors:
+            description: TibberBinarySensorEntityDescription | None = (
+                api_binary_sensors.get(sensor.id)
+            )
+            if description is None:
+                continue
+            entities.append(TibberDataAPIBinarySensor(coordinator, device, description))
+    async_add_entities(entities)
+
+
+class TibberDataAPIBinarySensor(
+    CoordinatorEntity[TibberDataAPICoordinator], BinarySensorEntity
+):
+    """Representation of a Tibber Data API binary sensor."""
+
+    _attr_has_entity_name = True
+    entity_description: TibberBinarySensorEntityDescription
+
+    def __init__(
+        self,
+        coordinator: TibberDataAPICoordinator,
+        device: TibberDevice,
+        entity_description: TibberBinarySensorEntityDescription,
+    ) -> None:
+        """Initialize the binary sensor."""
+        super().__init__(coordinator)
+
+        self._device_id: str = device.id
+        self.entity_description = entity_description
+
+        self._attr_unique_id = f"{device.external_id}_{self.entity_description.key}"
+
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, device.external_id)},
+            name=device.name,
+            manufacturer=device.brand,
+            model=device.model,
+        )
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return the state of the binary sensor."""
+        sensors = self.coordinator.sensors_by_device.get(self._device_id, {})
+        sensor = sensors[self.entity_description.key]
+        value: str | None = str(sensor.value) if sensor.value is not None else None
+        return self.entity_description.is_on_fn(value)

homeassistant/components/tibber/sensor.py

@@ -430,9 +430,6 @@ def _setup_data_api_sensors(
         for sensor in device.sensors:
             description: SensorEntityDescription | None = api_sensors.get(sensor.id)
             if description is None:
-                _LOGGER.debug(
-                    "Sensor %s not found in DATA_API_SENSORS, skipping", sensor
-                )
                 continue
             entities.append(TibberDataAPISensor(coordinator, device, description))
     async_add_entities(entities)

tests/components/tibber/test_binary_sensor.py

@@ -0,0 +1,150 @@
+"""Tests for the Tibber binary sensors."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import tibber
+
+from homeassistant.components.recorder import Recorder
+from homeassistant.components.tibber.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+
+from tests.common import MockConfigEntry
+
+
+def create_tibber_device_with_binary_sensors(
+    device_id: str = "device-id",
+    external_id: str = "external-id",
+    name: str = "Test Device",
+    brand: str = "Tibber",
+    model: str = "Gen1",
+    connector_status: str | None = "connected",
+    charging_status: str | None = "charging",
+    device_status: str | None = "on",
+    home_id: str = "home-id",
+) -> tibber.data_api.TibberDevice:
+    """Create a fake Tibber Data API device with binary sensor capabilities."""
+    device_data = {
+        "id": device_id,
+        "externalId": external_id,
+        "info": {
+            "name": name,
+            "brand": brand,
+            "model": model,
+        },
+        "capabilities": [
+            {
+                "id": "connector.status",
+                "value": connector_status,
+                "description": "Connector status",
+                "unit": "",
+            },
+            {
+                "id": "charging.status",
+                "value": charging_status,
+                "description": "Charging status",
+                "unit": "",
+            },
+            {
+                "id": "onOff",
+                "value": device_status,
+                "description": "Device status",
+                "unit": "",
+            },
+        ],
+    }
+    return tibber.data_api.TibberDevice(device_data, home_id=home_id)
+
+
+async def test_binary_sensors_are_created(
+    recorder_mock: Recorder,
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    data_api_client_mock: AsyncMock,
+    setup_credentials: None,
+    entity_registry: er.EntityRegistry,
+) -> None:
+    """Ensure binary sensors are created from Data API devices."""
+    device = create_tibber_device_with_binary_sensors()
+    data_api_client_mock.get_all_devices = AsyncMock(return_value={"device-id": device})
+    data_api_client_mock.update_devices = AsyncMock(return_value={"device-id": device})
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    connector_unique_id = "external-id_connector.status"
+    connector_entity_id = entity_registry.async_get_entity_id(
+        "binary_sensor", DOMAIN, connector_unique_id
+    )
+    assert connector_entity_id is not None
+    state = hass.states.get(connector_entity_id)
+    assert state is not None
+    assert state.state == "on"
+
+    charging_unique_id = "external-id_charging.status"
+    charging_entity_id = entity_registry.async_get_entity_id(
+        "binary_sensor", DOMAIN, charging_unique_id
+    )
+    assert charging_entity_id is not None
+    state = hass.states.get(charging_entity_id)
+    assert state is not None
+    assert state.state == "on"
+
+    device_unique_id = "external-id_onOff"
+    device_entity_id = entity_registry.async_get_entity_id(
+        "binary_sensor", DOMAIN, device_unique_id
+    )
+    assert device_entity_id is not None
+    state = hass.states.get(device_entity_id)
+    assert state is not None
+    assert state.state == "on"
+
+
+async def test_device_status_on(
+    recorder_mock: Recorder,
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    data_api_client_mock: AsyncMock,
+    setup_credentials: None,
+    entity_registry: er.EntityRegistry,
+) -> None:
+    """Test device status on state."""
+    device = create_tibber_device_with_binary_sensors(device_status="on")
+    data_api_client_mock.get_all_devices = AsyncMock(return_value={"device-id": device})
+    data_api_client_mock.update_devices = AsyncMock(return_value={"device-id": device})
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    unique_id = "external-id_onOff"
+    entity_id = entity_registry.async_get_entity_id("binary_sensor", DOMAIN, unique_id)
+    assert entity_id is not None
+    state = hass.states.get(entity_id)
+    assert state is not None
+    assert state.state == "on"
+
+
+async def test_device_status_off(
+    recorder_mock: Recorder,
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    data_api_client_mock: AsyncMock,
+    setup_credentials: None,
+    entity_registry: er.EntityRegistry,
+) -> None:
+    """Test device status off state."""
+    device = create_tibber_device_with_binary_sensors(device_status="off")
+    data_api_client_mock.get_all_devices = AsyncMock(return_value={"device-id": device})
+    data_api_client_mock.update_devices = AsyncMock(return_value={"device-id": device})
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    unique_id = "external-id_onOff"
+    entity_id = entity_registry.async_get_entity_id("binary_sensor", DOMAIN, unique_id)
+    assert entity_id is not None
+    state = hass.states.get(entity_id)
+    assert state is not None
+    assert state.state == "off"