Add LED skip start/end, rename standby_interval to keepalive_interval, remove migrations

LED skip: set first N and last M LEDs to black on a target. Color sources
(static, gradient, effect, color cycle) render across only the active
(non-skipped) LEDs. Processor pads with blacks before sending to device.

Rename standby_interval → keepalive_interval across all Python, API
schemas, and JS. from_dict falls back to old key for existing configs.

Remove legacy migration functions (_migrate_devices_to_targets,
_migrate_targets_to_color_strips) and legacy fields from target model.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-23 02:15:29 +03:00
parent f9a5fb68ed
commit e32bfab888
14 changed files with 168 additions and 163 deletions

View File

@@ -14,5 +14,5 @@ class ProcessingSettings:
brightness: float = 1.0
smoothing: float = 0.3
interpolation_mode: str = "average"
standby_interval: float = 1.0 # seconds between keepalive sends when screen is static
keepalive_interval: float = 1.0 # seconds between keepalive sends when screen is static
state_check_interval: int = DEFAULT_STATE_CHECK_INTERVAL

View File

@@ -269,8 +269,10 @@ class ProcessorManager:
device_id: str,
color_strip_source_id: str = "",
fps: int = 30,
standby_interval: float = 1.0,
keepalive_interval: float = 1.0,
state_check_interval: int = DEFAULT_STATE_CHECK_INTERVAL,
led_skip_start: int = 0,
led_skip_end: int = 0,
):
"""Register a WLED target processor."""
if target_id in self._processors:
@@ -283,8 +285,10 @@ class ProcessorManager:
device_id=device_id,
color_strip_source_id=color_strip_source_id,
fps=fps,
standby_interval=standby_interval,
keepalive_interval=keepalive_interval,
state_check_interval=state_check_interval,
led_skip_start=led_skip_start,
led_skip_end=led_skip_end,
ctx=self._build_context(),
)
self._processors[target_id] = proc

View File

@@ -41,16 +41,20 @@ class WledTargetProcessor(TargetProcessor):
device_id: str,
color_strip_source_id: str,
fps: int,
standby_interval: float,
keepalive_interval: float,
state_check_interval: int,
ctx: TargetContext,
led_skip_start: int = 0,
led_skip_end: int = 0,
ctx: TargetContext = None,
):
super().__init__(target_id, ctx)
self._device_id = device_id
self._color_strip_source_id = color_strip_source_id
self._target_fps = fps if fps > 0 else 30
self._standby_interval = standby_interval
self._keepalive_interval = keepalive_interval
self._state_check_interval = state_check_interval
self._led_skip_start = max(0, led_skip_start)
self._led_skip_end = max(0, led_skip_end)
# Runtime state (populated on start)
self._led_client: Optional[LEDClient] = None
@@ -126,7 +130,8 @@ class WledTargetProcessor(TargetProcessor):
)
from wled_controller.core.processing.effect_stream import EffectColorStripStream
if isinstance(stream, (StaticColorStripStream, GradientColorStripStream, ColorCycleColorStripStream, EffectColorStripStream)) and device_info.led_count > 0:
stream.configure(device_info.led_count)
effective_leds = device_info.led_count - self._led_skip_start - self._led_skip_end
stream.configure(max(1, effective_leds))
# Notify stream manager of our target FPS so it can adjust capture rate
css_manager.notify_target_fps(
@@ -209,10 +214,14 @@ class WledTargetProcessor(TargetProcessor):
css_manager.notify_target_fps(
self._color_strip_source_id, self._target_id, self._target_fps
)
if "standby_interval" in settings:
self._standby_interval = settings["standby_interval"]
if "keepalive_interval" in settings:
self._keepalive_interval = settings["keepalive_interval"]
if "state_check_interval" in settings:
self._state_check_interval = settings["state_check_interval"]
if "led_skip_start" in settings:
self._led_skip_start = max(0, settings["led_skip_start"])
if "led_skip_end" in settings:
self._led_skip_end = max(0, settings["led_skip_end"])
logger.info(f"Updated settings for target {self._target_id}")
def update_device(self, device_id: str) -> None:
@@ -293,6 +302,8 @@ class WledTargetProcessor(TargetProcessor):
"display_index": self._resolved_display_index,
"overlay_active": self._overlay_active,
"needs_keepalive": self._needs_keepalive,
"led_skip_start": self._led_skip_start,
"led_skip_end": self._led_skip_end,
"last_update": metrics.last_update,
"errors": [metrics.last_error] if metrics.last_error else [],
}
@@ -404,10 +415,24 @@ class WledTargetProcessor(TargetProcessor):
])
return result
def _apply_led_skip(self, colors: np.ndarray) -> np.ndarray:
"""Pad color array with black at start/end for skipped LEDs."""
s, e = self._led_skip_start, self._led_skip_end
if s <= 0 and e <= 0:
return colors
channels = colors.shape[1] if colors.ndim == 2 else 3
parts = []
if s > 0:
parts.append(np.zeros((s, channels), dtype=np.uint8))
parts.append(colors)
if e > 0:
parts.append(np.zeros((e, channels), dtype=np.uint8))
return np.vstack(parts)
async def _processing_loop(self) -> None:
"""Main processing loop — poll ColorStripStream → apply brightness → send."""
stream = self._color_strip_stream
standby_interval = self._standby_interval
keepalive_interval = self._keepalive_interval
fps_samples: collections.deque = collections.deque(maxlen=10)
send_timestamps: collections.deque = collections.deque()
@@ -415,6 +440,7 @@ class WledTargetProcessor(TargetProcessor):
last_send_time = 0.0
prev_frame_time_stamp = time.perf_counter()
loop = asyncio.get_running_loop()
effective_leds = max(1, (device_info.led_count if device_info else 0) - self._led_skip_start - self._led_skip_end)
# Short re-poll interval when the animation thread hasn't produced a new
# frame yet. The animation thread and this loop both target the same FPS
# but are unsynchronised; without a short re-poll the loop can miss a
@@ -471,12 +497,13 @@ class WledTargetProcessor(TargetProcessor):
if colors is prev_colors:
# Same frame — send keepalive if interval elapsed (only for devices that need it)
if self._needs_keepalive and prev_colors is not None and (loop_start - last_send_time) >= standby_interval:
if self._needs_keepalive and prev_colors is not None and (loop_start - last_send_time) >= keepalive_interval:
if not self._is_running or self._led_client is None:
break
kc = prev_colors
if device_info and device_info.led_count > 0:
kc = self._fit_to_device(kc, device_info.led_count)
kc = self._fit_to_device(kc, effective_leds)
kc = self._apply_led_skip(kc)
send_colors = self._apply_brightness(kc, device_info)
if self._led_client.supports_fast_send:
self._led_client.send_pixels_fast(send_colors)
@@ -496,9 +523,10 @@ class WledTargetProcessor(TargetProcessor):
prev_colors = colors
# Fit to this device's LED count (stream may be shared)
# Fit to effective LED count (excluding skipped) then pad with blacks
if device_info and device_info.led_count > 0:
colors = self._fit_to_device(colors, device_info.led_count)
colors = self._fit_to_device(colors, effective_leds)
colors = self._apply_led_skip(colors)
# Apply device software brightness
send_colors = self._apply_brightness(colors, device_info)