alexei.dolgolyov
ebec1bd16e
- Add BetterCam capture engine (DXGI Desktop Duplication, priority 4) - Fix missing picture_stream_id in get_device endpoint - Fix template delete validation to check streams instead of devices - Add description field to capture engine template UI - Default template name changed to "Default" with descriptive text - Display picker highlights selected display instead of primary - Fix modals closing when dragging text selection outside dialog - Rename "Engine Configuration" to "Configuration", hide when empty - Rename "Run Test" to "Run" across all test buttons - Always reserve space for vertical scrollbar - Redesign Stream Settings info panel with pill-style props - Fix processed stream showing internal ID instead of stream name - Update en/ru locale files Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
254 lines
7.5 KiB
Python
254 lines
7.5 KiB
Python
"""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.core.capture_engines.factory import EngineRegistry
|
|
from wled_controller.storage.template import CaptureTemplate
|
|
from wled_controller.utils import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class TemplateStore:
|
|
"""Storage for capture templates.
|
|
|
|
All templates are persisted to the JSON file.
|
|
On startup, if no templates exist, one is auto-created using the
|
|
highest-priority available engine.
|
|
"""
|
|
|
|
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()
|
|
self._ensure_initial_template()
|
|
|
|
def _ensure_initial_template(self) -> None:
|
|
"""Auto-create a template if none exist, using the best available engine."""
|
|
if self._templates:
|
|
return
|
|
|
|
best_engine = EngineRegistry.get_best_available_engine()
|
|
if not best_engine:
|
|
logger.warning("No capture engines available, cannot create initial template")
|
|
return
|
|
|
|
engine_class = EngineRegistry.get_engine(best_engine)
|
|
default_config = engine_class.get_default_config()
|
|
now = datetime.utcnow()
|
|
template_id = f"tpl_{uuid.uuid4().hex[:8]}"
|
|
|
|
template = CaptureTemplate(
|
|
id=template_id,
|
|
name="Default",
|
|
engine_type=best_engine,
|
|
engine_config=default_config,
|
|
created_at=now,
|
|
updated_at=now,
|
|
description=f"Default capture template using {best_engine.upper()} engine",
|
|
)
|
|
|
|
self._templates[template_id] = template
|
|
self._save()
|
|
logger.info(f"Auto-created initial template: {template.name} ({template_id}, engine={best_engine})")
|
|
|
|
def _load(self) -> None:
|
|
"""Load templates from file."""
|
|
if not self.file_path.exists():
|
|
return
|
|
|
|
try:
|
|
with open(self.file_path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
templates_data = data.get("templates", {})
|
|
loaded = 0
|
|
for template_id, template_dict in templates_data.items():
|
|
try:
|
|
template = CaptureTemplate.from_dict(template_dict)
|
|
self._templates[template_id] = template
|
|
loaded += 1
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Failed to load template {template_id}: {e}",
|
|
exc_info=True
|
|
)
|
|
|
|
if loaded > 0:
|
|
logger.info(f"Loaded {loaded} templates from storage")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load templates from {self.file_path}: {e}")
|
|
raise
|
|
|
|
logger.info(f"Template store initialized with {len(self._templates)} templates")
|
|
|
|
def _save(self) -> None:
|
|
"""Save all templates to file."""
|
|
try:
|
|
# Ensure directory exists
|
|
self.file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
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:
|
|
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,
|
|
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_type: 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_type: New engine type (optional)
|
|
engine_config: New engine config (optional)
|
|
description: New description (optional)
|
|
|
|
Returns:
|
|
Updated template
|
|
|
|
Raises:
|
|
ValueError: If template not found
|
|
"""
|
|
if template_id not in self._templates:
|
|
raise ValueError(f"Template not found: {template_id}")
|
|
|
|
template = self._templates[template_id]
|
|
|
|
# Update fields
|
|
if name is not None:
|
|
template.name = name
|
|
if engine_type is not None:
|
|
template.engine_type = engine_type
|
|
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
|
|
"""
|
|
if template_id not in self._templates:
|
|
raise ValueError(f"Template not found: {template_id}")
|
|
|
|
# Remove and save
|
|
del self._templates[template_id]
|
|
self._save()
|
|
|
|
logger.info(f"Deleted template: {template_id}")
|