"""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