diff --git a/apps/api/config.py b/apps/api/config.py new file mode 100644 index 0000000..a32f9d6 --- /dev/null +++ b/apps/api/config.py @@ -0,0 +1,146 @@ +"""Configuration loading and caching for API application. + +This module provides centralized configuration management for devices and layout, +with startup validation and in-memory caching for performance. +""" + +import logging +from pathlib import Path +from typing import Any + +import yaml + +from packages.home_capabilities.layout import UiLayout + +logger = logging.getLogger(__name__) + +# Global caches (loaded once at startup) +devices_cache: list[dict[str, Any]] = [] +layout_cache: UiLayout | None = None + + +def load_devices_from_file() -> list[dict[str, Any]]: + """Load devices from configuration file and validate. + + Returns: + list: List of device configurations + + Raises: + FileNotFoundError: If devices.yaml doesn't exist + KeyError: If any device is missing required homekit_aid field + ValueError: If devices.yaml is invalid or contains duplicate homekit_aid values + """ + config_path = Path(__file__).parent.parent.parent / "config" / "devices.yaml" + + if not config_path.exists(): + raise FileNotFoundError(f"devices.yaml not found at {config_path}") + + with open(config_path, "r") as f: + config = yaml.safe_load(f) + + if not config or "devices" not in config: + raise ValueError("devices.yaml must contain 'devices' key") + + # Normalize device entries: accept both 'id' and 'device_id', use 'device_id' internally + devices = config.get("devices", []) + for device in devices: + device["device_id"] = device.pop("device_id", device.pop("id", None)) + + # Validate required homekit_aid field + if "homekit_aid" not in device: + raise KeyError(f"Device {device.get('device_id', 'unknown')} is missing required 'homekit_aid' field") + + # Validate unique homekit_aid values + aids = [d["homekit_aid"] for d in devices] + if len(aids) != len(set(aids)): + duplicates = [aid for aid in aids if aids.count(aid) > 1] + raise ValueError(f"Duplicate homekit_aid values found: {set(duplicates)}") + + logger.info(f"Loaded {len(devices)} devices with unique homekit_aid values (range: {min(aids)}-{max(aids)})") + + return devices + + +def load_layout_from_file() -> UiLayout: + """Load UI layout from configuration file and validate. + + Returns: + UiLayout: Parsed and validated layout configuration + + Raises: + FileNotFoundError: If layout.yaml doesn't exist + ValueError: If layout validation fails + yaml.YAMLError: If YAML parsing fails + """ + config_path = Path(__file__).parent.parent.parent / "config" / "layout.yaml" + + if not config_path.exists(): + raise FileNotFoundError( + f"Layout configuration not found: {config_path}. " + f"Please create a layout.yaml file with room and device definitions." + ) + + try: + with open(config_path, "r", encoding="utf-8") as f: + data = yaml.safe_load(f) + except yaml.YAMLError as e: + raise yaml.YAMLError(f"Failed to parse YAML in {config_path}: {e}") + + if data is None: + raise ValueError(f"Layout file is empty: {config_path}") + + try: + layout = UiLayout(**data) + except Exception as e: + raise ValueError(f"Invalid layout configuration in {config_path}: {e}") + + total_devices = layout.total_devices() + room_names = [room.name for room in layout.rooms] + logger.info( + f"Loaded layout: {len(layout.rooms)} rooms, " + f"{total_devices} total devices (Rooms: {', '.join(room_names)})" + ) + + return layout + + +def load_devices() -> list[dict[str, Any]]: + """Get devices from in-memory cache. + + Returns: + list: List of device configurations (loaded at startup) + """ + return devices_cache + + +def load_layout() -> UiLayout: + """Get layout from in-memory cache. + + Returns: + UiLayout: Layout configuration (loaded at startup) + + Raises: + RuntimeError: If layout cache is not initialized + """ + if layout_cache is None: + raise RuntimeError("Layout cache not initialized. Application startup may have failed.") + return layout_cache + + +def initialize_config() -> None: + """Initialize configuration by loading devices and layout. + + This function should be called once during application startup. + + Raises: + Exception: If configuration loading or validation fails + """ + global devices_cache, layout_cache + + # Load devices with validation + devices_cache = load_devices_from_file() + + # Load layout with validation + layout_cache = load_layout_from_file() + + logger.info("Configuration initialization complete")