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

Add adaptive brightness value source with time-of-day and scene modes

Browse Source 
New "adaptive" value source type that automatically adjusts brightness
based on external conditions. Two sub-modes: time-of-day (schedule-based
interpolation with midnight wrap) and scene brightness (frame luminance
analysis via numpy BT.601 subsampling with EMA smoothing).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Admin
2026-02-24 15:14:30 +03:00
parent 48651f0a4e
commit d339dd3f90
11 changed files with 643 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
server/src/wled_controller/core/processing/processor_manager.py
View File

@@ -95,6 +95,7 @@ class ProcessorManager:
value_source_store=value_source_store,
audio_capture_manager=self._audio_capture_manager,
audio_source_store=audio_source_store,
live_stream_manager=self._live_stream_manager,
) if value_source_store else None
self._overlay_manager = OverlayManager()
self._event_queues: List[asyncio.Queue] = []

247
server/src/wled_controller/core/processing/value_stream.py
View File

@@ -1,12 +1,14 @@
"""Value stream — runtime scalar signal generators.
A ValueStream wraps a ValueSource config and computes a float (0.0–1.0)
on demand via ``get_value()``. Three concrete types:
on demand via ``get_value()``. Five concrete types:
StaticValueStream — returns a constant
AnimatedValueStream — evaluates a periodic waveform (sine/triangle/square/sawtooth)
AudioValueStream — polls audio analysis for RMS/peak/beat, applies
sensitivity and temporal smoothing
StaticValueStream — returns a constant
AnimatedValueStream — evaluates a periodic waveform (sine/triangle/square/sawtooth)
AudioValueStream — polls audio analysis for RMS/peak/beat, applies
sensitivity and temporal smoothing
TimeOfDayValueStream — interpolates brightness along a 24h schedule
SceneValueStream — derives brightness from a picture source's frame luminance
ValueStreams are cheap (trivial math or single poll), so they compute inline
in the caller's processing loop — no background threads required.
@@ -19,12 +21,16 @@ from __future__ import annotations
import math
import time
from typing import TYPE_CHECKING, Dict, Optional
from datetime import datetime
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
import numpy as np
from wled_controller.utils import get_logger
if TYPE_CHECKING:
from wled_controller.core.audio.audio_capture import AudioCaptureManager
from wled_controller.core.processing.live_stream_manager import LiveStreamManager
from wled_controller.storage.audio_source_store import AudioSourceStore
from wled_controller.storage.value_source import ValueSource
from wled_controller.storage.value_source_store import ValueSourceStore
@@ -277,6 +283,214 @@ class AudioValueStream(ValueStream):
)
# ---------------------------------------------------------------------------
# Time of Day
# ---------------------------------------------------------------------------
_MINUTES_PER_DAY = 1440 # 24 * 60
class TimeOfDayValueStream(ValueStream):
"""Interpolates brightness along a 24-hour schedule.
Schedule is a list of {"time": "HH:MM", "value": 0.0–1.0} dicts.
At runtime, finds the two surrounding points and linearly interpolates.
The schedule wraps around midnight.
"""
def __init__(
self,
schedule: List[dict],
min_value: float = 0.0,
max_value: float = 1.0,
):
self._points: List[Tuple[float, float]] = [] # (minutes, value)
self._min = min_value
self._max = max_value
self._parse_schedule(schedule)
def _parse_schedule(self, schedule: List[dict]) -> None:
"""Parse schedule into sorted (minutes, value) tuples."""
points = []
for entry in schedule:
t_str = entry.get("time", "00:00")
parts = t_str.split(":")
h = int(parts[0])
m = int(parts[1]) if len(parts) > 1 else 0
minutes = h * 60 + m
val = max(0.0, min(1.0, float(entry.get("value", 1.0))))
points.append((minutes, val))
points.sort(key=lambda p: p[0])
self._points = points
def get_value(self) -> float:
if len(self._points) < 2:
return self._max # fallback: full brightness
now = datetime.now()
current = now.hour * 60 + now.minute + now.second / 60.0
points = self._points
n = len(points)
# Find the first point whose time is > current
right_idx = 0
for i, (t, _) in enumerate(points):
if t > current:
right_idx = i
break
else:
right_idx = 0 # all points <= current: wrap to first
left_idx = (right_idx - 1) % n
t_left, v_left = points[left_idx]
t_right, v_right = points[right_idx]
# Compute interval length (handling midnight wrap)
if t_right > t_left:
interval = t_right - t_left
elapsed = current - t_left
else:
interval = (_MINUTES_PER_DAY - t_left) + t_right
elapsed = (current - t_left) if current >= t_left else (_MINUTES_PER_DAY - t_left) + current
frac = elapsed / interval if interval > 0 else 0.0
raw = v_left + frac * (v_right - v_left)
# Map to output range
return self._min + max(0.0, min(1.0, raw)) * (self._max - self._min)
def update_source(self, source: "ValueSource") -> None:
from wled_controller.storage.value_source import AdaptiveValueSource
if isinstance(source, AdaptiveValueSource) and source.adaptive_mode == "time_of_day":
self._parse_schedule(source.schedule)
self._min = source.min_value
self._max = source.max_value
# ---------------------------------------------------------------------------
# Scene
# ---------------------------------------------------------------------------
class SceneValueStream(ValueStream):
"""Derives brightness from a picture source's average frame luminance.
Acquires a LiveStream from LiveStreamManager on start(). On each
get_value() call, reads the latest frame, subsamples, computes mean
luminance (BT.601), applies sensitivity/smoothing/mapping.
Behaviors:
complement — dark scene → high brightness (ambient backlight visibility)
match — bright scene → high brightness
"""
def __init__(
self,
picture_source_id: str,
scene_behavior: str = "complement",
sensitivity: float = 1.0,
smoothing: float = 0.3,
min_value: float = 0.0,
max_value: float = 1.0,
live_stream_manager: Optional["LiveStreamManager"] = None,
):
self._picture_source_id = picture_source_id
self._behavior = scene_behavior
self._sensitivity = sensitivity
self._smoothing = smoothing
self._min = min_value
self._max = max_value
self._live_stream_manager = live_stream_manager
self._live_stream = None
self._prev_value = 0.5 # neutral start
def start(self) -> None:
if self._live_stream_manager and self._picture_source_id:
try:
self._live_stream = self._live_stream_manager.acquire(
self._picture_source_id
)
logger.info(
f"SceneValueStream acquired live stream for {self._picture_source_id}"
)
except Exception as e:
logger.warning(f"SceneValueStream failed to acquire live stream: {e}")
self._live_stream = None
def stop(self) -> None:
if self._live_stream is not None and self._live_stream_manager:
try:
self._live_stream_manager.release(self._picture_source_id)
except Exception as e:
logger.warning(f"SceneValueStream failed to release live stream: {e}")
self._live_stream = None
self._prev_value = 0.5
def get_value(self) -> float:
if self._live_stream is None:
return self._prev_value
frame = self._live_stream.get_latest_frame()
if frame is None:
return self._prev_value
# Fast luminance: subsample to ~64x64 via numpy stride (zero-copy view)
img = frame.image
h, w = img.shape[:2]
step_h = max(1, h // 64)
step_w = max(1, w // 64)
sampled = img[::step_h, ::step_w].astype(np.float32)
# BT.601 weighted luminance, normalized to [0, 1]
luminance = float(
(0.299 * sampled[:, :, 0] + 0.587 * sampled[:, :, 1] + 0.114 * sampled[:, :, 2]).mean()
) / 255.0
# Apply sensitivity
raw = min(1.0, luminance * self._sensitivity)
# Apply behavior
if self._behavior == "complement":
raw = 1.0 - raw
# Temporal smoothing (EMA)
smoothed = self._smoothing * self._prev_value + (1.0 - self._smoothing) * raw
self._prev_value = smoothed
# Map to output range
clamped = max(0.0, min(1.0, smoothed))
return self._min + clamped * (self._max - self._min)
def update_source(self, source: "ValueSource") -> None:
from wled_controller.storage.value_source import AdaptiveValueSource
if not isinstance(source, AdaptiveValueSource) or source.adaptive_mode != "scene":
return
self._behavior = source.scene_behavior
self._sensitivity = source.sensitivity
self._smoothing = source.smoothing
self._min = source.min_value
self._max = source.max_value
# If picture source changed, swap live streams
if source.picture_source_id != self._picture_source_id:
old_id = self._picture_source_id
self._picture_source_id = source.picture_source_id
if self._live_stream is not None and self._live_stream_manager:
self._live_stream_manager.release(old_id)
try:
self._live_stream = self._live_stream_manager.acquire(
self._picture_source_id
)
logger.info(
f"SceneValueStream swapped live stream: {old_id} → "
f"{self._picture_source_id}"
)
except Exception as e:
logger.warning(f"SceneValueStream failed to swap live stream: {e}")
self._live_stream = None
# ---------------------------------------------------------------------------
# Manager
# ---------------------------------------------------------------------------
@@ -297,10 +511,12 @@ class ValueStreamManager:
value_source_store: "ValueSourceStore",
audio_capture_manager: Optional["AudioCaptureManager"] = None,
audio_source_store: Optional["AudioSourceStore"] = None,
live_stream_manager: Optional["LiveStreamManager"] = None,
):
self._value_source_store = value_source_store
self._audio_capture_manager = audio_capture_manager
self._audio_source_store = audio_source_store
self._live_stream_manager = live_stream_manager
self._streams: Dict[str, ValueStream] = {}
def acquire(self, vs_id: str, consumer_id: str) -> ValueStream:
@@ -359,6 +575,7 @@ class ValueStreamManager:
def _create_stream(self, source: "ValueSource") -> ValueStream:
"""Factory: create the appropriate ValueStream for a ValueSource."""
from wled_controller.storage.value_source import (
AdaptiveValueSource,
AnimatedValueSource,
AudioValueSource,
StaticValueSource,
@@ -385,5 +602,23 @@ class ValueStreamManager:
audio_source_store=self._audio_source_store,
)
if isinstance(source, AdaptiveValueSource):
if source.adaptive_mode == "scene":
return SceneValueStream(
picture_source_id=source.picture_source_id,
scene_behavior=source.scene_behavior,
sensitivity=source.sensitivity,
smoothing=source.smoothing,
min_value=source.min_value,
max_value=source.max_value,
live_stream_manager=self._live_stream_manager,
)
# Default: time_of_day
return TimeOfDayValueStream(
schedule=source.schedule,
min_value=source.min_value,
max_value=source.max_value,
)
# Fallback
return StaticValueStream(value=1.0)