alexei.dolgolyov
67da014684
- Rewrite integration to target-centric model: each picture target becomes a HA device under a server hub with switch, FPS, and status sensors - Replace KC light entities with color sensors (hex state + RGB attributes) for better automation support via WebSocket real-time updates - Add WebSocket manager for Key Colors color streaming - Add KC per-stage timing metrics (calc_colors, broadcast) with rolling avg - Fix KC timing fields missing from API by adding them to Pydantic schema - Make start/stop processing idempotent to prevent intermittent 404 errors - Add HAOS localization support (en, ru) using translation_key system - Rename integration from "WLED Screen Controller" to "LED Screen Controller" - Remove obsolete select.py (display select) and README.md Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
274 lines
9.2 KiB
Python
274 lines
9.2 KiB
Python
"""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)
|