alexei.dolgolyov/wled-screen-controller-mixed
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rewrite HAOS integration: target-centric architecture with KC color sensors

Browse Source 
- 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>
This commit is contained in:
Admin
2026-02-17 13:01:40 +03:00
parent e92fe4eb0a
commit 67da014684
19 changed files with 772 additions and 746 deletions
Download Patch File Download Diff File Expand all files Collapse all files

244
custom_components/wled_screen_controller/sensor.py
View File

@@ -1,7 +1,8 @@
"""Sensor platform for WLED Screen Controller."""
"""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 (
@@ -10,13 +11,18 @@ from homeassistant.components.sensor import (
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
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__)
@@ -26,180 +32,242 @@ async def async_setup_entry(
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up WLED Screen Controller sensors."""
"""Set up LED Screen Controller sensors."""
data = hass.data[DOMAIN][entry.entry_id]
coordinator: WLEDScreenControllerCoordinator = data["coordinator"]
coordinator: WLEDScreenControllerCoordinator = data[DATA_COORDINATOR]
ws_manager: KeyColorsWebSocketManager = data[DATA_WS_MANAGER]
entities = []
if coordinator.data and "devices" in coordinator.data:
for device_id, device_data in coordinator.data["devices"].items():
device_info = device_data["info"]
# FPS sensor
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, device_id, device_info, entry.entry_id
)
WLEDScreenControllerFPSSensor(coordinator, target_id, entry.entry_id)
)
# Status sensor
entities.append(
WLEDScreenControllerStatusSensor(
coordinator, device_id, device_info, entry.entry_id
coordinator, target_id, entry.entry_id
)
)
# Frames processed sensor
entities.append(
WLEDScreenControllerFramesSensor(
coordinator, device_id, device_info, 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 WLED Screen Controller."""
"""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,
device_id: str,
device_info: dict[str, Any],
target_id: str,
entry_id: str,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._device_id = device_id
self._device_info = device_info
self._target_id = target_id
self._entry_id = entry_id
self._attr_unique_id = f"{device_id}_fps"
self._attr_name = "FPS"
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._device_id)},
}
return {"identifiers": {(DOMAIN, self._target_id)}}
@property
def native_value(self) -> float | None:
"""Return the FPS value."""
if not self.coordinator.data:
target_data = self._get_target_data()
if not target_data or not target_data.get("state"):
return None
device_data = self.coordinator.data["devices"].get(self._device_id)
if not device_data or not device_data.get("state"):
state = target_data["state"]
if not state.get("processing"):
return None
return device_data["state"].get("fps_actual")
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 {}
device_data = self.coordinator.data["devices"].get(self._device_id)
if not device_data or not device_data.get("state"):
return {}
return {
"target_fps": device_data["state"].get("fps_target"),
}
return None
return self.coordinator.data.get("targets", {}).get(self._target_id)
class WLEDScreenControllerStatusSensor(CoordinatorEntity, SensorEntity):
"""Status sensor for WLED Screen Controller."""
"""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,
device_id: str,
device_info: dict[str, Any],
target_id: str,
entry_id: str,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._device_id = device_id
self._device_info = device_info
self._target_id = target_id
self._entry_id = entry_id
self._attr_unique_id = f"{device_id}_status"
self._attr_name = "Status"
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._device_id)},
}
return {"identifiers": {(DOMAIN, self._target_id)}}
@property
def native_value(self) -> str:
"""Return the status."""
if not self.coordinator.data:
return "unknown"
device_data = self.coordinator.data["devices"].get(self._device_id)
if not device_data:
target_data = self._get_target_data()
if not target_data:
return "unavailable"
if device_data.get("state") and device_data["state"].get("processing"):
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
class WLEDScreenControllerFramesSensor(CoordinatorEntity, SensorEntity):
"""Frames processed sensor."""
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_state_class = SensorStateClass.TOTAL_INCREASING
_attr_icon = "mdi:counter"
_attr_icon = "mdi:palette"
def __init__(
self,
coordinator: WLEDScreenControllerCoordinator,
device_id: str,
device_info: dict[str, Any],
ws_manager: KeyColorsWebSocketManager,
target_id: str,
rectangle_name: str,
entry_id: str,
) -> None:
"""Initialize the sensor."""
"""Initialize the color sensor."""
super().__init__(coordinator)
self._device_id = device_id
self._device_info = device_info
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
self._attr_unique_id = f"{device_id}_frames"
self._attr_name = "Frames Processed"
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 {
"identifiers": {(DOMAIN, self._device_id)},
"r": r,
"g": g,
"b": b,
"brightness": brightness,
"rgb_color": [r, g, b],
}
@property
def native_value(self) -> int | None:
"""Return frames processed."""
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
device_data = self.coordinator.data["devices"].get(self._device_id)
if not device_data or not device_data.get("metrics"):
return None
return device_data["metrics"].get("frames_processed", 0)
return self.coordinator.data.get("targets", {}).get(self._target_id)