From ed2b48f2f567c449076427b5bc29e94eebcddb42 Mon Sep 17 00:00:00 2001 From: Justin Estrada Date: Wed, 17 Jun 2026 15:46:40 -0400 Subject: [PATCH] Unify frame sampling, add fast mode, and support bundled-app transcription MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace codec-specific I-frame extraction with a single binary-subdivision sampler that works for any codec. Frames are grabbed in coverage-priority order (endpoints → midpoint → quarters → eighths…) and capped by both a max_frames count and a time budget, so faster machines and smaller clips naturally get denser coverage without per-codec special-casing. - core/frames.py: remove ALL_INTRA_CODECS branch; add sample_frames (capped subdivision) and sample_fast_frames (N equidistant frames, no filtering). - core/schemas.py: new settings — max_frames, time_budget, fast_mode, fast_frame_count. - core/transcribe.py: mlx-whisper now falls back to a subprocess using a Python interpreter that has it installed, so transcription works in the packaged app; in-process path retained for running from source. - core/llm.py: thread logging into LLM clients; clearer transcript-summary logs and a retry on first failure. - AutoBin.spec: bundle mlx_whisper + huggingface_hub for the frozen build. - gui/: wire fast mode and the new sampling settings through the settings panel, orchestrator, and workers. .gitignore: ignore *.LRV test footage and node_modules / remotion out dir. Co-Authored-By: Claude Opus 4.8 --- .gitignore | 5 + AutoBin.spec | 20 +- core/frames.py | 437 +++++++++++++++---------------------- core/llm.py | 55 +++-- core/schemas.py | 5 +- core/transcribe.py | 271 +++++++++++++++++++++-- gui/main_window.py | 95 ++++++++- gui/orchestrator.py | 486 +++++++++++++++++++++++++++++++++++++----- gui/settings_panel.py | 41 +++- gui/workers.py | 123 +++++++++-- 10 files changed, 1157 insertions(+), 381 deletions(-) diff --git a/.gitignore b/.gitignore index 8fae348..cc371dd 100644 --- a/.gitignore +++ b/.gitignore @@ -34,10 +34,15 @@ env/ *.avi *.mkv *.mxf +*.LRV *_output/ output_frames/ TEST FOOTAGE/ +# Node (remotion experiments) +node_modules/ +remotion/out/ + # Whisper model cache (downloaded at runtime) whisper_models/ diff --git a/AutoBin.spec b/AutoBin.spec index f00737f..9462ff4 100644 --- a/AutoBin.spec +++ b/AutoBin.spec @@ -25,6 +25,15 @@ pydantic_submodules = collect_submodules("pydantic") # Collect skimage data files skimage_data = collect_data_files("skimage") +# Collect mlx-whisper + mlx (Apple Silicon transcription) +mlx_whisper_submodules = collect_submodules("mlx_whisper") +mlx_submodules = collect_submodules("mlx") +mlx_whisper_data = collect_data_files("mlx_whisper") +mlx_data = collect_data_files("mlx") + +# Collect huggingface_hub (for model downloads) +huggingface_submodules = collect_submodules("huggingface_hub") + a = Analysis( ["main.py"], pathex=[], @@ -32,10 +41,15 @@ a = Analysis( datas=[ *pyside6_data, *skimage_data, + *mlx_whisper_data, + *mlx_data, ], hiddenimports=[ *pyside6_submodules, *pydantic_submodules, + *mlx_whisper_submodules, + *mlx_submodules, + *huggingface_submodules, "core", "core.schemas", "core.frames", @@ -59,6 +73,11 @@ a = Analysis( "requests", "skimage", "skimage.metrics", + "mlx", + "mlx.core", + "mlx.nn", + "mlx_whisper", + "huggingface_hub", ], hookspath=[], hooksconfig={}, @@ -67,7 +86,6 @@ a = Analysis( "tkinter", "matplotlib", "test", - "unittest", "IPython", "jupyter", ], diff --git a/core/frames.py b/core/frames.py index e015667..ff1180a 100644 --- a/core/frames.py +++ b/core/frames.py @@ -1,14 +1,18 @@ """ Frame extraction and filtering pipeline. -Refactored from extract_iframes.py with progress callbacks for GUI integration. + +Unified approach: binary subdivision seeking works for ANY codec. +Frames are grabbed in coverage-priority order (midpoint → quarters → eighths…), +similarity-filtered to drop near-duplicates, and returned at VLM resolution. """ from __future__ import annotations +import collections import glob import json import os -import re +import shutil import subprocess import tempfile import time @@ -18,22 +22,6 @@ import numpy as np -# Codecs where every frame is an I-frame (all-intra / intermediate codecs). -# For these, -skip_frame nokey returns *every* frame, so we sample at -# fixed intervals instead. -ALL_INTRA_CODECS = { - "cfhd", # GoPro CineForm - "prores", # Apple ProRes (all variants) - "dnxhd", "dnxhr",# Avid DNxHD / DNxHR - "mjpeg", # Motion JPEG - "v210", # Uncompressed 10-bit - "rawvideo", # Raw - "huffyuv", # Huffman lossless - "ffv1", # FFV1 lossless - "jpeg2000", # JPEG 2000 -} - - # --------------------------------------------------------------------------- # Video Probing # --------------------------------------------------------------------------- @@ -75,73 +63,24 @@ def get_video_info(video_path: str) -> dict: # --------------------------------------------------------------------------- -# I-Frame Extraction +# Frame Sampling (binary subdivision) # --------------------------------------------------------------------------- -def extract_iframes(video_path: str, temp_dir: str, max_width: int = 640, - log: Callable[[str], None] | None = None) -> list[str]: - """ - Extract I-frames from video into temp_dir, downscaled to max_width for - fast comparison. Returns sorted list of paths. - """ - output_pattern = os.path.join(temp_dir, "iframe_%06d.png") - - cmd = [ - "ffmpeg", - "-hide_banner", - "-loglevel", "warning", - "-skip_frame", "nokey", - "-i", video_path, - "-vsync", "vfr", - "-frame_pts", "1", - "-vf", f"scale='min({max_width},iw)':-2", - "-q:v", "2", - output_pattern, - ] - - if log: - log("[extract] Finding I-frames...") - subprocess.run(cmd, capture_output=True, text=True) +def _build_subdivision_timestamps(duration: float) -> list[float]: + """Build timestamps in binary subdivision order: midpoint, quarters, eighths… - frames = sorted(glob.glob(os.path.join(temp_dir, "iframe_*.png"))) - if log: - log(f"[extract] Found {len(frames)} I-frames") - return frames - - -def sample_frames_by_seeking(video_path: str, temp_dir: str, - duration: float, time_budget: float = 10.0, - max_width: int = 640, - log: Callable[[str], None] | None = None) -> list[str]: - """ - Binary subdivision sampler for all-intra codecs (CineForm, ProRes, etc.). - - Instead of deciding a frame count upfront, this grabs frames in order of - maximum coverage: midpoint first, then quarter-points, then eighths, etc. - It keeps going until the time budget runs out. This way, faster machines - and smaller files naturally get more frames, and you always have the best - possible coverage for the time spent. - - The time_budget (seconds) is configurable in settings. + Returns timestamps ordered by coverage priority — the first N timestamps + always give the best possible spatial coverage across the clip. """ - import collections - deadline = time.time() + time_budget - - if log: - log(f"[extract] All-intra codec — subdividing clip (budget: {time_budget:.0f}s)...") - - # Build the subdivision queue: BFS over midpoints - # Start with the full range, grab its midpoint, split into two halves, repeat queue: collections.deque[tuple[float, float]] = collections.deque() queue.append((0.0, duration)) - # Always grab the very first and last frame first (endpoints) + # Endpoints first (always most useful) seek_order: list[float] = [0.0, max(0.0, duration - 0.5)] while queue: lo, hi = queue.popleft() mid = (lo + hi) / 2.0 - # Skip if this segment is too small to meaningfully subdivide if (hi - lo) < 1.0: continue seek_order.append(mid) @@ -150,28 +89,48 @@ def sample_frames_by_seeking(video_path: str, temp_dir: str, # Deduplicate while preserving order seen: set[float] = set() - unique_order: list[float] = [] + unique: list[float] = [] for ts in seek_order: rounded = round(ts, 2) if rounded not in seen: seen.add(rounded) - unique_order.append(rounded) + unique.append(rounded) + return unique + + +def sample_frames(video_path: str, output_dir: str, duration: float, + max_frames: int = 16, time_budget: float = 10.0, + max_width: int = 640, + log: Callable[[str], None] | None = None) -> list[str]: + """Sample frames via binary subdivision seeking. - # Seek frames until time runs out - grabbed: dict[float, str] = {} # timestamp -> path + Grabs frames in coverage-priority order until either *max_frames* or + *time_budget* (seconds) is reached, whichever comes first. Works for + any codec — no I-frame detection needed. + + Returns paths sorted chronologically. + """ + _log = log or (lambda m: None) + os.makedirs(output_dir, exist_ok=True) + deadline = time.time() + time_budget + + _log(f"[extract] Sampling frames (max {max_frames}, budget {time_budget:.0f}s)...") + + timestamps = _build_subdivision_timestamps(duration) + grabbed: dict[float, str] = {} count = 0 - for ts in unique_order: + for ts in timestamps: + if count >= max_frames: + _log(f"[extract] Frame cap ({max_frames}) reached after {count} frames") + break if time.time() >= deadline: - if log: - log(f"[extract] Time budget reached after {count} frames") + _log(f"[extract] Time budget reached after {count} frames") break - out_path = os.path.join(temp_dir, f"iframe_{count:06d}.png") + out_path = os.path.join(output_dir, f"frame_{count:04d}.png") cmd = [ - "ffmpeg", - "-hide_banner", - "-loglevel", "warning", + "ffmpeg", "-hide_banner", "-loglevel", "warning", "-ss", f"{ts:.3f}", "-i", video_path, "-vf", f"scale='min({max_width},iw)':-2", @@ -180,39 +139,62 @@ def sample_frames_by_seeking(video_path: str, temp_dir: str, out_path, ] subprocess.run(cmd, capture_output=True, text=True) - if os.path.exists(out_path): + if os.path.exists(out_path) and os.path.getsize(out_path) > 0: grabbed[ts] = out_path count += 1 - # Sort by timestamp so downstream filtering gets frames in order - sorted_timestamps = sorted(grabbed.keys()) - frames = [grabbed[ts] for ts in sorted_timestamps] + # Sort chronologically and rename so numbers are sequential + sorted_ts = sorted(grabbed.keys()) + frames: list[str] = [] + for i, ts in enumerate(sorted_ts): + old_path = grabbed[ts] + new_path = os.path.join(output_dir, f"sampled_{i:04d}.png") + os.rename(old_path, new_path) + frames.append(new_path) + + elapsed = time_budget - max(0, deadline - time.time()) + _log(f"[extract] Sampled {len(frames)} frames in {elapsed:.1f}s") + return frames - # Rename in chronological order so frame numbers make sense - renamed: list[str] = [] - for i, path in enumerate(frames): - new_path = os.path.join(temp_dir, f"iframe_{i:06d}_sorted.png") - os.rename(path, new_path) - renamed.append(new_path) - if log: - elapsed = time_budget - max(0, deadline - time.time()) - log(f"[extract] Sampled {len(renamed)} frames in {elapsed:.1f}s") - return renamed +def sample_fast_frames(video_path: str, output_dir: str, duration: float, + n_frames: int = 5, max_width: int = 640, + log: Callable[[str], None] | None = None) -> list[str]: + """Fast mode: grab exactly n_frames equidistant frames. No filtering.""" + _log = log or (lambda m: None) + _log(f"[extract] Fast mode — grabbing {n_frames} equidistant frames...") + os.makedirs(output_dir, exist_ok=True) + frames = [] -def get_frame_num(path: str) -> int: - """Extract frame number from 'iframe_000102.png' or 'iframe_000102_sorted.png' -> 102.""" - match = re.search(r"iframe_(\d+)(?:_sorted)?\.png", os.path.basename(path)) - return int(match.group(1)) if match else 0 + margin = min(1.0, duration * 0.05) + usable = duration - 2 * margin + if usable <= 0: + usable = duration + margin = 0 + for i in range(n_frames): + if n_frames == 1: + ts = duration / 2 + else: + ts = margin + (usable * i / (n_frames - 1)) -def is_dark(img: np.ndarray, threshold: float = 15.0) -> bool: - """Check if an image is dark/black based on mean brightness.""" - if img is None: - return True - gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) - return np.mean(gray) < threshold + out_path = os.path.join(output_dir, f"fast_{i:03d}.jpg") + cmd = [ + "ffmpeg", "-hide_banner", "-loglevel", "warning", + "-ss", f"{ts:.3f}", + "-i", video_path, + "-vf", f"scale='min({max_width},iw)':-2", + "-frames:v", "1", + "-q:v", "2", + out_path, + ] + subprocess.run(cmd, capture_output=True, text=True) + if os.path.exists(out_path): + frames.append(out_path) + + _log(f"[extract] Fast mode: got {len(frames)} frames") + return frames # --------------------------------------------------------------------------- @@ -260,21 +242,21 @@ def _phash(img): # --------------------------------------------------------------------------- -# Filtering & Auto-Tuning +# Similarity Filtering & Auto-Tuning # --------------------------------------------------------------------------- def _precompute_similarities(paths: list[str], metric_fn, - log: Callable[[str], None] | None = None) -> list[tuple[str, float]]: - """ - Read each image once, compute similarity to its predecessor. - Returns [(path, similarity_to_previous), ...] where the first entry - has similarity=0.0 (always kept). + log: Callable[[str], None] | None = None + ) -> list[tuple[str, float]]: + """Read each image once, compute similarity to predecessor. + + Returns [(path, similarity_to_previous), ...] — first entry has sim=0.0. """ if not paths: return [] result: list[tuple[str, float]] = [(paths[0], 0.0)] prev_img = cv2.imread(paths[0]) - for i, path in enumerate(paths[1:], 1): + for path in paths[1:]: curr_img = cv2.imread(path) if curr_img is None: continue @@ -287,34 +269,18 @@ def _precompute_similarities(paths: list[str], metric_fn, def _filter_from_scores(scores: list[tuple[str, float]], threshold: float) -> list[str]: - """ - Filter guide frames using pre-computed similarity scores. - A frame is kept when its similarity to the *last kept frame* is below threshold. + """Filter frames using pre-computed scores. - Because similarities are sequential (each vs its predecessor), we need to - track accumulated similarity — when consecutive frames are all similar, - skipping one means the next comparison is against a frame further back. - We re-simulate the keep/skip logic using the original sequential scores. + Keeps a frame when any frame in the run since the last kept frame had + similarity below threshold (indicating a scene change occurred). """ if not scores: return [] - # We always keep the first frame kept = [scores[0][0]] last_kept_idx = 0 for i in range(1, len(scores)): - # Compute similarity between current frame and last kept frame. - # If frames between last_kept and i were all skipped, we can't use - # the precomputed score directly (it's vs the immediate predecessor). - # However, for the common case the sequential score is a good proxy: - # if sim(i-1, i) is low, sim(last_kept, i) is also likely low. - # For perfect accuracy we'd need all-pairs, but that's O(n²). - # - # Optimization: use the minimum similarity in the run since last kept. - # If any frame in the run had low similarity to its predecessor, - # there was a scene change, so current frame differs from last kept. min_sim_in_run = min(scores[j][1] for j in range(last_kept_idx + 1, i + 1)) - if min_sim_in_run < threshold: kept.append(scores[i][0]) last_kept_idx = i @@ -322,30 +288,23 @@ def _filter_from_scores(scores: list[tuple[str, float]], threshold: float) -> li return kept -def filter_guide_paths(paths: list[str], threshold: float, metric_fn) -> list[str]: - """Keep only unique guide frames where similarity to previous drops below threshold.""" - if not paths: - return [] - kept = [paths[0]] - last_img = cv2.imread(paths[0]) - for path in paths[1:]: - curr_img = cv2.imread(path) - if curr_img is None: - continue - if metric_fn(last_img, curr_img) < threshold: - kept.append(path) - last_img = curr_img +def _filter_guide_frames(scores: list[tuple[str, float]], threshold: float, + log: Callable[[str], None] | None = None) -> list[str]: + """Filter using pre-computed scores and log result.""" + kept = _filter_from_scores(scores, threshold) + if log: + log(f"[filter] {len(kept)} unique frames after similarity filter " + f"(from {len(scores)}, threshold={threshold:.4f})") return kept -def auto_tune_threshold(paths: list[str], metric_fn, target_guides: int, +def auto_tune_threshold(scores: list[tuple[str, float]], target_guides: int, lo: float = 0.60, hi: float = 0.995, iterations: int = 15, log: Callable[[str], None] | None = None) -> float: + """Binary-search for threshold that produces closest to target_guides. + + Uses pre-computed scores (no re-reading images). """ - Binary-search for threshold that produces closest to target_guides. - Pre-computes all pairwise similarities once, then searches over cached scores. - """ - scores = _precompute_similarities(paths, metric_fn, log=log) if not scores: return (lo + hi) / 2 @@ -358,63 +317,10 @@ def auto_tune_threshold(paths: list[str], metric_fn, target_guides: int, else: hi = mid best_t = mid - return best_t - - -# --------------------------------------------------------------------------- -# Target Index Computation -# --------------------------------------------------------------------------- - -def determine_target_indices(guide_paths: list[str], total_frames: int, - offset: int = 10, - log: Callable[[str], None] | None = None) -> list[int]: - """ - For each guide I-frame, produce the frame offset before and after it. - First frame: if dark, only emit +offset. - """ - targets = set() - for i, path in enumerate(guide_paths): - idx = get_frame_num(path) - if i == 0: - img = cv2.imread(path) - if is_dark(img): - if log: - log(f"[logic] First frame ({idx}) is dark — using +{offset} only.") - targets.add(min(total_frames - 1, idx + offset)) - continue - targets.add(max(0, idx - offset)) - targets.add(min(total_frames - 1, idx + offset)) - return sorted(targets) - - -# --------------------------------------------------------------------------- -# Final Frame Extraction -# --------------------------------------------------------------------------- -def extract_final_frames(video_path: str, indices: list[int], output_dir: str, - log: Callable[[str], None] | None = None, - progress: Callable[[int, int], None] | None = None) -> list[str]: - """Seek and save exact frames at full resolution. Returns list of saved paths.""" - os.makedirs(output_dir, exist_ok=True) - cap = cv2.VideoCapture(video_path) - saved = [] - total = len(indices) - - for i, target_idx in enumerate(indices): - cap.set(cv2.CAP_PROP_POS_FRAMES, target_idx) - ret, frame = cap.read() - if not ret: - continue - dst = os.path.join(output_dir, f"frame_{i+1:03d}_orig{target_idx:06d}.jpg") - cv2.imwrite(dst, frame, [cv2.IMWRITE_JPEG_QUALITY, 95]) - saved.append(dst) - if progress: - progress(i + 1, total) - - cap.release() if log: - log(f"[output] Extracted {len(saved)} contextual frames to {output_dir}/") - return saved + log(f"[auto] Selected threshold: {best_t:.4f}") + return best_t # --------------------------------------------------------------------------- @@ -424,13 +330,33 @@ def extract_final_frames(video_path: str, indices: list[int], output_dir: str, def run_frame_pipeline(video_path: str, output_dir: str, threshold: float | None = None, target_fpm: float = 4.0, + max_frames: int = 16, + time_budget: float = 10.0, metric: str = "histogram", - offset: int = 10, log: Callable[[str], None] | None = None, - progress: Callable[[int, int], None] | None = None) -> list[str]: - """ - Full frame extraction pipeline. Returns list of saved frame paths. - If threshold is None, auto-tunes to target_fpm. + progress: Callable[[int, int], None] | None = None, + **_kwargs) -> list[str]: + """Unified frame extraction pipeline. + + 1. Probe video metadata + 2. Sample frames via binary subdivision (any codec) + 3. Similarity-filter to drop near-duplicates + 4. Copy final frames to output_dir + + Parameters + ---------- + max_frames : int + Hard cap on frames to sample (default 16). + time_budget : float + Max seconds to spend sampling (default 10s). + threshold : float or None + Similarity threshold. None → auto-tune to target_fpm. + target_fpm : float + Target frames-per-minute for auto-tuning (default 4.0). + metric : str + Similarity metric: histogram (fast), ssim, or phash. + + Returns list of saved frame paths. """ _log = log or (lambda msg: None) @@ -439,63 +365,50 @@ def run_frame_pipeline(video_path: str, output_dir: str, _log("[error] Could not read video metadata.") return [] - duration_min = info["duration"] / 60.0 + duration = info["duration"] + duration_min = duration / 60.0 _log(f"[info] {info['codec'].upper()} {info['width']}x{info['height']}, " f"{info['bitrate']/1e6:.1f} Mbps, {info['fps']:.0f} fps, " - f"{info['duration']:.1f}s ({duration_min:.1f} min)") - - metric_fn = METRICS[metric] - is_all_intra = info["codec"].lower() in ALL_INTRA_CODECS - - with tempfile.TemporaryDirectory(prefix="iframes_") as tmp: - if is_all_intra: - # All-intra codec: binary subdivision with time budget. - # Grabs midpoint, then quarters, then eighths, etc. - # until the time budget runs out. - iframe_paths = sample_frames_by_seeking( - video_path, tmp, info["duration"], - time_budget=10.0, log=log, - ) - else: - iframe_paths = extract_iframes(video_path, tmp, log=log) + f"{duration:.1f}s ({duration_min:.1f} min)") + + if duration <= 0: + _log("[error] Video has zero duration.") + return [] + + metric_fn = METRICS.get(metric, histogram_similarity) + + with tempfile.TemporaryDirectory(prefix="frames_") as tmp: + # Step 1: Sample frames via binary subdivision + sampled = sample_frames( + video_path, tmp, duration, + max_frames=max_frames, time_budget=time_budget, + log=log, + ) - if not iframe_paths: - _log("[error] No I-frames found.") + if not sampled: + _log("[error] No frames sampled.") return [] - # Auto-tune or use fixed threshold + # Step 2: Compute pairwise similarities (once) + scores = _precompute_similarities(sampled, metric_fn, log=log) + + # Step 3: Auto-tune or use fixed threshold, then filter if threshold is None: target_guides = max(2, int(target_fpm * duration_min / 2)) _log(f"[auto] Tuning for ~{target_fpm} frames/min (~{target_guides} guides)...") - threshold = auto_tune_threshold(iframe_paths, metric_fn, target_guides, log=log) - _log(f"[auto] Selected threshold: {threshold:.4f}") - else: - _log(f"[filter] Using fixed threshold: {threshold}") - - guide_paths = filter_guide_paths(iframe_paths, threshold, metric_fn) - _log(f"[filter] {len(guide_paths)} guide scenes identified.") - - if is_all_intra: - # For all-intra codecs, the sampled frames are already good - # enough for VLM analysis (640px). Copy the guide frames to - # the output dir instead of re-seeking into the huge file, - # which would be just as slow as the initial extraction. - import shutil - os.makedirs(output_dir, exist_ok=True) - saved = [] - for i, path in enumerate(guide_paths): - dst = os.path.join(output_dir, f"frame_{i+1:03d}.jpg") - shutil.copy2(path, dst) - saved.append(dst) - if progress: - progress(i + 1, len(guide_paths)) - _log(f"[output] Copied {len(saved)} guide frames to {output_dir}/") - else: - target_indices = determine_target_indices( - guide_paths, info["total_frames"], offset, log=log - ) - saved = extract_final_frames(video_path, target_indices, output_dir, - log=log, progress=progress) + threshold = auto_tune_threshold(scores, target_guides, log=log) + + guide_paths = _filter_guide_frames(scores, threshold, log=log) + + # Step 4: Copy filtered frames to output dir + os.makedirs(output_dir, exist_ok=True) + saved: list[str] = [] + for i, path in enumerate(guide_paths): + dst = os.path.join(output_dir, f"frame_{i+1:03d}.jpg") + shutil.copy2(path, dst) + saved.append(dst) + if progress: + progress(i + 1, len(guide_paths)) - _log(f"[done] {len(saved)} contextual frames saved ({len(saved)/duration_min:.1f}/min)") + _log(f"[done] {len(saved)} frames saved ({len(saved)/max(duration_min, 0.01):.1f}/min)") return saved diff --git a/core/llm.py b/core/llm.py index a3ff380..097fc1c 100644 --- a/core/llm.py +++ b/core/llm.py @@ -48,12 +48,18 @@ def _strip_schema_meta(schema_dict: dict) -> dict: def _clean_and_parse(content: str, schema: type[BaseModel]) -> BaseModel: """ Try to parse LLM output as JSON. Handles common issues: + - Qwen3 ... reasoning tags - Markdown code fences - Extra wrapping keys (e.g. {"description": {actual data}}) - Non-JSON text mixed in """ content = content.strip() + # Strip ... tags (Qwen3 thinking mode) + # Must happen BEFORE JSON extraction — think blocks often contain + # curly braces that confuse the regex-based JSON finder. + content = re.sub(r"[\s\S]*?", "", content).strip() + # Strip markdown code fences content = re.sub(r"^```(?:json)?\s*\n?", "", content) content = re.sub(r"\n?```\s*$", "", content) @@ -107,10 +113,11 @@ def complete_vision(self, prompt: str, image_paths: list[str], # --------------------------------------------------------------------------- class OllamaClient(LLMClient): - def __init__(self, settings: LLMSettings): + def __init__(self, settings: LLMSettings, log: Callable[[str], None] | None = None): self.model = settings.model self.base_url = settings.base_url.rstrip("/") self.vlm_resolution = settings.vlm_input_resolution + self._log = log or (lambda m: None) def _encode_image(self, path: str) -> str: """Read and optionally downscale image, return base64.""" @@ -143,9 +150,11 @@ def complete_text(self, prompt: str, schema: type[BaseModel]) -> BaseModel: "stream": False, "think": False, } + self._log(f"[ollama] complete_text → {self.model} ({schema.__name__})") resp = requests.post(f"{self.base_url}/api/chat", json=payload, timeout=600) resp.raise_for_status() content = resp.json()["message"]["content"] + self._log(f"[ollama] raw response ({len(content)} chars): {content[:150]!r}") return _clean_and_parse(content, schema) def complete_vision(self, prompt: str, image_paths: list[str], @@ -329,15 +338,16 @@ def complete_vision(self, prompt: str, image_paths: list[str], # Factory # --------------------------------------------------------------------------- -def get_client(settings: LLMSettings) -> LLMClient: +def get_client(settings: LLMSettings, + log: Callable[[str], None] | None = None) -> LLMClient: if settings.backend == "ollama": - return OllamaClient(settings) + return OllamaClient(settings, log=log) elif settings.backend == "openai": return OpenAIClient(settings) elif settings.backend == "anthropic": return AnthropicClient(settings) else: - return OllamaClient(settings) + return OllamaClient(settings, log=log) # --------------------------------------------------------------------------- @@ -461,7 +471,7 @@ def classify_clip(video_path: str, settings: LLMSettings, _log("[classify] Could not sample frames") return None, [] - client = get_client(settings) + client = get_client(settings, log=_log) try: result = client.complete_vision(CLASSIFY_PROMPT, frame_paths, ClipClassification) _log(f"[classify] {result.roll_type.upper()} | {result.shot_type} | " @@ -497,7 +507,7 @@ def refine_classification(transcript: str, classification: ClipClassification, return None _log("[refine] Refining classification with transcript context...") - client = get_client(settings) + client = get_client(settings, log=_log) prompt = REFINE_PROMPT.format( roll_type=classification.roll_type, subject=classification.subject, @@ -527,20 +537,35 @@ def run_llm_pipeline(transcript: str, frame_paths: list[str], Returns (TranscriptSummary, deduplicated_keywords). """ _log = log or (lambda m: None) - client = get_client(settings) + client = get_client(settings, log=_log) # Step 1: Summarize transcript summary = None - if transcript.strip(): - _log("[llm] Step 1: Summarizing transcript...") + transcript_clean = transcript.strip() + if transcript_clean: + _log(f"[llm] Step 1: Summarizing transcript ({len(transcript_clean)} chars, " + f"first 80: {transcript_clean[:80]!r})...") + prompt_text = TRANSCRIPT_PROMPT.format(transcript=transcript_clean[:8000]) + + # Attempt 1 try: - summary = client.complete_text( - TRANSCRIPT_PROMPT.format(transcript=transcript[:8000]), - TranscriptSummary, - ) - _log(f"[llm] Summary: {summary.title}") + summary = client.complete_text(prompt_text, TranscriptSummary) + _log(f"[llm] Summary OK: \"{summary.title}\" | " + f"{len(summary.summary)} char summary | " + f"{len(summary.topics)} topics") except Exception as e: - _log(f"[llm] Transcript summary failed: {e}") + _log(f"[llm] Transcript summary attempt 1 failed: {e}") + + # Attempt 2 — retry once (Ollama can be flaky on first call) + try: + _log("[llm] Retrying summary...") + summary = client.complete_text(prompt_text, TranscriptSummary) + _log(f"[llm] Summary OK (retry): \"{summary.title}\"") + except Exception as e2: + _log(f"[llm] Transcript summary attempt 2 failed: {e2}") + else: + _log(f"[llm] Step 1: Skipping summary — transcript is empty " + f"(raw len={len(transcript)}, stripped len={len(transcript_clean)})") # Step 2: Process image batches all_keywords: list[str] = [] diff --git a/core/schemas.py b/core/schemas.py index 89d8646..7f3041e 100644 --- a/core/schemas.py +++ b/core/schemas.py @@ -13,7 +13,10 @@ class IngestSettings(BaseModel): threshold: float | None = None # None = auto-tune metric: str = "histogram" # histogram | ssim | phash target_fpm: float = 4.0 - offset: int = 10 + max_frames: int = 16 # hard cap on sampled frames per clip + time_budget: float = 10.0 # max seconds to spend sampling frames + fast_mode: bool = False # 5 equidistant frames, tiny whisper, single LLM batch + fast_frame_count: int = 5 # number of frames in fast mode class LLMSettings(BaseModel): diff --git a/core/transcribe.py b/core/transcribe.py index 28e4fbe..264c2c5 100644 --- a/core/transcribe.py +++ b/core/transcribe.py @@ -110,12 +110,19 @@ def check_audio_level(video_path: str, threshold_db: float = -40.0, analyze_duration = min(duration, 1.0) start_time = 0 + # Select best audio stream (avoid ambisonic on .360 files) + stream_map = _find_best_audio_stream(video_path) + # Use ffmpeg astats to get RMS and peak levels cmd = [ "ffmpeg", "-hide_banner", "-loglevel", "error", "-ss", str(start_time), "-i", video_path, "-t", str(analyze_duration), + ] + if stream_map: + cmd += ["-map", stream_map] + cmd += [ "-vn", "-af", "astats=metadata=1:reset=0,ametadata=print:key=lavfi.astats.Overall.RMS_level:key=lavfi.astats.Overall.Peak_level", "-f", "null", "-", ] @@ -152,6 +159,10 @@ def check_audio_level(video_path: str, threshold_db: float = -40.0, "-ss", str(start_time), "-i", video_path, "-t", str(analyze_duration), + ] + if stream_map: + cmd2 += ["-map", stream_map] + cmd2 += [ "-vn", "-af", "volumedetect", "-f", "null", "-", ] @@ -319,8 +330,94 @@ def transcribe(self, audio_path: str, initial_prompt: str | None = None, ... +def _find_python_with_mlx_whisper() -> str | None: + """Find a Python interpreter that has mlx-whisper installed. + + Checks (in order): + 1. .venv next to this source file (running from source) + 2. .venv relative to the .app bundle location (PyInstaller) + 3. The Python that launched this process + 4. Common Homebrew / pyenv / system Python paths + 5. ``python3`` on PATH + """ + import shutil + import sys + + candidates: list[str] = [] + + # 1. Walk up from this source file to find .venv + here = os.path.dirname(os.path.abspath(__file__)) + for _i in range(5): # walk up to 5 levels + venv_py = os.path.join(here, ".venv", "bin", "python") + if os.path.isfile(venv_py): + candidates.append(venv_py) + break + here = os.path.dirname(here) + + # 2. Walk up from sys.executable (PyInstaller .app): + # e.g. dist/AutoBin.app/Contents/MacOS/AutoBin → project root + exe_dir = os.path.dirname(os.path.abspath(sys.executable)) + for _i in range(6): + venv_py = os.path.join(exe_dir, ".venv", "bin", "python") + if os.path.isfile(venv_py): + if venv_py not in candidates: + candidates.append(venv_py) + break + exe_dir = os.path.dirname(exe_dir) + + # 3. The Python that launched this process + if sys.executable and os.path.isfile(sys.executable): + candidates.append(sys.executable) + + # 4. Common locations + home = os.path.expanduser("~") + extra_paths = [ + os.path.join(home, "Documents", "VLM_I_FRAME_EXTRACTOR", ".venv", "bin", "python"), + "/opt/homebrew/bin/python3", + "/usr/local/bin/python3", + ] + # pyenv versions + pyenv_root = os.path.join(home, ".pyenv", "versions") + if os.path.isdir(pyenv_root): + try: + for ver_dir in sorted(os.listdir(pyenv_root), reverse=True): + py = os.path.join(pyenv_root, ver_dir, "bin", "python3") + if os.path.isfile(py): + extra_paths.append(py) + break + except OSError: + pass + + for p in extra_paths: + if os.path.isfile(p) and p not in candidates: + candidates.append(p) + + # 5. System python3 on PATH + sys_py = shutil.which("python3") + if sys_py and sys_py not in candidates: + candidates.append(sys_py) + + for py in candidates: + try: + r = subprocess.run( + [py, "-c", "import mlx_whisper; print('ok')"], + capture_output=True, text=True, timeout=10, + ) + if r.returncode == 0 and "ok" in r.stdout: + return py + except Exception: + continue + return None + + class MLXWhisperBackend(TranscriptionBackend): - """Uses mlx-whisper for Apple Silicon optimized transcription.""" + """Uses mlx-whisper for Apple Silicon optimized transcription. + + First tries an in-process import. If that fails (e.g. when running + inside a PyInstaller .app where native Metal libs can't be bundled), + falls back to running mlx-whisper in a subprocess using whichever + Python on the system has it installed. + """ def __init__(self, model_size: str = "base"): self.model_size = model_size @@ -328,25 +425,38 @@ def __init__(self, model_size: str = "base"): def transcribe(self, audio_path: str, initial_prompt: str | None = None, log: Callable[[str], None] | None = None) -> str: _log = log or (lambda m: None) + + # --- Attempt 1: in-process import (works when running from source) --- try: import mlx_whisper - except ImportError: - _log("[transcribe] mlx-whisper not installed. Run: pip install mlx-whisper") - return "" - - _log(f"[transcribe] MLX-Whisper ({self.model_size}) processing...") + return self._transcribe_in_process(mlx_whisper, audio_path, initial_prompt, _log) + except ImportError as exc: + _log(f"[transcribe] In-process mlx-whisper unavailable ({exc}), trying subprocess...") + except Exception as exc: + _log(f"[transcribe] In-process mlx-whisper failed ({exc}), trying subprocess...") + + # --- Attempt 2: subprocess fallback (for .app bundles) --- + return self._transcribe_subprocess(audio_path, initial_prompt, _log) + + # --------------------------------------------------------------------- # + def _transcribe_in_process(self, mlx_whisper, audio_path: str, + initial_prompt: str | None, + _log: Callable[[str], None]) -> str: + _log(f"[transcribe] MLX-Whisper ({self.model_size}) processing in-process...") if initial_prompt: _log(f"[transcribe] Vocabulary prompt: {initial_prompt[:80]}...") - # Use local model if downloaded, otherwise fall back to HF repo local_path = get_whisper_model_path(self.model_size) if is_whisper_model_downloaded(self.model_size): model_name = local_path _log(f"[transcribe] Using cached model: {local_path}") else: - model_name = MLX_WHISPER_MODELS.get(self.model_size, - f"mlx-community/whisper-{self.model_size}-mlx") - _log(f"[transcribe] Downloading model from HuggingFace (first run)...") + model_name = MLX_WHISPER_MODELS.get( + self.model_size, + f"mlx-community/whisper-{self.model_size}-mlx", + ) + _log("[transcribe] Downloading model from HuggingFace (first run)...") + kwargs = {"path_or_hf_repo": model_name} if initial_prompt: kwargs["initial_prompt"] = initial_prompt @@ -356,6 +466,62 @@ def transcribe(self, audio_path: str, initial_prompt: str | None = None, _log(f"[transcribe] Got {len(text)} chars of transcript") return text + # --------------------------------------------------------------------- # + def _transcribe_subprocess(self, audio_path: str, + initial_prompt: str | None, + _log: Callable[[str], None]) -> str: + py = _find_python_with_mlx_whisper() + if not py: + _log("[transcribe] ✗ No Python with mlx-whisper found. " + "Install it: pip install mlx-whisper") + return "" + _log(f"[transcribe] Using subprocess: {py}") + + local_path = get_whisper_model_path(self.model_size) + if is_whisper_model_downloaded(self.model_size): + model_name = local_path + else: + model_name = MLX_WHISPER_MODELS.get( + self.model_size, + f"mlx-community/whisper-{self.model_size}-mlx", + ) + + # Build a small Python script to run transcription and print the text + script = ( + "import json, sys, mlx_whisper\n" + f"kwargs = {{'path_or_hf_repo': {model_name!r}}}\n" + ) + if initial_prompt: + script += f"kwargs['initial_prompt'] = {initial_prompt!r}\n" + script += ( + f"result = mlx_whisper.transcribe({audio_path!r}, **kwargs)\n" + "print(json.dumps({'text': result.get('text', '')}))\n" + ) + + _log(f"[transcribe] MLX-Whisper ({self.model_size}) via subprocess...") + try: + proc = subprocess.run( + [py, "-c", script], + capture_output=True, text=True, + timeout=600, # 10 min max for long clips + ) + if proc.returncode != 0: + stderr_short = (proc.stderr or "")[:500] + _log(f"[transcribe] Subprocess failed (exit={proc.returncode}): {stderr_short}") + return "" + + import json + data = json.loads(proc.stdout.strip().split("\n")[-1]) + text = data.get("text", "").strip() + _log(f"[transcribe] Got {len(text)} chars of transcript") + return text + except subprocess.TimeoutExpired: + _log("[transcribe] Subprocess timed out (>10min)") + return "" + except Exception as exc: + _log(f"[transcribe] Subprocess error: {exc}") + return "" + class FasterWhisperBackend(TranscriptionBackend): """Uses faster-whisper as CPU fallback.""" @@ -393,19 +559,92 @@ def get_backend(settings: TranscriptionSettings) -> TranscriptionBackend: return MLXWhisperBackend(settings.model_size) -def extract_audio(video_path: str, output_dir: str | None = None) -> str: - """Extract audio from video to a temporary WAV file using ffmpeg.""" +def _find_best_audio_stream(video_path: str) -> str | None: + """Probe for the first stereo/mono AAC or PCM audio stream (skip ambisonic). + + GoPro .360 files have both a normal AAC track and an ambisonic + (4-channel) track. ffmpeg's default stream selection sometimes picks + the ambisonic one, which the resampler can't handle. This function + returns an ffmpeg map specifier like '0:1' for the first usable audio + stream, or None if only one audio stream exists (let ffmpeg decide). + """ + probe_cmd = [ + "ffprobe", "-hide_banner", "-loglevel", "error", + "-show_entries", "stream=index,codec_type,channels,codec_name", + "-select_streams", "a", + "-of", "csv=p=0", + video_path, + ] + try: + proc = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=10) + lines = [l.strip() for l in proc.stdout.strip().split("\n") if l.strip()] + except Exception: + return None + + if len(lines) <= 1: + return None # single audio stream, no need to force-select + + # Output format: index,codec_name,codec_type,channels + # Prefer the first stream with <= 2 channels (stereo or mono) + for line in lines: + parts = line.split(",") + if len(parts) < 4: + continue + idx = parts[0] + channels_str = parts[-1] # channels is always the last column + try: + channels = int(channels_str) + except ValueError: + continue + if channels <= 2: + return f"0:{idx}" + + return None # fallback: let ffmpeg decide + + +def extract_audio(video_path: str, output_dir: str | None = None, + log: Callable[[str], None] | None = None) -> str: + """Extract audio from video to a temporary WAV file using ffmpeg. + + Automatically selects the first stereo/mono audio stream to avoid + ambisonic tracks on multi-stream files (e.g. GoPro .360). + """ + _log = log or (lambda m: None) if output_dir is None: output_dir = tempfile.gettempdir() audio_path = os.path.join(output_dir, "audio.wav") + stream_map = _find_best_audio_stream(video_path) + cmd = [ "ffmpeg", "-hide_banner", "-loglevel", "warning", "-i", video_path, + ] + if stream_map: + cmd += ["-map", stream_map] + _log(f"[transcribe] Selecting audio stream {stream_map} (skipping ambisonic)") + cmd += [ "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", "-y", audio_path, ] - subprocess.run(cmd, capture_output=True, text=True) + proc = subprocess.run(cmd, capture_output=True, text=True) + + # Verify non-empty output + if not os.path.isfile(audio_path) or os.path.getsize(audio_path) == 0: + _log(f"[transcribe] Audio extraction produced empty file (ffmpeg exit={proc.returncode})") + if proc.stderr: + _log(f"[transcribe] ffmpeg: {proc.stderr[:300]}") + # Retry without stream map as fallback + if stream_map: + _log("[transcribe] Retrying without stream selection...") + cmd_fallback = [ + "ffmpeg", "-hide_banner", "-loglevel", "warning", + "-i", video_path, + "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", + "-y", audio_path, + ] + subprocess.run(cmd_fallback, capture_output=True, text=True) + return audio_path @@ -422,9 +661,9 @@ def transcribe_video(video_path: str, settings: TranscriptionSettings, _log(f"[transcribe] Custom vocabulary: {', '.join(vocab)}") with tempfile.TemporaryDirectory(prefix="transcribe_") as tmp: - audio_path = extract_audio(video_path, tmp) - if not os.path.isfile(audio_path): - _log("[transcribe] Failed to extract audio.") + audio_path = extract_audio(video_path, tmp, log=_log) + if not os.path.isfile(audio_path) or os.path.getsize(audio_path) == 0: + _log("[transcribe] Failed to extract audio (missing or empty).") return "" backend = get_backend(settings) diff --git a/gui/main_window.py b/gui/main_window.py index e9fe3a9..0da7ae6 100644 --- a/gui/main_window.py +++ b/gui/main_window.py @@ -143,12 +143,43 @@ def _build_ui(self): self.start_btn.clicked.connect(self._start_processing) btn_row.addWidget(self.start_btn) + self.pause_btn = QPushButton("Pause") + self.pause_btn.setEnabled(False) + self.pause_btn.setToolTip("Pause after current clip finishes") + self.pause_btn.clicked.connect(self._toggle_pause) + btn_row.addWidget(self.pause_btn) + self.stop_btn = QPushButton("Stop") self.stop_btn.setEnabled(False) + self.stop_btn.setToolTip("Stop after current clip finishes (no new clips)") self.stop_btn.clicked.connect(self._stop_processing) btn_row.addWidget(self.stop_btn) + self.cancel_btn = QPushButton("Cancel") + self.cancel_btn.setEnabled(False) + self.cancel_btn.setToolTip("Cancel current clip immediately and skip to next") + self.cancel_btn.setStyleSheet("color: #ff6666;") + self.cancel_btn.clicked.connect(self._cancel_current) + btn_row.addWidget(self.cancel_btn) + left_layout.addLayout(btn_row) + + # Second row: retry failed + btn_row2 = QHBoxLayout() + self.retry_btn = QPushButton("Retry Failed") + self.retry_btn.setEnabled(False) + self.retry_btn.setToolTip("Re-process all clips that failed or timed out") + self.retry_btn.clicked.connect(self._retry_failed) + btn_row2.addWidget(self.retry_btn) + + self.cancel_all_btn = QPushButton("Cancel All") + self.cancel_all_btn.setEnabled(False) + self.cancel_all_btn.setToolTip("Cancel current clip and stop the entire queue") + self.cancel_all_btn.setStyleSheet("color: #ff4444;") + self.cancel_all_btn.clicked.connect(self._cancel_all) + btn_row2.addWidget(self.cancel_all_btn) + + left_layout.addLayout(btn_row2) splitter.addWidget(left) # Center: tabs (progress + settings) @@ -197,6 +228,9 @@ def _connect_orchestrator(self): o.video_error.connect(self._on_video_error) o.multicam_groups_found.connect(self._on_multicam_groups_found) o.queue_completed.connect(self._on_queue_completed) + o.video_failed.connect(self._on_video_failed) + o.paused.connect(self._on_paused) + o.resumed.connect(self._on_resumed) o.queue_progress.connect(self._on_queue_progress) def _ensure_output_folder(self) -> bool: @@ -246,6 +280,11 @@ def _start_processing(self): self.start_btn.setEnabled(False) self.stop_btn.setEnabled(True) + self.cancel_btn.setEnabled(True) + self.cancel_all_btn.setEnabled(True) + self.pause_btn.setEnabled(True) + self.pause_btn.setText("Pause") + self.retry_btn.setEnabled(False) self.tabs.setCurrentWidget(self.progress_panel) self.progress_panel.log("Starting pipeline...") @@ -253,9 +292,48 @@ def _start_processing(self): def _stop_processing(self): self._orchestrator.stop() + self._set_idle_buttons() + self.progress_panel.log("Stopped by user.") + + def _cancel_current(self): + self._orchestrator.cancel_current() + + def _cancel_all(self): + self._orchestrator.cancel_all() + self._set_idle_buttons() + + def _toggle_pause(self): + if self._orchestrator.is_paused: + self._orchestrator.resume() + else: + self._orchestrator.pause() + + def _retry_failed(self): + self.start_btn.setEnabled(False) + self.stop_btn.setEnabled(True) + self.cancel_btn.setEnabled(True) + self.cancel_all_btn.setEnabled(True) + self.pause_btn.setEnabled(True) + self.pause_btn.setText("Pause") + self.retry_btn.setEnabled(False) + self._orchestrator.retry_failed() + + def _set_idle_buttons(self): + """Reset all control buttons to idle state.""" self.start_btn.setEnabled(True) self.stop_btn.setEnabled(False) - self.progress_panel.log("Stopped by user.") + self.cancel_btn.setEnabled(False) + self.cancel_all_btn.setEnabled(False) + self.pause_btn.setEnabled(False) + self.pause_btn.setText("Pause") + # Enable retry if there are failed clips + self.retry_btn.setEnabled(bool(self._orchestrator.get_failed())) + + def _on_paused(self): + self.pause_btn.setText("Resume") + + def _on_resumed(self): + self.pause_btn.setText("Pause") # -- Queue click handler -- @@ -359,6 +437,11 @@ def _on_video_error(self, index: int, msg: str): self.queue_panel.set_item_status(index, "[ERR]") self.progress_panel.log(f"[error] {msg}") + def _on_video_failed(self, index: int, path: str, reason: str): + """A clip was marked FAILED (timeout or cancelled).""" + self.queue_panel.set_item_status(index, "[FAIL]") + self.progress_panel.log(f"[FAIL] {os.path.basename(path)}: {reason}") + def _on_multicam_groups_found(self, groups: list): """Store multicam groups and update the current metadata view.""" self._multicam_groups = groups @@ -379,10 +462,14 @@ def _on_multicam_groups_found(self, groups: list): def _on_queue_completed(self): self._active_index = -1 - self.start_btn.setEnabled(True) - self.stop_btn.setEnabled(False) + self._set_idle_buttons() self.progress_panel.set_complete() - self.progress_panel.log("All videos processed.") + + failed = self._orchestrator.get_failed() + if failed: + self.progress_panel.log(f"All videos processed. {len(failed)} clip(s) FAILED — use Retry Failed to re-process.") + else: + self.progress_panel.log("All videos processed.") # Show timing stats per file type stats = self._orchestrator.get_timing_stats() diff --git a/gui/orchestrator.py b/gui/orchestrator.py index a0ab70d..e18cd0a 100644 --- a/gui/orchestrator.py +++ b/gui/orchestrator.py @@ -7,7 +7,7 @@ import tempfile import time -from PySide6.QtCore import QObject, Signal +from PySide6.QtCore import QObject, QTimer, Signal from core.resolve_export import export_csv, export_combined_csv from core.schemas import ( @@ -18,18 +18,23 @@ AudioCheckWorker, ClipClassificationWorker, ClipRefinementWorker, + FastFrameWorker, FrameExtractionWorker, LLMWorker, MultiCamDetectionWorker, TranscriptionWorker, ) +# Default per-video timeout: 10 minutes +DEFAULT_VIDEO_TIMEOUT_S = 600 + class PipelineOrchestrator(QObject): # Signals for GUI updates log = Signal(str) video_started = Signal(int, str) # queue_index, video_path video_skipped = Signal(int, str) # queue_index, video_path (already done) + video_failed = Signal(int, str, str) # queue_index, video_path, reason frame_progress = Signal(int, int) transcript_progress = Signal(int, int) llm_progress = Signal(int, int) @@ -45,6 +50,8 @@ class PipelineOrchestrator(QObject): multicam_groups_found = Signal(list) # list of MultiCamGroup queue_completed = Signal() queue_progress = Signal(int, int, float) # completed, total, est_remaining_s + paused = Signal() # emitted when pipeline pauses + resumed = Signal() # emitted when pipeline resumes def __init__(self, settings: AppSettings, parent=None): super().__init__(parent) @@ -52,6 +59,14 @@ def __init__(self, settings: AppSettings, parent=None): self._queue: list[str] = [] self._current_index = 0 self._running = False + self._paused = False + self._cancelling = False # True during cancel-current-clip + + # Failed clips: index → reason + self._failed: dict[int, str] = {} + + # Per-video timeout (seconds). 0 = no timeout. + self.video_timeout_s: float = DEFAULT_VIDEO_TIMEOUT_S # Per-video state self._temp_frames_dir: str | None = None @@ -64,6 +79,7 @@ def __init__(self, settings: AppSettings, parent=None): self._frames_done = False self._transcript_done = False self._classify_done = False + self._llm_started = False # Step timing for current video self._step_timing = StepTiming() @@ -83,11 +99,21 @@ def __init__(self, settings: AppSettings, parent=None): # Store completed results for click-to-view self._results: dict[int, VideoResult] = {} + # Per-video watchdog timer (runs on the main/GUI thread) + self._watchdog = QTimer(self) + self._watchdog.setSingleShot(True) + self._watchdog.timeout.connect(self._on_video_timeout) + + # ------------------------------------------------------------------ + # Queue management + # ------------------------------------------------------------------ + def set_queue(self, video_paths: list[str], folder_tags_map: dict[int, list[str]] | None = None): self._queue = list(video_paths) self._current_index = 0 self._results.clear() + self._failed.clear() self._folder_tags_map = folder_tags_map or {} self._completed_count = 0 self._completed_times.clear() @@ -96,31 +122,215 @@ def get_result(self, index: int) -> VideoResult | None: """Get completed result for a queue index (for click-to-view).""" return self._results.get(index) + def get_failed(self) -> dict[int, str]: + """Return dict of failed clip indices → reason strings.""" + return dict(self._failed) + + # ------------------------------------------------------------------ + # Control: start / stop / cancel / pause / resume / retry + # ------------------------------------------------------------------ + def start(self): if not self._queue: return self._running = True + self._paused = False + self._cancelling = False self._current_index = 0 self._completed_count = 0 self._completed_times.clear() + self._failed.clear() self._process_next() def stop(self): + """Stop after the current clip finishes (no new clips started).""" self._running = False + self._paused = False + self._watchdog.stop() + + def cancel_current(self): + """Cancel the current clip immediately → mark FAILED → move on.""" + if not self._running: + return + self._cancelling = True + self._watchdog.stop() + self.log.emit("[cancel] Cancelling current clip...") + self._kill_active_workers() + # Mark as failed and advance + video_path = self._queue[self._current_index] + reason = "Cancelled by user" + self._failed[self._current_index] = reason + self.video_failed.emit(self._current_index, video_path, reason) + self.log.emit(f"[cancel] {os.path.basename(video_path)} marked as FAILED") + self._cleanup_temp() + self._cancelling = False + self._current_index += 1 + self._completed_count += 1 + est = self._estimate_remaining() + self.queue_progress.emit(self._completed_count, len(self._queue), est) + self._process_next() + + def cancel_all(self): + """Cancel the current clip and stop the queue entirely.""" + self._running = False + self._paused = False + self._cancelling = True + self._watchdog.stop() + self.log.emit("[cancel] Cancelling all processing...") + self._kill_active_workers() + # Mark current clip as failed if in progress + if self._current_index < len(self._queue): + video_path = self._queue[self._current_index] + reason = "Cancelled by user (cancel all)" + self._failed[self._current_index] = reason + self.video_failed.emit(self._current_index, video_path, reason) + self._cleanup_temp() + self._cancelling = False + # Emit remaining as failed? No — just stop. User can retry later. + self.queue_completed.emit() + + def pause(self): + """Pause after the current clip finishes.""" + if self._running and not self._paused: + self._paused = True + self.log.emit("[pause] Pipeline will pause after current clip finishes.") + self.paused.emit() + + def resume(self): + """Resume a paused pipeline.""" + if self._paused: + self._paused = False + self.log.emit("[resume] Pipeline resumed.") + self.resumed.emit() + self._process_next() + + def retry_failed(self): + """Re-queue all failed clips and start processing.""" + if not self._failed: + self.log.emit("[retry] No failed clips to retry.") + return + failed_indices = sorted(self._failed.keys()) + self.log.emit(f"[retry] Retrying {len(failed_indices)} failed clip(s)...") + # Remove their CSVs so they don't get skipped + output_folder = self._get_output_folder() + for idx in failed_indices: + video_path = self._queue[idx] + if output_folder: + csv_path = os.path.join(output_folder, f"{self._csv_stem(video_path)}.csv") + if os.path.isfile(csv_path): + os.remove(csv_path) + # Build a new queue of just the failed clips + retry_paths = [self._queue[i] for i in failed_indices] + retry_tags = { + new_i: self._folder_tags_map.get(old_i, []) + for new_i, old_i in enumerate(failed_indices) + } + self._failed.clear() + self.set_queue(retry_paths, retry_tags) + self._running = True + self._paused = False + self._process_next() + + @property + def is_paused(self) -> bool: + return self._paused + + @property + def is_running(self) -> bool: + return self._running + + # ------------------------------------------------------------------ + # Worker termination helpers + # ------------------------------------------------------------------ + + def _kill_active_workers(self): + """Cancel and wait for all active workers (with hard kill fallback).""" + workers = self.get_active_workers() + # First: cooperative cancel + for w in workers: + w.cancel() + # Wait up to 5s for graceful exit + for w in workers: + if w.isRunning(): + w.wait(5000) + # Hard terminate anything still stuck + for w in workers: + if w.isRunning(): + self.log.emit(f"[cancel] Force-terminating {type(w).__name__}") + w.terminate() + w.wait(2000) + + def _cleanup_temp(self): + """Remove temp frames directory for the current clip.""" + if self._temp_frames_dir and os.path.isdir(self._temp_frames_dir): + shutil.rmtree(self._temp_frames_dir, ignore_errors=True) + if self._classify_frame_paths: + tmp_dir = os.path.dirname(self._classify_frame_paths[0]) + if os.path.isdir(tmp_dir) and tmp_dir.startswith(tempfile.gettempdir()): + shutil.rmtree(tmp_dir, ignore_errors=True) + + # ------------------------------------------------------------------ + # Watchdog (per-video timeout) + # ------------------------------------------------------------------ + + def _start_watchdog(self): + """Start the per-video timeout watchdog.""" + if self.video_timeout_s > 0: + self._watchdog.start(int(self.video_timeout_s * 1000)) + + def _on_video_timeout(self): + """Fired when a clip exceeds the timeout. Mark FAILED and skip.""" + if not self._running or self._current_index >= len(self._queue): + return + video_path = self._queue[self._current_index] + elapsed = time.monotonic() - self._video_start_time + reason = f"Timed out after {elapsed:.0f}s (limit: {self.video_timeout_s:.0f}s)" + self.log.emit(f"[timeout] {os.path.basename(video_path)}: {reason}") + self._kill_active_workers() + self._failed[self._current_index] = reason + self.video_failed.emit(self._current_index, video_path, reason) + self._cleanup_temp() + self._current_index += 1 + self._completed_count += 1 + est = self._estimate_remaining() + self.queue_progress.emit(self._completed_count, len(self._queue), est) + self._process_next() + + # ------------------------------------------------------------------ + # CSV helpers + # ------------------------------------------------------------------ def _get_output_folder(self) -> str: """Return the centralized output folder for CSVs.""" return self.settings.export.output_folder + @staticmethod + def _csv_stem(video_path: str) -> str: + """Build a unique CSV stem from a video path, including extension. + + 'GX010024.mov' → 'GX010024_mov' + 'GX010024.360' → 'GX010024_360' + This prevents .mov/.LRV/.360 variants from overwriting each other. + """ + base = os.path.basename(video_path) + name, ext = os.path.splitext(base) + ext_clean = ext.lstrip(".").lower() + if ext_clean: + return f"{name}_{ext_clean}" + return name + def _csv_exists_for(self, video_path: str) -> bool: """Check if a CSV has already been exported for this video.""" output_folder = self._get_output_folder() if not output_folder: return False - video_name = os.path.splitext(os.path.basename(video_path))[0] - csv_path = os.path.join(output_folder, f"{video_name}.csv") + csv_path = os.path.join(output_folder, f"{self._csv_stem(video_path)}.csv") return os.path.isfile(csv_path) + # ------------------------------------------------------------------ + # Overrides + # ------------------------------------------------------------------ + def apply_overrides(self, index: int, overrides: UserOverrides): """Apply user overrides to a completed result and re-export CSV.""" result = self._results.get(index) @@ -188,8 +398,7 @@ def apply_overrides(self, index: int, overrides: UserOverrides): # Re-export CSV output_folder = self._get_output_folder() if output_folder: - video_name = os.path.splitext(os.path.basename(result.video_path))[0] - csv_path = os.path.join(output_folder, f"{video_name}.csv") + csv_path = os.path.join(output_folder, f"{self._csv_stem(result.video_path)}.csv") export_csv(result, csv_path) self.log.emit(f"[export] CSV updated: {csv_path}") @@ -197,6 +406,10 @@ def apply_overrides(self, index: int, overrides: UserOverrides): if len(self._results) > 1: self._export_combined_csv() + # ------------------------------------------------------------------ + # Multi-cam + # ------------------------------------------------------------------ + def _run_multicam_detection(self): """Run multi-cam detection after all clips are processed.""" self.log.emit("\n" + "=" * 50) @@ -229,20 +442,25 @@ def _on_multicam_done(self, groups: list): if output_folder: for result in self._results.values(): if result.multicam_group_id: - video_name = os.path.splitext( - os.path.basename(result.video_path) - )[0] - csv_path = os.path.join(output_folder, f"{video_name}.csv") + csv_path = os.path.join( + output_folder, + f"{self._csv_stem(result.video_path)}.csv", + ) export_csv(result, csv_path) self.multicam_groups_found.emit(groups) - self.queue_completed.emit() - self._running = False + self._finish_queue() def _on_multicam_error(self, msg: str): self.log.emit(f"[warning] Multi-cam detection failed: {msg}") - self.queue_completed.emit() + self._finish_queue() + + def _finish_queue(self): + """Final cleanup when the entire queue is done.""" + self._watchdog.stop() self._running = False + self._paused = False + self.queue_completed.emit() def _export_combined_csv(self): """Write a single combined CSV with all completed results.""" @@ -256,6 +474,10 @@ def _export_combined_csv(self): export_combined_csv(results, combined_path) self.log.emit(f"[export] Combined CSV saved: {combined_path} ({len(results)} clips)") + # ------------------------------------------------------------------ + # ETA + # ------------------------------------------------------------------ + def _estimate_remaining(self) -> float: """Estimate remaining seconds based on average per-video time.""" if not self._completed_times: @@ -264,9 +486,27 @@ def _estimate_remaining(self) -> float: remaining_count = len(self._queue) - self._current_index return avg * remaining_count + # ------------------------------------------------------------------ + # Main pipeline loop + # ------------------------------------------------------------------ + def _process_next(self): + # Pause check — don't start next clip, but don't mark as done + if self._paused and self._running: + self.log.emit("[pause] Pipeline paused. Use Resume to continue.") + return + if not self._running or self._current_index >= len(self._queue): + self._watchdog.stop() self._export_combined_csv() + + # Log failed summary + if self._failed: + self.log.emit(f"\n[summary] {len(self._failed)} clip(s) FAILED:") + for idx, reason in sorted(self._failed.items()): + name = os.path.basename(self._queue[idx]) if idx < len(self._queue) else f"#{idx}" + self.log.emit(f" - {name}: {reason}") + # Run multi-cam detection if we have 2+ results with transcripts results_with_transcripts = [ r for r in self._results.values() @@ -275,8 +515,7 @@ def _process_next(self): if len(results_with_transcripts) >= 2: self._run_multicam_detection() else: - self.queue_completed.emit() - self._running = False + self._finish_queue() return video_path = self._queue[self._current_index] @@ -307,39 +546,131 @@ def _process_next(self): self._frames_done = False self._transcript_done = False self._classify_done = False + self._llm_started = False + self._cancelling = False # Reset step timing self._step_timing = StepTiming() self._video_start_time = time.monotonic() + # Start per-video watchdog + self._start_watchdog() + # Temp dir for frames (cleaned up after LLM processing) self._temp_frames_dir = tempfile.mkdtemp(prefix="framex_") - # Start audio check + classification in parallel (both are fast) - self._classify_start = time.monotonic() - self._classify_worker = ClipClassificationWorker( - video_path, self.settings, keep_frames=True - ) - self._classify_worker.log.connect(self.log.emit) - self._classify_worker.finished.connect(self._on_classify_done) - self._classify_worker.error.connect(self._on_classify_error) - self._classify_worker.start() - - if self.settings.transcription.audio_check: - self._audio_check_start = time.monotonic() - self._audio_worker = AudioCheckWorker(video_path, self.settings) - self._audio_worker.log.connect(self.log.emit) - self._audio_worker.finished.connect(self._on_audio_check_done) - self._audio_worker.error.connect(self._on_audio_check_error) - self._audio_worker.start() - else: - # Skip audio check, go straight to transcription + fast_mode = self.settings.ingest.fast_mode + + if fast_mode: + # --- FAST MODE --- + # Grab 5 equidistant frames (used for classification + keywords) + # Classification runs on those same frames after they arrive + # Transcribe with tiny model in parallel + self.log.emit("[fast] Fast mode enabled — 5 frames, tiny whisper, single LLM batch") + + self._frames_start = time.monotonic() + self._fast_frame_worker = FastFrameWorker( + video_path, + os.path.join(self._temp_frames_dir, "frames"), + n_frames=self.settings.ingest.fast_frame_count, + max_width=self.settings.llm.vlm_input_resolution, + ) + self._fast_frame_worker.log.connect(self.log.emit) + self._fast_frame_worker.finished.connect(self._on_fast_frames_done) + self._fast_frame_worker.error.connect(self._on_frame_error) + self._fast_frame_worker.start() + + # Start transcription in parallel (no audio check in fast mode) self._start_transcription(video_path) + # classify_done stays False — will be set when _on_fast_classify_done fires + # This prevents the LLM from firing before frames + classification are ready + + else: + # --- NORMAL MODE --- + # Start audio check + classification in parallel (both are fast) + self._classify_start = time.monotonic() + self._classify_worker = ClipClassificationWorker( + video_path, self.settings, keep_frames=True + ) + self._classify_worker.log.connect(self.log.emit) + self._classify_worker.finished.connect(self._on_classify_done) + self._classify_worker.error.connect(self._on_classify_error) + self._classify_worker.start() + + if self.settings.transcription.audio_check: + self._audio_check_start = time.monotonic() + self._audio_worker = AudioCheckWorker(video_path, self.settings) + self._audio_worker.log.connect(self.log.emit) + self._audio_worker.finished.connect(self._on_audio_check_done) + self._audio_worker.error.connect(self._on_audio_check_error) + self._audio_worker.start() + else: + # Skip audio check, go straight to transcription + self._start_transcription(video_path) + + # ------------------------------------------------------------------ + # Fast-mode frame/classify handlers + # ------------------------------------------------------------------ + + def _on_fast_frames_done(self, paths: list[str]): + """Fast mode: 5 frames grabbed, used for both classification and keywords.""" + if self._cancelling: + return + self._step_timing.frame_extraction_s = time.monotonic() - self._frames_start + self._frame_paths = paths + self._classify_frame_paths = paths + self._frames_done = True + self.frame_count_update.emit(len(paths)) + self.frames_available.emit(paths) + self.log.emit(f"[fast] Got {len(paths)} frames for full pipeline") + + # In fast mode, run classification on these same frames + if paths: + self._classify_start = time.monotonic() + self._fast_classify_worker = ClipClassificationWorker( + self._queue[self._current_index], self.settings, + keep_frames=False, # we already have frames + ) + self._fast_classify_worker.log.connect(self.log.emit) + self._fast_classify_worker.finished.connect(self._on_fast_classify_done) + self._fast_classify_worker.error.connect(self._on_classify_error) + self._fast_classify_worker.start() + else: + # No frames — mark classify as done so LLM gate opens + self._classify_done = True + self.log.emit("[fast] No frames extracted — skipping classification") + self._check_ready_for_llm() + + def _on_fast_classify_done(self, result, frame_paths: list[str]): + """Fast mode classification done — don't replace our existing frames.""" + if self._cancelling: + return + self._step_timing.classification_s = time.monotonic() - self._classify_start + self._classification = result + self._classify_done = True + self.classification_done.emit(result) + if result: + self.log.emit(f"[fast] Classification: {result.roll_type} | {result.shot_type}") + self._check_ready_for_llm() + + # ------------------------------------------------------------------ + # Normal-mode handlers + # ------------------------------------------------------------------ + def _start_transcription(self, video_path: str): """Launch the transcription worker.""" self._transcribe_start = time.monotonic() - self._transcript_worker = TranscriptionWorker(video_path, self.settings) + + # In fast mode, override to tiny whisper model + settings = self.settings + if settings.ingest.fast_mode: + from copy import deepcopy + settings = deepcopy(self.settings) + settings.transcription.model_size = "tiny" + self.log.emit("[fast] Using whisper-tiny for transcription") + + self._transcript_worker = TranscriptionWorker(video_path, settings) self._transcript_worker.log.connect(self.log.emit) self._transcript_worker.progress.connect(self.transcript_progress.emit) self._transcript_worker.finished.connect(self._on_transcript_done) @@ -347,6 +678,8 @@ def _start_transcription(self, video_path: str): self._transcript_worker.start() def _on_audio_check_done(self, result: dict): + if self._cancelling: + return self._step_timing.audio_check_s = time.monotonic() - self._audio_check_start self._audio_check_result = AudioCheckResult(**result) video_path = self._queue[self._current_index] @@ -354,22 +687,20 @@ def _on_audio_check_done(self, result: dict): if result.get("has_audio", False): self._start_transcription(video_path) else: - # BUG FIX: Always transcribe anyway, just log the warning. - # The audio check being negative doesn't mean the transcript is useless — - # 360 cameras and other sources may have speech that fails the - # speech-ratio check due to ambient noise characteristics. - # We still transcribe and let the downstream length checks - # (len > 20 for refinement, strip() for summary) decide. self.log.emit("[orchestrator] Audio check: low/no speech detected. " "Transcribing anyway (downstream checks will filter).") self._start_transcription(video_path) def _on_audio_check_error(self, msg: str): + if self._cancelling: + return self.log.emit(f"[warning] Audio check failed: {msg} — transcribing anyway") video_path = self._queue[self._current_index] self._start_transcription(video_path) def _on_classify_done(self, result: ClipClassification | None, frame_paths: list[str]): + if self._cancelling: + return self._step_timing.classification_s = time.monotonic() - self._classify_start self._classification = result self._classify_frame_paths = frame_paths @@ -408,6 +739,8 @@ def _on_classify_done(self, result: ClipClassification | None, frame_paths: list self._frame_worker.start() def _on_classify_error(self, msg: str): + if self._cancelling: + return self.log.emit(f"[warning] Classification failed: {msg}") self._classification = None self._classify_done = True @@ -426,6 +759,8 @@ def _on_classify_error(self, msg: str): self._frame_worker.start() def _on_frames_done(self, paths: list[str]): + if self._cancelling: + return self._step_timing.frame_extraction_s = time.monotonic() - self._frames_start self._frame_paths = paths self._frames_done = True @@ -435,12 +770,16 @@ def _on_frames_done(self, paths: list[str]): self._check_ready_for_llm() def _on_frame_error(self, msg: str): + if self._cancelling: + return self.log.emit(f"[error] Frame extraction failed: {msg}") self._frames_done = True self.frame_count_update.emit(0) self._check_ready_for_llm() def _on_transcript_done(self, text: str): + if self._cancelling: + return self._step_timing.transcription_s = time.monotonic() - self._transcribe_start self._transcript = text self._transcript_done = True @@ -449,22 +788,42 @@ def _on_transcript_done(self, text: str): self._check_ready_for_llm() def _on_transcript_error(self, msg: str): + if self._cancelling: + return self.log.emit(f"[warning] Transcription failed: {msg}") self._transcript = "" self._transcript_done = True self.transcript_text_done.emit("") self._check_ready_for_llm() + # ------------------------------------------------------------------ + # LLM gate + # ------------------------------------------------------------------ + def _check_ready_for_llm(self): if not (self._frames_done and self._transcript_done and self._classify_done): + self.log.emit( + f"[orchestrator] LLM gate check: frames={self._frames_done} " + f"transcript={self._transcript_done} classify={self._classify_done} — waiting" + ) return + # Guard: prevent double-fire if multiple signals arrive after all flags are set + if self._llm_started: + self.log.emit("[orchestrator] LLM already started — ignoring duplicate gate trigger") + return + self._llm_started = True + if not self._frame_paths and not self._transcript: self.log.emit("[warning] No frames or transcript — skipping LLM.") self._finalize_video(None, []) return - self.log.emit("[orchestrator] Starting LLM pipeline...") + self.log.emit( + f"[orchestrator] Starting LLM pipeline — " + f"transcript={len(self._transcript)} chars, " + f"frames={len(self._frame_paths)}" + ) self._llm_start = time.monotonic() self._llm_worker = LLMWorker( self._transcript, self._frame_paths, self.settings @@ -475,10 +834,24 @@ def _check_ready_for_llm(self): self._llm_worker.error.connect(self._on_llm_error) self._llm_worker.start() + # ------------------------------------------------------------------ + # LLM + refinement handlers + # ------------------------------------------------------------------ + def _on_llm_done(self, summary, keywords: list[str]): + if self._cancelling: + return self._step_timing.llm_s = time.monotonic() - self._llm_start if summary: self.transcript_summary_done.emit(summary) + self.log.emit(f"[orchestrator] Summary: \"{summary.title}\" | " + f"{len(summary.summary)} char description | " + f"{len(summary.topics)} topics | " + f"{len(keywords)} keywords") + else: + self.log.emit(f"[orchestrator] ⚠ No transcript summary produced! " + f"transcript_len={len(self._transcript)}, " + f"frames={len(self._frame_paths)}") self.keywords_done.emit(keywords) if self._classification and self._transcript and len(self._transcript.strip()) > 20: @@ -500,6 +873,8 @@ def _on_llm_done(self, summary, keywords: list[str]): def _on_refinement_done(self, refinement: ClipRefinement | None, summary: TranscriptSummary | None, keywords: list[str]): + if self._cancelling: + return self._step_timing.refinement_s = time.monotonic() - self._refine_start self._refinement = refinement if refinement: @@ -513,14 +888,23 @@ def _on_refinement_done(self, refinement: ClipRefinement | None, def _on_refinement_error(self, msg: str, summary: TranscriptSummary | None, keywords: list[str]): + if self._cancelling: + return self.log.emit(f"[warning] Refinement failed: {msg}") self._finalize_video(summary, keywords) def _on_llm_error(self, msg: str): + if self._cancelling: + return self.log.emit(f"[error] LLM pipeline failed: {msg}") self._finalize_video(None, []) + # ------------------------------------------------------------------ + # Finalization + # ------------------------------------------------------------------ + def _finalize_video(self, summary: TranscriptSummary | None, keywords: list[str]): + self._watchdog.stop() video_path = self._queue[self._current_index] # Finalize total timing @@ -593,32 +977,28 @@ def _finalize_video(self, summary: TranscriptSummary | None, keywords: list[str] output_folder = self._get_output_folder() if output_folder and (keywords or self._classification): os.makedirs(output_folder, exist_ok=True) - video_name = os.path.splitext(os.path.basename(video_path))[0] - csv_path = os.path.join(output_folder, f"{video_name}.csv") + csv_path = os.path.join(output_folder, f"{self._csv_stem(video_path)}.csv") export_csv(result, csv_path) self.log.emit(f"[export] CSV saved: {csv_path}") # Clean up temp frames - if self._temp_frames_dir and os.path.isdir(self._temp_frames_dir): - shutil.rmtree(self._temp_frames_dir, ignore_errors=True) - - # Clean up classification temp frames - if self._classify_frame_paths: - tmp_dir = os.path.dirname(self._classify_frame_paths[0]) - if os.path.isdir(tmp_dir) and tmp_dir.startswith(tempfile.gettempdir()): - shutil.rmtree(tmp_dir, ignore_errors=True) + self._cleanup_temp() self.video_completed.emit(self._current_index, result) self._current_index += 1 self._process_next() + # ------------------------------------------------------------------ + # Introspection + # ------------------------------------------------------------------ + def get_active_workers(self) -> list: """Return list of currently running QThread workers.""" workers = [] for attr in ("_audio_worker", "_classify_worker", "_transcript_worker", - "_frame_worker", "_llm_worker", "_refine_worker", - "_multicam_worker"): + "_frame_worker", "_fast_frame_worker", "_fast_classify_worker", + "_llm_worker", "_refine_worker", "_multicam_worker"): w = getattr(self, attr, None) if w is not None and w.isRunning(): workers.append(w) diff --git a/gui/settings_panel.py b/gui/settings_panel.py index 2547164..61003d2 100644 --- a/gui/settings_panel.py +++ b/gui/settings_panel.py @@ -4,6 +4,7 @@ from PySide6.QtCore import QThread, Signal from PySide6.QtWidgets import ( + QCheckBox, QComboBox, QDoubleSpinBox, QFileDialog, @@ -70,9 +71,32 @@ def _build_ingest_tab(self): self.metric_combo.addItems(["histogram", "ssim", "phash"]) form.addRow("Metric:", self.metric_combo) - self.offset_spin = QSpinBox() - self.offset_spin.setRange(1, 60) - form.addRow("Frame Offset:", self.offset_spin) + self.max_frames_spin = QSpinBox() + self.max_frames_spin.setRange(4, 64) + self.max_frames_spin.setToolTip("Hard cap on frames sampled per clip") + form.addRow("Max Frames:", self.max_frames_spin) + + self.time_budget_spin = QDoubleSpinBox() + self.time_budget_spin.setRange(2.0, 60.0) + self.time_budget_spin.setSingleStep(1.0) + self.time_budget_spin.setDecimals(0) + self.time_budget_spin.setSuffix("s") + self.time_budget_spin.setToolTip("Max seconds to spend sampling frames per clip") + form.addRow("Time Budget:", self.time_budget_spin) + + # Fast mode toggle + self.fast_mode_cb = QCheckBox("Fast mode — fewer frames, tiny whisper, single LLM batch") + self.fast_mode_cb.setToolTip( + "Grabs equidistant frames (no similarity filtering).\n" + "Uses whisper-tiny for transcription and skips audio check.\n" + "Much faster but slightly less detailed metadata." + ) + form.addRow("Fast Mode:", self.fast_mode_cb) + + self.fast_frame_count_spin = QSpinBox() + self.fast_frame_count_spin.setRange(3, 32) + self.fast_frame_count_spin.setToolTip("Number of equidistant frames to grab in fast mode") + form.addRow("Fast Mode Frames:", self.fast_frame_count_spin) return w @@ -164,7 +188,6 @@ def _build_transcription_tab(self): form.addRow("", vocab_hint) # Audio check settings - from PySide6.QtWidgets import QCheckBox self.audio_check_cb = QCheckBox("Check audio level before transcribing") self.audio_check_cb.setToolTip("Skip transcription on silent/ambient-only clips to avoid hallucinated text") form.addRow("Audio Check:", self.audio_check_cb) @@ -260,7 +283,10 @@ def _load_from_settings(self): self.threshold_spin.setValue(s.ingest.threshold) self.target_fpm_spin.setValue(s.ingest.target_fpm) self.metric_combo.setCurrentText(s.ingest.metric) - self.offset_spin.setValue(s.ingest.offset) + self.max_frames_spin.setValue(s.ingest.max_frames) + self.time_budget_spin.setValue(s.ingest.time_budget) + self.fast_mode_cb.setChecked(s.ingest.fast_mode) + self.fast_frame_count_spin.setValue(s.ingest.fast_frame_count) # LLM self.llm_backend.setCurrentText(s.llm.backend) @@ -294,7 +320,10 @@ def save_to_settings(self) -> AppSettings: s.ingest.threshold = self.threshold_spin.value() s.ingest.target_fpm = self.target_fpm_spin.value() s.ingest.metric = self.metric_combo.currentText() - s.ingest.offset = self.offset_spin.value() + s.ingest.max_frames = self.max_frames_spin.value() + s.ingest.time_budget = self.time_budget_spin.value() + s.ingest.fast_mode = self.fast_mode_cb.isChecked() + s.ingest.fast_frame_count = self.fast_frame_count_spin.value() # LLM s.llm.backend = self.llm_backend.currentText() diff --git a/gui/workers.py b/gui/workers.py index cadb540..a1a25ee 100644 --- a/gui/workers.py +++ b/gui/workers.py @@ -3,17 +3,46 @@ from __future__ import annotations import tempfile +import time from PySide6.QtCore import QThread, Signal -from core.frames import run_frame_pipeline +from core.frames import run_frame_pipeline, sample_fast_frames, get_video_info from core.llm import classify_clip, refine_classification, run_llm_pipeline from core.multicam import find_multicam_groups from core.schemas import AppSettings, ClipClassification, ClipRefinement, TranscriptSummary, VideoResult from core.transcribe import check_audio_level, transcribe_video -class AudioCheckWorker(QThread): +# --------------------------------------------------------------------------- +# Cancellable base class +# --------------------------------------------------------------------------- + +class CancellableWorker(QThread): + """Base QThread with a cooperative cancellation flag. + + Workers should periodically check ``self.is_cancelled`` and exit early + when it returns True. The core functions can't check this directly, + but the long-running subprocess calls already have timeouts, and + :pymethod:`terminate` acts as a hard stop for truly stuck threads. + """ + + _cancelled: bool = False + + def cancel(self): + """Request cooperative cancellation.""" + self._cancelled = True + + @property + def is_cancelled(self): + return self._cancelled + + +# --------------------------------------------------------------------------- +# Workers +# --------------------------------------------------------------------------- + +class AudioCheckWorker(CancellableWorker): """Quick ffmpeg-based audio level check. Runs before transcription.""" log = Signal(str) finished = Signal(dict) # {has_audio, rms_db, peak_db, speech_ratio} @@ -31,12 +60,14 @@ def run(self): threshold_db=self.settings.transcription.noise_floor_db, log=lambda msg: self.log.emit(msg), ) - self.finished.emit(result) + if not self.is_cancelled: + self.finished.emit(result) except Exception as e: - self.error.emit(str(e)) + if not self.is_cancelled: + self.error.emit(str(e)) -class ClipClassificationWorker(QThread): +class ClipClassificationWorker(CancellableWorker): log = Signal(str) finished = Signal(object, list) # (ClipClassification | None, frame_paths) error = Signal(str) @@ -55,12 +86,14 @@ def run(self): log=lambda msg: self.log.emit(msg), keep_frames=self.keep_frames, ) - self.finished.emit(result, frame_paths) + if not self.is_cancelled: + self.finished.emit(result, frame_paths) except Exception as e: - self.error.emit(str(e)) + if not self.is_cancelled: + self.error.emit(str(e)) -class ClipRefinementWorker(QThread): +class ClipRefinementWorker(CancellableWorker): log = Signal(str) finished = Signal(object) # ClipRefinement | None error = Signal(str) @@ -80,12 +113,14 @@ def run(self): self.settings.llm, log=lambda msg: self.log.emit(msg), ) - self.finished.emit(result) + if not self.is_cancelled: + self.finished.emit(result) except Exception as e: - self.error.emit(str(e)) + if not self.is_cancelled: + self.error.emit(str(e)) -class FrameExtractionWorker(QThread): +class FrameExtractionWorker(CancellableWorker): log = Signal(str) progress = Signal(int, int) finished = Signal(list) # list of frame paths @@ -105,17 +140,20 @@ def run(self): output_dir=self.output_dir, threshold=ingest.threshold, target_fpm=ingest.target_fpm, + max_frames=ingest.max_frames, + time_budget=ingest.time_budget, metric=ingest.metric, - offset=ingest.offset, log=lambda msg: self.log.emit(msg), progress=lambda cur, tot: self.progress.emit(cur, tot), ) - self.finished.emit(saved) + if not self.is_cancelled: + self.finished.emit(saved) except Exception as e: - self.error.emit(str(e)) + if not self.is_cancelled: + self.error.emit(str(e)) -class TranscriptionWorker(QThread): +class TranscriptionWorker(CancellableWorker): log = Signal(str) progress = Signal(int, int) finished = Signal(str) # transcript text @@ -137,12 +175,14 @@ def run(self): log=lambda msg: self.log.emit(msg), ) self.progress.emit(1, 1) - self.finished.emit(text) + if not self.is_cancelled: + self.finished.emit(text) except Exception as e: - self.error.emit(str(e)) + if not self.is_cancelled: + self.error.emit(str(e)) -class LLMWorker(QThread): +class LLMWorker(CancellableWorker): log = Signal(str) progress = Signal(int, int) finished = Signal(object, list) # (TranscriptSummary | None, keywords) @@ -163,12 +203,47 @@ def run(self): log=lambda msg: self.log.emit(msg), progress=lambda cur, tot: self.progress.emit(cur, tot), ) - self.finished.emit(summary, keywords) + if not self.is_cancelled: + self.finished.emit(summary, keywords) + except Exception as e: + if not self.is_cancelled: + self.error.emit(str(e)) + + +class FastFrameWorker(CancellableWorker): + """Grab N equidistant frames via seeking. Used in fast mode.""" + log = Signal(str) + finished = Signal(list) # list of frame paths + error = Signal(str) + + def __init__(self, video_path: str, output_dir: str, + n_frames: int = 5, max_width: int = 640): + super().__init__() + self.video_path = video_path + self.output_dir = output_dir + self.n_frames = n_frames + self.max_width = max_width + + def run(self): + try: + info = get_video_info(self.video_path) + duration = info.get("duration", 0) + if duration <= 0: + self.error.emit("Could not determine video duration") + return + frames = sample_fast_frames( + self.video_path, self.output_dir, duration, + n_frames=self.n_frames, max_width=self.max_width, + log=lambda msg: self.log.emit(msg), + ) + if not self.is_cancelled: + self.finished.emit(frames) except Exception as e: - self.error.emit(str(e)) + if not self.is_cancelled: + self.error.emit(str(e)) -class MultiCamDetectionWorker(QThread): +class MultiCamDetectionWorker(CancellableWorker): """Post-queue worker that compares transcripts to find multi-cam groups.""" log = Signal(str) progress = Signal(int, int) @@ -188,6 +263,8 @@ def run(self): log=lambda msg: self.log.emit(msg), progress=lambda cur, tot: self.progress.emit(cur, tot), ) - self.finished.emit(groups) + if not self.is_cancelled: + self.finished.emit(groups) except Exception as e: - self.error.emit(str(e)) + if not self.is_cancelled: + self.error.emit(str(e))