alexei.dolgolyov/ledgrab-haos-integration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(light): periodic color keep-alive for api_input lights

Browse Source 
Background task re-pushes the current color at ~timeout/2 (clamped
0.5-30s) while the light is on, so the server's idle timer never
expires and the strip stays at the picked color instead of reverting
to fallback_color.

Drops the length: 9999 magic value from segment payloads now that the
server defaults length to the remainder of the strip — requires the
matching server change in led-grab.
This commit is contained in:
Admin
2026-04-26 23:34:06 +03:00
parent 30085c00b9
commit ca5b0a8014
1 changed files with 122 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/ledgrab/light.py
+122 -27
View File
@@ -1,6 +1,7 @@
"""Light platform for LED Screen Controller (api_input CSS sources)."""
from __future__ import annotations
import asyncio
import logging
from typing import Any
@@ -21,6 +22,13 @@ from .coordinator import LedGrabCoordinator
_LOGGER = logging.getLogger(__name__)
# Lower bound on the keep-alive interval (seconds). Refreshing faster than
# this is wasteful and can flood the server for very short timeouts.
MIN_KEEPALIVE_INTERVAL = 0.5
# Upper bound — even with very long timeouts, refresh occasionally so the
# strip recovers within reasonable time after a server-side restart.
MAX_KEEPALIVE_INTERVAL = 30.0
async def async_setup_entry(
hass: HomeAssistant,
@@ -51,6 +59,11 @@ class ApiInputLight(CoordinatorEntity, LightEntity, RestoreEntity):
``fallback_color`` — so the chosen default color persists across off/on
cycles. The on/off state itself is restored across HA restarts via
``RestoreEntity``.
While the light is on, a background task re-pushes the current color at
roughly half the source's ``timeout`` so the server's idle timer never
expires and the strip stays at the picked color rather than reverting to
``fallback_color``.
"""
_attr_has_entity_name = True
@@ -84,24 +97,34 @@ class ApiInputLight(CoordinatorEntity, LightEntity, RestoreEntity):
)
self._is_on: bool = has_color
self._brightness: int = 255
self._keepalive_task: asyncio.Task[None] | None = None
async def async_added_to_hass(self) -> None:
"""Restore on/off state and last color from the previous HA session."""
await super().async_added_to_hass()
last_state = await self.async_get_last_state()
if last_state is None:
return
self._is_on = last_state.state == "on"
last_rgb = last_state.attributes.get("rgb_color")
if (
isinstance(last_rgb, (list, tuple))
and len(last_rgb) == 3
and all(isinstance(c, (int, float)) for c in last_rgb)
):
self._rgb_color = tuple(int(c) for c in last_rgb) # type: ignore[assignment]
last_brightness = last_state.attributes.get("brightness")
if isinstance(last_brightness, (int, float)):
self._brightness = int(last_brightness)
if last_state is not None:
self._is_on = last_state.state == "on"
last_rgb = last_state.attributes.get("rgb_color")
if (
isinstance(last_rgb, (list, tuple))
and len(last_rgb) == 3
and all(isinstance(c, (int, float)) for c in last_rgb)
):
self._rgb_color = tuple(int(c) for c in last_rgb) # type: ignore[assignment]
last_brightness = last_state.attributes.get("brightness")
if isinstance(last_brightness, (int, float)):
self._brightness = int(last_brightness)
# If we restored to "on", resume the keep-alive so the server stays
# at the picked color rather than drifting to fallback_color.
if self._is_on:
self._start_keepalive()
async def async_will_remove_from_hass(self) -> None:
"""Cancel the keep-alive task when the entity is removed/unloaded."""
self._stop_keepalive()
await super().async_will_remove_from_hass()
@property
def device_info(self) -> dict[str, Any]:
@@ -135,20 +158,18 @@ class ApiInputLight(CoordinatorEntity, LightEntity, RestoreEntity):
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)]
scaled = self._scaled_color()
await self.coordinator.push_segments(
self._source_id,
[{"start": 0, "length": 9999, "mode": "solid", "color": scaled}],
[{"mode": "solid", "color": scaled}],
)
# Update fallback_color so the color persists beyond the timeout
# Update fallback_color so the color persists if the keep-alive ever
# misses a beat (server restart, integration unload, etc.).
await self.coordinator.update_source(
self._source_id, fallback_color=scaled,
)
self._is_on = True
self._start_keepalive()
self.async_write_ha_state()
async def async_turn_off(self, **kwargs: Any) -> None:
@@ -157,21 +178,95 @@ class ApiInputLight(CoordinatorEntity, LightEntity, RestoreEntity):
``fallback_color`` is intentionally left untouched so the chosen
default color survives off/on cycles.
"""
self._stop_keepalive()
await self.coordinator.push_segments(
self._source_id,
[{"start": 0, "length": 9999, "mode": "solid", "color": [0, 0, 0]}],
[{"mode": "solid", "color": [0, 0, 0]}],
)
self._is_on = False
self.async_write_ha_state()
def _scaled_color(self) -> list[int]:
"""Return the current RGB color scaled by brightness."""
scale = self._brightness / 255
r, g, b = self._rgb_color
return [round(r * scale), round(g * scale), round(b * scale)]
def _get_fallback_color(self) -> list[int]:
"""Read fallback_color from the source config in coordinator data."""
if not self.coordinator.data:
source = self._get_source()
if source is None:
return [0, 0, 0]
fallback = source.get("fallback_color")
if fallback and len(fallback) >= 3:
return list(fallback[:3])
return [0, 0, 0]
def _get_source(self) -> dict[str, Any] | None:
"""Return the cached source config dict, or None if unavailable."""
if not self.coordinator.data:
return None
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]
return source
return None
def _get_keepalive_interval(self) -> float:
"""Pick a refresh interval based on the source's ``timeout`` field.
``timeout`` is stored as a ``BindableFloat`` — either a plain number
or ``{"value": float, "source_id": str}`` when bound. A timeout of 0
means "never expire" on the server, but we still refresh
periodically so the strip recovers from server-side restarts.
"""
source = self._get_source()
timeout: float | None = None
if source is not None:
raw = source.get("timeout")
if isinstance(raw, dict):
value = raw.get("value")
if isinstance(value, (int, float)):
timeout = float(value)
elif isinstance(raw, (int, float)):
timeout = float(raw)
if timeout is None or timeout <= 0:
return MAX_KEEPALIVE_INTERVAL
# Refresh at half the timeout so we have a full timeout-window of
# slack for retry / network jitter before the server gives up.
interval = timeout / 2.0
return max(MIN_KEEPALIVE_INTERVAL, min(MAX_KEEPALIVE_INTERVAL, interval))
def _start_keepalive(self) -> None:
"""(Re)start the background re-push task."""
self._stop_keepalive()
self._keepalive_task = self.hass.loop.create_task(self._keepalive_loop())
def _stop_keepalive(self) -> None:
"""Cancel the background re-push task if running."""
task = self._keepalive_task
self._keepalive_task = None
if task is not None and not task.done():
task.cancel()
async def _keepalive_loop(self) -> None:
"""Periodically re-push the current color while the light is on."""
try:
while self._is_on:
await asyncio.sleep(self._get_keepalive_interval())
if not self._is_on:
return
try:
await self.coordinator.push_segments(
self._source_id,
[{"mode": "solid", "color": self._scaled_color()}],
)
except Exception as err: # pragma: no cover — transient API errors
_LOGGER.debug(
"Keep-alive push failed for %s: %s",
self._source_id,
err,
)
except asyncio.CancelledError:
pass