From fdb73c9fc9d2a3244a55d9c1925ff7175ccd8efb Mon Sep 17 00:00:00 2001 From: "alexei.dolgolyov" Date: Mon, 9 Feb 2026 19:23:20 +0300 Subject: [PATCH] Fix MSS template test: add engine initialization and create TemplateStore - Add engine.initialize() call in test endpoint to fix "Engine not initialized" error for MSS and DXcam - Create template.py with CaptureTemplate dataclass for template data model - Create template_store.py with TemplateStore class for template CRUD operations - TemplateStore loads from capture_templates.json and provides get_all, create, get, update, delete methods Fixes MSS capture test failing with "Engine not initialized" error. WGC worked because it auto-initializes in capture_display(), but MSS and DXcam require explicit initialization. Co-Authored-By: Claude Sonnet 4.5 --- server/src/wled_controller/api/routes.py | 3 +- .../src/wled_controller/storage/template.py | 61 +++++ .../wled_controller/storage/template_store.py | 221 ++++++++++++++++++ 3 files changed, 284 insertions(+), 1 deletion(-) create mode 100644 server/src/wled_controller/storage/template.py create mode 100644 server/src/wled_controller/storage/template_store.py diff --git a/server/src/wled_controller/api/routes.py b/server/src/wled_controller/api/routes.py index bd872c1..2070f06 100644 --- a/server/src/wled_controller/api/routes.py +++ b/server/src/wled_controller/api/routes.py @@ -930,8 +930,9 @@ async def test_template( ) ) - # Create engine (initialization happens on first capture) + # Create and initialize engine engine = EngineRegistry.create_engine(test_request.engine_type, test_request.engine_config) + engine.initialize() # Run sustained capture test logger.info(f"Starting {test_request.capture_duration}s capture test with {test_request.engine_type}") diff --git a/server/src/wled_controller/storage/template.py b/server/src/wled_controller/storage/template.py new file mode 100644 index 0000000..5374fb3 --- /dev/null +++ b/server/src/wled_controller/storage/template.py @@ -0,0 +1,61 @@ +"""Capture template data model.""" + +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Dict, Optional + + +@dataclass +class CaptureTemplate: + """Represents a screen capture template configuration.""" + + id: str + name: str + engine_type: str + engine_config: Dict[str, Any] + is_default: bool + created_at: datetime + updated_at: datetime + description: Optional[str] = None + + def to_dict(self) -> dict: + """Convert template to dictionary. + + Returns: + Dictionary representation + """ + return { + "id": self.id, + "name": self.name, + "engine_type": self.engine_type, + "engine_config": self.engine_config, + "is_default": self.is_default, + "created_at": self.created_at.isoformat(), + "updated_at": self.updated_at.isoformat(), + "description": self.description, + } + + @classmethod + def from_dict(cls, data: dict) -> "CaptureTemplate": + """Create template from dictionary. + + Args: + data: Dictionary with template data + + Returns: + CaptureTemplate instance + """ + return cls( + id=data["id"], + name=data["name"], + engine_type=data["engine_type"], + engine_config=data.get("engine_config", {}), + is_default=data.get("is_default", False), + created_at=datetime.fromisoformat(data["created_at"]) + if isinstance(data.get("created_at"), str) + else data.get("created_at", datetime.utcnow()), + updated_at=datetime.fromisoformat(data["updated_at"]) + if isinstance(data.get("updated_at"), str) + else data.get("updated_at", datetime.utcnow()), + description=data.get("description"), + ) diff --git a/server/src/wled_controller/storage/template_store.py b/server/src/wled_controller/storage/template_store.py new file mode 100644 index 0000000..c951ba8 --- /dev/null +++ b/server/src/wled_controller/storage/template_store.py @@ -0,0 +1,221 @@ +"""Template 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.template import CaptureTemplate +from wled_controller.utils import get_logger + +logger = get_logger(__name__) + + +class TemplateStore: + """Storage for capture templates.""" + + def __init__(self, file_path: str): + """Initialize template store. + + Args: + file_path: Path to templates JSON file + """ + self.file_path = Path(file_path) + self._templates: Dict[str, CaptureTemplate] = {} + self._load() + + def _load(self) -> None: + """Load templates from file.""" + if not self.file_path.exists(): + logger.warning(f"Templates file not found: {self.file_path}") + self._save() # Create empty file + return + + try: + with open(self.file_path, "r", encoding="utf-8") as f: + data = json.load(f) + + templates_data = data.get("templates", {}) + for template_id, template_dict in templates_data.items(): + try: + template = CaptureTemplate.from_dict(template_dict) + self._templates[template_id] = template + except Exception as e: + logger.error( + f"Failed to load template {template_id}: {e}", + exc_info=True + ) + + logger.info(f"Loaded {len(self._templates)} templates from storage") + logger.info(f"Template store initialized with {len(self._templates)} templates") + + except Exception as e: + logger.error(f"Failed to load templates from {self.file_path}: {e}") + raise + + def _save(self) -> None: + """Save templates to file.""" + try: + # Ensure directory exists + self.file_path.parent.mkdir(parents=True, exist_ok=True) + + # Convert templates to dict format + templates_dict = { + template_id: template.to_dict() + for template_id, template in self._templates.items() + } + + data = { + "version": "1.0.0", + "templates": templates_dict, + } + + # Write to file + 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 templates to {self.file_path}: {e}") + raise + + def get_all_templates(self) -> List[CaptureTemplate]: + """Get all templates. + + Returns: + List of all templates + """ + return list(self._templates.values()) + + def get_template(self, template_id: str) -> CaptureTemplate: + """Get template by ID. + + Args: + template_id: Template ID + + Returns: + Template instance + + Raises: + ValueError: If template not found + """ + if template_id not in self._templates: + raise ValueError(f"Template not found: {template_id}") + return self._templates[template_id] + + def create_template( + self, + name: str, + engine_type: str, + engine_config: Dict[str, any], + description: Optional[str] = None, + ) -> CaptureTemplate: + """Create a new template. + + Args: + name: Template name + engine_type: Engine type (mss, dxcam, wgc) + engine_config: Engine-specific configuration + description: Optional description + + Returns: + Created template + + Raises: + ValueError: If template with same name exists + """ + # Check for duplicate name + for template in self._templates.values(): + if template.name == name and not template.is_default: + raise ValueError(f"Template with name '{name}' already exists") + + # Generate new ID + template_id = f"tpl_{uuid.uuid4().hex[:8]}" + + # Create template + now = datetime.utcnow() + template = CaptureTemplate( + id=template_id, + name=name, + engine_type=engine_type, + engine_config=engine_config, + is_default=False, + created_at=now, + updated_at=now, + description=description, + ) + + # Store and save + self._templates[template_id] = template + self._save() + + logger.info(f"Created template: {name} ({template_id})") + return template + + def update_template( + self, + template_id: str, + name: Optional[str] = None, + engine_config: Optional[Dict[str, any]] = None, + description: Optional[str] = None, + ) -> CaptureTemplate: + """Update an existing template. + + Args: + template_id: Template ID + name: New name (optional) + engine_config: New engine config (optional) + description: New description (optional) + + Returns: + Updated template + + Raises: + ValueError: If template not found or is a default template + """ + if template_id not in self._templates: + raise ValueError(f"Template not found: {template_id}") + + template = self._templates[template_id] + + if template.is_default: + raise ValueError("Cannot modify default templates") + + # Update fields + if name is not None: + template.name = name + if engine_config is not None: + template.engine_config = engine_config + if description is not None: + template.description = description + + template.updated_at = datetime.utcnow() + + # Save + self._save() + + logger.info(f"Updated template: {template_id}") + return template + + def delete_template(self, template_id: str) -> None: + """Delete a template. + + Args: + template_id: Template ID + + Raises: + ValueError: If template not found or is a default template + """ + if template_id not in self._templates: + raise ValueError(f"Template not found: {template_id}") + + template = self._templates[template_id] + + if template.is_default: + raise ValueError("Cannot delete default templates") + + # Remove and save + del self._templates[template_id] + self._save() + + logger.info(f"Deleted template: {template_id}")