Debug replay (#22212)

* debug replay implementation

* fix masks after dev rebase

* fix squash merge issues

* fix

* fix

* fix

* no need to write debug replay camera to config

* camera and filter button and dropdown

* add filters

* add ability to edit motion and object config for debug replay

* add debug draw overlay to debug replay

* add guard to prevent crash when camera is no longer in camera_states

* fix overflow due to radix absolutely positioned elements

* increase number of messages

* ensure deep_merge replaces existing list values when override is true

* add back button

* add debug replay to explore and review menus

* clean up

* clean up

* update instructions to prevent exposing exception info

* fix typing

* refactor output logic

* refactor with helper function

* move init to function for consistency

frigate/api/app.py

@@ -49,12 +49,13 @@ from frigate.stats.prometheus import get_metrics, update_metrics
 from frigate.types import JobStatusTypesEnum
 from frigate.util.builtin import (
     clean_camera_user_pass,
+    deep_merge,
     flatten_config_data,
     load_labels,
     process_config_query_string,
     update_yaml_file_bulk,
 )
-from frigate.util.config import find_config_file
+from frigate.util.config import apply_section_update, find_config_file
 from frigate.util.schema import get_config_schema
 from frigate.util.services import (
     get_nvidia_driver_info,
@@ -422,9 +423,100 @@ def config_save(save_option: str, body: Any = Body(media_type="text/plain")):
         )
 
 
+def _config_set_in_memory(request: Request, body: AppConfigSetBody) -> JSONResponse:
+    """Apply config changes in-memory only, without writing to YAML.
+
+    Used for temporary config changes like debug replay camera tuning.
+    Updates the in-memory Pydantic config and publishes ZMQ updates,
+    bypassing YAML parsing entirely.
+    """
+    try:
+        updates = {}
+        if body.config_data:
+            updates = flatten_config_data(body.config_data)
+            updates = {k: ("" if v is None else v) for k, v in updates.items()}
+
+        if not updates:
+            return JSONResponse(
+                content={"success": False, "message": "No configuration data provided"},
+                status_code=400,
+            )
+
+        config: FrigateConfig = request.app.frigate_config
+
+        # Group flat key paths into nested per-camera, per-section dicts
+        grouped: dict[str, dict[str, dict]] = {}
+        for key_path, value in updates.items():
+            parts = key_path.split(".")
+            if len(parts) < 3 or parts[0] != "cameras":
+                continue
+
+            cam, section = parts[1], parts[2]
+            grouped.setdefault(cam, {}).setdefault(section, {})
+
+            # Build nested dict from remaining path (e.g. "filters.person.threshold")
+            target = grouped[cam][section]
+            for part in parts[3:-1]:
+                target = target.setdefault(part, {})
+            if len(parts) > 3:
+                target[parts[-1]] = value
+            elif isinstance(value, dict):
+                grouped[cam][section] = deep_merge(
+                    grouped[cam][section], value, override=True
+                )
+            else:
+                grouped[cam][section] = value
+
+        # Apply each section update
+        for cam_name, sections in grouped.items():
+            camera_config = config.cameras.get(cam_name)
+            if not camera_config:
+                return JSONResponse(
+                    content={
+                        "success": False,
+                        "message": f"Camera '{cam_name}' not found",
+                    },
+                    status_code=400,
+                )
+
+            for section_name, update in sections.items():
+                err = apply_section_update(camera_config, section_name, update)
+                if err is not None:
+                    return JSONResponse(
+                        content={"success": False, "message": err},
+                        status_code=400,
+                    )
+
+        # Publish ZMQ updates so processing threads pick up changes
+        if body.update_topic and body.update_topic.startswith("config/cameras/"):
+            _, _, camera, field = body.update_topic.split("/")
+            settings = getattr(config.cameras.get(camera, None), field, None)
+
+            if settings is not None:
+                request.app.config_publisher.publish_update(
+                    CameraConfigUpdateTopic(CameraConfigUpdateEnum[field], camera),
+                    settings,
+                )
+
+        return JSONResponse(
+            content={"success": True, "message": "Config applied in-memory"},
+            status_code=200,
+        )
+    except Exception as e:
+        logger.error(f"Error applying config in-memory: {e}")
+        return JSONResponse(
+            content={"success": False, "message": "Error applying config"},
+            status_code=500,
+        )
+
+
 @router.put("/config/set", dependencies=[Depends(require_role(["admin"]))])
 def config_set(request: Request, body: AppConfigSetBody):
     config_file = find_config_file()
+
+    if body.skip_save:
+        return _config_set_in_memory(request, body)
+
     lock = FileLock(f"{config_file}.lock", timeout=5)
 
     try:

frigate/api/debug_replay.py

@@ -0,0 +1,176 @@
+"""Debug replay API endpoints."""
+
+import asyncio
+import logging
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+
+from frigate.api.auth import require_role
+from frigate.api.defs.tags import Tags
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=[Tags.app])
+
+
+class DebugReplayStartBody(BaseModel):
+    """Request body for starting a debug replay session."""
+
+    camera: str = Field(title="Source camera name")
+    start_time: float = Field(title="Start timestamp")
+    end_time: float = Field(title="End timestamp")
+
+
+class DebugReplayStartResponse(BaseModel):
+    """Response for starting a debug replay session."""
+
+    success: bool
+    replay_camera: str
+
+
+class DebugReplayStatusResponse(BaseModel):
+    """Response for debug replay status."""
+
+    active: bool
+    replay_camera: str | None = None
+    source_camera: str | None = None
+    start_time: float | None = None
+    end_time: float | None = None
+    live_ready: bool = False
+
+
+class DebugReplayStopResponse(BaseModel):
+    """Response for stopping a debug replay session."""
+
+    success: bool
+
+
+@router.post(
+    "/debug_replay/start",
+    response_model=DebugReplayStartResponse,
+    dependencies=[Depends(require_role(["admin"]))],
+    summary="Start debug replay",
+    description="Start a debug replay session from camera recordings.",
+)
+async def start_debug_replay(request: Request, body: DebugReplayStartBody):
+    """Start a debug replay session."""
+    replay_manager = request.app.replay_manager
+
+    if replay_manager.active:
+        return JSONResponse(
+            content={
+                "success": False,
+                "message": "A replay session is already active",
+            },
+            status_code=409,
+        )
+
+    try:
+        replay_camera = await asyncio.to_thread(
+            replay_manager.start,
+            source_camera=body.camera,
+            start_ts=body.start_time,
+            end_ts=body.end_time,
+            frigate_config=request.app.frigate_config,
+            config_publisher=request.app.config_publisher,
+        )
+    except ValueError:
+        logger.exception("Invalid parameters for debug replay start request")
+        return JSONResponse(
+            content={
+                "success": False,
+                "message": "Invalid debug replay request parameters",
+            },
+            status_code=400,
+        )
+    except RuntimeError:
+        logger.exception("Error while starting debug replay session")
+        return JSONResponse(
+            content={
+                "success": False,
+                "message": "An internal error occurred while starting debug replay",
+            },
+            status_code=500,
+        )
+
+    return DebugReplayStartResponse(
+        success=True,
+        replay_camera=replay_camera,
+    )
+
+
+@router.get(
+    "/debug_replay/status",
+    response_model=DebugReplayStatusResponse,
+    dependencies=[Depends(require_role(["admin"]))],
+    summary="Get debug replay status",
+    description="Get the status of the current debug replay session.",
+)
+def get_debug_replay_status(request: Request):
+    """Get the current replay session status."""
+    replay_manager = request.app.replay_manager
+
+    live_ready = False
+    replay_camera = replay_manager.replay_camera_name
+
+    if replay_manager.active and replay_camera:
+        frame_processor = request.app.detected_frames_processor
+        frame = frame_processor.get_current_frame(replay_camera)
+
+        if frame is not None:
+            frame_time = frame_processor.get_current_frame_time(replay_camera)
+            camera_config = request.app.frigate_config.cameras.get(replay_camera)
+            retry_interval = 10
+
+            if camera_config is not None:
+                retry_interval = float(camera_config.ffmpeg.retry_interval or 10)
+
+            live_ready = datetime.now().timestamp() <= frame_time + retry_interval
+
+    return DebugReplayStatusResponse(
+        active=replay_manager.active,
+        replay_camera=replay_camera,
+        source_camera=replay_manager.source_camera,
+        start_time=replay_manager.start_ts,
+        end_time=replay_manager.end_ts,
+        live_ready=live_ready,
+    )
+
+
+@router.post(
+    "/debug_replay/stop",
+    response_model=DebugReplayStopResponse,
+    dependencies=[Depends(require_role(["admin"]))],
+    summary="Stop debug replay",
+    description="Stop the active debug replay session and clean up all artifacts.",
+)
+async def stop_debug_replay(request: Request):
+    """Stop the active replay session."""
+    replay_manager = request.app.replay_manager
+
+    if not replay_manager.active:
+        return JSONResponse(
+            content={"success": False, "message": "No active replay session"},
+            status_code=400,
+        )
+
+    try:
+        await asyncio.to_thread(
+            replay_manager.stop,
+            frigate_config=request.app.frigate_config,
+            config_publisher=request.app.config_publisher,
+        )
+    except (ValueError, RuntimeError, OSError) as e:
+        logger.error("Error stopping replay: %s", e)
+        return JSONResponse(
+            content={
+                "success": False,
+                "message": "Failed to stop replay session due to an internal error.",
+            },
+            status_code=500,
+        )
+
+    return DebugReplayStopResponse(success=True)

frigate/api/defs/request/app_body.py

@@ -7,6 +7,7 @@ class AppConfigSetBody(BaseModel):
     requires_restart: int = 1
     update_topic: str | None = None
     config_data: Optional[Dict[str, Any]] = None
+    skip_save: bool = False
 
 
 class AppPutPasswordBody(BaseModel):

frigate/api/fastapi_app.py

@@ -18,6 +18,7 @@ from frigate.api import (
     camera,
     chat,
     classification,
+    debug_replay,
     event,
     export,
     media,
@@ -32,6 +33,7 @@ from frigate.comms.event_metadata_updater import (
 )
 from frigate.config import FrigateConfig
 from frigate.config.camera.updater import CameraConfigUpdatePublisher
+from frigate.debug_replay import DebugReplayManager
 from frigate.embeddings import EmbeddingsContext
 from frigate.genai import GenAIClientManager
 from frigate.ptz.onvif import OnvifController
@@ -65,6 +67,7 @@ def create_fastapi_app(
     stats_emitter: StatsEmitter,
     event_metadata_updater: EventMetadataPublisher,
     config_publisher: CameraConfigUpdatePublisher,
+    replay_manager: DebugReplayManager,
     enforce_default_admin: bool = True,
 ):
     logger.info("Starting FastAPI app")
@@ -133,6 +136,7 @@ def create_fastapi_app(
     app.include_router(event.router)
     app.include_router(media.router)
     app.include_router(record.router)
+    app.include_router(debug_replay.router)
     # App Properties
     app.frigate_config = frigate_config
     app.genai_manager = GenAIClientManager(frigate_config)
@@ -144,6 +148,7 @@ def create_fastapi_app(
     app.stats_emitter = stats_emitter
     app.event_metadata_updater = event_metadata_updater
     app.config_publisher = config_publisher
+    app.replay_manager = replay_manager
 
     if frigate_config.auth.enabled:
         secret = get_jwt_secret()