"""Sensor platform for LED Screen Controller.""" from __future__ import annotations import logging from collections.abc import Callable from typing import Any from homeassistant.components.sensor import ( SensorDeviceClass, SensorEntity, SensorStateClass, ) from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import ( DOMAIN, TARGET_TYPE_KEY_COLORS, DATA_COORDINATOR, DATA_WS_MANAGER, ) from .coordinator import WLEDScreenControllerCoordinator from .ws_manager import KeyColorsWebSocketManager _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: WLEDScreenControllerCoordinator = data[DATA_COORDINATOR] ws_manager: KeyColorsWebSocketManager = data[DATA_WS_MANAGER] entities: list[SensorEntity] = [] if coordinator.data and "targets" in coordinator.data: for target_id, target_data in coordinator.data["targets"].items(): entities.append( WLEDScreenControllerFPSSensor(coordinator, target_id, entry.entry_id) ) entities.append( WLEDScreenControllerStatusSensor( coordinator, target_id, entry.entry_id ) ) # Add color sensors for Key Colors targets info = target_data["info"] if info.get("target_type") == TARGET_TYPE_KEY_COLORS: rectangles = target_data.get("rectangles", []) for rect in rectangles: entities.append( WLEDScreenControllerColorSensor( coordinator=coordinator, ws_manager=ws_manager, target_id=target_id, rectangle_name=rect["name"], entry_id=entry.entry_id, ) ) async_add_entities(entities) class WLEDScreenControllerFPSSensor(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: WLEDScreenControllerCoordinator, 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 WLEDScreenControllerStatusSensor(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: WLEDScreenControllerCoordinator, 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 WLEDScreenControllerColorSensor(CoordinatorEntity, SensorEntity): """Color sensor reporting the extracted screen color for a Key Colors rectangle.""" _attr_has_entity_name = True _attr_icon = "mdi:palette" def __init__( self, coordinator: WLEDScreenControllerCoordinator, ws_manager: KeyColorsWebSocketManager, target_id: str, rectangle_name: str, entry_id: str, ) -> None: """Initialize the color sensor.""" super().__init__(coordinator) self._target_id = target_id self._rectangle_name = rectangle_name self._ws_manager = ws_manager self._entry_id = entry_id self._unregister_ws: Callable[[], None] | None = None sanitized = rectangle_name.lower().replace(" ", "_").replace("-", "_") self._attr_unique_id = f"{target_id}_{sanitized}_color" self._attr_translation_key = "rectangle_color" self._attr_translation_placeholders = {"rectangle_name": rectangle_name} @property def device_info(self) -> dict[str, Any]: """Return device information.""" return {"identifiers": {(DOMAIN, self._target_id)}} async def async_added_to_hass(self) -> None: """Register WS callback when entity is added.""" await super().async_added_to_hass() self._unregister_ws = self._ws_manager.register_callback( self._target_id, self._handle_color_update ) async def async_will_remove_from_hass(self) -> None: """Unregister WS callback when entity is removed.""" if self._unregister_ws: self._unregister_ws() self._unregister_ws = None await super().async_will_remove_from_hass() def _handle_color_update(self, colors: dict) -> None: """Handle incoming color update from WebSocket.""" if self._rectangle_name in colors: self.async_write_ha_state() @property def native_value(self) -> str | None: """Return the hex color string (e.g. #FF8800).""" color = self._get_color() if color is None: return None return f"#{color['r']:02X}{color['g']:02X}{color['b']:02X}" @property def extra_state_attributes(self) -> dict[str, Any]: """Return r, g, b, brightness as attributes.""" color = self._get_color() if color is None: return {} r, g, b = color["r"], color["g"], color["b"] brightness = int(0.299 * r + 0.587 * g + 0.114 * b) return { "r": r, "g": g, "b": b, "brightness": brightness, "rgb_color": [r, g, b], } @property def available(self) -> bool: """Return if entity is available.""" return self._get_target_data() is not None def _get_color(self) -> dict[str, int] | None: """Get the current color for this rectangle from WS manager.""" target_data = self._get_target_data() if not target_data or not target_data.get("state"): return None if not target_data["state"].get("processing"): return None colors = self._ws_manager.get_latest_colors(self._target_id) return colors.get(self._rectangle_name) 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)