From dca6f6b3355b4bc799cd17b17267a8bfdf0c6ece Mon Sep 17 00:00:00 2001 From: jetson Date: Thu, 17 Sep 2026 16:00:02 +0700 Subject: [PATCH] =?UTF-8?q?feat(api):=20video=20preview=20endpoints=20?= =?UTF-8?q?=E2=80=94=20thumbnail,=20sample=20frames,=20live=20frame?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app.py | 48 ++++++++++++++++++ src/preview.py | 84 +++++++++++++++++++++++++++++++ tests/test_preview.py | 114 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 246 insertions(+) create mode 100644 src/preview.py create mode 100644 tests/test_preview.py diff --git a/app.py b/app.py index b8626eb..e7bd958 100644 --- a/app.py +++ b/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/") +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//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//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")) diff --git a/src/preview.py b/src/preview.py new file mode 100644 index 0000000..9f4508d --- /dev/null +++ b/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() diff --git a/tests/test_preview.py b/tests/test_preview.py new file mode 100644 index 0000000..6608886 --- /dev/null +++ b/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