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

Add Key Colors target type for extracting colors from screen regions

Browse Source 
Introduce a new "key_colors" target type alongside WLED targets, enabling
real-time color extraction from configurable screen rectangles with
average/median/dominant modes, temporal smoothing, and WebSocket streaming.

- Split WledPictureTarget into its own module, add KeyColorsPictureTarget
- Add KC target lifecycle to ProcessorManager (register, start/stop, processing loop)
- Extend API routes and schemas for KC targets (CRUD, settings, state, metrics, colors)
- Add WebSocket endpoint for real-time color updates with auth
- Add KC sub-tab in Targets UI with editor modal and live color swatches
- Add EN and RU translations for all key colors strings

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Admin
2026-02-12 16:43:09 +03:00
parent 3d2393e474
commit 5f9bc9a37e
13 changed files with 1525 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

215
server/src/wled_controller/api/routes/picture_targets.py
View File

@@ -1,6 +1,8 @@
"""Picture target routes: CRUD, processing control, settings, state, metrics."""
from fastapi import APIRouter, HTTPException, Depends
import secrets
from fastapi import APIRouter, HTTPException, Depends, Query, WebSocket, WebSocketDisconnect
from wled_controller.api.auth import AuthRequired
from wled_controller.api.dependencies import (
@@ -9,6 +11,10 @@ from wled_controller.api.dependencies import (
get_processor_manager,
)
from wled_controller.api.schemas.picture_targets import (
ExtractedColorResponse,
KeyColorRectangleSchema,
KeyColorsResponse,
KeyColorsSettingsSchema,
PictureTargetCreate,
PictureTargetListResponse,
PictureTargetResponse,
@@ -17,9 +23,15 @@ from wled_controller.api.schemas.picture_targets import (
TargetMetricsResponse,
TargetProcessingState,
)
from wled_controller.config import config
from wled_controller.core.processor_manager import ProcessorManager, ProcessingSettings
from wled_controller.storage import DeviceStore
from wled_controller.storage.picture_target import WledPictureTarget
from wled_controller.storage.wled_picture_target import WledPictureTarget
from wled_controller.storage.key_colors_picture_target import (
KeyColorRectangle,
KeyColorsSettings,
KeyColorsPictureTarget,
)
from wled_controller.storage.picture_target_store import PictureTargetStore
from wled_controller.utils import get_logger
@@ -66,20 +78,66 @@ def _settings_to_schema(settings: ProcessingSettings) -> ProcessingSettingsSchem
)
def _kc_settings_to_schema(settings: KeyColorsSettings) -> KeyColorsSettingsSchema:
"""Convert core KeyColorsSettings to schema."""
return KeyColorsSettingsSchema(
fps=settings.fps,
interpolation_mode=settings.interpolation_mode,
smoothing=settings.smoothing,
rectangles=[
KeyColorRectangleSchema(name=r.name, x=r.x, y=r.y, width=r.width, height=r.height)
for r in settings.rectangles
],
)
def _kc_schema_to_settings(schema: KeyColorsSettingsSchema) -> KeyColorsSettings:
"""Convert schema KeyColorsSettings to core."""
return KeyColorsSettings(
fps=schema.fps,
interpolation_mode=schema.interpolation_mode,
smoothing=schema.smoothing,
rectangles=[
KeyColorRectangle(name=r.name, x=r.x, y=r.y, width=r.width, height=r.height)
for r in schema.rectangles
],
)
def _target_to_response(target) -> PictureTargetResponse:
"""Convert a PictureTarget to PictureTargetResponse."""
settings_schema = _settings_to_schema(target.settings) if isinstance(target, WledPictureTarget) else ProcessingSettingsSchema()
return PictureTargetResponse(
id=target.id,
name=target.name,
target_type=target.target_type,
device_id=target.device_id if isinstance(target, WledPictureTarget) else "",
picture_source_id=target.picture_source_id if isinstance(target, WledPictureTarget) else "",
settings=settings_schema,
description=target.description,
created_at=target.created_at,
updated_at=target.updated_at,
)
if isinstance(target, WledPictureTarget):
return PictureTargetResponse(
id=target.id,
name=target.name,
target_type=target.target_type,
device_id=target.device_id,
picture_source_id=target.picture_source_id,
settings=_settings_to_schema(target.settings),
description=target.description,
created_at=target.created_at,
updated_at=target.updated_at,
)
elif isinstance(target, KeyColorsPictureTarget):
return PictureTargetResponse(
id=target.id,
name=target.name,
target_type=target.target_type,
picture_source_id=target.picture_source_id,
key_colors_settings=_kc_settings_to_schema(target.settings),
description=target.description,
created_at=target.created_at,
updated_at=target.updated_at,
)
else:
return PictureTargetResponse(
id=target.id,
name=target.name,
target_type=target.target_type,
description=target.description,
created_at=target.created_at,
updated_at=target.updated_at,
)
# ===== CRUD ENDPOINTS =====
@@ -102,6 +160,7 @@ async def create_target(
# Convert settings
core_settings = _settings_to_core(data.settings) if data.settings else None
kc_settings = _kc_schema_to_settings(data.key_colors_settings) if data.key_colors_settings else None
# Create in store
target = target_store.create_target(
@@ -110,6 +169,7 @@ async def create_target(
device_id=data.device_id,
picture_source_id=data.picture_source_id,
settings=core_settings,
key_colors_settings=kc_settings,
description=data.description,
)
@@ -124,6 +184,15 @@ async def create_target(
)
except ValueError as e:
logger.warning(f"Could not register target {target.id} in processor manager: {e}")
elif isinstance(target, KeyColorsPictureTarget):
try:
manager.add_kc_target(
target_id=target.id,
picture_source_id=target.picture_source_id,
settings=target.settings,
)
except ValueError as e:
logger.warning(f"Could not register KC target {target.id}: {e}")
return _target_to_response(target)
@@ -180,6 +249,7 @@ async def update_target(
# Convert settings
core_settings = _settings_to_core(data.settings) if data.settings else None
kc_settings = _kc_schema_to_settings(data.key_colors_settings) if data.key_colors_settings else None
# Update in store
target = target_store.update_target(
@@ -188,6 +258,7 @@ async def update_target(
device_id=data.device_id,
picture_source_id=data.picture_source_id,
settings=core_settings,
key_colors_settings=kc_settings,
description=data.description,
)
@@ -201,7 +272,14 @@ async def update_target(
if data.device_id is not None:
manager.update_target_device(target_id, target.device_id)
except ValueError:
# Target may not be registered in manager yet
pass
elif isinstance(target, KeyColorsPictureTarget):
try:
if data.key_colors_settings is not None:
manager.update_kc_target_settings(target_id, target.settings)
if data.picture_source_id is not None:
manager.update_kc_target_source(target_id, target.picture_source_id)
except ValueError:
pass
return _target_to_response(target)
@@ -224,16 +302,21 @@ async def delete_target(
):
"""Delete a picture target. Stops processing first if active."""
try:
# Stop processing if running
# Stop processing if running (WLED or KC)
try:
if manager.is_target_processing(target_id):
if manager.is_kc_target(target_id):
await manager.stop_kc_processing(target_id)
elif manager.is_target_processing(target_id):
await manager.stop_processing(target_id)
except ValueError:
pass
# Remove from manager
# Remove from manager (WLED or KC)
try:
manager.remove_target(target_id)
if manager.is_kc_target(target_id):
manager.remove_kc_target(target_id)
else:
manager.remove_target(target_id)
except (ValueError, RuntimeError):
pass
@@ -260,10 +343,13 @@ async def start_processing(
):
"""Start processing for a picture target."""
try:
# Verify target exists
target_store.get_target(target_id)
# Verify target exists and dispatch by type
target = target_store.get_target(target_id)
await manager.start_processing(target_id)
if isinstance(target, KeyColorsPictureTarget):
await manager.start_kc_processing(target_id)
else:
await manager.start_processing(target_id)
logger.info(f"Started processing for target {target_id}")
return {"status": "started", "target_id": target_id}
@@ -285,7 +371,10 @@ async def stop_processing(
):
"""Stop processing for a picture target."""
try:
await manager.stop_processing(target_id)
if manager.is_kc_target(target_id):
await manager.stop_kc_processing(target_id)
else:
await manager.stop_processing(target_id)
logger.info(f"Stopped processing for target {target_id}")
return {"status": "stopped", "target_id": target_id}
@@ -307,7 +396,10 @@ async def get_target_state(
):
"""Get current processing state for a target."""
try:
state = manager.get_target_state(target_id)
if manager.is_kc_target(target_id):
state = manager.get_kc_target_state(target_id)
else:
state = manager.get_target_state(target_id)
return TargetProcessingState(**state)
except ValueError as e:
@@ -317,7 +409,7 @@ async def get_target_state(
raise HTTPException(status_code=500, detail=str(e))
@router.get("/api/v1/picture-targets/{target_id}/settings", response_model=ProcessingSettingsSchema, tags=["Settings"])
@router.get("/api/v1/picture-targets/{target_id}/settings", tags=["Settings"])
async def get_target_settings(
target_id: str,
_auth: AuthRequired,
@@ -326,6 +418,8 @@ async def get_target_settings(
"""Get processing settings for a target."""
try:
target = target_store.get_target(target_id)
if isinstance(target, KeyColorsPictureTarget):
return _kc_settings_to_schema(target.settings)
if isinstance(target, WledPictureTarget):
return _settings_to_schema(target.settings)
return ProcessingSettingsSchema()
@@ -405,7 +499,10 @@ async def get_target_metrics(
):
"""Get processing metrics for a target."""
try:
metrics = manager.get_target_metrics(target_id)
if manager.is_kc_target(target_id):
metrics = manager.get_kc_target_metrics(target_id)
else:
metrics = manager.get_target_metrics(target_id)
return TargetMetricsResponse(**metrics)
except ValueError as e:
@@ -413,3 +510,69 @@ async def get_target_metrics(
except Exception as e:
logger.error(f"Failed to get target metrics: {e}")
raise HTTPException(status_code=500, detail=str(e))
# ===== KEY COLORS ENDPOINTS =====
@router.get("/api/v1/picture-targets/{target_id}/colors", response_model=KeyColorsResponse, tags=["Key Colors"])
async def get_target_colors(
target_id: str,
_auth: AuthRequired,
manager: ProcessorManager = Depends(get_processor_manager),
):
"""Get latest extracted colors for a key-colors target (polling)."""
try:
raw_colors = manager.get_kc_latest_colors(target_id)
colors = {}
for name, (r, g, b) in raw_colors.items():
colors[name] = ExtractedColorResponse(
r=r, g=g, b=b,
hex=f"#{r:02x}{g:02x}{b:02x}",
)
from datetime import datetime
return KeyColorsResponse(
target_id=target_id,
colors=colors,
timestamp=datetime.utcnow(),
)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
@router.websocket("/api/v1/picture-targets/{target_id}/ws")
async def target_colors_ws(
websocket: WebSocket,
target_id: str,
token: str = Query(""),
):
"""WebSocket for real-time key color updates. Auth via ?token=<api_key>."""
# Authenticate
authenticated = False
if token and config.auth.api_keys:
for _label, api_key in config.auth.api_keys.items():
if secrets.compare_digest(token, api_key):
authenticated = True
break
if not authenticated:
await websocket.close(code=4001, reason="Unauthorized")
return
await websocket.accept()
manager = get_processor_manager()
try:
manager.add_kc_ws_client(target_id, websocket)
except ValueError:
await websocket.close(code=4004, reason="Target not found")
return
try:
while True:
# Keep alive — wait for client messages (or disconnect)
await websocket.receive_text()
except WebSocketDisconnect:
pass
finally:
manager.remove_kc_ws_client(target_id, websocket)

55
server/src/wled_controller/api/schemas/picture_targets.py
View File

@@ -35,14 +35,51 @@ class ProcessingSettings(BaseModel):
)
class KeyColorRectangleSchema(BaseModel):
"""A named rectangle for key color extraction (relative coords 0.0-1.0)."""
name: str = Field(description="Rectangle name", min_length=1, max_length=50)
x: float = Field(default=0.0, description="Left edge (0.0-1.0)", ge=0.0, le=1.0)
y: float = Field(default=0.0, description="Top edge (0.0-1.0)", ge=0.0, le=1.0)
width: float = Field(default=1.0, description="Width (0.0-1.0)", gt=0.0, le=1.0)
height: float = Field(default=1.0, description="Height (0.0-1.0)", gt=0.0, le=1.0)
class KeyColorsSettingsSchema(BaseModel):
"""Settings for key colors extraction."""
fps: int = Field(default=10, description="Extraction rate (1-60)", ge=1, le=60)
interpolation_mode: str = Field(default="average", description="Color mode (average, median, dominant)")
smoothing: float = Field(default=0.3, description="Temporal smoothing (0.0-1.0)", ge=0.0, le=1.0)
rectangles: List[KeyColorRectangleSchema] = Field(default_factory=list, description="Rectangles to extract colors from")
class ExtractedColorResponse(BaseModel):
"""A single extracted color."""
r: int = Field(description="Red (0-255)")
g: int = Field(description="Green (0-255)")
b: int = Field(description="Blue (0-255)")
hex: str = Field(description="Hex color (#rrggbb)")
class KeyColorsResponse(BaseModel):
"""Extracted key colors for a target."""
target_id: str = Field(description="Target ID")
colors: Dict[str, ExtractedColorResponse] = Field(description="Rectangle name -> color")
timestamp: Optional[datetime] = Field(None, description="Extraction timestamp")
class PictureTargetCreate(BaseModel):
"""Request to create a picture target."""
name: str = Field(description="Target name", min_length=1, max_length=100)
target_type: str = Field(default="wled", description="Target type (wled)")
target_type: str = Field(default="wled", description="Target type (wled, key_colors)")
device_id: str = Field(default="", description="WLED device ID")
picture_source_id: str = Field(default="", description="Picture source ID")
settings: Optional[ProcessingSettings] = Field(None, description="Processing settings")
settings: Optional[ProcessingSettings] = Field(None, description="Processing settings (for wled targets)")
key_colors_settings: Optional[KeyColorsSettingsSchema] = Field(None, description="Key colors settings (for key_colors targets)")
description: Optional[str] = Field(None, description="Optional description", max_length=500)
@@ -52,7 +89,8 @@ class PictureTargetUpdate(BaseModel):
name: Optional[str] = Field(None, description="Target name", min_length=1, max_length=100)
device_id: Optional[str] = Field(None, description="WLED device ID")
picture_source_id: Optional[str] = Field(None, description="Picture source ID")
settings: Optional[ProcessingSettings] = Field(None, description="Processing settings")
settings: Optional[ProcessingSettings] = Field(None, description="Processing settings (for wled targets)")
key_colors_settings: Optional[KeyColorsSettingsSchema] = Field(None, description="Key colors settings (for key_colors targets)")
description: Optional[str] = Field(None, description="Optional description", max_length=500)
@@ -64,7 +102,8 @@ class PictureTargetResponse(BaseModel):
target_type: str = Field(description="Target type")
device_id: str = Field(default="", description="WLED device ID")
picture_source_id: str = Field(default="", description="Picture source ID")
settings: ProcessingSettings = Field(description="Processing settings")
settings: Optional[ProcessingSettings] = Field(None, description="Processing settings (wled)")
key_colors_settings: Optional[KeyColorsSettingsSchema] = Field(None, description="Key colors settings (key_colors)")
description: Optional[str] = Field(None, description="Description")
created_at: datetime = Field(description="Creation timestamp")
updated_at: datetime = Field(description="Last update timestamp")
@@ -81,11 +120,11 @@ class TargetProcessingState(BaseModel):
"""Processing state for a picture target."""
target_id: str = Field(description="Target ID")
device_id: str = Field(description="Device ID")
device_id: Optional[str] = Field(None, description="Device ID")
processing: bool = Field(description="Whether processing is active")
fps_actual: Optional[float] = Field(None, description="Actual FPS achieved")
fps_target: int = Field(description="Target FPS")
display_index: int = Field(description="Current display index")
fps_target: int = Field(default=0, description="Target FPS")
display_index: int = Field(default=0, description="Current display index")
last_update: Optional[datetime] = Field(None, description="Last successful update")
errors: List[str] = Field(default_factory=list, description="Recent errors")
wled_online: bool = Field(default=False, description="Whether WLED device is reachable")
@@ -103,7 +142,7 @@ class TargetMetricsResponse(BaseModel):
"""Target metrics response."""
target_id: str = Field(description="Target ID")
device_id: str = Field(description="Device ID")
device_id: Optional[str] = Field(None, description="Device ID")
processing: bool = Field(description="Whether processing is active")
fps_actual: Optional[float] = Field(None, description="Actual FPS")
fps_target: int = Field(description="Target FPS")