""" Configuration models and loaders for groups and scenes. This module provides Pydantic models for validating groups.yaml and scenes.yaml, along with loader functions that parse YAML files into typed configuration objects. """ from pathlib import Path from typing import Any import yaml from pydantic import BaseModel, Field, field_validator, model_validator # ============================================================================ # GROUP MODELS # ============================================================================ class GroupSelector(BaseModel): """Selector for automatically adding devices to a group.""" type: str = Field(..., description="Device type (e.g., 'light', 'thermostat')") room: str | None = Field(None, description="Filter by room name") tags: list[str] | None = Field(None, description="Filter by device tags") class GroupConfig(BaseModel): """Configuration for a device group.""" id: str = Field(..., description="Unique group identifier") name: str = Field(..., description="Human-readable group name") selector: GroupSelector | None = Field(None, description="Auto-select devices by criteria") device_ids: list[str] = Field(default_factory=list, description="Explicit device IDs") capabilities: dict[str, bool] = Field( default_factory=dict, description="Supported capabilities (e.g., {'brightness': True})" ) class GroupsConfigRoot(BaseModel): """Root configuration for groups.yaml.""" version: int = Field(..., description="Configuration schema version") groups: list[GroupConfig] = Field(default_factory=list, description="List of groups") @field_validator('groups') @classmethod def validate_unique_ids(cls, groups: list[GroupConfig]) -> list[GroupConfig]: """Ensure all group IDs are unique.""" ids = [g.id for g in groups] duplicates = [id for id in ids if ids.count(id) > 1] if duplicates: raise ValueError(f"Duplicate group IDs found: {set(duplicates)}") return groups # ============================================================================ # SCENE MODELS # ============================================================================ class SceneSelector(BaseModel): """Selector for targeting devices in a scene step.""" type: str | None = Field(None, description="Device type (e.g., 'light', 'outlet')") room: str | None = Field(None, description="Filter by room name") tags: list[str] | None = Field(None, description="Filter by device tags") class SceneStep(BaseModel): """A single step in a scene execution.""" selector: SceneSelector | None = Field(None, description="Select devices by criteria") group_id: str | None = Field(None, description="Target a specific group") action: dict[str, Any] = Field(..., description="Action to execute (type + payload)") delay_ms: int | None = Field(None, description="Delay before next step (milliseconds)") @model_validator(mode='after') def validate_selector_or_group(self) -> 'SceneStep': """Ensure either selector OR group_id is specified, but not both.""" has_selector = self.selector is not None has_group = self.group_id is not None if not has_selector and not has_group: raise ValueError("SceneStep must have either 'selector' or 'group_id'") if has_selector and has_group: raise ValueError("SceneStep cannot have both 'selector' and 'group_id'") return self class SceneConfig(BaseModel): """Configuration for a scene.""" id: str = Field(..., description="Unique scene identifier") name: str = Field(..., description="Human-readable scene name") steps: list[SceneStep] = Field(..., description="Ordered list of actions") class ScenesConfigRoot(BaseModel): """Root configuration for scenes.yaml.""" version: int = Field(..., description="Configuration schema version") scenes: list[SceneConfig] = Field(default_factory=list, description="List of scenes") @field_validator('scenes') @classmethod def validate_unique_ids(cls, scenes: list[SceneConfig]) -> list[SceneConfig]: """Ensure all scene IDs are unique.""" ids = [s.id for s in scenes] duplicates = [id for id in ids if ids.count(id) > 1] if duplicates: raise ValueError(f"Duplicate scene IDs found: {set(duplicates)}") return scenes # ============================================================================ # LOADER FUNCTIONS # ============================================================================ def load_groups(path: Path | str) -> GroupsConfigRoot: """ Load and validate groups configuration from YAML file. Args: path: Path to groups.yaml file Returns: Validated GroupsConfigRoot object Raises: FileNotFoundError: If config file doesn't exist ValidationError: If configuration is invalid ValueError: If duplicate group IDs are found or YAML is empty """ path = Path(path) if not path.exists(): raise FileNotFoundError(f"Groups config file not found: {path}") with open(path, 'r', encoding='utf-8') as f: data = yaml.safe_load(f) if data is None: raise ValueError(f"Groups config file is empty: {path}") return GroupsConfigRoot.model_validate(data) def load_scenes(path: Path | str) -> ScenesConfigRoot: """ Load and validate scenes configuration from YAML file. Args: path: Path to scenes.yaml file Returns: Validated ScenesConfigRoot object Raises: FileNotFoundError: If config file doesn't exist ValidationError: If configuration is invalid ValueError: If duplicate scene IDs, invalid steps, or empty YAML are found """ path = Path(path) if not path.exists(): raise FileNotFoundError(f"Scenes config file not found: {path}") with open(path, 'r', encoding='utf-8') as f: data = yaml.safe_load(f) if data is None: raise ValueError(f"Scenes config file is empty: {path}") return ScenesConfigRoot.model_validate(data) # ============================================================================ # CONVENIENCE FUNCTIONS # ============================================================================ def get_group_by_id(config: GroupsConfigRoot, group_id: str) -> GroupConfig | None: """Find a group by its ID.""" for group in config.groups: if group.id == group_id: return group return None def get_scene_by_id(config: ScenesConfigRoot, scene_id: str) -> SceneConfig | None: """Find a scene by its ID.""" for scene in config.scenes: if scene.id == scene_id: return scene return None # ============================================================================ # EXAMPLE USAGE # ============================================================================ if __name__ == "__main__": from pathlib import Path # Example: Load groups configuration try: groups_path = Path(__file__).parent.parent / "config" / "groups.yaml" groups = load_groups(groups_path) print(f"✓ Loaded {len(groups.groups)} groups (version {groups.version})") for group in groups.groups: print(f" - {group.id}: {group.name}") if group.selector: print(f" Selector: type={group.selector.type}, room={group.selector.room}") if group.device_ids: print(f" Devices: {', '.join(group.device_ids)}") except Exception as e: print(f"✗ Error loading groups: {e}") print() # Example: Load scenes configuration try: scenes_path = Path(__file__).parent.parent / "config" / "scenes.yaml" scenes = load_scenes(scenes_path) print(f"✓ Loaded {len(scenes.scenes)} scenes (version {scenes.version})") for scene in scenes.scenes: print(f" - {scene.id}: {scene.name} ({len(scene.steps)} steps)") for i, step in enumerate(scene.steps, 1): if step.selector: print(f" Step {i}: selector type={step.selector.type}") elif step.group_id: print(f" Step {i}: group_id={step.group_id}") print(f" Action: {step.action}") except Exception as e: print(f"✗ Error loading scenes: {e}")