feat(api): video preview endpoints — thumbnail, sample frames, live frame

app.py

@@ -14,6 +14,7 @@ from werkzeug.utils import secure_filename
 
 from src.job import JobQueue
 from src.model_registry import scan_models
+from src.preview import extract_thumbnail, extract_sample_frames
 
 load_dotenv()
 
@@ -175,6 +176,53 @@ def api_job_detail(job_id):
     })
 
 
+@app.route("/preview/thumb/<job_id>")
+def preview_thumb(job_id):
+    """Serve a thumbnail JPEG for a job's video (cached in job output_dir)."""
+    job = job_queue.get_job(job_id)
+    if job is None:
+        return "Job not found", 404
+    thumb_path = os.path.join(job.output_dir, "thumbnail.jpg")
+    if not os.path.isfile(thumb_path):
+        try:
+            extract_thumbnail(job.video_path, thumb_path)
+        except (FileNotFoundError, RuntimeError):
+            return "Could not generate thumbnail", 404
+    return send_file(thumb_path, mimetype="image/jpeg")
+
+
+@app.route("/api/jobs/<job_id>/samples")
+def api_job_samples(job_id):
+    """Return JSON array of sample frame paths (extract if not cached)."""
+    job = job_queue.get_job(job_id)
+    if job is None:
+        return jsonify({"error": "not found"}), 404
+    samples_dir = os.path.join(job.output_dir, "samples")
+    cache_file = os.path.join(samples_dir, ".cached")
+    if not os.path.isfile(cache_file):
+        try:
+            paths = extract_sample_frames(job.video_path, samples_dir)
+            with open(cache_file, "w") as f:
+                f.write("\n".join(paths))
+        except (FileNotFoundError, RuntimeError):
+            return jsonify({"error": "could not generate samples"}), 404
+    else:
+        with open(cache_file) as f:
+            paths = [line.strip() for line in f if line.strip()]
+    rel_paths = [os.path.relpath(p, job.output_dir) for p in paths]
+    return jsonify(rel_paths)
+
+
+@app.route("/api/jobs/<job_id>/frame")
+def api_job_frame(job_id):
+    """Return current latest_frame as JPEG (live preview during processing)."""
+    job = job_queue.get_job(job_id)
+    if job is None:
+        return "Job not found", 404
+    # latest_frame is not set yet (Task 5 adds it) — return 404 for now
+    return "No frame available", 404
+
+
 def main():
     host = os.getenv("WEB_HOST", "0.0.0.0")
     port = int(os.getenv("WEB_PORT", "9000"))

src/preview.py

@@ -0,0 +1,84 @@
+# src/preview.py
+"""Video preview utilities — probe, thumbnails, sample frames."""
+
+from __future__ import annotations
+
+import os
+
+import cv2
+
+
+def probe_video(video_path: str) -> dict:
+    """Return video metadata: {width, height, fps, duration, frame_count}."""
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        raise FileNotFoundError(f"Cannot open video: {video_path}")
+    try:
+        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        fps = cap.get(cv2.CAP_PROP_FPS)
+        frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+        duration = frame_count / fps if fps > 0 else 0.0
+        return {
+            "width": width,
+            "height": height,
+            "fps": fps,
+            "duration": duration,
+            "frame_count": frame_count,
+        }
+    finally:
+        cap.release()
+
+
+def extract_thumbnail(
+    video_path: str, output_path: str, time_sec: float = 1.0
+) -> str:
+    """Extract a single frame as JPEG. Returns the output_path."""
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        raise FileNotFoundError(f"Cannot open video: {video_path}")
+    try:
+        fps = cap.get(cv2.CAP_PROP_FPS)
+        if fps > 0:
+            frame_idx = int(time_sec * fps)
+        else:
+            frame_idx = 0
+        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
+        ret, frame = cap.read()
+        if not ret:
+            raise RuntimeError(f"Failed to read frame at {time_sec}s")
+        os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
+        cv2.imwrite(output_path, frame)
+        return output_path
+    finally:
+        cap.release()
+
+
+def extract_sample_frames(
+    video_path: str, output_dir: str, count: int = 6
+) -> list[str]:
+    """Extract evenly-spaced frames as JPEGs. Returns list of paths."""
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        raise FileNotFoundError(f"Cannot open video: {video_path}")
+    try:
+        frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+        if frame_count <= 0:
+            return []
+        indices = [
+            int(i * (frame_count - 1) / (count - 1)) if count > 1 else 0
+            for i in range(count)
+        ]
+        os.makedirs(output_dir, exist_ok=True)
+        paths: list[str] = []
+        for idx, frame_idx in enumerate(indices):
+            cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
+            ret, frame = cap.read()
+            if not ret:
+                continue
+            out_path = os.path.join(output_dir, f"sample_{idx:02d}.jpg")
+            cv2.imwrite(out_path, frame)
+            paths.append(out_path)
+        return paths
+    finally:
+        cap.release()

tests/test_preview.py

@@ -0,0 +1,114 @@
+# tests/test_preview.py
+"""Tests for video preview utilities and routes."""
+
+import os
+import tempfile
+
+import cv2
+import numpy as np
+import pytest
+
+from src.preview import probe_video, extract_thumbnail, extract_sample_frames
+
+
+@pytest.fixture
+def sample_video(tmp_path):
+    """Create a small test video (10 frames, 25 fps, 320x240)."""
+    video_path = str(tmp_path / "test.mp4")
+    fourcc = cv2.VideoWriter_fourcc(*"mp4v")
+    writer = cv2.VideoWriter(video_path, fourcc, 25.0, (320, 240))
+    for i in range(10):
+        frame = np.full((240, 320, 3), [(i * 25) % 256, 100, 200], dtype=np.uint8)
+        writer.write(frame)
+    writer.release()
+    return video_path
+
+
+# --- Unit tests for preview.py ---
+
+def test_probe_video(sample_video, tmp_path):
+    info = probe_video(sample_video)
+    assert info["width"] == 320
+    assert info["height"] == 240
+    assert info["fps"] == 25.0
+    assert info["frame_count"] == 10
+    assert info["duration"] == pytest.approx(0.4, abs=0.05)
+
+
+def test_probe_video_not_found():
+    with pytest.raises(FileNotFoundError):
+        probe_video("/nonexistent/video.mp4")
+
+
+def test_extract_thumbnail(sample_video, tmp_path):
+    out = str(tmp_path / "thumb.jpg")
+    result = extract_thumbnail(sample_video, out, time_sec=0.1)
+    assert result == out
+    assert os.path.isfile(out)
+    assert os.path.getsize(out) > 0
+
+
+def test_extract_thumbnail_custom_time(sample_video, tmp_path):
+    out = str(tmp_path / "thumb.jpg")
+    result = extract_thumbnail(sample_video, out, time_sec=0.0)
+    assert os.path.isfile(result)
+
+
+def test_extract_thumbnail_not_found():
+    with pytest.raises(FileNotFoundError):
+        extract_thumbnail("/nonexistent/video.mp4", "/tmp/thumb.jpg")
+
+
+def test_extract_sample_frames(sample_video, tmp_path):
+    out_dir = str(tmp_path / "samples")
+    paths = extract_sample_frames(sample_video, out_dir, count=3)
+    assert len(paths) == 3
+    for p in paths:
+        assert os.path.isfile(p)
+        assert p.endswith(".jpg")
+
+
+def test_extract_sample_frames_default(sample_video, tmp_path):
+    out_dir = str(tmp_path / "samples")
+    paths = extract_sample_frames(sample_video, out_dir)
+    assert len(paths) == 6
+
+
+def test_extract_sample_frames_not_found():
+    with pytest.raises(FileNotFoundError):
+        extract_sample_frames("/nonexistent/video.mp4", "/tmp/samples")
+
+
+# --- Route tests ---
+
+@pytest.fixture
+def client():
+    from app import app
+    app.config["TESTING"] = True
+    with app.test_client() as c:
+        yield c
+
+
+def test_preview_thumb_nonexistent(client):
+    resp = client.get("/preview/thumb/nonexistent")
+    assert resp.status_code == 404
+
+
+def test_api_job_samples_nonexistent(client):
+    resp = client.get("/api/jobs/nonexistent/samples")
+    assert resp.status_code == 404
+    data = resp.get_json()
+    assert "error" in data
+
+
+def test_api_job_frame_nonexistent(client):
+    resp = client.get("/api/jobs/nonexistent/frame")
+    assert resp.status_code == 404
+
+
+def test_api_job_frame_no_latest_frame(client):
+    """Frame route returns 404 when no latest_frame is set (Task 5 TBD)."""
+    from app import job_queue
+    job = job_queue.add_job(video_path="/tmp/test.mp4", model_configs=[])
+    resp = client.get(f"/api/jobs/{job.job_id}/frame")
+    assert resp.status_code == 404