room_cal_ha/custom_components/room_cal_ha/binary_sensor.py

from __future__ import annotations

from datetime import timedelta

from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util import dt as dt_util

from .const import (
  CONF_REFRESH_INTERVAL,
  CONF_ROOM_NAME,
  CONF_SYNC_MODE,
  DEFAULT_REFRESH_INTERVAL,
  DOMAIN,
  SYNC_MODE_MANUAL,
)
from .coordinator import CoordinatorData, RoomCalCoordinator


async def async_setup_entry(
  hass: HomeAssistant,
  entry: ConfigEntry,
  async_add_entities: AddEntitiesCallback,
) -> None:
  coordinator: RoomCalCoordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
  device_name = f"{entry.data.get(CONF_ROOM_NAME, entry.title)} Room State"
  device_info = DeviceInfo(
    identifiers={(DOMAIN, f"{entry.entry_id}_room_state")},
    name=device_name,
  )
  async_add_entities([StaleSensor(coordinator, entry, device_info)])


class StaleSensor(CoordinatorEntity[RoomCalCoordinator], BinarySensorEntity):
  _attr_has_entity_name = True
  _attr_name = "Stale"
  _attr_icon = "mdi:clock-alert-outline"

  def __init__(
    self,
    coordinator: RoomCalCoordinator,
    entry: ConfigEntry,
    device_info: DeviceInfo,
  ) -> None:
    super().__init__(coordinator)
    self._entry = entry
    self._attr_unique_id = f"{entry.entry_id}_stale"
    self._attr_device_info = device_info

  @property
  def is_on(self) -> bool | None:
    if self._entry.options.get(CONF_SYNC_MODE, SYNC_MODE_MANUAL) == SYNC_MODE_MANUAL:
      return None

    interval = self._entry.options.get(CONF_REFRESH_INTERVAL, DEFAULT_REFRESH_INTERVAL)
    threshold = timedelta(minutes=interval * 2)
    now = dt_util.utcnow()
    data: CoordinatorData = self.coordinator.data

    for ts in (data.current_last_refresh, data.next_last_refresh):
      if ts is None or (now - ts) > threshold:
        return True

    return False