"""Dependency injection for API routes. Uses a registry dict instead of individual module-level globals. All getter function signatures remain unchanged for FastAPI Depends() compatibility. """ from typing import Any, Dict, Type, TypeVar 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.output_target_store import OutputTargetStore 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 from wled_controller.storage.value_source_store import ValueSourceStore from wled_controller.storage.automation_store import AutomationStore from wled_controller.storage.scene_preset_store import ScenePresetStore from wled_controller.storage.sync_clock_store import SyncClockStore from wled_controller.storage.color_strip_processing_template_store import ColorStripProcessingTemplateStore from wled_controller.core.automations.automation_engine import AutomationEngine from wled_controller.core.backup.auto_backup import AutoBackupEngine from wled_controller.core.processing.sync_clock_manager import SyncClockManager T = TypeVar("T") # Central dependency registry — keyed by type or string label _deps: Dict[str, Any] = {} def _get(key: str, label: str) -> Any: """Get a dependency by key, raising RuntimeError if not initialized.""" dep = _deps.get(key) if dep is None: raise RuntimeError(f"{label} not initialized") return dep # ── Typed getters (unchanged signatures for FastAPI Depends()) ────────── def get_device_store() -> DeviceStore: return _get("device_store", "Device store") def get_template_store() -> TemplateStore: return _get("template_store", "Template store") def get_pp_template_store() -> PostprocessingTemplateStore: return _get("pp_template_store", "Postprocessing template store") def get_pattern_template_store() -> PatternTemplateStore: return _get("pattern_template_store", "Pattern template store") def get_picture_source_store() -> PictureSourceStore: return _get("picture_source_store", "Picture source store") def get_output_target_store() -> OutputTargetStore: return _get("output_target_store", "Output target store") def get_color_strip_store() -> ColorStripStore: return _get("color_strip_store", "Color strip store") def get_audio_source_store() -> AudioSourceStore: return _get("audio_source_store", "Audio source store") def get_audio_template_store() -> AudioTemplateStore: return _get("audio_template_store", "Audio template store") def get_value_source_store() -> ValueSourceStore: return _get("value_source_store", "Value source store") def get_processor_manager() -> ProcessorManager: return _get("processor_manager", "Processor manager") def get_automation_store() -> AutomationStore: return _get("automation_store", "Automation store") def get_scene_preset_store() -> ScenePresetStore: return _get("scene_preset_store", "Scene preset store") def get_automation_engine() -> AutomationEngine: return _get("automation_engine", "Automation engine") def get_auto_backup_engine() -> AutoBackupEngine: return _get("auto_backup_engine", "Auto-backup engine") def get_sync_clock_store() -> SyncClockStore: return _get("sync_clock_store", "Sync clock store") def get_sync_clock_manager() -> SyncClockManager: return _get("sync_clock_manager", "Sync clock manager") def get_cspt_store() -> ColorStripProcessingTemplateStore: return _get("cspt_store", "Color strip processing template store") # ── Event helper ──────────────────────────────────────────────────────── def fire_entity_event(entity_type: str, action: str, entity_id: str) -> None: """Fire an entity_changed event via the ProcessorManager event bus. Args: entity_type: e.g. "device", "output_target", "color_strip_source" action: "created", "updated", or "deleted" entity_id: The entity's unique ID """ pm = _deps.get("processor_manager") if pm is not None: pm.fire_event({ "type": "entity_changed", "entity_type": entity_type, "action": action, "id": entity_id, }) # ── Initialization ────────────────────────────────────────────────────── def init_dependencies( device_store: DeviceStore, template_store: TemplateStore, processor_manager: ProcessorManager, pp_template_store: PostprocessingTemplateStore | None = None, pattern_template_store: PatternTemplateStore | None = None, picture_source_store: PictureSourceStore | None = None, output_target_store: OutputTargetStore | None = None, color_strip_store: ColorStripStore | None = None, audio_source_store: AudioSourceStore | None = None, audio_template_store: AudioTemplateStore | None = None, value_source_store: ValueSourceStore | None = None, automation_store: AutomationStore | None = None, scene_preset_store: ScenePresetStore | None = None, automation_engine: AutomationEngine | None = None, auto_backup_engine: AutoBackupEngine | None = None, sync_clock_store: SyncClockStore | None = None, sync_clock_manager: SyncClockManager | None = None, cspt_store: ColorStripProcessingTemplateStore | None = None, ): """Initialize global dependencies.""" _deps.update({ "device_store": device_store, "template_store": template_store, "processor_manager": processor_manager, "pp_template_store": pp_template_store, "pattern_template_store": pattern_template_store, "picture_source_store": picture_source_store, "output_target_store": output_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, "automation_store": automation_store, "scene_preset_store": scene_preset_store, "automation_engine": automation_engine, "auto_backup_engine": auto_backup_engine, "sync_clock_store": sync_clock_store, "sync_clock_manager": sync_clock_manager, "cspt_store": cspt_store, })