"""FastAPI application entry point.""" import sys from contextlib import asynccontextmanager from pathlib import Path from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from starlette.requests import Request from wled_controller import __version__ from wled_controller.api import router from wled_controller.api.dependencies import init_dependencies from wled_controller.config import get_config from wled_controller.core.processing.processor_manager import ProcessorManager from wled_controller.storage import DeviceStore from wled_controller.storage.template_store import TemplateStore from wled_controller.storage.postprocessing_template_store import PostprocessingTemplateStore from wled_controller.storage.pattern_template_store import PatternTemplateStore from wled_controller.storage.picture_source_store import PictureSourceStore from wled_controller.storage.picture_target_store import PictureTargetStore from wled_controller.storage.color_strip_store import ColorStripStore from wled_controller.storage.audio_source_store import AudioSourceStore from wled_controller.storage.audio_template_store import AudioTemplateStore import wled_controller.core.audio # noqa: F401 — trigger engine auto-registration from wled_controller.storage.value_source_store import ValueSourceStore from wled_controller.storage.profile_store import ProfileStore from wled_controller.core.profiles.profile_engine import ProfileEngine from wled_controller.core.backup.auto_backup import AutoBackupEngine from wled_controller.api.routes.system import STORE_MAP from wled_controller.utils import setup_logging, get_logger # Initialize logging setup_logging() logger = get_logger(__name__) # Get configuration config = get_config() # Initialize storage and processing device_store = DeviceStore(config.storage.devices_file) template_store = TemplateStore(config.storage.templates_file) pp_template_store = PostprocessingTemplateStore(config.storage.postprocessing_templates_file) picture_source_store = PictureSourceStore(config.storage.picture_sources_file) picture_target_store = PictureTargetStore(config.storage.picture_targets_file) pattern_template_store = PatternTemplateStore(config.storage.pattern_templates_file) color_strip_store = ColorStripStore(config.storage.color_strip_sources_file) audio_source_store = AudioSourceStore(config.storage.audio_sources_file) audio_template_store = AudioTemplateStore(config.storage.audio_templates_file) value_source_store = ValueSourceStore(config.storage.value_sources_file) profile_store = ProfileStore(config.storage.profiles_file) # Migrate embedded audio config from CSS entities to audio sources audio_source_store.migrate_from_css(color_strip_store) # Assign default audio template to multichannel sources that have none audio_source_store.migrate_add_default_template(audio_template_store) processor_manager = ProcessorManager( picture_source_store=picture_source_store, capture_template_store=template_store, pp_template_store=pp_template_store, pattern_template_store=pattern_template_store, device_store=device_store, color_strip_store=color_strip_store, audio_source_store=audio_source_store, value_source_store=value_source_store, audio_template_store=audio_template_store, ) @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan manager. Handles startup and shutdown events. """ # Startup logger.info(f"Starting LED Grab v{__version__}") logger.info(f"Python version: {sys.version}") logger.info(f"Server listening on {config.server.host}:{config.server.port}") # Validate authentication configuration if not config.auth.api_keys: logger.error("=" * 70) logger.error("CRITICAL: No API keys configured!") logger.error("Authentication is REQUIRED for all API requests.") logger.error("Please add API keys to your configuration:") logger.error(" 1. Generate keys: openssl rand -hex 32") logger.error(" 2. Add to config/default_config.yaml under auth.api_keys") logger.error(" 3. Format: label: \"your-generated-key\"") logger.error("=" * 70) raise RuntimeError("No API keys configured - server cannot start without authentication") # Log authentication status logger.info(f"API Authentication: ENFORCED ({len(config.auth.api_keys)} clients configured)") client_labels = ", ".join(config.auth.api_keys.keys()) logger.info(f"Authorized clients: {client_labels}") logger.info("All API requests require valid Bearer token authentication") # Create profile engine (needs processor_manager) profile_engine = ProfileEngine(profile_store, processor_manager) # Create auto-backup engine auto_backup_engine = AutoBackupEngine( settings_path=Path("data/auto_backup_settings.json"), backup_dir=Path("data/backups"), store_map=STORE_MAP, storage_config=config.storage, ) # Initialize API dependencies init_dependencies( device_store, template_store, processor_manager, pp_template_store=pp_template_store, pattern_template_store=pattern_template_store, picture_source_store=picture_source_store, picture_target_store=picture_target_store, color_strip_store=color_strip_store, audio_source_store=audio_source_store, audio_template_store=audio_template_store, value_source_store=value_source_store, profile_store=profile_store, profile_engine=profile_engine, auto_backup_engine=auto_backup_engine, ) # Register devices in processor manager for health monitoring devices = device_store.get_all_devices() for device in devices: try: processor_manager.add_device( device_id=device.id, device_url=device.url, led_count=device.led_count, device_type=device.device_type, baud_rate=device.baud_rate, software_brightness=device.software_brightness, auto_shutdown=device.auto_shutdown, ) logger.info(f"Registered device: {device.name} ({device.id})") except Exception as e: logger.error(f"Failed to register device {device.id}: {e}") logger.info(f"Registered {len(devices)} devices for health monitoring") # Register picture targets in processor manager targets = picture_target_store.get_all_targets() registered_targets = 0 for target in targets: try: target.register_with_manager(processor_manager) registered_targets += 1 logger.info(f"Registered target: {target.name} ({target.id})") except Exception as e: logger.error(f"Failed to register target {target.id}: {e}") logger.info(f"Registered {registered_targets} picture target(s)") # Start background health monitoring for all devices await processor_manager.start_health_monitoring() # Start profile engine (evaluates conditions and auto-starts/stops targets) await profile_engine.start() # Start auto-backup engine (periodic configuration backups) await auto_backup_engine.start() # Auto-start targets with auto_start=True auto_started = 0 for target in targets: if getattr(target, "auto_start", False): try: await processor_manager.start_processing(target.id) auto_started += 1 logger.info(f"Auto-started target: {target.name} ({target.id})") except Exception as e: logger.warning(f"Failed to auto-start target {target.id}: {e}") if auto_started: logger.info(f"Auto-started {auto_started} target(s)") yield # Shutdown logger.info("Shutting down LED Grab") # Stop auto-backup engine try: await auto_backup_engine.stop() except Exception as e: logger.error(f"Error stopping auto-backup engine: {e}") # Stop profile engine first (deactivates profile-managed targets) try: await profile_engine.stop() logger.info("Stopped profile engine") except Exception as e: logger.error(f"Error stopping profile engine: {e}") # Stop all processing try: await processor_manager.stop_all() logger.info("Stopped all processors") except Exception as e: logger.error(f"Error stopping processors: {e}") # Create FastAPI application app = FastAPI( title="LED Grab", description="Control WLED devices based on screen content for ambient lighting", version=__version__, lifespan=lifespan, docs_url="/docs", redoc_url="/redoc", openapi_url="/openapi.json", ) # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=config.server.cors_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Include API routes app.include_router(router) # Mount static files static_path = Path(__file__).parent / "static" if static_path.exists(): app.mount("/static", StaticFiles(directory=str(static_path)), name="static") logger.info(f"Mounted static files from {static_path}") else: logger.warning(f"Static files directory not found: {static_path}") # Jinja2 templates templates_path = Path(__file__).parent / "templates" templates = Jinja2Templates(directory=str(templates_path)) @app.exception_handler(Exception) async def global_exception_handler(request, exc): """Global exception handler for unhandled errors.""" logger.error(f"Unhandled exception: {exc}", exc_info=True) return JSONResponse( status_code=500, content={ "error": "InternalServerError", "message": "An unexpected error occurred", "detail": str(exc) if config.server.log_level == "DEBUG" else None, }, ) @app.get("/") async def root(request: Request): """Serve the web UI dashboard.""" return templates.TemplateResponse(request, "index.html") if __name__ == "__main__": import uvicorn uvicorn.run( "wled_controller.main:app", host=config.server.host, port=config.server.port, log_level=config.server.log_level.lower(), reload=False, # Disabled due to watchfiles infinite reload loop )