binary_sensor.py

"""SEM Solar Energy Management binary sensors."""

from __future__ import annotations

import logging
import time
from typing import Any, Dict

from homeassistant.components.binary_sensor import (
    BinarySensorEntity,
    BinarySensorEntityDescription,
    BinarySensorDeviceClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity

from .consts.core import SENSOR_DARK_READ_GRACE_S
from .coordinator import SEMCoordinator
from .coordinator.install_modules import kept_descriptions, presence_of
from .sensor import _cleanup_stale_entities

type SEMConfigEntry = ConfigEntry[SEMCoordinator]


def coarse_cycles(n) -> int:
    """(#829) Exact for the first five cycles (the diagnosis window #589 pins),
    then rounded DOWN to one significant figure: 5, 10, 20 … 3000. A standing
    mismatch stops writing a state row per cycle while staying legible."""
    try:
        n = int(n or 0)
    except (TypeError, ValueError):
        return 0
    if n <= 5:
        return n
    if n < 10:
        return 5
    mag = 10 ** (len(str(n)) - 1)
    return (n // mag) * mag

_LOGGER = logging.getLogger(__name__)

PARALLEL_UPDATES = 0  # Coordinator handles all updates

# (#1201) Each of these flags is built from a power, and is None while every
# read of that power is dark (``_build_system_status``). It keeps its last
# value through the dark-read grace, as the power and the status beside it
# do (sensor.py), so a short blink does not flip it. Then it shows unknown,
# never Off. Every other binary sensor shows its value at once.
HELD_FLAG_KEYS = frozenset({
    "solar_active", "battery_charging", "battery_discharging",
    "grid_export_active",
})

BINARY_SENSOR_TYPES = [
    BinarySensorEntityDescription(
        key="ev_connected",
        device_class=BinarySensorDeviceClass.PLUG,
    ),
    BinarySensorEntityDescription(
        key="ev_charging",
        device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
    ),
    BinarySensorEntityDescription(
        key="battery_charging",
        device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
    ),
    BinarySensorEntityDescription(
        key="battery_discharging",
        device_class=BinarySensorDeviceClass.BATTERY,
    ),
    BinarySensorEntityDescription(
        key="grid_export_active",
        device_class=BinarySensorDeviceClass.POWER,
    ),
    BinarySensorEntityDescription(
        key="solar_active",
        device_class=BinarySensorDeviceClass.POWER,
    ),
    # (#559 Phase 0) debounced surplus availability for user automations
    # (no device class — POWER/RUNNING would show misleading icons)
    BinarySensorEntityDescription(
        key="surplus_available",
    ),
    # Phase 0: Forecast available
    BinarySensorEntityDescription(
        key="forecast_available",
        device_class=BinarySensorDeviceClass.CONNECTIVITY,
    ),
    # Phase 1: Dynamic tariff active
    BinarySensorEntityDescription(
        key="tariff_is_dynamic",
    ),
    # Phase 2: Heat pump solar boost active
    BinarySensorEntityDescription(
        key="heat_pump_solar_boost",
    ),
    # #437: heat pump registered with the surplus controller (presence
    # flag for the dashboard auto-hide — sg_ready_state defaults to 2
    # even when no controller exists, so the dashboard needs an
    # explicit signal to distinguish "unconfigured" from "in normal").
    BinarySensorEntityDescription(
        key="heat_pump_registered",
        device_class=BinarySensorDeviceClass.CONNECTIVITY,
    ),
    # #590 — layered-trace health: ON when a control OR perception layer-boundary
    # fault has persisted (a subsystem decided to act but reality disagreed, or a
    # sign reading contradicts its energy counters). The single surface for the
    # trace + the retired sign-contradiction sensors; attributes name the fault.
    BinarySensorEntityDescription(
        key="layer_mismatch",
        device_class=BinarySensorDeviceClass.PROBLEM,
        entity_category=EntityCategory.DIAGNOSTIC,
    ),
]


async def async_setup_entry(
    hass: HomeAssistant, entry: SEMConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
    """Set up SEM Solar Energy Management binary sensors."""
    coordinator: SEMCoordinator = entry.runtime_data

    # (#923) Only the binary sensors of modules this install has — UNKNOWN keeps.
    static_descriptions = kept_descriptions(
        "binary_sensor", BINARY_SENSOR_TYPES, presence_of(coordinator))
    entities = [
        SEMSolarBinarySensor(coordinator, description, entry)
        for description in static_descriptions
    ]

    # Per-charger binary sensors (#193)
    full_config = {**entry.data, **entry.options}
    ev_chargers = full_config.get("ev_chargers", [])
    per_charger_descriptions = []
    for charger_cfg in ev_chargers:
        cid = charger_cfg.get("id", "ev_charger")
        description = BinarySensorEntityDescription(
            key=f"charger_{cid}_connected",
            device_class=BinarySensorDeviceClass.PLUG,
        )
        per_charger_descriptions.append(description)
        entities.append(SEMSolarBinarySensor(coordinator, description, entry))

    async_add_entities(entities)

    # (#923) binary_sensor had no stale sweep at all, so a removed or
    # module-gated key lingered in the registry forever. Same sweep the
    # sensor platform runs, same two unique_id formats.
    _cleanup_stale_entities(
        hass, entry, list(static_descriptions) + per_charger_descriptions,
        "binary_sensor")


class SEMSolarBinarySensor(CoordinatorEntity, BinarySensorEntity):
    """EMS Solar Optimizer binary sensor entity."""

    _attr_has_entity_name = True

    # Disabled by default
    DISABLED_BY_DEFAULT = {
        "forecast_available", "tariff_is_dynamic",
        "heat_pump_solar_boost",
    }

    # Diagnostic sensors
    DIAGNOSTIC_SENSORS = {
        "forecast_available", "tariff_is_dynamic",
    }

    # (#1201) The dark-read grace, as ``SEMSolarSensor`` keeps it.
    # Overridable clock for tests.
    _now_monotonic = staticmethod(time.monotonic)
    _held_is_on = None
    _last_good_value = None
    _last_good_at = None
    _stale_s = 0

    def __init__(
        self,
        coordinator: SEMCoordinator,
        description: BinarySensorEntityDescription,
        entry: ConfigEntry,
    ) -> None:
        """Initialize the binary sensor entity."""
        super().__init__(coordinator)
        self.entity_description = description
        self._attr_unique_id = f"sem_{description.key}"
        self._attr_translation_key = description.key
        # Force stable entity ID regardless of HA language
        self.entity_id = f"binary_sensor.sem_{description.key}"
        self._entry = entry

        if description.key in self.DIAGNOSTIC_SENSORS:
            self._attr_entity_category = EntityCategory.DIAGNOSTIC

        if description.key in self.DISABLED_BY_DEFAULT:
            self._attr_entity_registry_enabled_default = False

        # Start unavailable until first coordinator update (#70)
        self._first_update_received = False

        if coordinator.data:
            self._update_from_coordinator()

    def _update_from_coordinator(self) -> None:
        """(#1201) Hold a ``HELD_FLAG_KEYS`` flag while its power is dark:
        its last value for SENSOR_DARK_READ_GRACE_S, with ``stale_s``, then
        None (unknown). A flag never read stays None."""
        if (self.entity_description.key not in HELD_FLAG_KEYS
                or not self.coordinator.data):
            return
        # a missing key is no read either: ``to_dict`` always publishes these
        value = self.coordinator.data.get(self.entity_description.key)
        now = float(self._now_monotonic())
        if value is None and self._last_good_value is not None:
            age = now - float(self._last_good_at or now)
            if age <= SENSOR_DARK_READ_GRACE_S:
                self._stale_s = int(age)
                self._held_is_on = self._last_good_value
                return
            self._last_good_value = None
            self._last_good_at = None
        self._stale_s = 0
        if isinstance(value, bool):
            self._last_good_value = value
            self._last_good_at = now
        self._held_is_on = value

    @callback
    def _handle_coordinator_update(self) -> None:
        """Handle updated data from the coordinator."""
        self._update_from_coordinator()
        super()._handle_coordinator_update()

    @property
    def available(self) -> bool:
        """Return if entity is available."""
        if not self._first_update_received:
            if self.coordinator.last_update_success and self.coordinator.data is not None:
                self._first_update_received = True
            else:
                return False
        return self.coordinator.last_update_success and self.coordinator.data is not None

    @property
    def is_on(self) -> bool | None:
        """Return true if the binary sensor is on."""
        if not self.coordinator.data:
            return None

        key = self.entity_description.key
        if key in HELD_FLAG_KEYS:
            return self._held_is_on
        return self.coordinator.data.get(key, False)

    @property
    def extra_state_attributes(self) -> Dict[str, Any] | None:
        """#590 — for the layered-trace health sensor, name the persisted fault
        (which subsystem, e.g. ``perception:battery_sign``, and for how many
        cycles) so it's diagnosable without the diagnose dump. (#1201) A held
        flag says for how long with ``stale_s``, as the sensors do."""
        if self._stale_s:
            return {"stale_s": int(self._stale_s)}
        if self.entity_description.key != "layer_mismatch" or not self.coordinator.data:
            return None
        return {
            "subsystem": self.coordinator.data.get("layer_mismatch_subsystem"),
            # (#829) exact while diagnosing, coarse once standing — a per-cycle
            # counter on attributes wrote a row every cycle the fault persisted.
            "persisted_cycles": coarse_cycles(
                self.coordinator.data.get("layer_mismatch_cycles", 0)),
        }

    @property
    def device_info(self) -> Dict[str, Any]:
        """Return device information."""
        return self.coordinator.device_info