hOn/custom_components/hon/binary_sensor.py

import logging
from dataclasses import dataclass

from pyhon import HonConnection

from homeassistant.components.binary_sensor import BinarySensorEntityDescription, BinarySensorDeviceClass, \
    BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import callback
from .const import DOMAIN
from .hon import HonCoordinator, HonEntity

_LOGGER = logging.getLogger(__name__)


@dataclass
class HonBinarySensorEntityDescriptionMixin:
    on_value: str = ""


@dataclass
class HonBinarySensorEntityDescription(HonBinarySensorEntityDescriptionMixin, BinarySensorEntityDescription):
    pass


BINARY_SENSORS: dict[str, tuple[HonBinarySensorEntityDescription, ...]] = {
    "WM": (
        HonBinarySensorEntityDescription(
            key="lastConnEvent.category",
            name="Connection",
            device_class=BinarySensorDeviceClass.CONNECTIVITY,
            on_value="CONNECTED",
        ),
        HonBinarySensorEntityDescription(
            key="doorLockStatus",
            name="Door Locked",
            device_class=BinarySensorDeviceClass.DOOR,
            on_value="0",
        ),
    )
}


async def async_setup_entry(hass, entry: ConfigEntry, async_add_entities) -> None:
    hon: HonConnection = hass.data[DOMAIN][entry.unique_id]
    coordinators = hass.data[DOMAIN]["coordinators"]
    appliances = []
    for device in hon.devices:
        if device.mac_address in coordinators:
            coordinator = hass.data[DOMAIN]["coordinators"][device.mac_address]
        else:
            coordinator = HonCoordinator(hass, device)
            hass.data[DOMAIN]["coordinators"][device.mac_address] = coordinator
        await coordinator.async_config_entry_first_refresh()

        if descriptions := BINARY_SENSORS.get(device.appliance_type_name):
            for description in descriptions:
                if not device.data.get(description.key):
                    _LOGGER.info("Can't setup %s", description.key)
                    continue
                appliances.extend([
                    HonBinarySensorEntity(hass, coordinator, entry, device, description)]
                )

    async_add_entities(appliances)


class HonBinarySensorEntity(HonEntity, BinarySensorEntity):
    entity_description: HonBinarySensorEntityDescription

    def __init__(self, hass, coordinator, entry, device, description) -> None:
        super().__init__(hass, entry, coordinator, device)

        self._coordinator = coordinator

        self.entity_description = description
        self._attr_unique_id = f"{super().unique_id}{description.key}"

    @property
    def is_on(self) -> bool:
        return self._device.data.get(self.entity_description.key, "") == self.entity_description.on_value

    @callback
    def _handle_coordinator_update(self):
        self._attr_native_value = self._device.data.get(self.entity_description.key, "")
        self.async_write_ha_state()