Add profile conditions, scene presets, MQTT integration, and Scenes tab
Feature 1 — Profile Conditions: time-of-day, system idle (Win32 GetLastInputInfo), and display state (GUID_CONSOLE_DISPLAY_STATE) condition types for automatic profile activation. Feature 2 — Scene Presets: snapshot/restore system that captures target running states, device brightness, and profile enables. Server-side capture with 5-step activation order. Dedicated Scenes tab with CardSection-based card grid, command palette integration, and dashboard quick-activate section. Feature 3 — MQTT Integration: MQTTService singleton with aiomqtt, MQTTLEDClient device provider for pixel output, MQTT profile condition type with topic/payload matching, and frontend support for MQTT device type and condition editor. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -22,6 +22,14 @@ class Condition:
|
||||
return AlwaysCondition.from_dict(data)
|
||||
if ct == "application":
|
||||
return ApplicationCondition.from_dict(data)
|
||||
if ct == "time_of_day":
|
||||
return TimeOfDayCondition.from_dict(data)
|
||||
if ct == "system_idle":
|
||||
return SystemIdleCondition.from_dict(data)
|
||||
if ct == "display_state":
|
||||
return DisplayStateCondition.from_dict(data)
|
||||
if ct == "mqtt":
|
||||
return MQTTCondition.from_dict(data)
|
||||
raise ValueError(f"Unknown condition type: {ct}")
|
||||
|
||||
|
||||
@@ -58,6 +66,98 @@ class ApplicationCondition(Condition):
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class TimeOfDayCondition(Condition):
|
||||
"""Activate during a specific time range (server local time).
|
||||
|
||||
Supports overnight ranges: if start_time > end_time, the range wraps
|
||||
around midnight (e.g. 22:00 → 06:00).
|
||||
"""
|
||||
|
||||
condition_type: str = "time_of_day"
|
||||
start_time: str = "00:00" # HH:MM
|
||||
end_time: str = "23:59" # HH:MM
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = super().to_dict()
|
||||
d["start_time"] = self.start_time
|
||||
d["end_time"] = self.end_time
|
||||
return d
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "TimeOfDayCondition":
|
||||
return cls(
|
||||
start_time=data.get("start_time", "00:00"),
|
||||
end_time=data.get("end_time", "23:59"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class SystemIdleCondition(Condition):
|
||||
"""Activate based on system idle time (keyboard/mouse inactivity)."""
|
||||
|
||||
condition_type: str = "system_idle"
|
||||
idle_minutes: int = 5
|
||||
when_idle: bool = True # True = active when idle; False = active when NOT idle
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = super().to_dict()
|
||||
d["idle_minutes"] = self.idle_minutes
|
||||
d["when_idle"] = self.when_idle
|
||||
return d
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "SystemIdleCondition":
|
||||
return cls(
|
||||
idle_minutes=data.get("idle_minutes", 5),
|
||||
when_idle=data.get("when_idle", True),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class DisplayStateCondition(Condition):
|
||||
"""Activate based on display/monitor power state."""
|
||||
|
||||
condition_type: str = "display_state"
|
||||
state: str = "on" # "on" | "off"
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = super().to_dict()
|
||||
d["state"] = self.state
|
||||
return d
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "DisplayStateCondition":
|
||||
return cls(
|
||||
state=data.get("state", "on"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class MQTTCondition(Condition):
|
||||
"""Activate based on an MQTT topic value."""
|
||||
|
||||
condition_type: str = "mqtt"
|
||||
topic: str = ""
|
||||
payload: str = ""
|
||||
match_mode: str = "exact" # "exact" | "contains" | "regex"
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = super().to_dict()
|
||||
d["topic"] = self.topic
|
||||
d["payload"] = self.payload
|
||||
d["match_mode"] = self.match_mode
|
||||
return d
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "MQTTCondition":
|
||||
return cls(
|
||||
topic=data.get("topic", ""),
|
||||
payload=data.get("payload", ""),
|
||||
match_mode=data.get("match_mode", "exact"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Profile:
|
||||
"""Automation profile that activates targets based on conditions."""
|
||||
|
||||
125
server/src/wled_controller/storage/scene_preset.py
Normal file
125
server/src/wled_controller/storage/scene_preset.py
Normal file
@@ -0,0 +1,125 @@
|
||||
"""Scene preset data models — snapshot of current system state."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class TargetSnapshot:
|
||||
"""Snapshot of a single target's mutable state."""
|
||||
|
||||
target_id: str
|
||||
running: bool = False
|
||||
color_strip_source_id: str = ""
|
||||
brightness_value_source_id: str = ""
|
||||
fps: int = 30
|
||||
auto_start: bool = False
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"target_id": self.target_id,
|
||||
"running": self.running,
|
||||
"color_strip_source_id": self.color_strip_source_id,
|
||||
"brightness_value_source_id": self.brightness_value_source_id,
|
||||
"fps": self.fps,
|
||||
"auto_start": self.auto_start,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "TargetSnapshot":
|
||||
return cls(
|
||||
target_id=data["target_id"],
|
||||
running=data.get("running", False),
|
||||
color_strip_source_id=data.get("color_strip_source_id", ""),
|
||||
brightness_value_source_id=data.get("brightness_value_source_id", ""),
|
||||
fps=data.get("fps", 30),
|
||||
auto_start=data.get("auto_start", False),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class DeviceBrightnessSnapshot:
|
||||
"""Snapshot of a device's software brightness."""
|
||||
|
||||
device_id: str
|
||||
software_brightness: int = 255
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"device_id": self.device_id,
|
||||
"software_brightness": self.software_brightness,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "DeviceBrightnessSnapshot":
|
||||
return cls(
|
||||
device_id=data["device_id"],
|
||||
software_brightness=data.get("software_brightness", 255),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProfileSnapshot:
|
||||
"""Snapshot of a profile's enabled state."""
|
||||
|
||||
profile_id: str
|
||||
enabled: bool = True
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"profile_id": self.profile_id,
|
||||
"enabled": self.enabled,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "ProfileSnapshot":
|
||||
return cls(
|
||||
profile_id=data["profile_id"],
|
||||
enabled=data.get("enabled", True),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ScenePreset:
|
||||
"""A named snapshot of system state that can be restored."""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
description: str = ""
|
||||
color: str = "#4fc3f7" # accent color for the card
|
||||
targets: List[TargetSnapshot] = field(default_factory=list)
|
||||
devices: List[DeviceBrightnessSnapshot] = field(default_factory=list)
|
||||
profiles: List[ProfileSnapshot] = field(default_factory=list)
|
||||
order: int = 0
|
||||
created_at: datetime = field(default_factory=datetime.utcnow)
|
||||
updated_at: datetime = field(default_factory=datetime.utcnow)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"name": self.name,
|
||||
"description": self.description,
|
||||
"color": self.color,
|
||||
"targets": [t.to_dict() for t in self.targets],
|
||||
"devices": [d.to_dict() for d in self.devices],
|
||||
"profiles": [p.to_dict() for p in self.profiles],
|
||||
"order": self.order,
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"updated_at": self.updated_at.isoformat(),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "ScenePreset":
|
||||
return cls(
|
||||
id=data["id"],
|
||||
name=data["name"],
|
||||
description=data.get("description", ""),
|
||||
color=data.get("color", "#4fc3f7"),
|
||||
targets=[TargetSnapshot.from_dict(t) for t in data.get("targets", [])],
|
||||
devices=[DeviceBrightnessSnapshot.from_dict(d) for d in data.get("devices", [])],
|
||||
profiles=[ProfileSnapshot.from_dict(p) for p in data.get("profiles", [])],
|
||||
order=data.get("order", 0),
|
||||
created_at=datetime.fromisoformat(data.get("created_at", datetime.utcnow().isoformat())),
|
||||
updated_at=datetime.fromisoformat(data.get("updated_at", datetime.utcnow().isoformat())),
|
||||
)
|
||||
134
server/src/wled_controller/storage/scene_preset_store.py
Normal file
134
server/src/wled_controller/storage/scene_preset_store.py
Normal file
@@ -0,0 +1,134 @@
|
||||
"""Scene preset storage using JSON files."""
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from wled_controller.storage.scene_preset import ScenePreset
|
||||
from wled_controller.utils import atomic_write_json, get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class ScenePresetStore:
|
||||
"""Persistent storage for scene presets."""
|
||||
|
||||
def __init__(self, file_path: str):
|
||||
self.file_path = Path(file_path)
|
||||
self._presets: Dict[str, ScenePreset] = {}
|
||||
self._load()
|
||||
|
||||
def _load(self) -> None:
|
||||
if not self.file_path.exists():
|
||||
return
|
||||
|
||||
try:
|
||||
with open(self.file_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
presets_data = data.get("scene_presets", {})
|
||||
loaded = 0
|
||||
for preset_id, preset_dict in presets_data.items():
|
||||
try:
|
||||
preset = ScenePreset.from_dict(preset_dict)
|
||||
self._presets[preset_id] = preset
|
||||
loaded += 1
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load scene preset {preset_id}: {e}", exc_info=True)
|
||||
|
||||
if loaded > 0:
|
||||
logger.info(f"Loaded {loaded} scene presets from storage")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load scene presets from {self.file_path}: {e}")
|
||||
raise
|
||||
|
||||
logger.info(f"Scene preset store initialized with {len(self._presets)} presets")
|
||||
|
||||
def _save(self) -> None:
|
||||
try:
|
||||
data = {
|
||||
"version": "1.0.0",
|
||||
"scene_presets": {
|
||||
pid: p.to_dict() for pid, p in self._presets.items()
|
||||
},
|
||||
}
|
||||
atomic_write_json(self.file_path, data)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to save scene presets to {self.file_path}: {e}")
|
||||
raise
|
||||
|
||||
def get_all_presets(self) -> List[ScenePreset]:
|
||||
return sorted(self._presets.values(), key=lambda p: p.order)
|
||||
|
||||
def get_preset(self, preset_id: str) -> ScenePreset:
|
||||
if preset_id not in self._presets:
|
||||
raise ValueError(f"Scene preset not found: {preset_id}")
|
||||
return self._presets[preset_id]
|
||||
|
||||
def create_preset(self, preset: ScenePreset) -> ScenePreset:
|
||||
for p in self._presets.values():
|
||||
if p.name == preset.name:
|
||||
raise ValueError(f"Scene preset with name '{preset.name}' already exists")
|
||||
|
||||
self._presets[preset.id] = preset
|
||||
self._save()
|
||||
logger.info(f"Created scene preset: {preset.name} ({preset.id})")
|
||||
return preset
|
||||
|
||||
def update_preset(
|
||||
self,
|
||||
preset_id: str,
|
||||
name: Optional[str] = None,
|
||||
description: Optional[str] = None,
|
||||
color: Optional[str] = None,
|
||||
order: Optional[int] = None,
|
||||
) -> ScenePreset:
|
||||
if preset_id not in self._presets:
|
||||
raise ValueError(f"Scene preset not found: {preset_id}")
|
||||
|
||||
preset = self._presets[preset_id]
|
||||
|
||||
if name is not None:
|
||||
for pid, p in self._presets.items():
|
||||
if pid != preset_id and p.name == name:
|
||||
raise ValueError(f"Scene preset with name '{name}' already exists")
|
||||
preset.name = name
|
||||
if description is not None:
|
||||
preset.description = description
|
||||
if color is not None:
|
||||
preset.color = color
|
||||
if order is not None:
|
||||
preset.order = order
|
||||
|
||||
preset.updated_at = datetime.utcnow()
|
||||
self._save()
|
||||
logger.info(f"Updated scene preset: {preset_id}")
|
||||
return preset
|
||||
|
||||
def recapture_preset(self, preset_id: str, preset: ScenePreset) -> ScenePreset:
|
||||
"""Replace snapshot data of an existing preset (recapture current state)."""
|
||||
if preset_id not in self._presets:
|
||||
raise ValueError(f"Scene preset not found: {preset_id}")
|
||||
|
||||
existing = self._presets[preset_id]
|
||||
existing.targets = preset.targets
|
||||
existing.devices = preset.devices
|
||||
existing.profiles = preset.profiles
|
||||
existing.updated_at = datetime.utcnow()
|
||||
self._save()
|
||||
logger.info(f"Recaptured scene preset: {preset_id}")
|
||||
return existing
|
||||
|
||||
def delete_preset(self, preset_id: str) -> None:
|
||||
if preset_id not in self._presets:
|
||||
raise ValueError(f"Scene preset not found: {preset_id}")
|
||||
|
||||
del self._presets[preset_id]
|
||||
self._save()
|
||||
logger.info(f"Deleted scene preset: {preset_id}")
|
||||
|
||||
def count(self) -> int:
|
||||
return len(self._presets)
|
||||
Reference in New Issue
Block a user