Refactor capture engine architecture, rename PictureStream to PictureSource, and split API modules

- Separate CaptureEngine into stateless factory + stateful CaptureStream session
- Add LiveStream/LiveStreamManager for shared capture with reference counting
- Rename PictureStream to PictureSource across storage, API, and UI
- Remove legacy migration logic and unused compatibility code
- Split monolithic routes.py (1935 lines) into 5 focused route modules
- Split schemas.py (480 lines) into 7 schema modules with re-exports
- Extract dependency injection into dedicated dependencies.py

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-12 14:27:00 +03:00
parent b8389f080a
commit c3828e10fa
42 changed files with 4047 additions and 3797 deletions

View File

@@ -0,0 +1,107 @@
"""Pydantic schemas for API request and response models."""
from .common import (
CaptureImage,
BorderExtraction,
ErrorResponse,
PerformanceMetrics,
TemplateTestResponse,
)
from .system import (
DisplayInfo,
DisplayListResponse,
HealthResponse,
VersionResponse,
)
from .devices import (
Calibration,
CalibrationTestModeRequest,
CalibrationTestModeResponse,
ColorCorrection,
DeviceCreate,
DeviceListResponse,
DeviceResponse,
DeviceUpdate,
MetricsResponse,
ProcessingSettings,
ProcessingState,
)
from .templates import (
EngineInfo,
EngineListResponse,
TemplateAssignment,
TemplateCreate,
TemplateListResponse,
TemplateResponse,
TemplateTestRequest,
TemplateUpdate,
)
from .filters import (
FilterInstanceSchema,
FilterOptionDefSchema,
FilterTypeListResponse,
FilterTypeResponse,
)
from .postprocessing import (
PostprocessingTemplateCreate,
PostprocessingTemplateListResponse,
PostprocessingTemplateResponse,
PostprocessingTemplateUpdate,
PPTemplateTestRequest,
)
from .picture_sources import (
ImageValidateRequest,
ImageValidateResponse,
PictureSourceCreate,
PictureSourceListResponse,
PictureSourceResponse,
PictureSourceTestRequest,
PictureSourceUpdate,
)
__all__ = [
"CaptureImage",
"BorderExtraction",
"ErrorResponse",
"PerformanceMetrics",
"TemplateTestResponse",
"DisplayInfo",
"DisplayListResponse",
"HealthResponse",
"VersionResponse",
"Calibration",
"CalibrationTestModeRequest",
"CalibrationTestModeResponse",
"ColorCorrection",
"DeviceCreate",
"DeviceListResponse",
"DeviceResponse",
"DeviceUpdate",
"MetricsResponse",
"ProcessingSettings",
"ProcessingState",
"EngineInfo",
"EngineListResponse",
"TemplateAssignment",
"TemplateCreate",
"TemplateListResponse",
"TemplateResponse",
"TemplateTestRequest",
"TemplateUpdate",
"FilterInstanceSchema",
"FilterOptionDefSchema",
"FilterTypeListResponse",
"FilterTypeResponse",
"PostprocessingTemplateCreate",
"PostprocessingTemplateListResponse",
"PostprocessingTemplateResponse",
"PostprocessingTemplateUpdate",
"PPTemplateTestRequest",
"ImageValidateRequest",
"ImageValidateResponse",
"PictureSourceCreate",
"PictureSourceListResponse",
"PictureSourceResponse",
"PictureSourceTestRequest",
"PictureSourceUpdate",
]

View File

@@ -0,0 +1,52 @@
"""Shared schemas used across multiple route modules."""
from datetime import datetime
from typing import Dict, List, Optional
from pydantic import BaseModel, Field
class ErrorResponse(BaseModel):
"""Error response."""
error: str = Field(description="Error type")
message: str = Field(description="Error message")
detail: Optional[Dict] = Field(None, description="Additional error details")
timestamp: datetime = Field(default_factory=datetime.utcnow, description="Error timestamp")
class CaptureImage(BaseModel):
"""Captured image with metadata."""
image: str = Field(description="Base64-encoded thumbnail image data")
full_image: Optional[str] = Field(None, description="Base64-encoded full-resolution image data")
width: int = Field(description="Original image width in pixels")
height: int = Field(description="Original image height in pixels")
thumbnail_width: Optional[int] = Field(None, description="Thumbnail width (if resized)")
thumbnail_height: Optional[int] = Field(None, description="Thumbnail height (if resized)")
class BorderExtraction(BaseModel):
"""Extracted border images."""
top: str = Field(description="Base64-encoded top border image")
right: str = Field(description="Base64-encoded right border image")
bottom: str = Field(description="Base64-encoded bottom border image")
left: str = Field(description="Base64-encoded left border image")
class PerformanceMetrics(BaseModel):
"""Performance metrics for template test."""
capture_duration_s: float = Field(description="Total capture duration in seconds")
frame_count: int = Field(description="Number of frames captured")
actual_fps: float = Field(description="Actual FPS (frame_count / duration)")
avg_capture_time_ms: float = Field(description="Average time per frame capture in milliseconds")
class TemplateTestResponse(BaseModel):
"""Response from template test."""
full_capture: CaptureImage = Field(description="Full screen capture with thumbnail")
border_extraction: Optional[BorderExtraction] = Field(None, description="Extracted border images (deprecated)")
performance: PerformanceMetrics = Field(description="Performance metrics")

View File

@@ -0,0 +1,161 @@
"""Device-related schemas (CRUD, settings, calibration, processing state, metrics)."""
from datetime import datetime
from typing import Dict, List, Literal, Optional
from pydantic import BaseModel, Field
from wled_controller.core.processor_manager import DEFAULT_STATE_CHECK_INTERVAL
class DeviceCreate(BaseModel):
"""Request to create/attach a WLED device."""
name: str = Field(description="Device name", min_length=1, max_length=100)
url: str = Field(description="WLED device URL (e.g., http://192.168.1.100)")
class DeviceUpdate(BaseModel):
"""Request to update device information."""
name: Optional[str] = Field(None, description="Device name", min_length=1, max_length=100)
url: Optional[str] = Field(None, description="WLED device URL")
enabled: Optional[bool] = Field(None, description="Whether device is enabled")
picture_source_id: Optional[str] = Field(None, description="Picture source ID")
class ColorCorrection(BaseModel):
"""Color correction settings."""
gamma: float = Field(default=2.2, description="Gamma correction", ge=0.1, le=5.0)
saturation: float = Field(default=1.0, description="Saturation multiplier", ge=0.0, le=2.0)
brightness: float = Field(default=1.0, description="Brightness multiplier", ge=0.0, le=1.0)
class ProcessingSettings(BaseModel):
"""Processing settings for a device."""
display_index: int = Field(default=0, description="Display to capture", ge=0)
fps: int = Field(default=30, description="Target frames per second", ge=10, le=90)
border_width: int = Field(default=10, description="Border width in pixels", ge=1, le=100)
interpolation_mode: str = Field(default="average", description="LED color interpolation mode (average, median, dominant)")
brightness: float = Field(default=1.0, description="Global brightness (0.0-1.0)", ge=0.0, le=1.0)
smoothing: float = Field(default=0.3, description="Temporal smoothing factor (0.0=none, 1.0=full)", ge=0.0, le=1.0)
state_check_interval: int = Field(
default=DEFAULT_STATE_CHECK_INTERVAL, ge=5, le=600,
description="Seconds between WLED health checks"
)
color_correction: Optional[ColorCorrection] = Field(
default_factory=ColorCorrection,
description="Color correction settings"
)
class Calibration(BaseModel):
"""Calibration configuration for pixel-to-LED mapping."""
layout: Literal["clockwise", "counterclockwise"] = Field(
default="clockwise",
description="LED strip layout direction"
)
start_position: Literal["top_left", "top_right", "bottom_left", "bottom_right"] = Field(
default="bottom_left",
description="Starting corner of the LED strip"
)
offset: int = Field(
default=0,
ge=0,
description="Number of LEDs from physical LED 0 to start corner (along strip direction)"
)
leds_top: int = Field(default=0, ge=0, description="Number of LEDs on the top edge")
leds_right: int = Field(default=0, ge=0, description="Number of LEDs on the right edge")
leds_bottom: int = Field(default=0, ge=0, description="Number of LEDs on the bottom edge")
leds_left: int = Field(default=0, ge=0, description="Number of LEDs on the left edge")
# Per-edge span: fraction of screen side covered by LEDs (0.0-1.0)
span_top_start: float = Field(default=0.0, ge=0.0, le=1.0, description="Start of top edge coverage")
span_top_end: float = Field(default=1.0, ge=0.0, le=1.0, description="End of top edge coverage")
span_right_start: float = Field(default=0.0, ge=0.0, le=1.0, description="Start of right edge coverage")
span_right_end: float = Field(default=1.0, ge=0.0, le=1.0, description="End of right edge coverage")
span_bottom_start: float = Field(default=0.0, ge=0.0, le=1.0, description="Start of bottom edge coverage")
span_bottom_end: float = Field(default=1.0, ge=0.0, le=1.0, description="End of bottom edge coverage")
span_left_start: float = Field(default=0.0, ge=0.0, le=1.0, description="Start of left edge coverage")
span_left_end: float = Field(default=1.0, ge=0.0, le=1.0, description="End of left edge coverage")
class CalibrationTestModeRequest(BaseModel):
"""Request to set calibration test mode with multiple edges."""
edges: Dict[str, List[int]] = Field(
default_factory=dict,
description="Map of active edge names to RGB colors. "
"E.g. {'top': [255, 0, 0], 'left': [255, 255, 0]}. "
"Empty dict = exit test mode."
)
class CalibrationTestModeResponse(BaseModel):
"""Response for calibration test mode."""
test_mode: bool = Field(description="Whether test mode is active")
active_edges: List[str] = Field(default_factory=list, description="Currently lit edges")
device_id: str = Field(description="Device ID")
class DeviceResponse(BaseModel):
"""Device information response."""
id: str = Field(description="Device ID")
name: str = Field(description="Device name")
url: str = Field(description="WLED device URL")
led_count: int = Field(description="Total number of LEDs")
enabled: bool = Field(description="Whether device is enabled")
status: Literal["connected", "disconnected", "error"] = Field(
description="Connection status"
)
settings: ProcessingSettings = Field(description="Processing settings")
calibration: Optional[Calibration] = Field(None, description="Calibration configuration")
picture_source_id: str = Field(default="", description="ID of assigned picture source")
created_at: datetime = Field(description="Creation timestamp")
updated_at: datetime = Field(description="Last update timestamp")
class DeviceListResponse(BaseModel):
"""List of devices response."""
devices: List[DeviceResponse] = Field(description="List of devices")
count: int = Field(description="Number of devices")
class ProcessingState(BaseModel):
"""Processing state for a device."""
device_id: str = Field(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")
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")
wled_latency_ms: Optional[float] = Field(None, description="WLED health check latency in ms")
wled_name: Optional[str] = Field(None, description="WLED device name")
wled_version: Optional[str] = Field(None, description="WLED firmware version")
wled_led_count: Optional[int] = Field(None, description="LED count reported by WLED device")
wled_rgbw: Optional[bool] = Field(None, description="Whether WLED device uses RGBW LEDs")
wled_led_type: Optional[str] = Field(None, description="LED chip type (e.g. WS2812B, SK6812 RGBW)")
wled_last_checked: Optional[datetime] = Field(None, description="Last health check time")
wled_error: Optional[str] = Field(None, description="Last health check error")
class MetricsResponse(BaseModel):
"""Device metrics response."""
device_id: str = Field(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")
uptime_seconds: float = Field(description="Processing uptime in seconds")
frames_processed: int = Field(description="Total frames processed")
errors_count: int = Field(description="Total error count")
last_error: Optional[str] = Field(None, description="Last error message")
last_update: Optional[datetime] = Field(None, description="Last update timestamp")

View File

@@ -0,0 +1,39 @@
"""Filter-related schemas."""
from typing import Any, Dict, List
from pydantic import BaseModel, Field
class FilterInstanceSchema(BaseModel):
"""A single filter instance with its configuration."""
filter_id: str = Field(description="Filter type identifier")
options: Dict[str, Any] = Field(default_factory=dict, description="Filter-specific options")
class FilterOptionDefSchema(BaseModel):
"""Describes a configurable option for a filter type."""
key: str = Field(description="Option key")
label: str = Field(description="Display label")
type: str = Field(description="Option type (float or int)")
default: Any = Field(description="Default value")
min_value: Any = Field(description="Minimum value")
max_value: Any = Field(description="Maximum value")
step: Any = Field(description="Step increment")
class FilterTypeResponse(BaseModel):
"""Available filter type with its options schema."""
filter_id: str = Field(description="Filter type identifier")
filter_name: str = Field(description="Display name")
options_schema: List[FilterOptionDefSchema] = Field(description="Configurable options")
class FilterTypeListResponse(BaseModel):
"""List of available filter types."""
filters: List[FilterTypeResponse] = Field(description="Available filter types")
count: int = Field(description="Number of filter types")

View File

@@ -0,0 +1,80 @@
"""Picture source schemas."""
from datetime import datetime
from typing import List, Literal, Optional
from pydantic import BaseModel, Field
class PictureSourceCreate(BaseModel):
"""Request to create a picture source."""
name: str = Field(description="Stream name", min_length=1, max_length=100)
stream_type: Literal["raw", "processed", "static_image"] = Field(description="Stream type")
display_index: Optional[int] = Field(None, description="Display index (raw streams)", ge=0)
capture_template_id: Optional[str] = Field(None, description="Capture template ID (raw streams)")
target_fps: Optional[int] = Field(None, description="Target FPS (raw streams)", ge=10, le=90)
source_stream_id: Optional[str] = Field(None, description="Source stream ID (processed streams)")
postprocessing_template_id: Optional[str] = Field(None, description="Postprocessing template ID (processed streams)")
image_source: Optional[str] = Field(None, description="Image URL or file path (static_image streams)")
description: Optional[str] = Field(None, description="Stream description", max_length=500)
class PictureSourceUpdate(BaseModel):
"""Request to update a picture source."""
name: Optional[str] = Field(None, description="Stream name", min_length=1, max_length=100)
display_index: Optional[int] = Field(None, description="Display index (raw streams)", ge=0)
capture_template_id: Optional[str] = Field(None, description="Capture template ID (raw streams)")
target_fps: Optional[int] = Field(None, description="Target FPS (raw streams)", ge=10, le=90)
source_stream_id: Optional[str] = Field(None, description="Source stream ID (processed streams)")
postprocessing_template_id: Optional[str] = Field(None, description="Postprocessing template ID (processed streams)")
image_source: Optional[str] = Field(None, description="Image URL or file path (static_image streams)")
description: Optional[str] = Field(None, description="Stream description", max_length=500)
class PictureSourceResponse(BaseModel):
"""Picture source information response."""
id: str = Field(description="Stream ID")
name: str = Field(description="Stream name")
stream_type: str = Field(description="Stream type (raw, processed, or static_image)")
display_index: Optional[int] = Field(None, description="Display index")
capture_template_id: Optional[str] = Field(None, description="Capture template ID")
target_fps: Optional[int] = Field(None, description="Target FPS")
source_stream_id: Optional[str] = Field(None, description="Source stream ID")
postprocessing_template_id: Optional[str] = Field(None, description="Postprocessing template ID")
image_source: Optional[str] = Field(None, description="Image URL or file path")
created_at: datetime = Field(description="Creation timestamp")
updated_at: datetime = Field(description="Last update timestamp")
description: Optional[str] = Field(None, description="Stream description")
class PictureSourceListResponse(BaseModel):
"""List of picture sources response."""
streams: List[PictureSourceResponse] = Field(description="List of picture sources")
count: int = Field(description="Number of streams")
class PictureSourceTestRequest(BaseModel):
"""Request to test a picture source."""
capture_duration: float = Field(default=5.0, ge=1.0, le=30.0, description="Duration to capture in seconds")
border_width: int = Field(default=10, ge=1, le=100, description="Border width in pixels for preview")
class ImageValidateRequest(BaseModel):
"""Request to validate an image source (URL or file path)."""
image_source: str = Field(description="Image URL or local file path")
class ImageValidateResponse(BaseModel):
"""Response from image validation."""
valid: bool = Field(description="Whether the image source is accessible and valid")
width: Optional[int] = Field(None, description="Image width in pixels")
height: Optional[int] = Field(None, description="Image height in pixels")
preview: Optional[str] = Field(None, description="Base64-encoded JPEG thumbnail")
error: Optional[str] = Field(None, description="Error message if invalid")

View File

@@ -0,0 +1,49 @@
"""Postprocessing template schemas."""
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field
from .filters import FilterInstanceSchema
class PostprocessingTemplateCreate(BaseModel):
"""Request to create a postprocessing template."""
name: str = Field(description="Template name", min_length=1, max_length=100)
filters: List[FilterInstanceSchema] = Field(default_factory=list, description="Ordered list of filter instances")
description: Optional[str] = Field(None, description="Template description", max_length=500)
class PostprocessingTemplateUpdate(BaseModel):
"""Request to update a postprocessing template."""
name: Optional[str] = Field(None, description="Template name", min_length=1, max_length=100)
filters: Optional[List[FilterInstanceSchema]] = Field(None, description="Ordered list of filter instances")
description: Optional[str] = Field(None, description="Template description", max_length=500)
class PostprocessingTemplateResponse(BaseModel):
"""Postprocessing template information response."""
id: str = Field(description="Template ID")
name: str = Field(description="Template name")
filters: List[FilterInstanceSchema] = Field(description="Ordered list of filter instances")
created_at: datetime = Field(description="Creation timestamp")
updated_at: datetime = Field(description="Last update timestamp")
description: Optional[str] = Field(None, description="Template description")
class PostprocessingTemplateListResponse(BaseModel):
"""List of postprocessing templates response."""
templates: List[PostprocessingTemplateResponse] = Field(description="List of postprocessing templates")
count: int = Field(description="Number of templates")
class PPTemplateTestRequest(BaseModel):
"""Request to test a postprocessing template against a source stream."""
source_stream_id: str = Field(description="ID of the source picture source to capture from")
capture_duration: float = Field(default=5.0, ge=1.0, le=30.0, description="Duration to capture in seconds")

View File

@@ -0,0 +1,42 @@
"""System-related schemas (health, version, displays)."""
from datetime import datetime
from typing import List, Literal
from pydantic import BaseModel, Field
class HealthResponse(BaseModel):
"""Health check response."""
status: Literal["healthy", "unhealthy"] = Field(description="Service health status")
timestamp: datetime = Field(description="Current server time")
version: str = Field(description="Application version")
class VersionResponse(BaseModel):
"""Version information response."""
version: str = Field(description="Application version")
python_version: str = Field(description="Python version")
api_version: str = Field(description="API version")
class DisplayInfo(BaseModel):
"""Display/monitor information."""
index: int = Field(description="Display index")
name: str = Field(description="Display name")
width: int = Field(description="Display width in pixels")
height: int = Field(description="Display height in pixels")
x: int = Field(description="Display X position")
y: int = Field(description="Display Y position")
is_primary: bool = Field(default=False, description="Whether this is the primary display")
refresh_rate: int = Field(description="Display refresh rate in Hz")
class DisplayListResponse(BaseModel):
"""List of available displays."""
displays: List[DisplayInfo] = Field(description="Available displays")
count: int = Field(description="Number of displays")

View File

@@ -0,0 +1,75 @@
"""Capture template and engine schemas."""
from datetime import datetime
from typing import Dict, List, Optional
from pydantic import BaseModel, Field
class TemplateCreate(BaseModel):
"""Request to create a capture template."""
name: str = Field(description="Template name", min_length=1, max_length=100)
engine_type: str = Field(description="Engine type (e.g., 'mss', 'dxcam', 'wgc')", min_length=1)
engine_config: Dict = Field(default_factory=dict, description="Engine-specific configuration")
description: Optional[str] = Field(None, description="Template description", max_length=500)
class TemplateUpdate(BaseModel):
"""Request to update a template."""
name: Optional[str] = Field(None, description="Template name", min_length=1, max_length=100)
engine_type: Optional[str] = Field(None, description="Capture engine type (mss, dxcam, wgc)")
engine_config: Optional[Dict] = Field(None, description="Engine-specific configuration")
description: Optional[str] = Field(None, description="Template description", max_length=500)
class TemplateResponse(BaseModel):
"""Template information response."""
id: str = Field(description="Template ID")
name: str = Field(description="Template name")
engine_type: str = Field(description="Engine type identifier")
engine_config: Dict = Field(description="Engine-specific configuration")
created_at: datetime = Field(description="Creation timestamp")
updated_at: datetime = Field(description="Last update timestamp")
description: Optional[str] = Field(None, description="Template description")
class TemplateListResponse(BaseModel):
"""List of templates response."""
templates: List[TemplateResponse] = Field(description="List of templates")
count: int = Field(description="Number of templates")
class EngineInfo(BaseModel):
"""Capture engine information."""
type: str = Field(description="Engine type identifier (e.g., 'mss', 'dxcam')")
name: str = Field(description="Human-readable engine name")
default_config: Dict = Field(description="Default configuration for this engine")
available: bool = Field(description="Whether engine is available on this system")
class EngineListResponse(BaseModel):
"""List of available engines response."""
engines: List[EngineInfo] = Field(description="Available capture engines")
count: int = Field(description="Number of engines")
class TemplateAssignment(BaseModel):
"""Request to assign template to device."""
template_id: str = Field(description="Template ID to assign")
class TemplateTestRequest(BaseModel):
"""Request to test a capture template."""
engine_type: str = Field(description="Capture engine type to test")
engine_config: Dict = Field(default={}, description="Engine configuration")
display_index: int = Field(description="Display index to capture")
border_width: int = Field(default=10, ge=1, le=100, description="Border width in pixels")
capture_duration: float = Field(default=5.0, ge=1.0, le=30.0, description="Duration to capture in seconds")