- Fix indentation bug causing scenes device to not register - Use nonlocal tracking to prevent infinite reload loops on target/scene changes - Guard WS start/stop to avoid redundant connections - Parallel device brightness fetching via asyncio.gather - Route set_leds service to correct coordinator by source ID - Remove stale pattern cache, reuse single timeout object - Fix translations structure for light/select entities - Unregister service when last config entry unloaded Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
152 lines
5.1 KiB
Python
152 lines
5.1 KiB
Python
"""Light platform for LED Screen Controller (api_input CSS sources)."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from homeassistant.components.light import (
|
|
ATTR_BRIGHTNESS,
|
|
ATTR_RGB_COLOR,
|
|
ColorMode,
|
|
LightEntity,
|
|
)
|
|
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, DATA_COORDINATOR
|
|
from .coordinator import WLEDScreenControllerCoordinator
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: ConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up LED Screen Controller api_input lights."""
|
|
data = hass.data[DOMAIN][entry.entry_id]
|
|
coordinator: WLEDScreenControllerCoordinator = data[DATA_COORDINATOR]
|
|
|
|
entities = []
|
|
if coordinator.data:
|
|
for source in coordinator.data.get("css_sources", []):
|
|
if source.get("source_type") == "api_input":
|
|
entities.append(
|
|
ApiInputLight(coordinator, source, entry.entry_id)
|
|
)
|
|
|
|
async_add_entities(entities)
|
|
|
|
|
|
class ApiInputLight(CoordinatorEntity, LightEntity):
|
|
"""Representation of an api_input CSS source as a light entity."""
|
|
|
|
_attr_has_entity_name = True
|
|
_attr_color_mode = ColorMode.RGB
|
|
_attr_supported_color_modes = {ColorMode.RGB}
|
|
_attr_translation_key = "api_input_light"
|
|
_attr_icon = "mdi:led-strip-variant"
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: WLEDScreenControllerCoordinator,
|
|
source: dict[str, Any],
|
|
entry_id: str,
|
|
) -> None:
|
|
"""Initialize the light."""
|
|
super().__init__(coordinator)
|
|
self._source_id: str = source["id"]
|
|
self._source_name: str = source.get("name", self._source_id)
|
|
self._entry_id = entry_id
|
|
self._attr_unique_id = f"{self._source_id}_light"
|
|
|
|
# Restore state from fallback_color
|
|
fallback = self._get_fallback_color()
|
|
is_off = fallback == [0, 0, 0]
|
|
self._is_on: bool = not is_off
|
|
self._rgb_color: tuple[int, int, int] = (
|
|
(255, 255, 255) if is_off else tuple(fallback) # type: ignore[arg-type]
|
|
)
|
|
self._brightness: int = 255
|
|
|
|
@property
|
|
def device_info(self) -> dict[str, Any]:
|
|
"""Return device information — one virtual device per api_input source."""
|
|
return {
|
|
"identifiers": {(DOMAIN, self._source_id)},
|
|
"name": self._source_name,
|
|
"manufacturer": "WLED Screen Controller",
|
|
"model": "API Input CSS Source",
|
|
}
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
"""Return the entity name."""
|
|
return self._source_name
|
|
|
|
@property
|
|
def is_on(self) -> bool:
|
|
"""Return true if the light is on."""
|
|
return self._is_on
|
|
|
|
@property
|
|
def rgb_color(self) -> tuple[int, int, int]:
|
|
"""Return the current RGB color."""
|
|
return self._rgb_color
|
|
|
|
@property
|
|
def brightness(self) -> int:
|
|
"""Return the current brightness (0-255)."""
|
|
return self._brightness
|
|
|
|
async def async_turn_on(self, **kwargs: Any) -> None:
|
|
"""Turn on the light, optionally setting color and brightness."""
|
|
if ATTR_RGB_COLOR in kwargs:
|
|
self._rgb_color = kwargs[ATTR_RGB_COLOR]
|
|
if ATTR_BRIGHTNESS in kwargs:
|
|
self._brightness = kwargs[ATTR_BRIGHTNESS]
|
|
|
|
# Scale RGB by brightness
|
|
scale = self._brightness / 255
|
|
r, g, b = self._rgb_color
|
|
scaled = [round(r * scale), round(g * scale), round(b * scale)]
|
|
|
|
await self.coordinator.push_segments(
|
|
self._source_id,
|
|
[{"start": 0, "length": 9999, "mode": "solid", "color": scaled}],
|
|
)
|
|
# Update fallback_color so the color persists beyond the timeout
|
|
await self.coordinator.update_source(
|
|
self._source_id, fallback_color=scaled,
|
|
)
|
|
self._is_on = True
|
|
self.async_write_ha_state()
|
|
|
|
async def async_turn_off(self, **kwargs: Any) -> None:
|
|
"""Turn off the light by pushing black and setting fallback to black."""
|
|
off_color = [0, 0, 0]
|
|
await self.coordinator.push_segments(
|
|
self._source_id,
|
|
[{"start": 0, "length": 9999, "mode": "solid", "color": off_color}],
|
|
)
|
|
await self.coordinator.update_source(
|
|
self._source_id, fallback_color=off_color,
|
|
)
|
|
self._is_on = False
|
|
self.async_write_ha_state()
|
|
|
|
def _get_fallback_color(self) -> list[int]:
|
|
"""Read fallback_color from the source config in coordinator data."""
|
|
if not self.coordinator.data:
|
|
return [0, 0, 0]
|
|
for source in self.coordinator.data.get("css_sources", []):
|
|
if source.get("id") == self._source_id:
|
|
fallback = source.get("fallback_color")
|
|
if fallback and len(fallback) >= 3:
|
|
return list(fallback[:3])
|
|
break
|
|
return [0, 0, 0]
|