287 lines
9.1 KiB
Python
287 lines
9.1 KiB
Python
"""Group and scene resolution logic."""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any, TypedDict
|
|
|
|
from packages.home_capabilities import (
|
|
GroupConfig,
|
|
GroupsConfigRoot,
|
|
SceneStep,
|
|
get_group_by_id,
|
|
load_layout,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ============================================================================
|
|
# TYPE DEFINITIONS
|
|
# ============================================================================
|
|
|
|
class DeviceDTO(TypedDict, total=False):
|
|
"""Device Data Transfer Object.
|
|
|
|
Represents a device as returned by /devices endpoint or load_devices().
|
|
|
|
Required fields:
|
|
device_id: Unique device identifier
|
|
type: Device type (light, thermostat, relay, etc.)
|
|
|
|
Optional fields:
|
|
name: Human-readable device name
|
|
features: Device capabilities (power, brightness, etc.)
|
|
technology: MQTT, zigbee2mqtt, simulator, etc.
|
|
topics: MQTT topic configuration
|
|
metadata: Additional device information
|
|
"""
|
|
device_id: str
|
|
type: str
|
|
name: str
|
|
features: dict[str, Any]
|
|
technology: str
|
|
topics: dict[str, str]
|
|
metadata: dict[str, Any]
|
|
|
|
|
|
# ============================================================================
|
|
# DEVICE-ROOM MAPPING
|
|
# ============================================================================
|
|
|
|
# Global cache for device -> room mapping
|
|
_device_room_cache: dict[str, str] = {}
|
|
|
|
|
|
def load_device_rooms(path: str | Path | None = None) -> dict[str, str]:
|
|
"""
|
|
Load device-to-room mapping from layout configuration.
|
|
|
|
This function extracts a mapping of device_id -> room_name from the layout.yaml
|
|
file, which is useful for resolving selectors like {room: "Küche"}.
|
|
|
|
Args:
|
|
path: Optional path to layout.yaml. If None, uses default path
|
|
(config/layout.yaml relative to workspace root)
|
|
|
|
Returns:
|
|
Dictionary mapping device_id to room_name. Returns empty dict if:
|
|
- layout.yaml doesn't exist
|
|
- layout.yaml is malformed
|
|
- layout.yaml is empty
|
|
|
|
Example:
|
|
>>> mapping = load_device_rooms()
|
|
>>> mapping['kueche_lampe1']
|
|
'Küche'
|
|
"""
|
|
global _device_room_cache
|
|
|
|
try:
|
|
# Load the layout using existing function
|
|
layout = load_layout(path)
|
|
|
|
# Build device -> room mapping
|
|
device_rooms: dict[str, str] = {}
|
|
for room in layout.rooms:
|
|
for device in room.devices:
|
|
device_rooms[device.device_id] = room.name
|
|
|
|
# Update global cache
|
|
_device_room_cache = device_rooms.copy()
|
|
|
|
logger.info(f"Loaded device-room mapping: {len(device_rooms)} devices")
|
|
return device_rooms
|
|
|
|
except (FileNotFoundError, ValueError, Exception) as e:
|
|
logger.warning(f"Failed to load device-room mapping: {e}")
|
|
logger.warning("Returning empty device-room mapping")
|
|
_device_room_cache = {}
|
|
return {}
|
|
|
|
|
|
def get_room(device_id: str) -> str | None:
|
|
"""
|
|
Get the room name for a given device ID.
|
|
|
|
This function uses the cached device-room mapping loaded by load_device_rooms().
|
|
If the cache is empty, it will attempt to load it first.
|
|
|
|
Args:
|
|
device_id: The device identifier to lookup
|
|
|
|
Returns:
|
|
Room name if device is found, None otherwise
|
|
|
|
Example:
|
|
>>> get_room('kueche_lampe1')
|
|
'Küche'
|
|
>>> get_room('nonexistent_device')
|
|
None
|
|
"""
|
|
# Check if cache is populated
|
|
if not _device_room_cache:
|
|
logger.debug("Device-room cache empty, loading from layout...")
|
|
# Load mapping (this updates the global _device_room_cache)
|
|
load_device_rooms()
|
|
|
|
# Access the cache after potential reload
|
|
return _device_room_cache.get(device_id)
|
|
|
|
|
|
def clear_room_cache() -> None:
|
|
"""
|
|
Clear the cached device-room mapping.
|
|
|
|
This is useful for testing or when the layout configuration has changed
|
|
and needs to be reloaded.
|
|
"""
|
|
_device_room_cache.clear()
|
|
logger.debug("Cleared device-room cache")
|
|
|
|
|
|
# ============================================================================
|
|
# GROUP & SCENE RESOLUTION
|
|
# ============================================================================
|
|
|
|
def resolve_group_devices(
|
|
group: GroupConfig,
|
|
devices: list[DeviceDTO],
|
|
device_rooms: dict[str, str]
|
|
) -> list[DeviceDTO]:
|
|
"""
|
|
Resolve devices for a group based on device_ids or selector.
|
|
|
|
Args:
|
|
group: Group configuration with device_ids or selector
|
|
devices: List of all available devices
|
|
device_rooms: Mapping of device_id -> room_name
|
|
|
|
Returns:
|
|
List of devices matching the group criteria (no duplicates)
|
|
|
|
Example:
|
|
>>> # Group with explicit device_ids
|
|
>>> group = GroupConfig(id="test", name="Test", device_ids=["lamp1", "lamp2"])
|
|
>>> resolve_group_devices(group, all_devices, {})
|
|
[{"device_id": "lamp1", ...}, {"device_id": "lamp2", ...}]
|
|
|
|
>>> # Group with selector (all lights in kitchen)
|
|
>>> group = GroupConfig(
|
|
... id="kitchen_lights",
|
|
... name="Kitchen Lights",
|
|
... selector=GroupSelector(type="light", room="Küche")
|
|
... )
|
|
>>> resolve_group_devices(group, all_devices, device_rooms)
|
|
[{"device_id": "kueche_deckenlampe", ...}, ...]
|
|
"""
|
|
# Case 1: Explicit device_ids
|
|
if group.device_ids:
|
|
device_id_set = set(group.device_ids)
|
|
return [d for d in devices if d["device_id"] in device_id_set]
|
|
|
|
# Case 2: Selector-based filtering
|
|
if group.selector:
|
|
filtered = []
|
|
|
|
for device in devices:
|
|
# Filter by type (required in selector)
|
|
if device["type"] != group.selector.type:
|
|
continue
|
|
|
|
# Filter by room (optional)
|
|
if group.selector.room:
|
|
device_room = device_rooms.get(device["device_id"])
|
|
if device_room != group.selector.room:
|
|
continue
|
|
|
|
# Filter by tags (optional, future feature)
|
|
# if group.selector.tags:
|
|
# device_tags = device.get("metadata", {}).get("tags", [])
|
|
# if not any(tag in device_tags for tag in group.selector.tags):
|
|
# continue
|
|
|
|
filtered.append(device)
|
|
|
|
return filtered
|
|
|
|
# No device_ids and no selector → empty list
|
|
return []
|
|
|
|
|
|
def resolve_scene_step_devices(
|
|
step: SceneStep,
|
|
groups_config: GroupsConfigRoot,
|
|
devices: list[DeviceDTO],
|
|
device_rooms: dict[str, str]
|
|
) -> list[DeviceDTO]:
|
|
"""
|
|
Resolve devices for a scene step based on group_id or selector.
|
|
|
|
Args:
|
|
step: Scene step with group_id or selector
|
|
groups_config: Groups configuration for group lookup
|
|
devices: List of all available devices
|
|
device_rooms: Mapping of device_id -> room_name
|
|
|
|
Returns:
|
|
List of devices matching the step criteria
|
|
|
|
Raises:
|
|
ValueError: If group_id is specified but group not found
|
|
|
|
Example:
|
|
>>> # Step with group_id
|
|
>>> step = SceneStep(group_id="kitchen_lights", action={...})
|
|
>>> resolve_scene_step_devices(step, groups_cfg, all_devices, device_rooms)
|
|
[{"device_id": "kueche_deckenlampe", ...}, ...]
|
|
|
|
>>> # Step with selector
|
|
>>> step = SceneStep(
|
|
... selector=SceneSelector(type="light", room="Küche"),
|
|
... action={...}
|
|
... )
|
|
>>> resolve_scene_step_devices(step, groups_cfg, all_devices, device_rooms)
|
|
[{"device_id": "kueche_deckenlampe", ...}, ...]
|
|
"""
|
|
# Case 1: Group reference
|
|
if step.group_id:
|
|
# Look up the group
|
|
group = get_group_by_id(groups_config, step.group_id)
|
|
|
|
if not group:
|
|
raise ValueError(
|
|
f"Scene step references unknown group_id: '{step.group_id}'. "
|
|
f"Available groups: {[g.id for g in groups_config.groups]}"
|
|
)
|
|
|
|
# Resolve the group's devices
|
|
return resolve_group_devices(group, devices, device_rooms)
|
|
|
|
# Case 2: Direct selector
|
|
if step.selector:
|
|
filtered = []
|
|
|
|
for device in devices:
|
|
# Filter by type (optional in scene selector)
|
|
if step.selector.type and device["type"] != step.selector.type:
|
|
continue
|
|
|
|
# Filter by room (optional)
|
|
if step.selector.room:
|
|
device_room = device_rooms.get(device["device_id"])
|
|
if device_room != step.selector.room:
|
|
continue
|
|
|
|
# Filter by tags (optional, future feature)
|
|
# if step.selector.tags:
|
|
# device_tags = device.get("metadata", {}).get("tags", [])
|
|
# if not any(tag in device_tags for tag in step.selector.tags):
|
|
# continue
|
|
|
|
filtered.append(device)
|
|
|
|
return filtered
|
|
|
|
# Should not reach here due to SceneStep validation (must have group_id or selector)
|
|
return []
|