core/homeassistant/components/smarttub/binary_sensor.py

"""Platform for binary sensor integration."""

from __future__ import annotations

import logging
from typing import Any

from smarttub import Spa, SpaError, SpaReminder
import voluptuous as vol

from homeassistant.components.binary_sensor import (
    BinarySensorDeviceClass,
    BinarySensorEntity,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_platform
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import VolDictType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator

from .const import ATTR_ERRORS, ATTR_REMINDERS, ATTR_SENSORS
from .controller import SmartTubConfigEntry
from .entity import (
    SmartTubEntity,
    SmartTubExternalSensorBase,
    SmartTubOnboardSensorBase,
)

# whether the reminder has been snoozed (bool)
ATTR_REMINDER_SNOOZED = "snoozed"

ATTR_ERROR_CODE = "error_code"
ATTR_ERROR_TITLE = "error_title"
ATTR_ERROR_DESCRIPTION = "error_description"
ATTR_ERROR_TYPE = "error_type"
ATTR_CREATED_AT = "created_at"
ATTR_UPDATED_AT = "updated_at"

# how many days to snooze the reminder for
ATTR_REMINDER_DAYS = "days"
RESET_REMINDER_SCHEMA: VolDictType = {
    vol.Required(ATTR_REMINDER_DAYS): vol.All(
        vol.Coerce(int), vol.Range(min=30, max=365)
    )
}
SNOOZE_REMINDER_SCHEMA: VolDictType = {
    vol.Required(ATTR_REMINDER_DAYS): vol.All(
        vol.Coerce(int), vol.Range(min=10, max=120)
    )
}

_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(
    hass: HomeAssistant,
    entry: SmartTubConfigEntry,
    async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
    """Set up binary sensor entities for the binary sensors in the tub."""

    controller = entry.runtime_data

    entities: list[BinarySensorEntity] = []
    for spa in controller.spas:
        entities.append(SmartTubOnline(controller.coordinator, spa))
        entities.append(SmartTubError(controller.coordinator, spa))
        entities.extend(
            SmartTubReminder(controller.coordinator, spa, reminder)
            for reminder in controller.coordinator.data[spa.id][ATTR_REMINDERS].values()
        )
        for sensor in controller.coordinator.data[spa.id][ATTR_SENSORS].values():
            name = sensor.name.strip("{}")
            if name.startswith("cover-"):
                entities.append(
                    SmartTubCoverSensor(controller.coordinator, spa, sensor)
                )

    async_add_entities(entities)

    platform = entity_platform.async_get_current_platform()

    platform.async_register_entity_service(
        "snooze_reminder",
        SNOOZE_REMINDER_SCHEMA,
        "async_snooze",
    )
    platform.async_register_entity_service(
        "reset_reminder",
        RESET_REMINDER_SCHEMA,
        "async_reset",
    )


class SmartTubOnline(SmartTubOnboardSensorBase, BinarySensorEntity):
    """A binary sensor indicating whether the spa is currently online (connected to the cloud)."""

    _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
    # This seems to be very noisy and not generally useful, so disable by default.
    _attr_entity_registry_enabled_default = False

    def __init__(
        self, coordinator: DataUpdateCoordinator[dict[str, Any]], spa: Spa
    ) -> None:
        """Initialize the entity."""
        super().__init__(coordinator, spa, "Online", "online")

    @property
    def is_on(self) -> bool:
        """Return true if the binary sensor is on."""
        return self._state is True


class SmartTubReminder(SmartTubEntity, BinarySensorEntity):
    """Reminders for maintenance actions."""

    _attr_device_class = BinarySensorDeviceClass.PROBLEM

    def __init__(
        self,
        coordinator: DataUpdateCoordinator[dict[str, Any]],
        spa: Spa,
        reminder: SpaReminder,
    ) -> None:
        """Initialize the entity."""
        super().__init__(
            coordinator,
            spa,
            f"{reminder.name.title()} Reminder",
        )
        self.reminder_id = reminder.id
        self._attr_unique_id = f"{spa.id}-reminder-{reminder.id}"

    @property
    def reminder(self) -> SpaReminder:
        """Return the underlying SpaReminder object for this entity."""
        return self.coordinator.data[self.spa.id][ATTR_REMINDERS][self.reminder_id]

    @property
    def is_on(self) -> bool:
        """Return whether the specified maintenance action needs to be taken."""
        return self.reminder.remaining_days == 0

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        """Return the state attributes."""
        return {
            ATTR_REMINDER_SNOOZED: self.reminder.snoozed,
            ATTR_REMINDER_DAYS: self.reminder.remaining_days,
        }

    async def async_snooze(self, days):
        """Snooze this reminder for the specified number of days."""
        await self.reminder.snooze(days)
        await self.coordinator.async_request_refresh()

    async def async_reset(self, days):
        """Dismiss this reminder, and reset it to the specified number of days."""
        await self.reminder.reset(days)
        await self.coordinator.async_request_refresh()


class SmartTubError(SmartTubEntity, BinarySensorEntity):
    """Indicates whether an error code is present.

    There may be 0 or more errors. If there are >0, we show the first one.
    """

    _attr_device_class = BinarySensorDeviceClass.PROBLEM

    def __init__(
        self, coordinator: DataUpdateCoordinator[dict[str, Any]], spa: Spa
    ) -> None:
        """Initialize the entity."""
        super().__init__(
            coordinator,
            spa,
            "Error",
        )

    @property
    def error(self) -> SpaError | None:
        """Return the underlying SpaError object for this entity."""
        errors = self.coordinator.data[self.spa.id][ATTR_ERRORS]
        if len(errors) == 0:
            return None
        return errors[0]

    @property
    def is_on(self) -> bool:
        """Return true if an error is signaled."""
        return self.error is not None

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        """Return the state attributes."""
        if (error := self.error) is None:
            return {}

        return {
            ATTR_ERROR_CODE: error.code,
            ATTR_ERROR_TITLE: error.title,
            ATTR_ERROR_DESCRIPTION: error.description,
            ATTR_ERROR_TYPE: error.error_type,
            ATTR_CREATED_AT: error.created_at.isoformat(),
            ATTR_UPDATED_AT: error.updated_at.isoformat(),
        }


class SmartTubCoverSensor(SmartTubExternalSensorBase, BinarySensorEntity):
    """Wireless magnetic cover sensor."""

    _attr_device_class = BinarySensorDeviceClass.OPENING

    @property
    def is_on(self) -> bool:
        """Return False if the cover is closed, True if open."""
        # magnet is True when the cover is closed, False when open
        # device class OPENING wants True to mean open, False to mean closed
        return not self.sensor.magnet