Compare commits

..

2 Commits

Author SHA1 Message Date
77dd342c4c Add software brightness control for Adalight devices
Emulates hardware brightness by multiplying pixel values before serial
send. Stored per-device and persisted across restarts.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-18 10:56:19 +03:00
27c97c3141 Add color correction postprocessing filter
New filter with color temperature (2000-10000K) and per-channel RGB
gain controls. Uses LUT-based processing for fast per-frame application.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-18 10:56:13 +03:00
8 changed files with 150 additions and 4 deletions

View File

@@ -323,8 +323,9 @@ async def get_device_brightness(
device_id: str,
_auth: AuthRequired,
store: DeviceStore = Depends(get_device_store),
manager: ProcessorManager = Depends(get_processor_manager),
):
"""Get current brightness from the WLED device."""
"""Get current brightness from the device."""
device = store.get_device(device_id)
if not device:
raise HTTPException(status_code=404, detail=f"Device {device_id} not found")
@@ -332,6 +333,8 @@ async def get_device_brightness(
raise HTTPException(status_code=400, detail=f"Brightness control is not supported for {device.device_type} devices")
try:
if device.device_type == "adalight":
return {"brightness": device.software_brightness}
provider = get_provider(device.device_type)
bri = await provider.get_brightness(device.url)
return {"brightness": bri}
@@ -346,8 +349,9 @@ async def set_device_brightness(
body: dict,
_auth: AuthRequired,
store: DeviceStore = Depends(get_device_store),
manager: ProcessorManager = Depends(get_processor_manager),
):
"""Set brightness on the WLED device directly."""
"""Set brightness on the device."""
device = store.get_device(device_id)
if not device:
raise HTTPException(status_code=404, detail=f"Device {device_id} not found")
@@ -359,6 +363,14 @@ async def set_device_brightness(
raise HTTPException(status_code=400, detail="brightness must be an integer 0-255")
try:
if device.device_type == "adalight":
device.software_brightness = bri
device.updated_at = __import__("datetime").datetime.utcnow()
store.save()
# Update runtime state so the processing loop picks it up
if device_id in manager._devices:
manager._devices[device_id].software_brightness = bri
return {"brightness": bri}
provider = get_provider(device.device_type)
await provider.set_brightness(device.url, bri)
return {"brightness": bri}

View File

@@ -22,10 +22,10 @@ class AdalightDeviceProvider(LEDDeviceProvider):
@property
def capabilities(self) -> set:
# No hardware brightness control, no standby required
# manual_led_count: user must specify LED count (can't auto-detect)
# power_control: can blank LEDs by sending all-black pixels
return {"manual_led_count", "power_control"}
# brightness_control: software brightness (multiplies pixel values before sending)
return {"manual_led_count", "power_control", "brightness_control"}
def create_client(self, url: str, **kwargs) -> LEDClient:
from wled_controller.core.adalight_client import AdalightClient

View File

@@ -1,5 +1,6 @@
"""Built-in postprocessing filters."""
import math
from typing import Any, Dict, List, Optional
import numpy as np
@@ -340,3 +341,119 @@ class FlipFilter(PostprocessingFilter):
else:
np.copyto(result, image[::-1])
return result
def _kelvin_to_rgb(kelvin: int) -> tuple:
"""Convert color temperature in Kelvin to normalized RGB multipliers.
Uses Tanner Helland's approximation, normalized so 6500K = (1, 1, 1).
"""
t = kelvin / 100.0
# Red
if t <= 66:
r = 255.0
else:
r = 329.698727446 * ((t - 60) ** -0.1332047592)
# Green
if t <= 66:
g = 99.4708025861 * math.log(t) - 161.1195681661
else:
g = 288.1221695283 * ((t - 60) ** -0.0755148492)
# Blue
if t >= 66:
b = 255.0
elif t <= 19:
b = 0.0
else:
b = 138.5177312231 * math.log(t - 10) - 305.0447927307
r = max(0.0, min(255.0, r))
g = max(0.0, min(255.0, g))
b = max(0.0, min(255.0, b))
return r / 255.0, g / 255.0, b / 255.0
# Pre-compute 6500K reference for normalization
_REF_R, _REF_G, _REF_B = _kelvin_to_rgb(6500)
@FilterRegistry.register
class ColorCorrectionFilter(PostprocessingFilter):
"""Adjusts color temperature and per-channel RGB gains using LUTs."""
filter_id = "color_correction"
filter_name = "Color Correction"
def __init__(self, options: Dict[str, Any]):
super().__init__(options)
temp = self.options["temperature"]
rg = self.options["red_gain"]
gg = self.options["green_gain"]
bg = self.options["blue_gain"]
# Color temperature → RGB multipliers, normalized to 6500K = (1,1,1)
tr, tg, tb = _kelvin_to_rgb(temp)
r_mult = (tr / _REF_R) * rg
g_mult = (tg / _REF_G) * gg
b_mult = (tb / _REF_B) * bg
# Build per-channel LUTs
src = np.arange(256, dtype=np.float32)
self._lut_r = np.clip(src * r_mult, 0, 255).astype(np.uint8)
self._lut_g = np.clip(src * g_mult, 0, 255).astype(np.uint8)
self._lut_b = np.clip(src * b_mult, 0, 255).astype(np.uint8)
self._is_neutral = (temp == 6500 and rg == 1.0 and gg == 1.0 and bg == 1.0)
@classmethod
def get_options_schema(cls) -> List[FilterOptionDef]:
return [
FilterOptionDef(
key="temperature",
label="Color Temperature (K)",
option_type="int",
default=6500,
min_value=2000,
max_value=10000,
step=100,
),
FilterOptionDef(
key="red_gain",
label="Red Gain",
option_type="float",
default=1.0,
min_value=0.0,
max_value=2.0,
step=0.05,
),
FilterOptionDef(
key="green_gain",
label="Green Gain",
option_type="float",
default=1.0,
min_value=0.0,
max_value=2.0,
step=0.05,
),
FilterOptionDef(
key="blue_gain",
label="Blue Gain",
option_type="float",
default=1.0,
min_value=0.0,
max_value=2.0,
step=0.05,
),
]
def process_image(self, image: np.ndarray, image_pool: ImagePool) -> Optional[np.ndarray]:
if self._is_neutral:
return None
image[:, :, 0] = self._lut_r[image[:, :, 0]]
image[:, :, 1] = self._lut_g[image[:, :, 1]]
image[:, :, 2] = self._lut_b[image[:, :, 2]]
return None

View File

@@ -166,6 +166,8 @@ class DeviceState:
baud_rate: Optional[int] = None
health: DeviceHealth = field(default_factory=DeviceHealth)
health_task: Optional[asyncio.Task] = None
# Software brightness for devices without hardware brightness (e.g. Adalight)
software_brightness: int = 255
# Calibration test mode (works independently of target processing)
test_mode_active: bool = False
test_mode_edges: Dict[str, Tuple[int, int, int]] = field(default_factory=dict)
@@ -281,6 +283,7 @@ class ProcessorManager:
calibration: Optional[CalibrationConfig] = None,
device_type: str = "wled",
baud_rate: Optional[int] = None,
software_brightness: int = 255,
):
"""Register a device for health monitoring.
@@ -291,6 +294,7 @@ class ProcessorManager:
calibration: Calibration config (creates default if None)
device_type: LED device type (e.g. "wled")
baud_rate: Serial baud rate (for adalight devices)
software_brightness: Software brightness 0-255 (for devices without hardware brightness)
"""
if device_id in self._devices:
raise ValueError(f"Device {device_id} already registered")
@@ -305,6 +309,7 @@ class ProcessorManager:
calibration=calibration,
device_type=device_type,
baud_rate=baud_rate,
software_brightness=software_brightness,
)
self._devices[device_id] = state
@@ -775,6 +780,8 @@ class ProcessorManager:
if not state.is_running or state.led_client is None:
break
brightness_value = int(led_brightness * 255)
if device_state and device_state.software_brightness < 255:
brightness_value = brightness_value * device_state.software_brightness // 255
if state.led_client.supports_fast_send:
state.led_client.send_pixels_fast(state.previous_colors, brightness=brightness_value)
else:
@@ -803,6 +810,8 @@ class ProcessorManager:
if not state.is_running or state.led_client is None:
break
brightness_value = int(led_brightness * 255)
if device_state and device_state.software_brightness < 255:
brightness_value = brightness_value * device_state.software_brightness // 255
t_send_start = time.perf_counter()
if state.led_client.supports_fast_send:
state.led_client.send_pixels_fast(led_colors, brightness=brightness_value)

View File

@@ -157,6 +157,7 @@ async def lifespan(app: FastAPI):
calibration=device.calibration,
device_type=device.device_type,
baud_rate=device.baud_rate,
software_brightness=device.software_brightness,
)
logger.info(f"Registered device: {device.name} ({device.id})")
except Exception as e:

View File

@@ -293,6 +293,7 @@
"filters.pixelate": "Pixelate",
"filters.auto_crop": "Auto Crop",
"filters.flip": "Flip",
"filters.color_correction": "Color Correction",
"postprocessing.description_label": "Description (optional):",
"postprocessing.description_placeholder": "Describe this template...",
"postprocessing.created": "Template created successfully",

View File

@@ -293,6 +293,7 @@
"filters.pixelate": "Пикселизация",
"filters.auto_crop": "Авто Обрезка",
"filters.flip": "Отражение",
"filters.color_correction": "Цветокоррекция",
"postprocessing.description_label": "Описание (необязательно):",
"postprocessing.description_placeholder": "Опишите этот шаблон...",
"postprocessing.created": "Шаблон успешно создан",

View File

@@ -33,6 +33,7 @@ class Device:
enabled: bool = True,
device_type: str = "wled",
baud_rate: Optional[int] = None,
software_brightness: int = 255,
calibration: Optional[CalibrationConfig] = None,
created_at: Optional[datetime] = None,
updated_at: Optional[datetime] = None,
@@ -44,6 +45,7 @@ class Device:
self.enabled = enabled
self.device_type = device_type
self.baud_rate = baud_rate
self.software_brightness = software_brightness
self.calibration = calibration or create_default_calibration(led_count)
self.created_at = created_at or datetime.utcnow()
self.updated_at = updated_at or datetime.utcnow()
@@ -63,6 +65,8 @@ class Device:
}
if self.baud_rate is not None:
d["baud_rate"] = self.baud_rate
if self.software_brightness != 255:
d["software_brightness"] = self.software_brightness
return d
@classmethod
@@ -87,6 +91,7 @@ class Device:
enabled=data.get("enabled", True),
device_type=data.get("device_type", "wled"),
baud_rate=data.get("baud_rate"),
software_brightness=data.get("software_brightness", 255),
calibration=calibration,
created_at=datetime.fromisoformat(data.get("created_at", datetime.utcnow().isoformat())),
updated_at=datetime.fromisoformat(data.get("updated_at", datetime.utcnow().isoformat())),