Add profile system for automatic target activation
Profiles monitor running processes and foreground windows to automatically start/stop targets when conditions are met. Includes profile engine, platform detector (WMI), REST API, process browser endpoint, and calibration persistence fix. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -243,7 +243,6 @@ class DeviceStore:
|
||||
device.url = url
|
||||
if led_count is not None:
|
||||
device.led_count = led_count
|
||||
device.calibration = create_default_calibration(led_count)
|
||||
if enabled is not None:
|
||||
device.enabled = enabled
|
||||
if baud_rate is not None:
|
||||
|
||||
91
server/src/wled_controller/storage/profile.py
Normal file
91
server/src/wled_controller/storage/profile.py
Normal file
@@ -0,0 +1,91 @@
|
||||
"""Profile and Condition data models."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class Condition:
|
||||
"""Base condition — polymorphic via condition_type discriminator."""
|
||||
|
||||
condition_type: str
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {"condition_type": self.condition_type}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Condition":
|
||||
"""Factory: dispatch to the correct subclass."""
|
||||
ct = data.get("condition_type", "")
|
||||
if ct == "application":
|
||||
return ApplicationCondition.from_dict(data)
|
||||
raise ValueError(f"Unknown condition type: {ct}")
|
||||
|
||||
|
||||
@dataclass
|
||||
class ApplicationCondition(Condition):
|
||||
"""Activate when specified applications are running or topmost."""
|
||||
|
||||
condition_type: str = "application"
|
||||
apps: List[str] = field(default_factory=list)
|
||||
match_type: str = "running" # "running" | "topmost"
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
d = super().to_dict()
|
||||
d["apps"] = list(self.apps)
|
||||
d["match_type"] = self.match_type
|
||||
return d
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "ApplicationCondition":
|
||||
return cls(
|
||||
apps=data.get("apps", []),
|
||||
match_type=data.get("match_type", "running"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Profile:
|
||||
"""Automation profile that activates targets based on conditions."""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
enabled: bool
|
||||
condition_logic: str # "or" | "and"
|
||||
conditions: List[Condition]
|
||||
target_ids: List[str]
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"id": self.id,
|
||||
"name": self.name,
|
||||
"enabled": self.enabled,
|
||||
"condition_logic": self.condition_logic,
|
||||
"conditions": [c.to_dict() for c in self.conditions],
|
||||
"target_ids": list(self.target_ids),
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"updated_at": self.updated_at.isoformat(),
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> "Profile":
|
||||
conditions = []
|
||||
for c_data in data.get("conditions", []):
|
||||
try:
|
||||
conditions.append(Condition.from_dict(c_data))
|
||||
except ValueError:
|
||||
pass # skip unknown condition types on load
|
||||
|
||||
return cls(
|
||||
id=data["id"],
|
||||
name=data["name"],
|
||||
enabled=data.get("enabled", True),
|
||||
condition_logic=data.get("condition_logic", "or"),
|
||||
conditions=conditions,
|
||||
target_ids=data.get("target_ids", []),
|
||||
created_at=datetime.fromisoformat(data.get("created_at", datetime.utcnow().isoformat())),
|
||||
updated_at=datetime.fromisoformat(data.get("updated_at", datetime.utcnow().isoformat())),
|
||||
)
|
||||
145
server/src/wled_controller/storage/profile_store.py
Normal file
145
server/src/wled_controller/storage/profile_store.py
Normal file
@@ -0,0 +1,145 @@
|
||||
"""Profile 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.profile import Condition, Profile
|
||||
from wled_controller.utils import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class ProfileStore:
|
||||
"""Persistent storage for automation profiles."""
|
||||
|
||||
def __init__(self, file_path: str):
|
||||
self.file_path = Path(file_path)
|
||||
self._profiles: Dict[str, Profile] = {}
|
||||
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)
|
||||
|
||||
profiles_data = data.get("profiles", {})
|
||||
loaded = 0
|
||||
for profile_id, profile_dict in profiles_data.items():
|
||||
try:
|
||||
profile = Profile.from_dict(profile_dict)
|
||||
self._profiles[profile_id] = profile
|
||||
loaded += 1
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load profile {profile_id}: {e}", exc_info=True)
|
||||
|
||||
if loaded > 0:
|
||||
logger.info(f"Loaded {loaded} profiles from storage")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load profiles from {self.file_path}: {e}")
|
||||
raise
|
||||
|
||||
logger.info(f"Profile store initialized with {len(self._profiles)} profiles")
|
||||
|
||||
def _save(self) -> None:
|
||||
try:
|
||||
self.file_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
data = {
|
||||
"version": "1.0.0",
|
||||
"profiles": {
|
||||
pid: p.to_dict() for pid, p in self._profiles.items()
|
||||
},
|
||||
}
|
||||
|
||||
with open(self.file_path, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to save profiles to {self.file_path}: {e}")
|
||||
raise
|
||||
|
||||
def get_all_profiles(self) -> List[Profile]:
|
||||
return list(self._profiles.values())
|
||||
|
||||
def get_profile(self, profile_id: str) -> Profile:
|
||||
if profile_id not in self._profiles:
|
||||
raise ValueError(f"Profile not found: {profile_id}")
|
||||
return self._profiles[profile_id]
|
||||
|
||||
def create_profile(
|
||||
self,
|
||||
name: str,
|
||||
enabled: bool = True,
|
||||
condition_logic: str = "or",
|
||||
conditions: Optional[List[Condition]] = None,
|
||||
target_ids: Optional[List[str]] = None,
|
||||
) -> Profile:
|
||||
profile_id = f"prof_{uuid.uuid4().hex[:8]}"
|
||||
now = datetime.utcnow()
|
||||
|
||||
profile = Profile(
|
||||
id=profile_id,
|
||||
name=name,
|
||||
enabled=enabled,
|
||||
condition_logic=condition_logic,
|
||||
conditions=conditions or [],
|
||||
target_ids=target_ids or [],
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
|
||||
self._profiles[profile_id] = profile
|
||||
self._save()
|
||||
|
||||
logger.info(f"Created profile: {name} ({profile_id})")
|
||||
return profile
|
||||
|
||||
def update_profile(
|
||||
self,
|
||||
profile_id: str,
|
||||
name: Optional[str] = None,
|
||||
enabled: Optional[bool] = None,
|
||||
condition_logic: Optional[str] = None,
|
||||
conditions: Optional[List[Condition]] = None,
|
||||
target_ids: Optional[List[str]] = None,
|
||||
) -> Profile:
|
||||
if profile_id not in self._profiles:
|
||||
raise ValueError(f"Profile not found: {profile_id}")
|
||||
|
||||
profile = self._profiles[profile_id]
|
||||
|
||||
if name is not None:
|
||||
profile.name = name
|
||||
if enabled is not None:
|
||||
profile.enabled = enabled
|
||||
if condition_logic is not None:
|
||||
profile.condition_logic = condition_logic
|
||||
if conditions is not None:
|
||||
profile.conditions = conditions
|
||||
if target_ids is not None:
|
||||
profile.target_ids = target_ids
|
||||
|
||||
profile.updated_at = datetime.utcnow()
|
||||
self._save()
|
||||
|
||||
logger.info(f"Updated profile: {profile_id}")
|
||||
return profile
|
||||
|
||||
def delete_profile(self, profile_id: str) -> None:
|
||||
if profile_id not in self._profiles:
|
||||
raise ValueError(f"Profile not found: {profile_id}")
|
||||
|
||||
del self._profiles[profile_id]
|
||||
self._save()
|
||||
|
||||
logger.info(f"Deleted profile: {profile_id}")
|
||||
|
||||
def count(self) -> int:
|
||||
return len(self._profiles)
|
||||
Reference in New Issue
Block a user