groups and scenes initial
This commit is contained in:
@@ -10,7 +10,25 @@ from packages.home_capabilities.temp_humidity_sensor import CAP_VERSION as TEMP_
|
||||
from packages.home_capabilities.temp_humidity_sensor import TempHumidityState
|
||||
from packages.home_capabilities.relay import CAP_VERSION as RELAY_VERSION
|
||||
from packages.home_capabilities.relay import RelayState
|
||||
from packages.home_capabilities.layout import DeviceTile, Room, UiLayout, load_layout
|
||||
from packages.home_capabilities.layout import (
|
||||
DeviceTile,
|
||||
Room,
|
||||
UiLayout,
|
||||
load_layout,
|
||||
)
|
||||
from packages.home_capabilities.groups_scenes import (
|
||||
GroupConfig,
|
||||
GroupsConfigRoot,
|
||||
GroupSelector,
|
||||
SceneConfig,
|
||||
ScenesConfigRoot,
|
||||
SceneSelector,
|
||||
SceneStep,
|
||||
get_group_by_id,
|
||||
get_scene_by_id,
|
||||
load_groups,
|
||||
load_scenes,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"LightState",
|
||||
@@ -27,4 +45,15 @@ __all__ = [
|
||||
"Room",
|
||||
"UiLayout",
|
||||
"load_layout",
|
||||
"GroupConfig",
|
||||
"GroupsConfigRoot",
|
||||
"GroupSelector",
|
||||
"SceneConfig",
|
||||
"ScenesConfigRoot",
|
||||
"SceneSelector",
|
||||
"SceneStep",
|
||||
"get_group_by_id",
|
||||
"get_scene_by_id",
|
||||
"load_groups",
|
||||
"load_scenes",
|
||||
]
|
||||
|
||||
229
packages/home_capabilities/groups_scenes.py
Normal file
229
packages/home_capabilities/groups_scenes.py
Normal file
@@ -0,0 +1,229 @@
|
||||
"""
|
||||
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}")
|
||||
Reference in New Issue
Block a user