Improve device cards, stream/template UI, and add PP template testing
- Move WLED UI button into URL badge as clickable link on device cards - Remove version label from device cards - Show PP template name on processed stream cards - Display filter chain as pills on processing template cards - Add processing template test with source stream selector - Pre-load PP templates when viewing streams to fix race condition - Add ESC key handling for all modals - Add filter chain CSS styles Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -53,6 +53,7 @@ from wled_controller.api.schemas import (
|
||||
PictureStreamResponse,
|
||||
PictureStreamListResponse,
|
||||
PictureStreamTestRequest,
|
||||
PPTemplateTestRequest,
|
||||
)
|
||||
from wled_controller.config import get_config
|
||||
from wled_controller.core.processor_manager import ProcessorManager, ProcessingSettings
|
||||
@@ -1255,6 +1256,172 @@ async def delete_pp_template(
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
@router.post("/api/v1/postprocessing-templates/{template_id}/test", response_model=TemplateTestResponse, tags=["Postprocessing Templates"])
|
||||
async def test_pp_template(
|
||||
template_id: str,
|
||||
test_request: PPTemplateTestRequest,
|
||||
_auth: AuthRequired,
|
||||
pp_store: PostprocessingTemplateStore = Depends(get_pp_template_store),
|
||||
stream_store: PictureStreamStore = Depends(get_picture_stream_store),
|
||||
template_store: TemplateStore = Depends(get_template_store),
|
||||
processor_manager: ProcessorManager = Depends(get_processor_manager),
|
||||
device_store: DeviceStore = Depends(get_device_store),
|
||||
):
|
||||
"""Test a postprocessing template by capturing from a source stream and applying filters."""
|
||||
engine = None
|
||||
try:
|
||||
# Get the PP template
|
||||
try:
|
||||
pp_template = pp_store.get_template(template_id)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
|
||||
# Resolve source stream chain to get the raw stream
|
||||
try:
|
||||
chain = stream_store.resolve_stream_chain(test_request.source_stream_id)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
|
||||
raw_stream = chain["raw_stream"]
|
||||
|
||||
# Get capture template from raw stream
|
||||
try:
|
||||
capture_template = template_store.get_template(raw_stream.capture_template_id)
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"Capture template not found: {raw_stream.capture_template_id}",
|
||||
)
|
||||
|
||||
display_index = raw_stream.display_index
|
||||
|
||||
# Validate engine
|
||||
if capture_template.engine_type not in EngineRegistry.get_available_engines():
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"Engine '{capture_template.engine_type}' is not available on this system",
|
||||
)
|
||||
|
||||
# Check display lock
|
||||
locked_device_id = processor_manager.get_display_lock_info(display_index)
|
||||
if locked_device_id:
|
||||
try:
|
||||
device = device_store.get_device(locked_device_id)
|
||||
device_name = device.name
|
||||
except Exception:
|
||||
device_name = locked_device_id
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"Display {display_index} is currently being captured by device '{device_name}'. "
|
||||
f"Please stop the device processing before testing.",
|
||||
)
|
||||
|
||||
# Create engine and run test
|
||||
engine = EngineRegistry.create_engine(capture_template.engine_type, capture_template.engine_config)
|
||||
|
||||
logger.info(f"Starting {test_request.capture_duration}s PP template test for {template_id} using stream {test_request.source_stream_id}")
|
||||
|
||||
frame_count = 0
|
||||
total_capture_time = 0.0
|
||||
last_frame = None
|
||||
|
||||
start_time = time.perf_counter()
|
||||
end_time = start_time + test_request.capture_duration
|
||||
|
||||
while time.perf_counter() < end_time:
|
||||
capture_start = time.perf_counter()
|
||||
screen_capture = engine.capture_display(display_index)
|
||||
capture_elapsed = time.perf_counter() - capture_start
|
||||
|
||||
total_capture_time += capture_elapsed
|
||||
frame_count += 1
|
||||
last_frame = screen_capture
|
||||
|
||||
actual_duration = time.perf_counter() - start_time
|
||||
|
||||
if last_frame is None:
|
||||
raise RuntimeError("No frames captured during test")
|
||||
|
||||
# Convert to PIL Image
|
||||
if isinstance(last_frame.image, np.ndarray):
|
||||
pil_image = Image.fromarray(last_frame.image)
|
||||
else:
|
||||
raise ValueError("Unexpected image format from engine")
|
||||
|
||||
# Create thumbnail
|
||||
thumbnail_width = 640
|
||||
aspect_ratio = pil_image.height / pil_image.width
|
||||
thumbnail_height = int(thumbnail_width * aspect_ratio)
|
||||
thumbnail = pil_image.copy()
|
||||
thumbnail.thumbnail((thumbnail_width, thumbnail_height), Image.Resampling.LANCZOS)
|
||||
|
||||
# Apply postprocessing filters
|
||||
if pp_template.filters:
|
||||
pool = ImagePool()
|
||||
|
||||
def apply_filters(img):
|
||||
arr = np.array(img)
|
||||
for fi in pp_template.filters:
|
||||
f = FilterRegistry.create_instance(fi.filter_id, fi.options)
|
||||
result = f.process_image(arr, pool)
|
||||
if result is not None:
|
||||
arr = result
|
||||
return Image.fromarray(arr)
|
||||
|
||||
thumbnail = apply_filters(thumbnail)
|
||||
pil_image = apply_filters(pil_image)
|
||||
|
||||
# Encode thumbnail
|
||||
img_buffer = io.BytesIO()
|
||||
thumbnail.save(img_buffer, format='JPEG', quality=85)
|
||||
img_buffer.seek(0)
|
||||
thumbnail_b64 = base64.b64encode(img_buffer.getvalue()).decode('utf-8')
|
||||
thumbnail_data_uri = f"data:image/jpeg;base64,{thumbnail_b64}"
|
||||
|
||||
# Encode full-resolution image
|
||||
full_buffer = io.BytesIO()
|
||||
pil_image.save(full_buffer, format='JPEG', quality=90)
|
||||
full_buffer.seek(0)
|
||||
full_b64 = base64.b64encode(full_buffer.getvalue()).decode('utf-8')
|
||||
full_data_uri = f"data:image/jpeg;base64,{full_b64}"
|
||||
|
||||
actual_fps = frame_count / actual_duration if actual_duration > 0 else 0
|
||||
avg_capture_time_ms = (total_capture_time / frame_count * 1000) if frame_count > 0 else 0
|
||||
width, height = pil_image.size
|
||||
|
||||
return TemplateTestResponse(
|
||||
full_capture=CaptureImage(
|
||||
image=thumbnail_data_uri,
|
||||
full_image=full_data_uri,
|
||||
width=width,
|
||||
height=height,
|
||||
thumbnail_width=thumbnail_width,
|
||||
thumbnail_height=thumbnail_height,
|
||||
),
|
||||
border_extraction=None,
|
||||
performance=PerformanceMetrics(
|
||||
capture_duration_s=actual_duration,
|
||||
frame_count=frame_count,
|
||||
actual_fps=actual_fps,
|
||||
avg_capture_time_ms=avg_capture_time_ms,
|
||||
),
|
||||
)
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
logger.error(f"Postprocessing template test failed: {e}")
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
finally:
|
||||
if engine:
|
||||
try:
|
||||
engine.release()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ===== PICTURE STREAM ENDPOINTS =====
|
||||
|
||||
def _stream_to_response(s) -> PictureStreamResponse:
|
||||
|
||||
@@ -454,3 +454,10 @@ class PictureStreamTestRequest(BaseModel):
|
||||
|
||||
capture_duration: float = Field(default=5.0, ge=1.0, le=30.0, description="Duration to capture in seconds")
|
||||
border_width: int = Field(default=10, ge=1, le=100, description="Border width in pixels for preview")
|
||||
|
||||
|
||||
class PPTemplateTestRequest(BaseModel):
|
||||
"""Request to test a postprocessing template against a source stream."""
|
||||
|
||||
source_stream_id: str = Field(description="ID of the source picture stream to capture from")
|
||||
capture_duration: float = Field(default=5.0, ge=1.0, le=30.0, description="Duration to capture in seconds")
|
||||
|
||||
Reference in New Issue
Block a user