Add browser compatible video finalization

Dockerfile

@@ -8,7 +8,7 @@ RUN npm run build && npm prune --omit=dev
 
 FROM python:3.10-slim-bookworm
 ENV NODE_ENV=production PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1 HF_HOME=/home/appuser/.cache/huggingface PORT=7860
-RUN apt-get update && apt-get install -y --no-install-recommends libglib2.0-0 libgl1 \
+RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg libglib2.0-0 libgl1 \
     && rm -rf /var/lib/apt/lists/* \
     && useradd --create-home --uid 1000 appuser
 COPY --from=node-build /usr/local/bin/node /usr/local/bin/node

app.py

@@ -7,6 +7,8 @@ import math
 import os
 import re
 import shutil
+import subprocess
+import threading
 import tempfile
 import time
 from concurrent.futures import ThreadPoolExecutor
@@ -237,6 +239,132 @@ def create_video_poster(source, destination):
         capture.release()
 
 
+def probe_video(path):
+    result = subprocess.run(
+        ["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)],
+        check=True, capture_output=True, text=True, timeout=60,
+    )
+    metadata = json.loads(result.stdout)
+    video = next((stream for stream in metadata.get("streams", []) if stream.get("codec_type") == "video"), None)
+    if not video:
+        raise RuntimeError(f"No video stream found in {path.name}")
+    def numeric(value):
+        try:
+            parsed = float(value)
+            return parsed if math.isfinite(parsed) else 0
+        except (TypeError, ValueError):
+            return 0
+
+    duration = numeric(video.get("duration")) or numeric(metadata.get("format", {}).get("duration"))
+    fps_text = video.get("avg_frame_rate", "0/1")
+    try:
+        numerator, denominator = fps_text.split("/", 1)
+        denominator_value = float(denominator)
+        fps = float(numerator) / denominator_value if denominator_value > 0 else 0
+    except (ValueError, ZeroDivisionError):
+        fps = 0
+    try:
+        total_frames = int(video.get("nb_frames"))
+    except (TypeError, ValueError):
+        total_frames = round(duration * fps)
+    compatible = (
+        Path(path).suffix.lower() == ".mp4"
+        and video.get("codec_name") == "h264"
+        and video.get("pix_fmt") == "yuv420p"
+        and duration > 0
+    )
+    return {"duration": duration, "frames": total_frames, "compatible": compatible}
+
+
+def finalize_videos(work, cameras, job):
+    video_dir = work / "videos"
+    inputs = []
+    for order in cameras:
+        matches = [path for path in video_dir.glob(f"camera_{order}.*") if path.is_file()]
+        if len(matches) != 1:
+            raise RuntimeError(f"Expected one source video for camera {order}")
+        metadata = probe_video(matches[0])
+        inputs.append((order, matches[0], metadata))
+
+    total_duration = sum(metadata["duration"] for _, _, metadata in inputs)
+    if total_duration <= 0:
+        raise RuntimeError("Could not determine video duration for finalization")
+    completed_duration = 0.0
+
+    def report(order, message, local_progress, frames_done=0, frames_total=0):
+        weighted = (completed_duration + inputs_by_order[order]["duration"] * local_progress) / total_duration
+        job.update(
+            status="running", stage="finalizing", progress=90 + 8 * weighted,
+            phase=message, videoProgress=round(local_progress * 100),
+            videoFramesProcessed=frames_done, videoFramesTotal=frames_total,
+            updatedAt=int(time.time() * 1000),
+        )
+
+    inputs_by_order = {order: metadata for order, _, metadata in inputs}
+    for index, (order, source, metadata) in enumerate(inputs, start=1):
+        if job.get("cancel"):
+            raise InterruptedError()
+        frames_total = metadata["frames"]
+        if metadata["compatible"]:
+            report(order, f"Video {index}/{len(inputs)} already browser-ready", 1, frames_total, frames_total)
+            completed_duration += metadata["duration"]
+            continue
+
+        target = video_dir / f"camera_{order}.mp4"
+        temporary = video_dir / f"camera_{order}.finalizing.mp4"
+        temporary.unlink(missing_ok=True)
+        command = [
+            "ffmpeg", "-hide_banner", "-loglevel", "error", "-nostats", "-y", "-i", str(source),
+            "-map", "0:v:0", "-an", "-sn", "-dn", "-vf", "scale=trunc(iw/2)*2:trunc(ih/2)*2",
+            "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-pix_fmt", "yuv420p",
+            "-movflags", "+faststart", "-progress", "pipe:1", str(temporary),
+        ]
+        process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1)
+        stderr_lines = []
+        stderr_reader = threading.Thread(target=lambda: stderr_lines.extend(process.stderr), daemon=True)
+        stderr_reader.start()
+        last_frames = 0
+        try:
+            for line in process.stdout:
+                if job.get("cancel"):
+                    process.terminate()
+                    raise InterruptedError()
+                key, separator, value = line.strip().partition("=")
+                if not separator:
+                    continue
+                if key == "frame":
+                    try:
+                        last_frames = int(value)
+                    except ValueError:
+                        pass
+                elif key == "out_time_us":
+                    try:
+                        local_progress = min(0.99, max(0, int(value) / 1_000_000 / metadata["duration"]))
+                        report(order, f"Preparing video {index}/{len(inputs)}", local_progress, last_frames, frames_total)
+                    except ValueError:
+                        pass
+            return_code = process.wait()
+            stderr_reader.join(timeout=5)
+            if return_code != 0:
+                raise RuntimeError("Video conversion failed: " + "".join(stderr_lines)[-1000:])
+            if not temporary.is_file() or temporary.stat().st_size == 0:
+                raise RuntimeError("Video conversion produced an empty file")
+            verified = probe_video(temporary)
+            if not verified["compatible"]:
+                raise RuntimeError("Converted video failed browser compatibility verification")
+            os.replace(temporary, target)
+            if source != target:
+                source.unlink()
+        except Exception:
+            if process.poll() is None:
+                process.kill()
+                process.wait()
+            temporary.unlink(missing_ok=True)
+            raise
+        report(order, f"Video {index}/{len(inputs)} verified", 1, frames_total, frames_total)
+        completed_duration += metadata["duration"]
+
+
 def run_job(client_id, work, cameras):
     job = jobs[client_id]
     session_id = job["sessionId"]
@@ -260,7 +388,10 @@ def run_job(client_id, work, cameras):
         if job.get("cancel"):
             raise InterruptedError()
         (work / "results.json").write_text(json.dumps(json_value(reports), allow_nan=False), encoding="utf-8")
-        job.update(progress=95, phase="Saving", stage="saving")
+        finalize_videos(work, cameras, job)
+        if job.get("cancel"):
+            raise InterruptedError()
+        job.update(progress=98, phase="Saving results", stage="saving", videoProgress=100)
         hub().upload_folder(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
             folder_path=str(work), path_in_repo=f"sessions/{session_id}",
             allow_patterns=["results.json", "videos/*", "posters/*.webp"])