"""Value source routes: CRUD for value sources.""" import asyncio import secrets from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, WebSocketDisconnect from wled_controller.api.auth import AuthRequired from wled_controller.api.dependencies import ( get_picture_target_store, get_processor_manager, get_value_source_store, ) from wled_controller.config import get_config from wled_controller.api.schemas.value_sources import ( ValueSourceCreate, ValueSourceListResponse, ValueSourceResponse, ValueSourceUpdate, ) from wled_controller.storage.value_source import ValueSource from wled_controller.storage.value_source_store import ValueSourceStore from wled_controller.storage.picture_target_store import PictureTargetStore from wled_controller.core.processing.processor_manager import ProcessorManager from wled_controller.utils import get_logger logger = get_logger(__name__) router = APIRouter() def _to_response(source: ValueSource) -> ValueSourceResponse: """Convert a ValueSource to a ValueSourceResponse.""" d = source.to_dict() return ValueSourceResponse( id=d["id"], name=d["name"], source_type=d["source_type"], value=d.get("value"), waveform=d.get("waveform"), speed=d.get("speed"), min_value=d.get("min_value"), max_value=d.get("max_value"), audio_source_id=d.get("audio_source_id"), mode=d.get("mode"), sensitivity=d.get("sensitivity"), smoothing=d.get("smoothing"), auto_gain=d.get("auto_gain"), schedule=d.get("schedule"), picture_source_id=d.get("picture_source_id"), scene_behavior=d.get("scene_behavior"), description=d.get("description"), created_at=source.created_at, updated_at=source.updated_at, ) @router.get("/api/v1/value-sources", response_model=ValueSourceListResponse, tags=["Value Sources"]) async def list_value_sources( _auth: AuthRequired, source_type: Optional[str] = Query(None, description="Filter by source_type: static, animated, audio, adaptive_time, or adaptive_scene"), store: ValueSourceStore = Depends(get_value_source_store), ): """List all value sources, optionally filtered by type.""" sources = store.get_all_sources() if source_type: sources = [s for s in sources if s.source_type == source_type] return ValueSourceListResponse( sources=[_to_response(s) for s in sources], count=len(sources), ) @router.post("/api/v1/value-sources", response_model=ValueSourceResponse, status_code=201, tags=["Value Sources"]) async def create_value_source( data: ValueSourceCreate, _auth: AuthRequired, store: ValueSourceStore = Depends(get_value_source_store), ): """Create a new value source.""" try: source = store.create_source( name=data.name, source_type=data.source_type, value=data.value, waveform=data.waveform, speed=data.speed, min_value=data.min_value, max_value=data.max_value, audio_source_id=data.audio_source_id, mode=data.mode, sensitivity=data.sensitivity, smoothing=data.smoothing, description=data.description, schedule=data.schedule, picture_source_id=data.picture_source_id, scene_behavior=data.scene_behavior, auto_gain=data.auto_gain, ) return _to_response(source) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) @router.get("/api/v1/value-sources/{source_id}", response_model=ValueSourceResponse, tags=["Value Sources"]) async def get_value_source( source_id: str, _auth: AuthRequired, store: ValueSourceStore = Depends(get_value_source_store), ): """Get a value source by ID.""" try: source = store.get_source(source_id) return _to_response(source) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) @router.put("/api/v1/value-sources/{source_id}", response_model=ValueSourceResponse, tags=["Value Sources"]) async def update_value_source( source_id: str, data: ValueSourceUpdate, _auth: AuthRequired, store: ValueSourceStore = Depends(get_value_source_store), pm: ProcessorManager = Depends(get_processor_manager), ): """Update an existing value source.""" try: source = store.update_source( source_id=source_id, name=data.name, value=data.value, waveform=data.waveform, speed=data.speed, min_value=data.min_value, max_value=data.max_value, audio_source_id=data.audio_source_id, mode=data.mode, sensitivity=data.sensitivity, smoothing=data.smoothing, description=data.description, schedule=data.schedule, picture_source_id=data.picture_source_id, scene_behavior=data.scene_behavior, auto_gain=data.auto_gain, ) # Hot-reload running value streams pm.update_value_source(source_id) return _to_response(source) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) @router.delete("/api/v1/value-sources/{source_id}", tags=["Value Sources"]) async def delete_value_source( source_id: str, _auth: AuthRequired, store: ValueSourceStore = Depends(get_value_source_store), target_store: PictureTargetStore = Depends(get_picture_target_store), ): """Delete a value source.""" try: # Check if any targets reference this value source from wled_controller.storage.wled_picture_target import WledPictureTarget for target in target_store.get_all_targets(): if isinstance(target, WledPictureTarget): if getattr(target, "brightness_value_source_id", "") == source_id: raise ValueError( f"Cannot delete: referenced by target '{target.name}'" ) store.delete_source(source_id) return {"status": "deleted", "id": source_id} except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) # ===== REAL-TIME VALUE SOURCE TEST WEBSOCKET ===== @router.websocket("/api/v1/value-sources/{source_id}/test/ws") async def test_value_source_ws( websocket: WebSocket, source_id: str, token: str = Query(""), ): """WebSocket for real-time value source output. Auth via ?token=. Acquires a ValueStream for the given source, polls get_value() at ~20 Hz, and streams {value: float} JSON to the client. """ # Authenticate authenticated = False cfg = get_config() if token and cfg.auth.api_keys: for _label, api_key in cfg.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 # Validate source exists store = get_value_source_store() try: store.get_source(source_id) except ValueError as e: await websocket.close(code=4004, reason=str(e)) return # Acquire a value stream manager = get_processor_manager() vsm = manager.value_stream_manager if vsm is None: await websocket.close(code=4003, reason="Value stream manager not available") return try: stream = vsm.acquire(source_id) except Exception as e: await websocket.close(code=4003, reason=str(e)) return await websocket.accept() logger.info(f"Value source test WebSocket connected for {source_id}") try: while True: value = stream.get_value() await websocket.send_json({"value": round(value, 4)}) await asyncio.sleep(0.05) except WebSocketDisconnect: pass except Exception as e: logger.error(f"Value source test WebSocket error for {source_id}: {e}") finally: vsm.release(source_id) logger.info(f"Value source test WebSocket disconnected for {source_id}")