config file loading

apps/api/main.py

@@ -24,9 +24,11 @@ from packages.home_capabilities import (
     ContactState,
     TempHumidityState,
     RelayState,
-    load_layout,
 )
 
+# Import configuration management
+from apps.api.config import initialize_config, load_devices, load_layout
+
 # Import resolvers (must be before router imports to avoid circular dependency)
 from apps.api.resolvers import (
     DeviceDTO,
@@ -51,9 +53,6 @@ logger = logging.getLogger(__name__)
 # Will be populated from Redis pub/sub messages
 device_states: dict[str, dict[str, Any]] = {}
 
-# Devices configuration cache (loaded once at startup)
-devices_cache: list[dict[str, Any]] = []
-
 # Background task reference
 background_task: asyncio.Task | None = None
 
@@ -102,24 +101,6 @@ async def get_device_state(device_id: str):
     except KeyError:
         raise HTTPException(status_code=404, detail="Device state not found")
 
-# --- Minimal-invasive: Einzelgerät-Layout-Endpunkt ---
-@app.get("/devices/{device_id}/layout")
-async def get_device_layout(device_id: str):
-    """Gibt die layout-spezifischen Informationen für ein einzelnes Gerät zurück."""
-    layout = load_layout()
-    for room in layout.get("rooms", []):
-        for device in room.get("devices", []):
-            if device.get("device_id") == device_id:
-                # Rückgabe: Layout-Infos + Raumname
-                return {
-                    "device_id": device_id,
-                    "room": room.get("name"),
-                    "title": device.get("title"),
-                    "icon": device.get("icon"),
-                    "rank": device.get("rank"),
-                }
-    raise HTTPException(status_code=404, detail="Device layout not found")
-
 
 @app.get("/health")
 async def health() -> dict[str, str]:
@@ -184,7 +165,7 @@ async def redis_state_listener():
 @app.on_event("startup")
 async def startup_event():
     """Start background tasks on application startup."""
-    global background_task, devices_cache
+    global background_task
     
     # Include routers
     from apps.api.routes.groups_scenes import router as groups_scenes_router
@@ -193,11 +174,11 @@ async def startup_event():
     app.include_router(groups_scenes_router, prefix="")
     app.include_router(rooms_router, prefix="")
     
-    # Load and validate devices configuration
+    # Load and validate configuration (devices + layout)
     try:
-        devices_cache = load_devices_from_file()
+        initialize_config()
     except Exception as e:
-        logger.error(f"Failed to load devices configuration: {e}")
+        logger.error(f"Failed to initialize configuration: {e}")
         raise  # Fatal error - application will not start
     
     background_task = asyncio.create_task(redis_state_listener())
@@ -252,57 +233,6 @@ class DeviceInfo(BaseModel):
 
 
 # Configuration helpers
-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_devices() -> list[dict[str, Any]]:
-    """Get devices from in-memory cache.
-    
-    Returns:
-        list: List of device configurations (loaded at startup)
-    """
-    return devices_cache
-
-
 def get_mqtt_settings() -> tuple[str, int]:
     """Get MQTT broker settings from environment.
     

apps/api/resolvers.py

@@ -4,12 +4,12 @@ import logging
 from pathlib import Path
 from typing import Any, TypedDict
 
+from apps.api.config import load_layout
 from packages.home_capabilities import (
     GroupConfig,
     GroupsConfigRoot,
     SceneStep,
     get_group_by_id,
-    load_layout,
 )
 
 logger = logging.getLogger(__name__)

apps/api/routes/rooms.py

@@ -12,7 +12,7 @@ from typing import Any
 from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel
 
-from packages.home_capabilities import load_layout
+from apps.api.config import load_layout
 
 logger = logging.getLogger(__name__)