230 lines
8.4 KiB
Python
230 lines
8.4 KiB
Python
"""
|
|
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}")
|