Files
wled-screen-controller-mixed/custom_components/wled_screen_controller/light.py
alexei.dolgolyov be356f30eb Fix HAOS light color reverting after timeout
When HA sets a color via turn_on, also update the source's fallback_color
to match. This way when the api_input timeout fires (default 5s), the
stream reverts to the same color instead of black. turn_off resets
fallback to [0,0,0].

Added coordinator.update_source() for PUT /color-strip-sources/{id}.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-17 14:52:50 +03:00

148 lines
4.9 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"
# Local state — not derived from coordinator data
self._is_on: bool = False
self._rgb_color: tuple[int, int, int] = (255, 255, 255)
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]