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 <noreply@anthropic.com>
This commit is contained in:
221
server/src/wled_controller/storage/template_store.py
Normal file
221
server/src/wled_controller/storage/template_store.py
Normal file
@@ -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}")
|
||||
Reference in New Issue
Block a user