core/homeassistant/components/flume/binary_sensor.py

"""Flume binary sensors."""

from dataclasses import dataclass
from typing import override

from homeassistant.components.binary_sensor import (
    BinarySensorDeviceClass,
    BinarySensorEntity,
    BinarySensorEntityDescription,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback

from .const import (
    FLUME_TYPE_BRIDGE,
    FLUME_TYPE_SENSOR,
    KEY_DEVICE_ID,
    KEY_DEVICE_LOCATION,
    KEY_DEVICE_LOCATION_NAME,
    KEY_DEVICE_TYPE,
    NOTIFICATION_HIGH_FLOW,
    NOTIFICATION_LEAK_DETECTED,
)
from .coordinator import (
    FlumeConfigEntry,
    FlumeDeviceConnectionUpdateCoordinator,
    FlumeNotificationDataUpdateCoordinator,
)
from .entity import FlumeEntity
from .util import get_valid_flume_devices

BINARY_SENSOR_DESCRIPTION_CONNECTED = BinarySensorEntityDescription(
    key="connected", device_class=BinarySensorDeviceClass.CONNECTIVITY
)

BINARY_SENSOR_DESCRIPTION_LOW_BATTERY = BinarySensorEntityDescription(
    key="low_battery",
    entity_category=EntityCategory.DIAGNOSTIC,
    device_class=BinarySensorDeviceClass.BATTERY,
)

# Levels reported by the devices endpoint. Unlisted values map to unknown
# rather than to a healthy battery.
BATTERY_LEVEL_IS_LOW = {"low": True, "medium": False, "high": False}


@dataclass(frozen=True, kw_only=True)
class FlumeBinarySensorEntityDescription(BinarySensorEntityDescription):
    """Describes a binary sensor entity."""

    event_rule: str


FLUME_BINARY_NOTIFICATION_SENSORS: tuple[FlumeBinarySensorEntityDescription, ...] = (
    FlumeBinarySensorEntityDescription(
        key="leak",
        translation_key="leak",
        entity_category=EntityCategory.DIAGNOSTIC,
        event_rule=NOTIFICATION_LEAK_DETECTED,
    ),
    FlumeBinarySensorEntityDescription(
        key="flow",
        translation_key="flow",
        entity_category=EntityCategory.DIAGNOSTIC,
        event_rule=NOTIFICATION_HIGH_FLOW,
    ),
)


async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: FlumeConfigEntry,
    async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
    """Set up a Flume binary sensor.."""
    flume_domain_data = config_entry.runtime_data
    flume_devices = flume_domain_data.devices

    flume_entity_list: list[
        FlumeNotificationBinarySensor
        | FlumeConnectionBinarySensor
        | FlumeLowBatteryBinarySensor
    ] = []

    connection_coordinator = FlumeDeviceConnectionUpdateCoordinator(
        hass=hass, config_entry=config_entry, flume_devices=flume_devices
    )
    notification_coordinator = flume_domain_data.notifications_coordinator
    flume_devices = get_valid_flume_devices(flume_devices)
    for device in flume_devices:
        device_id = device[KEY_DEVICE_ID]
        device_location_name = device[KEY_DEVICE_LOCATION][KEY_DEVICE_LOCATION_NAME]

        connection_sensor = FlumeConnectionBinarySensor(
            coordinator=connection_coordinator,
            description=BINARY_SENSOR_DESCRIPTION_CONNECTED,
            device_id=device_id,
            location_name=device_location_name,
            is_bridge=(device[KEY_DEVICE_TYPE] is FLUME_TYPE_BRIDGE),
        )

        flume_entity_list.append(connection_sensor)

        if device[KEY_DEVICE_TYPE] != FLUME_TYPE_SENSOR:
            continue

        flume_entity_list.append(
            FlumeLowBatteryBinarySensor(
                coordinator=connection_coordinator,
                description=BINARY_SENSOR_DESCRIPTION_LOW_BATTERY,
                device_id=device_id,
                location_name=device_location_name,
            )
        )

        # Build notification sensors
        flume_entity_list.extend(
            [
                FlumeNotificationBinarySensor(
                    coordinator=notification_coordinator,
                    description=description,
                    device_id=device_id,
                    location_name=device_location_name,
                )
                for description in FLUME_BINARY_NOTIFICATION_SENSORS
            ]
        )

    async_add_entities(flume_entity_list)


class FlumeNotificationBinarySensor(
    FlumeEntity[FlumeNotificationDataUpdateCoordinator], BinarySensorEntity
):
    """Binary sensor class."""

    entity_description: FlumeBinarySensorEntityDescription

    @property
    @override
    def is_on(self) -> bool:
        """Return on state."""
        return bool(
            (
                notifications := self.coordinator.active_notifications_by_device.get(
                    self.device_id
                )
            )
            and self.entity_description.event_rule in notifications
        )


class FlumeLowBatteryBinarySensor(
    FlumeEntity[FlumeDeviceConnectionUpdateCoordinator], BinarySensorEntity
):
    """Binary sensor class for the sensor battery level."""

    @property
    @override
    def is_on(self) -> bool | None:
        """Return True when the device reports a low battery."""
        return BATTERY_LEVEL_IS_LOW.get(
            self.coordinator.battery_level.get(self.device_id, "")
        )


class FlumeConnectionBinarySensor(
    FlumeEntity[FlumeDeviceConnectionUpdateCoordinator], BinarySensorEntity
):
    """Binary Sensor class for WIFI Connection status."""

    _attr_entity_category = EntityCategory.DIAGNOSTIC
    _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY

    @property
    @override
    def is_on(self) -> bool:
        """Return connection status."""
        return bool(
            (connected := self.coordinator.connected) and connected[self.device_id]
        )