localthings/custom_components/localthings/sensor.py

"""Sensor platform for Local Things."""

from __future__ import annotations

import time
from datetime import timedelta
from typing import cast

from homeassistant.components.sensor import SensorDeviceClass, SensorEntity, SensorStateClass
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
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 .const import (
    CONF_FINISH_TIME_HYSTERESIS_MINUTES,
    DEFAULT_FINISH_TIME_HYSTERESIS_MINUTES,
    DOMAIN,
)
from .coordinator import LocalThingsCoordinator
from .entity import LocalThingsEntity, _is_included
from .observe import MODE_OBSERVE, MODE_POLL
from .registry.entities import SensorDesc


async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][entry.entry_id]
    entities: list[SensorEntity] = [
        LocalThingsSensor(coordinator, b)
        for b in coordinator.bound
        if isinstance(b.desc, SensorDesc) and _is_included(b, coordinator)
    ]
    entities.append(LocalThingsConnectionModeSensor(coordinator))
    async_add_entities(entities)


class LocalThingsSensor(LocalThingsEntity, SensorEntity):
    def __init__(self, coordinator: LocalThingsCoordinator, bound) -> None:
        super().__init__(coordinator, bound)
        desc = cast(SensorDesc, bound.desc)
        self._attr_native_unit_of_measurement = desc.unit
        self._attr_device_class = (
            SensorDeviceClass(desc.device_class) if desc.device_class else None
        )
        self._attr_state_class = SensorStateClass(desc.state_class) if desc.state_class else None
        if desc.options:
            self._attr_options = list(desc.options)
        self._hysteresis_value = None
        self._sticky_value = None
        self._sticky_until: float | None = None
        self._sticky_armed = False

    @property
    def native_unit_of_measurement(self):
        desc = cast(SensorDesc, self._bound.desc)
        if desc.unit_fn is not None:
            return desc.unit_fn(self.coordinator.resource(self._bound.href))
        return self._attr_native_unit_of_measurement

    @property
    def native_value(self):
        raw = (self.coordinator.data or {}).get(self._state_key)
        desc = cast(SensorDesc, self._bound.desc)
        if desc.sticky_fn is not None:
            raw = self._apply_sticky(raw, desc)
        if not desc.hysteresis:
            return raw
        return self._apply_hysteresis(raw)

    def _apply_sticky(self, raw, desc: SensorDesc):
        """Freeze this entity at a value for up to `desc.sticky_seconds`
        after `desc.sticky_fn` next stops matching this href's live rep
        (issue #345 -- see operational.py's `_just_finished` for the
        motivating case). Entity-instance state only, exactly like
        `_hysteresis_value` above -- never written back to the coordinator
        cache, so write_fn, diagnostics, and the observe-mode sweep
        comparison keep seeing real device data throughout.

        Reads `sticky_fn` and friends against this href's own live rep,
        not the already-computed `raw`, so they can be independent of
        whatever rep_fn itself gates on -- notably `sticky_live_fn`, which
        is *not* `raw`: `raw` is rep_fn's own (possibly differently gated)
        result, e.g. progress's rep_fn shows "Idle" while paused, but a
        real progress value reported while paused (adding a sock mid-
        cycle) must still win over a stale hold, which means reading it
        ungated here rather than through that gate.

        Edge-triggered, not level-triggered: the window only (re)starts on
        a fresh False->True transition of `sticky_fn`, and -- this is the
        part level-triggering alone misses -- expiry is still checked on
        every call even while `sticky_fn` keeps matching. Without the
        latter, a firmware that leaves the underlying field stuck matching
        forever (the same class of quirk `_completion_minutes` already
        works around) would show the frozen value forever too, defeating
        "bounded, not indefinite".

        `sticky_bypass_fn`, when it matches, always passes
        `sticky_live_fn`'s result through and drops any hold -- for a
        condition where "not sticky right now" is ambiguous between "went
        idle, honor the hold" and "genuinely live, different data" (e.g. a
        new cycle's own real progress), which "consult the hold whenever
        sticky_fn is False" alone can't tell apart.
        """
        assert desc.sticky_fn is not None  # native_value only calls this when set
        rep = self.coordinator.resource(self._bound.href)
        now = time.monotonic()
        matches = desc.sticky_fn(rep)

        if matches:
            if not self._sticky_armed:
                self._sticky_value = (
                    desc.sticky_value_fn(rep) if desc.sticky_value_fn is not None else raw
                )
                self._sticky_until = now + desc.sticky_seconds
            self._sticky_armed = True
        else:
            self._sticky_armed = False
            if desc.sticky_bypass_fn is not None and desc.sticky_bypass_fn(rep):
                self._sticky_until = None
                return desc.sticky_live_fn(rep) if desc.sticky_live_fn is not None else raw

        if self._sticky_until is not None and now < self._sticky_until:
            return self._sticky_value
        return raw

    def _apply_hysteresis(self, raw):
        """Hold the last value this entity actually reported until a new one
        differs by at least the configured threshold, regardless of how long
        that difference has been building up (this is a deadband, not a
        time-based debounce).

        Values like finish_time are `now() + remaining`, recomputed from
        scratch every poll -- both wall-clock drift between the device's own
        remaining-time updates and the device revising its own estimate mid-
        cycle produce a stream of small, real changes that are individually
        meaningless but each trigger a recorder/logbook entry. A cycle
        ending (raw is None) or starting (cache empty) always passes through
        immediately -- only in-between jitter while a value already exists
        on both sides gets held back.
        """
        assert self.coordinator.config_entry is not None
        threshold_min = self.coordinator.config_entry.options.get(
            CONF_FINISH_TIME_HYSTERESIS_MINUTES, DEFAULT_FINISH_TIME_HYSTERESIS_MINUTES
        )
        if (
            threshold_min
            and raw is not None
            and self._hysteresis_value is not None
            and abs(raw - self._hysteresis_value) < timedelta(minutes=threshold_min)
        ):
            return self._hysteresis_value
        self._hysteresis_value = raw
        return raw


class LocalThingsConnectionModeSensor(CoordinatorEntity[LocalThingsCoordinator], SensorEntity):
    """Diagnostic sensor exposing whether this device is currently
    receiving push notifications (observe mode) or being polled only.
    Disabled by default — it's for troubleshooting, not everyday use."""

    _attr_has_entity_name = True
    _attr_translation_key = "connection_mode"
    _attr_entity_category = EntityCategory.DIAGNOSTIC
    _attr_entity_registry_enabled_default = False
    _attr_device_class = SensorDeviceClass.ENUM
    _attr_options = [MODE_OBSERVE, MODE_POLL]  # noqa: RUF012 -- HA `_attr_*` convention

    def __init__(self, coordinator: LocalThingsCoordinator) -> None:
        super().__init__(coordinator)
        self._attr_unique_id = f"{DOMAIN}_{coordinator.device_serial}_connection_mode"

    @property
    def device_info(self) -> DeviceInfo:
        return self.coordinator.device_info

    @property
    def native_value(self) -> str:
        return self.coordinator.observe_mode