ledgrab-haos-integration/custom_components/ledgrab/sensor.py

"""Sensor platform for LED Screen Controller."""

from __future__ import annotations

from datetime import datetime
import logging
from typing import Any

from homeassistant.components.sensor import (
    SensorDeviceClass,
    SensorEntity,
    SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    PERCENTAGE,
    UnitOfInformation,
    UnitOfTemperature,
    UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity

from .const import (
    DOMAIN,
    TARGET_TYPE_HA_LIGHT,
    DATA_COORDINATOR,
)
from .coordinator import LedGrabCoordinator
from .entity import LedGrabPlaylistEntity

_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up LED Screen Controller sensors."""
    data = hass.data[DOMAIN][entry.entry_id]
    coordinator: LedGrabCoordinator = data[DATA_COORDINATOR]

    entities: list[SensorEntity] = []
    if coordinator.data and "targets" in coordinator.data:
        for target_id, target_data in coordinator.data["targets"].items():
            entities.append(LedGrabFPSSensor(coordinator, target_id, entry.entry_id))
            entities.append(
                LedGrabStatusSensor(coordinator, target_id, entry.entry_id)
            )

            # Add mapped lights sensor for HA Light targets
            info = target_data["info"]
            if info.get("target_type") == TARGET_TYPE_HA_LIGHT:
                entities.append(HALightMappedLightsSensor(coordinator, target_id, entry.entry_id))

    if coordinator.data:
        for clock in coordinator.data.get("sync_clocks", []):
            entities.append(SyncClockElapsedSensor(coordinator, clock["id"], entry.entry_id))

        for playlist in coordinator.data.get("scene_playlists", []):
            entities.append(
                PlaylistCurrentSceneSensor(coordinator, playlist["id"], entry.entry_id)
            )
            entities.append(
                PlaylistItemsSensor(coordinator, playlist["id"], entry.entry_id)
            )

    entities.extend(_build_server_sensors(coordinator, entry.entry_id))

    async_add_entities(entities)


def _build_server_sensors(
    coordinator: LedGrabCoordinator,
    entry_id: str,
) -> list[SensorEntity]:
    """Build the server-level telemetry sensors.

    GPU/CPU-temp/battery sensors are only registered when the first refresh
    actually carries those readings — we don't want phantom unavailable
    entities on hardware that doesn't support a metric. A reload picks up
    metrics that come online later (e.g. user installs LibreHardwareMonitor).
    """
    system = (coordinator.data or {}).get("system") or {}
    perf = system.get("performance") or {}
    health = system.get("health")

    sensors: list[SensorEntity] = [
        ServerCpuPercentSensor(coordinator, entry_id),
        ServerRamPercentSensor(coordinator, entry_id),
        ServerAppCpuPercentSensor(coordinator, entry_id),
        ServerAppRamSensor(coordinator, entry_id),
        ServerVersionSensor(coordinator, entry_id),
    ]

    if health is not None:
        # Only meaningful when /health actually responded on first refresh.
        sensors.append(ServerLastRestartSensor(coordinator, entry_id))

    if perf.get("gpu") is not None:
        sensors.append(ServerGpuUtilizationSensor(coordinator, entry_id))
        sensors.append(ServerGpuTemperatureSensor(coordinator, entry_id))

    if perf.get("cpu_temp_c") is not None:
        sensors.append(ServerCpuTemperatureSensor(coordinator, entry_id))

    if perf.get("battery_percent") is not None:
        sensors.append(ServerBatterySensor(coordinator, entry_id))

    return sensors


class LedGrabFPSSensor(CoordinatorEntity, SensorEntity):
    """FPS sensor for a LED Screen Controller target."""

    _attr_has_entity_name = True
    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_native_unit_of_measurement = "FPS"
    _attr_icon = "mdi:speedometer"
    _attr_suggested_display_precision = 1

    def __init__(
        self,
        coordinator: LedGrabCoordinator,
        target_id: str,
        entry_id: str,
    ) -> None:
        """Initialize the sensor."""
        super().__init__(coordinator)
        self._target_id = target_id
        self._entry_id = entry_id
        self._attr_unique_id = f"{target_id}_fps"
        self._attr_translation_key = "fps"

    @property
    def device_info(self) -> dict[str, Any]:
        """Return device information."""
        return {"identifiers": {(DOMAIN, self._target_id)}}

    @property
    def native_value(self) -> float | None:
        """Return the FPS value."""
        target_data = self._get_target_data()
        if not target_data or not target_data.get("state"):
            return None
        state = target_data["state"]
        if not state.get("processing"):
            return None
        return state.get("fps_actual")

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        """Return additional attributes."""
        target_data = self._get_target_data()
        if not target_data or not target_data.get("state"):
            return {}
        return {"fps_target": target_data["state"].get("fps_target")}

    @property
    def available(self) -> bool:
        """Return if entity is available."""
        return self._get_target_data() is not None

    def _get_target_data(self) -> dict[str, Any] | None:
        if not self.coordinator.data:
            return None
        return self.coordinator.data.get("targets", {}).get(self._target_id)


class LedGrabStatusSensor(CoordinatorEntity, SensorEntity):
    """Status sensor for a LED Screen Controller target."""

    _attr_has_entity_name = True
    _attr_icon = "mdi:information-outline"
    _attr_device_class = SensorDeviceClass.ENUM
    _attr_options = ["processing", "idle", "error", "unavailable"]

    def __init__(
        self,
        coordinator: LedGrabCoordinator,
        target_id: str,
        entry_id: str,
    ) -> None:
        """Initialize the sensor."""
        super().__init__(coordinator)
        self._target_id = target_id
        self._entry_id = entry_id
        self._attr_unique_id = f"{target_id}_status"
        self._attr_translation_key = "status"

    @property
    def device_info(self) -> dict[str, Any]:
        """Return device information."""
        return {"identifiers": {(DOMAIN, self._target_id)}}

    @property
    def native_value(self) -> str:
        """Return the status."""
        target_data = self._get_target_data()
        if not target_data:
            return "unavailable"
        state = target_data.get("state")
        if not state:
            return "unavailable"
        if state.get("processing"):
            errors = state.get("errors", [])
            if errors:
                return "error"
            return "processing"
        return "idle"

    @property
    def available(self) -> bool:
        """Return if entity is available."""
        return self._get_target_data() is not None

    def _get_target_data(self) -> dict[str, Any] | None:
        if not self.coordinator.data:
            return None
        return self.coordinator.data.get("targets", {}).get(self._target_id)


class SyncClockElapsedSensor(CoordinatorEntity, SensorEntity):
    """Elapsed seconds since the sync clock last reset.

    Disabled by default — the value advances on every coordinator poll
    (~3s) while the clock is running, which would write a recorder row
    each refresh. Power users can enable it explicitly.
    """

    _attr_has_entity_name = True
    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_device_class = SensorDeviceClass.DURATION
    _attr_native_unit_of_measurement = UnitOfTime.SECONDS
    _attr_icon = "mdi:timer-outline"
    _attr_entity_category = EntityCategory.DIAGNOSTIC
    _attr_entity_registry_enabled_default = False
    _attr_suggested_display_precision = 1

    def __init__(
        self,
        coordinator: LedGrabCoordinator,
        clock_id: str,
        entry_id: str,
    ) -> None:
        super().__init__(coordinator)
        self._clock_id = clock_id
        self._entry_id = entry_id
        self._attr_unique_id = f"{clock_id}_elapsed"
        self._attr_translation_key = "sync_clock_elapsed"

    @property
    def device_info(self) -> dict[str, Any]:
        return {"identifiers": {(DOMAIN, self._clock_id)}}

    @property
    def native_value(self) -> float | None:
        clock = self._get_clock()
        if not clock:
            return None
        elapsed = clock.get("elapsed_time")
        return float(elapsed) if elapsed is not None else None

    @property
    def available(self) -> bool:
        return self._get_clock() is not None

    def _get_clock(self) -> dict[str, Any] | None:
        if not self.coordinator.data:
            return None
        for clock in self.coordinator.data.get("sync_clocks", []):
            if clock.get("id") == self._clock_id:
                return clock
        return None


class _PlaylistSensorBase(LedGrabPlaylistEntity, SensorEntity):
    """Sensor attached to a scene-playlist device.

    Device/availability/lookup plumbing lives in :class:`LedGrabPlaylistEntity`
    so it stays in sync with the playlist switch.
    """


class PlaylistCurrentSceneSensor(_PlaylistSensorBase):
    """Name of the scene preset this playlist is currently holding.

    Only this playlist's value populates while it is the active (cycling) one;
    every other playlist's sensor reads ``None`` (idle). The value advances as
    the engine steps to the next item.
    """

    _attr_icon = "mdi:playlist-star"

    def __init__(
        self,
        coordinator: LedGrabCoordinator,
        playlist_id: str,
        entry_id: str,
    ) -> None:
        super().__init__(coordinator, playlist_id, entry_id)
        self._attr_unique_id = f"{playlist_id}_current_scene"
        self._attr_translation_key = "playlist_current_scene"

    @property
    def native_value(self) -> str | None:
        state = self._running_state()
        if not state:
            return None
        preset_id = state.get("current_preset_id")
        if not preset_id:
            return None
        return self._resolve_preset_name(preset_id)

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        state = self._running_state()
        if not state:
            return {}
        return {
            "current_preset_id": state.get("current_preset_id"),
            "current_index": state.get("current_index"),
            "item_count": state.get("item_count"),
        }

    def _resolve_preset_name(self, preset_id: str) -> str:
        """Map a scene-preset id to its display name (fall back to the id)."""
        if self.coordinator.data:
            for preset in self.coordinator.data.get("scene_presets", []):
                if preset.get("id") == preset_id:
                    return preset.get("name", preset_id)
        return preset_id


class PlaylistItemsSensor(_PlaylistSensorBase):
    """Number of scene presets queued in a playlist (static config stat).

    No ``state_class`` — this is a configuration cardinality that only changes
    when the playlist is edited, so it shouldn't be enrolled in long-term
    statistics.
    """

    _attr_icon = "mdi:playlist-music"
    _attr_entity_category = EntityCategory.DIAGNOSTIC

    def __init__(
        self,
        coordinator: LedGrabCoordinator,
        playlist_id: str,
        entry_id: str,
    ) -> None:
        super().__init__(coordinator, playlist_id, entry_id)
        self._attr_unique_id = f"{playlist_id}_items"
        self._attr_translation_key = "playlist_items"

    @property
    def native_value(self) -> int | None:
        playlist = self._get_playlist()
        if playlist is None:
            return None
        return len(playlist.get("items") or [])


class HALightMappedLightsSensor(CoordinatorEntity, SensorEntity):
    """Sensor showing the number of mapped HA lights for an HA Light target."""

    _attr_has_entity_name = True
    _attr_icon = "mdi:lightbulb-group"

    def __init__(
        self,
        coordinator: LedGrabCoordinator,
        target_id: str,
        entry_id: str,
    ) -> None:
        """Initialize the sensor."""
        super().__init__(coordinator)
        self._target_id = target_id
        self._entry_id = entry_id
        self._attr_unique_id = f"{target_id}_mapped_lights"
        self._attr_translation_key = "mapped_lights"

    @property
    def device_info(self) -> dict[str, Any]:
        """Return device information."""
        return {"identifiers": {(DOMAIN, self._target_id)}}

    @property
    def native_value(self) -> int | None:
        """Return the number of mapped lights."""
        target_data = self._get_target_data()
        if not target_data:
            return None
        mappings = target_data.get("info", {}).get("light_mappings", [])
        return len(mappings)

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        """Return light mapping details as attributes."""
        target_data = self._get_target_data()
        if not target_data:
            return {}
        mappings = target_data.get("info", {}).get("light_mappings", [])
        entity_ids = [m.get("entity_id", "") for m in mappings]
        return {
            "entity_ids": entity_ids,
            "mappings": [
                {
                    "entity_id": m.get("entity_id", ""),
                    "led_start": m.get("led_start", 0),
                    "led_end": m.get("led_end", -1),
                    "brightness_scale": m.get("brightness_scale", 1.0),
                }
                for m in mappings
            ],
        }

    @property
    def available(self) -> bool:
        """Return if entity is available."""
        return self._get_target_data() is not None

    def _get_target_data(self) -> dict[str, Any] | None:
        if not self.coordinator.data:
            return None
        return self.coordinator.data.get("targets", {}).get(self._target_id)


# ─────────────────────────── Server-level sensors ───────────────────────────


class _ServerSensorBase(CoordinatorEntity, SensorEntity):
    """Shared plumbing for sensors attached to the synthetic Server device."""

    _attr_has_entity_name = True
    _attr_entity_category = EntityCategory.DIAGNOSTIC

    def __init__(
        self,
        coordinator: LedGrabCoordinator,
        entry_id: str,
        unique_suffix: str,
        translation_key: str,
    ) -> None:
        super().__init__(coordinator)
        self._entry_id = entry_id
        self._attr_unique_id = f"{entry_id}_server_{unique_suffix}"
        self._attr_translation_key = translation_key

    @property
    def device_info(self) -> dict[str, Any]:
        return {"identifiers": {(DOMAIN, f"{self._entry_id}_server")}}

    def _perf(self) -> dict[str, Any] | None:
        if not self.coordinator.data:
            return None
        system = self.coordinator.data.get("system") or {}
        return system.get("performance")

    @property
    def available(self) -> bool:
        # Performance metrics are missing whenever /api/v1/system/performance
        # fails — surface that as unavailable instead of stale numbers.
        return self._perf() is not None


class ServerCpuPercentSensor(_ServerSensorBase):
    """System-wide CPU usage on the led-grab server."""

    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_native_unit_of_measurement = PERCENTAGE
    _attr_icon = "mdi:cpu-64-bit"
    _attr_suggested_display_precision = 1

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "cpu_percent", "server_cpu_percent")

    @property
    def native_value(self) -> float | None:
        perf = self._perf()
        return None if perf is None else perf.get("cpu_percent")

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        perf = self._perf() or {}
        return {"cpu_name": perf.get("cpu_name")}


class ServerRamPercentSensor(_ServerSensorBase):
    """System-wide RAM usage on the led-grab server."""

    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_native_unit_of_measurement = PERCENTAGE
    _attr_icon = "mdi:memory"
    _attr_suggested_display_precision = 1

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "ram_percent", "server_ram_percent")

    @property
    def native_value(self) -> float | None:
        perf = self._perf()
        return None if perf is None else perf.get("ram_percent")

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        perf = self._perf() or {}
        return {
            "used_mb": perf.get("ram_used_mb"),
            "total_mb": perf.get("ram_total_mb"),
        }


class ServerAppCpuPercentSensor(_ServerSensorBase):
    """CPU usage of the led-grab process itself."""

    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_native_unit_of_measurement = PERCENTAGE
    _attr_icon = "mdi:application-cog"
    _attr_suggested_display_precision = 1
    _attr_entity_registry_enabled_default = False

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "app_cpu_percent", "server_app_cpu_percent")

    @property
    def native_value(self) -> float | None:
        perf = self._perf()
        return None if perf is None else perf.get("app_cpu_percent")


class ServerAppRamSensor(_ServerSensorBase):
    """Resident memory of the led-grab process."""

    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_device_class = SensorDeviceClass.DATA_SIZE
    _attr_native_unit_of_measurement = UnitOfInformation.MEGABYTES
    _attr_icon = "mdi:memory"
    _attr_suggested_display_precision = 1
    _attr_entity_registry_enabled_default = False

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "app_ram_mb", "server_app_ram")

    @property
    def native_value(self) -> float | None:
        perf = self._perf()
        return None if perf is None else perf.get("app_ram_mb")


class ServerGpuUtilizationSensor(_ServerSensorBase):
    """GPU utilization (NVML)."""

    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_native_unit_of_measurement = PERCENTAGE
    _attr_icon = "mdi:expansion-card"
    _attr_suggested_display_precision = 1

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "gpu_utilization", "server_gpu_utilization")

    @property
    def native_value(self) -> float | None:
        perf = self._perf()
        if perf is None:
            return None
        gpu = perf.get("gpu") or {}
        return gpu.get("utilization")

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        perf = self._perf() or {}
        gpu = perf.get("gpu") or {}
        return {
            "name": gpu.get("name"),
            "memory_used_mb": gpu.get("memory_used_mb"),
            "memory_total_mb": gpu.get("memory_total_mb"),
            "app_memory_mb": gpu.get("app_memory_mb"),
        }

    @property
    def available(self) -> bool:
        perf = self._perf()
        return perf is not None and perf.get("gpu") is not None


class ServerGpuTemperatureSensor(_ServerSensorBase):
    """GPU temperature (NVML)."""

    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_device_class = SensorDeviceClass.TEMPERATURE
    _attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS
    _attr_suggested_display_precision = 0

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "gpu_temp", "server_gpu_temp")

    @property
    def native_value(self) -> float | None:
        perf = self._perf()
        if perf is None:
            return None
        gpu = perf.get("gpu") or {}
        return gpu.get("temperature_c")

    @property
    def available(self) -> bool:
        perf = self._perf()
        return perf is not None and (perf.get("gpu") or {}).get("temperature_c") is not None


class ServerCpuTemperatureSensor(_ServerSensorBase):
    """Hottest CPU/SoC thermal-zone reading.

    Only registered when the first refresh actually reports a temperature —
    Windows needs LibreHardwareMonitor / OpenHardwareMonitor publishing WMI
    sensors, otherwise the field stays null.
    """

    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_device_class = SensorDeviceClass.TEMPERATURE
    _attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS
    _attr_suggested_display_precision = 0

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "cpu_temp", "server_cpu_temp")

    @property
    def native_value(self) -> float | None:
        perf = self._perf()
        return None if perf is None else perf.get("cpu_temp_c")


class ServerBatterySensor(_ServerSensorBase):
    """Battery charge level (laptops / portable hosts)."""

    _attr_state_class = SensorStateClass.MEASUREMENT
    _attr_device_class = SensorDeviceClass.BATTERY
    _attr_native_unit_of_measurement = PERCENTAGE
    _attr_suggested_display_precision = 0

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "battery", "server_battery")

    @property
    def native_value(self) -> float | None:
        perf = self._perf()
        return None if perf is None else perf.get("battery_percent")

    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        perf = self._perf() or {}
        return {"temperature_c": perf.get("battery_temp_c")}


class ServerLastRestartSensor(_ServerSensorBase):
    """Timestamp of the last server start.

    Derived from ``/health.uptime_seconds``; the coordinator caches the
    computed boot time and only refreshes it when the candidate moves more
    than a few seconds, so the recorder doesn't get poll-jitter writes.
    """

    _attr_device_class = SensorDeviceClass.TIMESTAMP
    _attr_icon = "mdi:restart"

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "last_restart", "server_last_restart")

    @property
    def native_value(self) -> datetime | None:
        if not self.coordinator.data:
            return None
        system = self.coordinator.data.get("system") or {}
        return system.get("boot_time")

    @property
    def available(self) -> bool:
        return self.native_value is not None


class ServerVersionSensor(_ServerSensorBase):
    """Reported led-grab server version."""

    _attr_icon = "mdi:tag-outline"

    def __init__(self, coordinator: LedGrabCoordinator, entry_id: str) -> None:
        super().__init__(coordinator, entry_id, "version", "server_version")

    @property
    def native_value(self) -> str | None:
        version = self.coordinator.server_version
        return version if version and version != "unknown" else None

    @property
    def available(self) -> bool:
        return self.native_value is not None