backend

frigate/api/defs/response/export_response.py

@@ -107,6 +107,14 @@ class ExportJobModel(BaseModel):
         default=None,
         description="Result metadata for completed jobs",
     )
+    current_step: str = Field(
+        default="queued",
+        description="Current execution step (queued, preparing, encoding, encoding_retry, finalizing)",
+    )
+    progress_percent: float = Field(
+        default=0.0,
+        description="Progress percentage of the current step (0.0 - 100.0)",
+    )
 
 
 ExportJobsResponse = List[ExportJobModel]

frigate/jobs/export.py

@@ -7,11 +7,13 @@ import time
 from dataclasses import dataclass
 from pathlib import Path
 from queue import Full, Queue
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 
 from peewee import DoesNotExist
 
+from frigate.comms.inter_process import InterProcessRequestor
 from frigate.config import FrigateConfig
+from frigate.const import UPDATE_JOB_STATE
 from frigate.jobs.job import Job
 from frigate.models import Export
 from frigate.record.export import PlaybackSourceEnum, RecordingExporter
@@ -23,6 +25,16 @@ logger = logging.getLogger(__name__)
 # Prevents a runaway client from unbounded memory growth.
 MAX_QUEUED_EXPORT_JOBS = 100
 
+# Minimum interval between progress broadcasts. FFmpeg can emit progress
+# events many times per second; we coalesce them so the WebSocket isn't
+# flooded with redundant updates.
+PROGRESS_BROADCAST_MIN_INTERVAL = 1.0
+
+# Delay before removing a completed job from the in-memory map. Gives the
+# frontend a chance to receive the final state via WebSocket before SWR
+# polling takes over.
+COMPLETED_JOB_CLEANUP_DELAY = 5.0
+
 
 class ExportQueueFullError(RuntimeError):
     """Raised when the export queue is at capacity."""
@@ -43,6 +55,8 @@ class ExportJob(Job):
     ffmpeg_input_args: Optional[str] = None
     ffmpeg_output_args: Optional[str] = None
     cpu_fallback: bool = False
+    current_step: str = "queued"
+    progress_percent: float = 0.0
 
     def to_dict(self) -> dict[str, Any]:
         """Convert to dictionary for API responses.
@@ -64,6 +78,8 @@ class ExportJob(Job):
             "end_time": self.end_time,
             "error_message": self.error_message,
             "results": self.results,
+            "current_step": self.current_step,
+            "progress_percent": self.progress_percent,
         }
 
 
@@ -91,6 +107,38 @@ class ExportQueueWorker(threading.Thread):
                 self.manager.queue.task_done()
 
 
+class JobStatePublisher:
+    """Publishes a single job state payload to the dispatcher.
+
+    Each call opens a short-lived :py:class:`InterProcessRequestor`, sends
+    the payload, and closes the socket. The short-lived design avoids
+    REQ/REP state corruption that would arise from sharing a single REQ
+    socket across the API thread and worker threads (REQ sockets must
+    strictly alternate send/recv).
+
+    With the 1s broadcast throttle in place, socket creation overhead is
+    negligible. The class also exists so tests can substitute a no-op
+    instance instead of stubbing ZMQ — see ``BaseTestHttp.setUp``.
+    """
+
+    def publish(self, payload: dict[str, Any]) -> None:
+        try:
+            requestor = InterProcessRequestor()
+        except Exception as err:
+            logger.warning("Failed to open job state requestor: %s", err)
+            return
+
+        try:
+            requestor.send_data(UPDATE_JOB_STATE, payload)
+        except Exception as err:
+            logger.debug("Job state broadcast failed: %s", err)
+        finally:
+            try:
+                requestor.stop()
+            except Exception:
+                pass
+
+
 class ExportJobManager:
     """Concurrency-limited manager for queued export jobs."""
 
@@ -99,6 +147,7 @@ class ExportJobManager:
         config: FrigateConfig,
         max_concurrent: int,
         max_queued: int = MAX_QUEUED_EXPORT_JOBS,
+        publisher: Optional[JobStatePublisher] = None,
     ) -> None:
         self.config = config
         self.max_concurrent = max(1, max_concurrent)
@@ -107,6 +156,68 @@ class ExportJobManager:
         self.lock = threading.Lock()
         self.workers: list[ExportQueueWorker] = []
         self.started = False
+        self.publisher = publisher if publisher is not None else JobStatePublisher()
+        self._last_broadcast_monotonic: float = 0.0
+        self._broadcast_throttle_lock = threading.Lock()
+
+    def _broadcast_all_jobs(self, force: bool = False) -> None:
+        """Publish aggregate export job state via the job_state WS topic.
+
+        When ``force`` is False, broadcasts within
+        ``PROGRESS_BROADCAST_MIN_INTERVAL`` of the previous one are skipped
+        to avoid flooding the WebSocket with rapid progress updates.
+        ``force`` bypasses the throttle and is used for status transitions
+        (enqueue/start/finish) where the frontend needs the latest state.
+        """
+        now = time.monotonic()
+        with self._broadcast_throttle_lock:
+            if (
+                not force
+                and now - self._last_broadcast_monotonic
+                < PROGRESS_BROADCAST_MIN_INTERVAL
+            ):
+                return
+            self._last_broadcast_monotonic = now
+
+        with self.lock:
+            active = [
+                j
+                for j in self.jobs.values()
+                if j.status in (JobStatusTypesEnum.queued, JobStatusTypesEnum.running)
+            ]
+
+        any_running = any(j.status == JobStatusTypesEnum.running for j in active)
+        payload: dict[str, Any] = {
+            "job_type": "export",
+            "status": "running" if any_running else "queued",
+            "results": {"jobs": [j.to_dict() for j in active]},
+        }
+
+        try:
+            self.publisher.publish(payload)
+        except Exception as err:
+            logger.warning("Publisher raised during job state broadcast: %s", err)
+
+    def _make_progress_callback(self, job: ExportJob) -> Callable[[str, float], None]:
+        """Build a callback the exporter can invoke during execution."""
+
+        def on_progress(step: str, percent: float) -> None:
+            job.current_step = step
+            job.progress_percent = percent
+            self._broadcast_all_jobs()
+
+        return on_progress
+
+    def _schedule_job_cleanup(self, job_id: str) -> None:
+        """Drop a completed job from ``self.jobs`` after a short delay."""
+
+        def cleanup() -> None:
+            with self.lock:
+                self.jobs.pop(job_id, None)
+
+        timer = threading.Timer(COMPLETED_JOB_CLEANUP_DELAY, cleanup)
+        timer.daemon = True
+        timer.start()
 
     def ensure_started(self) -> None:
         """Ensure worker threads are started exactly once."""
@@ -151,6 +262,8 @@ class ExportJobManager:
         with self.lock:
             self.jobs[job.id] = job
 
+        self._broadcast_all_jobs(force=True)
+
         return job.id
 
     def get_job(self, job_id: str) -> Optional[ExportJob]:
@@ -215,6 +328,7 @@ class ExportJobManager:
         """Execute a queued export job."""
         job.status = JobStatusTypesEnum.running
         job.start_time = time.time()
+        self._broadcast_all_jobs(force=True)
 
         exporter = RecordingExporter(
             self.config,
@@ -229,6 +343,7 @@ class ExportJobManager:
             job.ffmpeg_input_args,
             job.ffmpeg_output_args,
             job.cpu_fallback,
+            on_progress=self._make_progress_callback(job),
         )
 
         try:
@@ -257,6 +372,8 @@ class ExportJobManager:
             job.error_message = str(err)
         finally:
             job.end_time = time.time()
+            self._broadcast_all_jobs(force=True)
+            self._schedule_job_cleanup(job.id)
 
 
 _job_manager: Optional[ExportJobManager] = None

frigate/record/export.py

@@ -4,13 +4,14 @@ import datetime
 import logging
 import os
 import random
+import re
 import shutil
 import string
 import subprocess as sp
 import threading
 from enum import Enum
 from pathlib import Path
-from typing import Optional
+from typing import Callable, Optional
 
 from peewee import DoesNotExist
 
@@ -36,6 +37,10 @@ logger = logging.getLogger(__name__)
 DEFAULT_TIME_LAPSE_FFMPEG_ARGS = "-vf setpts=0.04*PTS -r 30"
 TIMELAPSE_DATA_INPUT_ARGS = "-an -skip_frame nokey"
 
+# Matches the setpts factor used in timelapse exports (e.g. setpts=0.04*PTS).
+# Captures the floating-point factor so we can scale expected duration.
+SETPTS_FACTOR_RE = re.compile(r"setpts=([0-9]*\.?[0-9]+)\*PTS")
+
 # ffmpeg flags that can read from or write to arbitrary files
 BLOCKED_FFMPEG_ARGS = frozenset(
     {
@@ -116,6 +121,7 @@ class RecordingExporter(threading.Thread):
         ffmpeg_input_args: Optional[str] = None,
         ffmpeg_output_args: Optional[str] = None,
         cpu_fallback: bool = False,
+        on_progress: Optional[Callable[[str, float], None]] = None,
     ) -> None:
         super().__init__()
         self.config = config
@@ -130,10 +136,213 @@ class RecordingExporter(threading.Thread):
         self.ffmpeg_input_args = ffmpeg_input_args
         self.ffmpeg_output_args = ffmpeg_output_args
         self.cpu_fallback = cpu_fallback
+        self.on_progress = on_progress
 
         # ensure export thumb dir
         Path(os.path.join(CLIPS_DIR, "export")).mkdir(exist_ok=True)
 
+    def _emit_progress(self, step: str, percent: float) -> None:
+        """Invoke the progress callback if one was supplied."""
+        if self.on_progress is None:
+            return
+        try:
+            self.on_progress(step, max(0.0, min(100.0, percent)))
+        except Exception:
+            logger.exception("Export progress callback failed")
+
+    def _expected_output_duration_seconds(self) -> float:
+        """Compute the expected duration of the output video in seconds.
+
+        Users often request a wide time range (e.g. a full hour) when only
+        a few minutes of recordings actually live on disk for that span,
+        so the requested range overstates the work and progress would
+        plateau very early. We sum the actual saved seconds from the
+        Recordings/Previews tables and use that as the input duration.
+        Timelapse exports then scale this by the setpts factor.
+        """
+        requested_duration = max(0.0, float(self.end_time - self.start_time))
+
+        recorded = self._sum_source_duration_seconds()
+        input_duration = (
+            recorded if recorded is not None and recorded > 0 else requested_duration
+        )
+
+        if not self.ffmpeg_output_args:
+            return input_duration
+
+        match = SETPTS_FACTOR_RE.search(self.ffmpeg_output_args)
+        if match is None:
+            return input_duration
+
+        try:
+            factor = float(match.group(1))
+        except ValueError:
+            return input_duration
+
+        if factor <= 0:
+            return input_duration
+
+        return input_duration * factor
+
+    def _sum_source_duration_seconds(self) -> Optional[float]:
+        """Sum saved-video seconds inside [start_time, end_time].
+
+        Queries Recordings or Previews depending on the playback source,
+        clamps each segment to the requested range, and returns the total.
+        Returns ``None`` on any error so the caller can fall back to the
+        requested range duration without losing progress reporting.
+        """
+        try:
+            if self.playback_source == PlaybackSourceEnum.recordings:
+                rows = (
+                    Recordings.select(Recordings.start_time, Recordings.end_time)
+                    .where(
+                        Recordings.start_time.between(self.start_time, self.end_time)
+                        | Recordings.end_time.between(self.start_time, self.end_time)
+                        | (
+                            (self.start_time > Recordings.start_time)
+                            & (self.end_time < Recordings.end_time)
+                        )
+                    )
+                    .where(Recordings.camera == self.camera)
+                    .iterator()
+                )
+            else:
+                rows = (
+                    Previews.select(Previews.start_time, Previews.end_time)
+                    .where(
+                        Previews.start_time.between(self.start_time, self.end_time)
+                        | Previews.end_time.between(self.start_time, self.end_time)
+                        | (
+                            (self.start_time > Previews.start_time)
+                            & (self.end_time < Previews.end_time)
+                        )
+                    )
+                    .where(Previews.camera == self.camera)
+                    .iterator()
+                )
+        except Exception:
+            logger.exception(
+                "Failed to sum source duration for export %s", self.export_id
+            )
+            return None
+
+        total = 0.0
+        try:
+            for row in rows:
+                clipped_start = max(float(row.start_time), float(self.start_time))
+                clipped_end = min(float(row.end_time), float(self.end_time))
+                if clipped_end > clipped_start:
+                    total += clipped_end - clipped_start
+        except Exception:
+            logger.exception(
+                "Failed to read recording rows for export %s", self.export_id
+            )
+            return None
+
+        return total
+
+    def _inject_progress_flags(self, ffmpeg_cmd: list[str]) -> list[str]:
+        """Insert FFmpeg progress reporting flags before the output path.
+
+        ``-progress pipe:2`` writes structured key=value lines to stderr,
+        ``-nostats`` suppresses the noisy default stats output.
+        """
+        if not ffmpeg_cmd:
+            return ffmpeg_cmd
+        return ffmpeg_cmd[:-1] + ["-progress", "pipe:2", "-nostats", ffmpeg_cmd[-1]]
+
+    def _run_ffmpeg_with_progress(
+        self,
+        ffmpeg_cmd: list[str],
+        playlist_lines: str | list[str],
+        step: str = "encoding",
+    ) -> tuple[int, str]:
+        """Run an FFmpeg export command, parsing progress events from stderr.
+
+        Returns ``(returncode, captured_stderr)``. Stdout is left attached to
+        the parent process so we don't have to drain it (and risk a deadlock
+        if the buffer fills). Progress percent is computed against the
+        expected output duration; values are clamped to [0, 100] inside
+        :py:meth:`_emit_progress`.
+        """
+        cmd = ["nice", "-n", str(PROCESS_PRIORITY_LOW)] + self._inject_progress_flags(
+            ffmpeg_cmd
+        )
+
+        if isinstance(playlist_lines, list):
+            stdin_payload = "\n".join(playlist_lines)
+        else:
+            stdin_payload = playlist_lines
+
+        expected_duration = self._expected_output_duration_seconds()
+
+        self._emit_progress(step, 0.0)
+
+        proc = sp.Popen(
+            cmd,
+            stdin=sp.PIPE,
+            stderr=sp.PIPE,
+            text=True,
+            encoding="ascii",
+            errors="replace",
+        )
+
+        assert proc.stdin is not None
+        assert proc.stderr is not None
+
+        try:
+            proc.stdin.write(stdin_payload)
+        except (BrokenPipeError, OSError):
+            # FFmpeg may have rejected the input early; still wait for it
+            # to terminate so the returncode is meaningful.
+            pass
+        finally:
+            try:
+                proc.stdin.close()
+            except (BrokenPipeError, OSError):
+                pass
+
+        captured: list[str] = []
+
+        try:
+            for raw_line in proc.stderr:
+                captured.append(raw_line)
+                line = raw_line.strip()
+
+                if not line:
+                    continue
+
+                if line.startswith("out_time_us="):
+                    if expected_duration <= 0:
+                        continue
+                    try:
+                        out_time_us = int(line.split("=", 1)[1])
+                    except (ValueError, IndexError):
+                        continue
+                    if out_time_us < 0:
+                        continue
+                    out_seconds = out_time_us / 1_000_000.0
+                    percent = (out_seconds / expected_duration) * 100.0
+                    self._emit_progress(step, percent)
+                elif line == "progress=end":
+                    self._emit_progress(step, 100.0)
+                    break
+        except Exception:
+            logger.exception("Failed reading FFmpeg progress for %s", self.export_id)
+
+        proc.wait()
+
+        # Drain any remaining stderr so callers can log it on failure.
+        try:
+            remaining = proc.stderr.read()
+            if remaining:
+                captured.append(remaining)
+        except Exception:
+            pass
+
+        return proc.returncode, "".join(captured)
+
     def get_datetime_from_timestamp(self, timestamp: int) -> str:
         # return in iso format
         return datetime.datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S")
@@ -406,6 +615,7 @@ class RecordingExporter(threading.Thread):
         logger.debug(
             f"Beginning export for {self.camera} from {self.start_time} to {self.end_time}"
         )
+        self._emit_progress("preparing", 0.0)
         export_name = (
             self.user_provided_name
             or f"{self.camera.replace('_', ' ')} {self.get_datetime_from_timestamp(self.start_time)} {self.get_datetime_from_timestamp(self.end_time)}"
@@ -443,16 +653,23 @@ class RecordingExporter(threading.Thread):
         except DoesNotExist:
             return
 
-        p = sp.run(
+        # When neither custom ffmpeg arg is set the default path uses
-            ["nice", "-n", str(PROCESS_PRIORITY_LOW)] + ffmpeg_cmd,
+        # `-c copy` (stream copy — no re-encoding). Report that as a
-            input="\n".join(playlist_lines),
+        # distinct step so the UI doesn't mislabel a remux as encoding.
-            encoding="ascii",
+        # The retry branch below always re-encodes because cpu_fallback
-            capture_output=True,
+        # requires custom args; it stays "encoding_retry".
+        is_stream_copy = (
+            self.ffmpeg_input_args is None and self.ffmpeg_output_args is None
+        )
+        initial_step = "copying" if is_stream_copy else "encoding"
+
+        returncode, stderr = self._run_ffmpeg_with_progress(
+            ffmpeg_cmd, playlist_lines, step=initial_step
         )
 
         # If export failed and cpu_fallback is enabled, retry without hwaccel
         if (
-            p.returncode != 0
+            returncode != 0
             and self.cpu_fallback
             and self.ffmpeg_input_args is not None
             and self.ffmpeg_output_args is not None
@@ -470,23 +687,21 @@ class RecordingExporter(threading.Thread):
                     video_path, use_hwaccel=False
                 )
 
-            p = sp.run(
+            returncode, stderr = self._run_ffmpeg_with_progress(
-                ["nice", "-n", str(PROCESS_PRIORITY_LOW)] + ffmpeg_cmd,
+                ffmpeg_cmd, playlist_lines, step="encoding_retry"
-                input="\n".join(playlist_lines),
-                encoding="ascii",
-                capture_output=True,
             )
 
-        if p.returncode != 0:
+        if returncode != 0:
             logger.error(
                 f"Failed to export {self.playback_source.value} for command {' '.join(ffmpeg_cmd)}"
             )
-            logger.error(p.stderr)
+            logger.error(stderr)
             Path(video_path).unlink(missing_ok=True)
             Export.delete().where(Export.id == self.export_id).execute()
             Path(thumb_path).unlink(missing_ok=True)
             return
         else:
+            self._emit_progress("finalizing", 100.0)
             Export.update({Export.in_progress: False}).where(
                 Export.id == self.export_id
             ).execute()