"""Saliency detection and crop path generation for video reframing.""" import shutil import sys from pathlib import Path import cv2 import numpy as np from scipy import ndimage from scipy.interpolate import interp1d from tqdm import tqdm from .config import settings from .ffmpeg_utils import get_cv2_rotation_code, probe_video, transcode_for_cv2 from .models import CropKeyframe, CropPath, Detection, VideoSegment class MediaPipeDetector: """MediaPipe-based face and pose detection for intelligent subject tracking.""" def __init__(self) -> None: """Initialize MediaPipe Face Mesh and Pose detectors.""" try: import mediapipe as mp self.mp_face_mesh = mp.solutions.face_mesh self.mp_pose = mp.solutions.pose # Initialize Face Mesh detector self.face_mesh = self.mp_face_mesh.FaceMesh( max_num_faces=settings.max_faces, refine_landmarks=False, min_detection_confidence=settings.min_face_confidence, min_tracking_confidence=settings.min_tracking_confidence, ) # Initialize Pose detector (fallback for when faces aren't visible) if settings.enable_pose_fallback: self.pose = self.mp_pose.Pose( static_image_mode=False, model_complexity=1, min_detection_confidence=settings.min_face_confidence, min_tracking_confidence=settings.min_tracking_confidence, ) else: self.pose = None self.available = True except ImportError: self.available = False def detect(self, frame: np.ndarray) -> tuple[int | None, int | None, float, str]: """ Detect human subjects in frame using face mesh or pose. Args: frame: Input BGR frame Returns: Tuple of (center_x, center_y, confidence, detection_method) Returns (None, None, 0.0, "none") if no detection """ if not self.available: return None, None, 0.0, "unavailable" frame_height, frame_width = frame.shape[:2] # Convert BGR to RGB (MediaPipe requires RGB) rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # Stage 1: Try Face Mesh detection face_results = self.face_mesh.process(rgb_frame) if face_results.multi_face_landmarks: # Get the best face (most central + largest) best_face = self._select_best_face( face_results.multi_face_landmarks, frame_width, frame_height ) # Get nose tip landmark (landmark #1) for center point nose_tip = best_face.landmark[1] # Convert normalized coordinates to pixel coordinates center_x = int(nose_tip.x * frame_width) center_y = int(nose_tip.y * frame_height) # Bias slightly upward for eye-level framing (40% of face height) # Get face bounding box for height calculation face_landmarks = best_face.landmark min_y = min(lm.y for lm in face_landmarks) max_y = max(lm.y for lm in face_landmarks) face_height = (max_y - min_y) * frame_height # Adjust center_y upward by 40% of face height center_y = int(center_y - face_height * 0.1) # High confidence for face detection confidence = 0.9 return center_x, center_y, confidence, "mediapipe_face" # Stage 2: Fall back to Pose detection if enabled if self.pose is not None: pose_results = self.pose.process(rgb_frame) if pose_results.pose_landmarks: landmarks = pose_results.pose_landmarks.landmark # Use nose landmark (index 0) as primary point nose = landmarks[0] # Check if nose is visible (confidence > threshold) if nose.visibility > settings.min_face_confidence: center_x = int(nose.x * frame_width) center_y = int(nose.y * frame_height) # Medium confidence for pose detection confidence = float(nose.visibility) return center_x, center_y, confidence, "mediapipe_pose" # If nose not visible, try shoulders midpoint left_shoulder = landmarks[11] right_shoulder = landmarks[12] if ( left_shoulder.visibility > settings.min_face_confidence and right_shoulder.visibility > settings.min_face_confidence ): # Calculate midpoint between shoulders center_x = int((left_shoulder.x + right_shoulder.x) / 2 * frame_width) center_y = int((left_shoulder.y + right_shoulder.y) / 2 * frame_height) # Bias upward for upper body framing center_y = int(center_y - frame_height * 0.05) # Medium confidence confidence = (left_shoulder.visibility + right_shoulder.visibility) / 2 return center_x, center_y, confidence, "mediapipe_pose" # No detection return None, None, 0.0, "none" def _select_best_face(self, faces: list, frame_width: int, frame_height: int) -> object: """ Select the best face from multiple detections. Prioritizes faces that are: 1. More central (60% weight) 2. Larger (40% weight) Args: faces: List of face landmark objects frame_width: Frame width in pixels frame_height: Frame height in pixels Returns: Best face landmark object """ if len(faces) == 1: return faces[0] center_x_frame = frame_width / 2 center_y_frame = frame_height / 2 best_face = None best_score = -1.0 for face in faces: # Calculate face center from landmarks landmarks = face.landmark avg_x = sum(lm.x for lm in landmarks) / len(landmarks) avg_y = sum(lm.y for lm in landmarks) / len(landmarks) face_center_x = avg_x * frame_width face_center_y = avg_y * frame_height # Centrality score distance = ( (face_center_x - center_x_frame) ** 2 + (face_center_y - center_y_frame) ** 2 ) ** 0.5 max_distance = (frame_width**2 + frame_height**2) ** 0.5 centrality_score = 1.0 - (distance / max_distance) # Size score (approximate face size from landmark spread) min_x = min(lm.x for lm in landmarks) max_x = max(lm.x for lm in landmarks) min_y = min(lm.y for lm in landmarks) max_y = max(lm.y for lm in landmarks) face_area = (max_x - min_x) * (max_y - min_y) size_score = face_area # Already normalized (0-1 range) # Combined score: 60% centrality, 40% size score = 0.6 * centrality_score + 0.4 * size_score if score > best_score: best_score = score best_face = face return best_face if best_face is not None else faces[0] def close(self) -> None: """Release MediaPipe resources.""" if self.available: if hasattr(self, "face_mesh"): self.face_mesh.close() if hasattr(self, "pose") and self.pose is not None: self.pose.close() class Reframer: """Handles saliency-based video reframing to 9:16 aspect ratio.""" def __init__( self, debug: bool = False, output_dir: Path | None = None, min_scene_detection_segment: float = 2.0, ) -> None: """Initialize reframer with saliency detector. Args: debug: If True, save visualization frames showing crop positions output_dir: Optional output directory where debug frames will also be copied min_scene_detection_segment: Segments shorter than this (seconds) skip PySceneDetect and use a single-pass equilibrium instead. Default 2.0s is suitable for the clip pipeline; cut-to-music passes 5.0s to avoid false scene cuts within short visual-peak clips. """ self.debug = debug self.output_dir = output_dir self.min_scene_detection_segment = min_scene_detection_segment self._last_detection_method = "unknown" self._last_confidence = 0.0 # Clean up root debug_frames folder on each debug run if self.debug: debug_dir = Path("debug_frames") if debug_dir.exists(): shutil.rmtree(debug_dir) debug_dir.mkdir(exist_ok=True) # Try to use PyAutoFlip if available, fallback to simple detection self.use_pyautoflip = False try: import pyautoflip # noqa: F401 self.use_pyautoflip = True except ImportError: pass # Initialize MediaPipe detection backend self.detector = MediaPipeDetector() print("Detection: MediaPipe (face mesh + pose)", file=sys.stderr) # Initialize Haar cascade face detector (fallback if MediaPipe unavailable) try: model_file = cv2.data.haarcascades + "haarcascade_frontalface_default.xml" self.face_cascade = cv2.CascadeClassifier(model_file) except Exception: self.face_cascade = None def _weighted_percentile( self, data: np.ndarray, weights: np.ndarray, percentile: float ) -> float: """Calculate weighted percentile (median when percentile=50).""" sorted_indices = np.argsort(data) sorted_data = data[sorted_indices] sorted_weights = weights[sorted_indices] cumsum = np.cumsum(sorted_weights) total = cumsum[-1] # Find value at percentile target = total * (percentile / 100.0) idx = np.searchsorted(cumsum, target) return float(sorted_data[min(idx, len(sorted_data) - 1)]) def _compute_equilibrium_from_samples( self, positions_x: list[int], positions_y: list[int], confidences: list[float], crop_width: int, crop_height: int, frame_width: int, frame_height: int, ) -> tuple[int, int, dict]: """ Calculate equilibrium position from detection samples. Args: positions_x: List of subject X positions positions_y: List of subject Y positions confidences: List of detection confidences crop_width: Crop window width crop_height: Crop window height frame_width: Video frame width frame_height: Video frame height Returns: (equilibrium_x, equilibrium_y, statistics_dict) """ if not positions_x: # Fallback: use frame center return ( (frame_width - crop_width) // 2, (frame_height - crop_height) // 2, {"method": "fallback_center"}, ) # Calculate equilibrium position using weighted median positions_x_arr = np.array(positions_x) positions_y_arr = np.array(positions_y) weights = np.array(confidences) # Weighted percentile for robustness (less affected by outliers) equilibrium_subject_x = self._weighted_percentile(positions_x_arr, weights, 50) equilibrium_subject_y = self._weighted_percentile(positions_y_arr, weights, 50) # Convert subject position to crop position (top-left corner) equilibrium_x = max( 0, min(int(equilibrium_subject_x - crop_width // 2), frame_width - crop_width) ) equilibrium_y = max( 0, min(int(equilibrium_subject_y - crop_height // 2), frame_height - crop_height) ) # Calculate statistics for debugging stats = { "method": "equilibrium", "num_samples": len(positions_x), "subject_x_min": int(np.min(positions_x_arr)), "subject_x_max": int(np.max(positions_x_arr)), "subject_x_range": int(np.max(positions_x_arr) - np.min(positions_x_arr)), "subject_x_median": int(equilibrium_subject_x), "subject_x_std": float(np.std(positions_x_arr)), "avg_confidence": float(np.mean(weights)), "equilibrium_crop_x": equilibrium_x, "equilibrium_crop_y": equilibrium_y, } return (equilibrium_x, equilibrium_y, stats) def _compute_equilibrium_from_detections( self, detections: list[Detection], sample_interval: float, crop_width: int, crop_height: int, frame_width: int, frame_height: int, ) -> tuple[int, int, dict]: """ Compute equilibrium position from pre-collected detections. Samples detections at regular intervals (every sample_interval seconds) and computes the weighted median position. Args: detections: List of Detection objects sample_interval: Interval for sampling (e.g., 0.5s) crop_width: Crop window width crop_height: Crop window height frame_width: Video frame width frame_height: Video frame height Returns: (equilibrium_x, equilibrium_y, statistics_dict) """ if not detections: return ( (frame_width - crop_width) // 2, (frame_height - crop_height) // 2, {"method": "fallback_center"}, ) # Sample detections at intervals positions_x = [] positions_y = [] confidences = [] last_sample_time = -float('inf') for detection in detections: if detection.center_x is None: continue # Sample at intervals if detection.time - last_sample_time >= sample_interval: positions_x.append(detection.center_x) positions_y.append(detection.center_y) confidences.append(detection.confidence) last_sample_time = detection.time # Use existing equilibrium computation logic return self._compute_equilibrium_from_samples( positions_x, positions_y, confidences, crop_width, crop_height, frame_width, frame_height, ) def _generate_keyframes_from_detections( self, detections: list[Detection], equilibrium_x: int, equilibrium_y: int, crop_width: int, crop_height: int, frame_width: int, frame_height: int, keyframe_interval: float, ) -> list[CropKeyframe]: """ Generate keyframes from pre-collected detections using equilibrium strategy. Args: detections: List of Detection objects for this scene equilibrium_x: Equilibrium crop X position equilibrium_y: Equilibrium crop Y position crop_width: Crop window width crop_height: Crop window height frame_width: Video frame width frame_height: Video frame height keyframe_interval: Interval for keyframes (e.g., 1.0s) Returns: List of CropKeyframe objects """ keyframes = [] if not detections: return keyframes # Current crop position (starts at equilibrium) current_x = equilibrium_x current_y = equilibrium_y # Sample detections at keyframe intervals last_keyframe_time = -float('inf') for detection in detections: if detection.time - last_keyframe_time < keyframe_interval: continue # Apply equilibrium-aware positioning if detection.center_x is not None: # Safe zone is always relative to the equilibrium position (the # intended resting place for this scene), NOT to current_x/y. # Measuring from current_x caused oscillation: once the crop drifted # to follow the subject, the shifted safe zone made the subject look # "safe", triggering an immediate snap back to equilibrium — a jump. safety_margin = settings.equilibrium_safety_margin safe_left = equilibrium_x + int(crop_width * safety_margin) safe_right = equilibrium_x + crop_width - int(crop_width * safety_margin) safe_top = equilibrium_y + int(crop_height * safety_margin) safe_bottom = equilibrium_y + crop_height - int(crop_height * safety_margin) in_safe_zone = ( safe_left <= detection.center_x <= safe_right and safe_top <= detection.center_y <= safe_bottom ) if in_safe_zone: # Subject safe - hold at equilibrium (no panning within scene) target_x = equilibrium_x target_y = equilibrium_y else: # Subject out of safe zone - nudge toward subject ideal_x = detection.center_x - crop_width // 2 ideal_y = detection.center_y - crop_height // 2 max_move = settings.equilibrium_max_movement dx = np.clip(ideal_x - current_x, -max_move, max_move) dy = np.clip(ideal_y - current_y, -max_move, max_move) target_x = current_x + int(dx) target_y = current_y + int(dy) else: # No detection - stay at current position target_x = current_x target_y = current_y # Clamp to valid range target_x = max(0, min(target_x, frame_width - crop_width)) target_y = max(0, min(target_y, frame_height - crop_height)) # Create keyframe keyframes.append(CropKeyframe( time=detection.time, x=target_x, y=target_y, width=crop_width, height=crop_height, confidence=detection.confidence, detection_method=detection.detection_method, scene_change=False, # Will be set by caller for first keyframe )) current_x = target_x current_y = target_y last_keyframe_time = detection.time return keyframes def _split_segment_into_scenes( self, video_path: Path, segment: VideoSegment, fps: float, ) -> list[VideoSegment]: """ Split a video segment into independent scenes using PySceneDetect. Uses AdaptiveDetector (content-aware) by default, with ContentDetector as fallback. If no scenes detected or errors occur, returns the original segment as a single-item list. Args: video_path: Path to video file segment: Original video segment to split fps: Video FPS Returns: List of VideoSegment objects, one per detected scene. Returns [segment] (single-item list) if no scenes detected or on error. """ try: from scenedetect import detect, AdaptiveDetector, ContentDetector start_time = segment.start end_time = segment.end # Skip scene detection for segments below the configured minimum length. if end_time - start_time < self.min_scene_detection_segment: print( f" Segment too short for scene detection " f"({end_time - start_time:.1f}s < " f"{self.min_scene_detection_segment:.1f}s), " f"using single-pass equilibrium", file=sys.stderr ) return [segment] # Choose detector based on configuration if settings.pyscenedetect_detector == "adaptive": detector = AdaptiveDetector( adaptive_threshold=settings.pyscenedetect_adaptive_threshold, min_scene_len=int(settings.pyscenedetect_min_scene_length * fps), luma_only=settings.pyscenedetect_luma_only, ) else: # content detector detector = ContentDetector( threshold=settings.pyscenedetect_content_threshold, min_scene_len=int(settings.pyscenedetect_min_scene_length * fps), luma_only=settings.pyscenedetect_luma_only, ) # Run scene detection scene_list = detect( str(video_path), detector, start_time=start_time, end_time=end_time, ) # Convert PySceneDetect scene list to VideoSegment objects if not scene_list or len(scene_list) == 0: return [segment] # No scenes detected, return original scenes = [] for i, (start_frame, end_frame) in enumerate(scene_list): scene_start = start_frame.get_seconds() scene_end = end_frame.get_seconds() # Ensure scene stays within original segment bounds scene_start = max(scene_start, segment.start) scene_end = min(scene_end, segment.end) # Skip if scene is too short (can happen at boundaries) if scene_end - scene_start < settings.pyscenedetect_min_scene_length: continue scenes.append(VideoSegment( start=scene_start, end=scene_end, score=segment.score, # Inherit from parent segment description=f"{segment.description} (scene {i+1})", )) return scenes if scenes else [segment] except ImportError: print( "Warning: PySceneDetect not available, falling back to single scene", file=sys.stderr ) return [segment] except Exception as e: print( f"Warning: Scene detection failed ({e}), falling back to single scene", file=sys.stderr ) return [segment] def find_scene_boundary_near( self, video_path: Path, timestamp: float, window_before: float = 2.0, window_after: float = 2.0, before_time: float | None = None, ) -> float | None: """ Find the nearest PySceneDetect scene cut within a window around timestamp. Runs detection on a narrow [timestamp-window_before, timestamp+window_after] window — cheap (~4s of video) compared to a full-video scan. Uses the same detector configuration as _split_segment_into_scenes. Args: video_path: Path to video file timestamp: Target time in seconds (e.g. a Gemini segment start) window_before: How many seconds before timestamp to search window_after: How many seconds after timestamp to search Returns: Nearest scene-cut timestamp within the window, or None if no cut found. """ try: from scenedetect import AdaptiveDetector, ContentDetector, detect metadata = probe_video(video_path) fps = metadata.fps search_start = max(0.0, timestamp - window_before) search_end = min(metadata.duration, timestamp + window_after) if search_end - search_start < 1.0: return None # Window too narrow to be meaningful if settings.pyscenedetect_detector == "adaptive": detector = AdaptiveDetector( adaptive_threshold=settings.pyscenedetect_adaptive_threshold, min_scene_len=int(settings.pyscenedetect_min_scene_length * fps), luma_only=settings.pyscenedetect_luma_only, ) else: detector = ContentDetector( threshold=settings.pyscenedetect_content_threshold, min_scene_len=int(settings.pyscenedetect_min_scene_length * fps), luma_only=settings.pyscenedetect_luma_only, ) scene_list = detect( str(video_path), detector, start_time=search_start, end_time=search_end, ) if not scene_list or len(scene_list) < 2: return None # No cut inside the window # Scene boundaries are the start of each scene after the first boundaries = [sf.get_seconds() for sf, _ in scene_list[1:]] if before_time is not None: # Return the latest boundary that precedes before_time — this is # the start of the scene that *contains* before_time (e.g. the # visual peak), avoiding short transitional shots before it. valid = [b for b in boundaries if b <= before_time] return max(valid) if valid else None return min(boundaries, key=lambda b: abs(b - timestamp)) except Exception: return None def _extract_detections_for_segment( self, video_path: Path, segment: VideoSegment, fps: float, ) -> list[Detection]: """ Extract raw detections for entire segment (single video pass). This method reads the video once and runs detection on each frame, returning all detection results with timestamps. This is more efficient than calling _process_segment_unified multiple times for scene-based processing. Args: video_path: Path to video file segment: Video segment to process fps: Video FPS Returns: List of Detection objects with timestamps """ detections = [] cap = cv2.VideoCapture(str(video_path)) if not cap.isOpened(): print(f"Error: Could not open video {video_path}", file=sys.stderr) return detections # Detect rotation so raw OpenCV frames are corrected to display orientation _rotate_code = get_cv2_rotation_code(video_path) # Seek to segment start cap.set(cv2.CAP_PROP_POS_MSEC, segment.start * 1000) frame_count = 0 current_time = segment.start frame_duration = 1.0 / fps while current_time <= segment.end: ret, frame = cap.read() if not ret: break if _rotate_code is not None: frame = cv2.rotate(frame, _rotate_code) # Run detection on this frame center_x, center_y, confidence, detection_method = self.detector.detect(frame) # Fallback detections if primary detector failed if center_x is None and self.face_cascade is not None: center_x, center_y = self._detect_faces(frame) confidence = 0.7 detection_method = "haar_face" if center_x is None: center_x, center_y = self._detect_edges_saliency(frame) confidence = 0.4 detection_method = "edges" # Store detection result detections.append(Detection( time=current_time, center_x=center_x, center_y=center_y, confidence=confidence, detection_method=detection_method, )) frame_count += 1 current_time = segment.start + (frame_count * frame_duration) cap.release() return detections def _process_segment_unified( self, video_path: Path, segment: VideoSegment, crop_width: int, crop_height: int, fps: float, ) -> tuple[list[CropKeyframe], int, int, dict]: """ Process segment in single pass: equilibrium + keyframe extraction. This unified function eliminates redundant video reading and detect() calls by combining equilibrium computation and keyframe extraction into one pass. Strategy: 1. Read each frame once 2. Call detect() once per frame 3. Collect equilibrium samples at intervals (every 0.5s) 4. Use two-phase approach: - Phase 1: Collect samples and initial equilibrium estimate - Phase 2: Apply equilibrium-aware positioning inline Args: video_path: Path to video file segment: Video segment to process crop_width: Width of crop window crop_height: Height of crop window fps: Video FPS Returns: (keyframes, equilibrium_x, equilibrium_y, eq_statistics) """ # Storage for equilibrium computation equilibrium_positions_x = [] equilibrium_positions_y = [] equilibrium_confidences = [] # Storage for keyframes keyframes = [] prev_frame = None prev_confidence = 0.0 last_scene_change_time = -999.0 # Video capture cap = cv2.VideoCapture(str(video_path)) cap.set(cv2.CAP_PROP_POS_MSEC, segment.start * 1000) # Correct for rotation metadata (e.g. iPhone portrait videos stored as landscape) _rotate_code = get_cv2_rotation_code(video_path) frame_count = 0 current_time = segment.start # Intervals equilibrium_interval = int(fps * settings.equilibrium_sample_interval) # Every 0.5s keyframe_interval = int(fps * settings.saliency_keyframe_interval) # Every 1.0s # Get frame dimensions; swap for ±90° rotation so downstream calculations # work in display space, not raw stored pixel space. frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) if _rotate_code in (cv2.ROTATE_90_CLOCKWISE, cv2.ROTATE_90_COUNTERCLOCKWISE): frame_width, frame_height = frame_height, frame_width # Equilibrium state (computed after initial samples) equilibrium_x = None equilibrium_y = None equilibrium_computed = False min_samples_for_equilibrium = 3 # Need at least 3 samples # Crop tracking state current_crop_x = None current_crop_y = None # Calculate total frames for progress bar total_frames = int((segment.end - segment.start) * fps) pbar = tqdm( total=total_frames, desc=" Processing segment (unified)", unit="frame", leave=False, disable=not sys.stderr.isatty(), ) # ═══════════════════════════════════════════════════ # SINGLE-PASS: Read frames once, process everything inline # ═══════════════════════════════════════════════════ while current_time < segment.end: ret, frame = cap.read() if not ret: break if _rotate_code is not None: frame = cv2.rotate(frame, _rotate_code) # ═══════════════════════════════════════════════════ # SINGLE detector.detect() call per frame # ═══════════════════════════════════════════════════ center_x, center_y, confidence, detection_method = self.detector.detect(frame) # Fallback detections if primary detector fails if center_x is None and self.face_cascade is not None: center_x, center_y = self._detect_faces(frame) confidence = 0.7 detection_method = "haar_face" # Collect equilibrium data (every 0.5s) if frame_count % equilibrium_interval == 0 and center_x is not None: equilibrium_positions_x.append(center_x) equilibrium_positions_y.append(center_y) equilibrium_confidences.append(confidence) # Compute equilibrium once we have enough samples if not equilibrium_computed and len(equilibrium_positions_x) >= min_samples_for_equilibrium: equilibrium_x, equilibrium_y, _ = self._compute_equilibrium_from_samples( equilibrium_positions_x, equilibrium_positions_y, equilibrium_confidences, crop_width, crop_height, frame_width, frame_height, ) current_crop_x = equilibrium_x current_crop_y = equilibrium_y equilibrium_computed = True # Extract keyframes (every 1.0s) - only after equilibrium is computed if frame_count % keyframe_interval == 0 and equilibrium_computed: # Scene detection is_scene_change = False if ( settings.enable_scene_detection and prev_frame is not None and current_time - last_scene_change_time > settings.scene_detection_min_interval ): is_scene_change, scene_change_confidence, scene_change_reason = ( self._detect_scene_change(frame, prev_frame, confidence, prev_confidence) ) if is_scene_change: last_scene_change_time = current_time if settings.scene_detection_debug: import logging logger = logging.getLogger(__name__) logger.info( f"Scene change at t={current_time:.2f}s " f"(confidence={scene_change_confidence:.3f}, " f"reason={scene_change_reason})" ) # Apply equilibrium-aware positioning using current detection if center_x is None: # No detection - stay at current position x, y = current_crop_x, current_crop_y self._last_detection_method = "equilibrium_hold" self._last_confidence = 0.2 else: # Safe zone is always relative to equilibrium (not current_crop). # Using current_crop caused oscillation: a drift away from # equilibrium shifted the safe zone, making the subject look # "safe" at the new crop position → immediate snap back to # equilibrium → visible jump on every other keyframe. safety_margin = settings.equilibrium_safety_margin safe_left = equilibrium_x + int(crop_width * safety_margin) safe_right = equilibrium_x + crop_width - int(crop_width * safety_margin) safe_top = equilibrium_y + int(crop_height * safety_margin) safe_bottom = equilibrium_y + crop_height - int(crop_height * safety_margin) in_safe_zone_x = safe_left <= center_x <= safe_right in_safe_zone_y = safe_top <= center_y <= safe_bottom if in_safe_zone_x and in_safe_zone_y: # Subject is safe - hold at equilibrium (no panning within scene) x, y = equilibrium_x, equilibrium_y else: # Subject approaching edge - calculate minimal adjustment desired_x = max(0, min(center_x - crop_width // 2, frame_width - crop_width)) desired_y = max( 0, min(center_y - crop_height // 2, frame_height - crop_height) ) # Move toward desired position with max movement constraint max_movement_eq = settings.equilibrium_max_movement dx = desired_x - current_crop_x dy = desired_y - current_crop_y distance = (dx**2 + dy**2) ** 0.5 if distance > max_movement_eq: scale = max_movement_eq / distance x = int(current_crop_x + dx * scale) y = int(current_crop_y + dy * scale) else: # Apply rubber band effect toward equilibrium equilibrium_weight = settings.equilibrium_rubber_band_strength x = int( desired_x * (1 - equilibrium_weight) + equilibrium_x * equilibrium_weight ) y = int( desired_y * (1 - equilibrium_weight) + equilibrium_y * equilibrium_weight ) self._last_detection_method = f"eq_{detection_method}" self._last_confidence = confidence # Create keyframe keyframe = CropKeyframe( time=current_time, x=x, y=y, width=crop_width, height=crop_height, confidence=self._last_confidence, detection_method=self._last_detection_method, scene_change=is_scene_change, ) keyframes.append(keyframe) # Update state current_crop_x, current_crop_y = x, y prev_confidence = confidence self._is_scene_change = is_scene_change # Debug visualization if self.debug: self._save_debug_frame( frame, x, y, crop_width, crop_height, current_time, equilibrium_x, equilibrium_y ) # Update prev_frame for scene detection (only store for keyframes) if frame_count % keyframe_interval == 0: prev_frame = frame.copy() frame_count += 1 current_time = segment.start + (frame_count / fps) pbar.update(1) pbar.close() cap.release() # Compute final equilibrium from all collected samples equilibrium_x, equilibrium_y, eq_stats = self._compute_equilibrium_from_samples( equilibrium_positions_x, equilibrium_positions_y, equilibrium_confidences, crop_width, crop_height, frame_width, frame_height, ) # Ensure at least one keyframe if not keyframes: keyframes.append( CropKeyframe( time=segment.start, x=equilibrium_x, y=equilibrium_y, width=crop_width, height=crop_height, confidence=0.1, detection_method="fallback_center", ) ) return keyframes, equilibrium_x, equilibrium_y, eq_stats def _detect_scene_change( self, frame_current: np.ndarray, frame_previous: np.ndarray, confidence_current: float, confidence_previous: float, ) -> tuple[bool, float, str]: """ Detect scene change using hybrid multi-signal approach. Returns: Tuple of (is_scene_change, confidence_score, reason) """ # Signal 1: Color histogram comparison (HSV space) hsv_current = cv2.cvtColor(frame_current, cv2.COLOR_BGR2HSV) hsv_previous = cv2.cvtColor(frame_previous, cv2.COLOR_BGR2HSV) # Calculate histograms (8x4x4 bins for speed) hist_current = cv2.calcHist( [hsv_current], [0, 1, 2], None, [8, 4, 4], [0, 180, 0, 256, 0, 256] ) hist_previous = cv2.calcHist( [hsv_previous], [0, 1, 2], None, [8, 4, 4], [0, 180, 0, 256, 0, 256] ) # Normalize cv2.normalize(hist_current, hist_current, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX) cv2.normalize(hist_previous, hist_previous, alpha=0, beta=1, norm_type=cv2.NORM_MINMAX) # Chi-square distance hist_distance = cv2.compareHist(hist_current, hist_previous, cv2.HISTCMP_CHISQR) hist_distance_norm = min(hist_distance / 2.0, 1.0) # Signal 2: Detection confidence drop confidence_drop = abs(confidence_current - confidence_previous) # Signal 3: Pixel difference (downsampled for speed) small_current = cv2.resize(frame_current, (64, 36)) small_previous = cv2.resize(frame_previous, (64, 36)) pixel_diff = ( np.mean(np.abs(small_current.astype(float) - small_previous.astype(float))) / 255.0 ) # Decision logic hist_threshold = settings.scene_histogram_threshold conf_drop_threshold = settings.scene_confidence_drop_threshold pixel_threshold = settings.scene_pixel_diff_threshold # Primary: Strong histogram change if hist_distance_norm > hist_threshold: return True, hist_distance_norm, "histogram" # Secondary: Confidence drop + moderate histogram change if confidence_drop > conf_drop_threshold and hist_distance_norm > hist_threshold * 0.7: return True, confidence_drop, "confidence_drop" # Tertiary: Extreme pixel difference if pixel_diff > pixel_threshold: return True, pixel_diff, "pixel_difference" return False, 0.0, "none" def generate_crop_path( self, video_path: Path, segment: VideoSegment, target_aspect: tuple[int, int] = (9, 16), ) -> CropPath: """ Generate smooth crop path for a video segment. Args: video_path: Path to input video segment: Video segment to process target_aspect: Target aspect ratio (width, height) Returns: CropPath with keyframes for smooth transitions """ # Store segment so _save_debug_frame can annotate type/score/description self._current_segment = segment # Transcode to H.264 if the codec is unsupported by OpenCV's bundled FFmpeg # (e.g. AV1). The system ffmpeg binary has libdav1d; OpenCV's does not. cv2_path, _cv2_tmp = transcode_for_cv2(video_path) try: return self._generate_crop_path_impl(cv2_path, segment, target_aspect) finally: if _cv2_tmp: cv2_path.unlink(missing_ok=True) def _generate_crop_path_impl( self, video_path: Path, segment: VideoSegment, target_aspect: tuple[int, int], ) -> CropPath: """Internal implementation of generate_crop_path after codec normalisation.""" # Get video metadata metadata = probe_video(video_path) # Calculate crop dimensions for 9:16 from any input aspect ratio crop_height = metadata.height crop_width = int(crop_height * (target_aspect[0] / target_aspect[1])) # Handle edge case: if video is narrower than target aspect ratio if crop_width > metadata.width: crop_width = metadata.width crop_height = int(crop_width * (target_aspect[1] / target_aspect[0])) print(f"Crop mode: SCENE_EQUILIBRIUM (per-scene stability)", file=sys.stderr) # Step 1: Split segment into scenes scenes = self._split_segment_into_scenes(video_path, segment, metadata.fps) if len(scenes) == 1: print(" No scene changes detected, processing as single equilibrium", file=sys.stderr) # Use the same two-pass pipeline as multi-scene: collect ALL detections # first so equilibrium is computed from the full clip, not just the first # 1-2 seconds. _process_segment_unified froze equilibrium after 3 samples # which caused keyframe drift and visible jumps when those early samples # didn't represent the subject's position throughout the clip. all_detections = self._extract_detections_for_segment( video_path, segment, metadata.fps ) eq_x, eq_y, eq_stats = self._compute_equilibrium_from_detections( all_detections, settings.equilibrium_sample_interval, crop_width, crop_height, metadata.width, metadata.height, ) print( f" Equilibrium: ({eq_x}, {eq_y}), " f"samples: {eq_stats.get('num_samples', 0)}", file=sys.stderr, ) raw_keyframes = self._generate_keyframes_from_detections( all_detections, eq_x, eq_y, crop_width, crop_height, metadata.width, metadata.height, settings.saliency_keyframe_interval, ) if self.debug: self._save_debug_frames_from_keyframes(video_path, raw_keyframes) else: print(f" Processing {len(scenes)} independent scenes", file=sys.stderr) # Extract detections for ENTIRE segment (single video pass) print(f" Extracting detections for entire segment...", file=sys.stderr) all_detections = self._extract_detections_for_segment( video_path, segment, metadata.fps ) print(f" Collected {len(all_detections)} detection samples", file=sys.stderr) # Step 3: Process each scene with pre-collected detections all_keyframes: list[CropKeyframe] = [] prev_scene_eq_x, prev_scene_eq_y = None, None for i, scene in enumerate(scenes, 1): scene_duration = scene.end - scene.start print( f" Scene {i}/{len(scenes)}: " f"{scene.start:.1f}s - {scene.end:.1f}s " f"({scene_duration:.1f}s)", file=sys.stderr ) # Filter detections for this scene (with tolerance for floating-point rounding) frame_tolerance = 0.5 / metadata.fps scene_detections = [ d for d in all_detections if (scene.start - frame_tolerance) <= d.time <= (scene.end + frame_tolerance) ] if not scene_detections: print(f" Warning: No detections in scene {i}", file=sys.stderr) continue # Compute equilibrium for this scene scene_eq_x, scene_eq_y, scene_stats = self._compute_equilibrium_from_detections( scene_detections, settings.equilibrium_sample_interval, crop_width, crop_height, metadata.width, metadata.height, ) # For scenes after the first, insert bridging keyframes at the boundary if i > 1 and prev_scene_eq_x is not None: epsilon = 0.001 # 1ms before the boundary all_keyframes.append(CropKeyframe( time=scene.start - epsilon, x=prev_scene_eq_x, y=prev_scene_eq_y, width=crop_width, height=crop_height, confidence=1.0, detection_method="scene_equilibrium", scene_change=False, )) all_keyframes.append(CropKeyframe( time=scene.start, x=scene_eq_x, y=scene_eq_y, width=crop_width, height=crop_height, confidence=1.0, detection_method="scene_equilibrium", scene_change=True, )) # Generate keyframes for this scene scene_keyframes = self._generate_keyframes_from_detections( scene_detections, scene_eq_x, scene_eq_y, crop_width, crop_height, metadata.width, metadata.height, settings.saliency_keyframe_interval, ) # Skip first keyframe if it's redundant with the boundary keyframe if scene_keyframes and i > 1: first_kf_time = scene_keyframes[0].time if abs(first_kf_time - scene.start) < (0.5 / metadata.fps): scene_keyframes = scene_keyframes[1:] all_keyframes.extend(scene_keyframes) print( f" Equilibrium: ({scene_eq_x}, {scene_eq_y}), " f"keyframes: {len(scene_keyframes)}, " f"samples: {scene_stats.get('num_samples', 0)}", file=sys.stderr ) prev_scene_eq_x, prev_scene_eq_y = scene_eq_x, scene_eq_y raw_keyframes = all_keyframes if not raw_keyframes: # Every scene had no detections (or each scene's sole keyframe was # stripped by the boundary-deduplication guard above). Fall back to # a single centred keyframe so downstream code always has at least one. print( " Warning: no keyframes produced for any scene — " "using center-crop fallback", file=sys.stderr, ) raw_keyframes = [ CropKeyframe( time=segment.start, x=(metadata.width - crop_width) // 2, y=(metadata.height - crop_height) // 2, width=crop_width, height=crop_height, confidence=0.1, detection_method="fallback_center", ) ] # Save debug frames for multi-scene path (single-scene path saves # them inline inside _process_segment_unified during frame reading). if self.debug: self._save_debug_frames_from_keyframes(video_path, raw_keyframes) # Apply smoothing smooth_keyframes = self._smooth_keyframes(raw_keyframes) return CropPath(segment=segment, keyframes=smooth_keyframes) def _detect_faces(self, frame: np.ndarray) -> tuple[int | None, int | None]: """ Detect faces in frame and return center of largest/most central face. Returns: Tuple of (center_x, center_y) or (None, None) if no faces found """ if self.face_cascade is None: return None, None gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) # Detect faces with multiple scale attempts for better detection # Using minNeighbors=3 (down from 5) for more lenient detection in music videos # where lighting, angles, and motion can challenge face detection faces = self.face_cascade.detectMultiScale( gray, scaleFactor=1.1, minNeighbors=3, # More lenient to catch faces at angles/movement minSize=(30, 30), flags=cv2.CASCADE_SCALE_IMAGE, ) if len(faces) == 0: return None, None # If multiple faces, prefer the most central and largest face # This handles scenes with multiple people better (artist vs backup dancers) if len(faces) > 1: frame_height, frame_width = gray.shape center_x_frame = frame_width // 2 center_y_frame = frame_height // 2 # Score each face by combination of size and centrality def face_score(face): fx, fy, fw, fh = face face_center_x = fx + fw // 2 face_center_y = fy + fh // 2 # Distance from frame center (normalized) distance = ( (face_center_x - center_x_frame) ** 2 + (face_center_y - center_y_frame) ** 2 ) ** 0.5 max_distance = (frame_width**2 + frame_height**2) ** 0.5 centrality_score = 1.0 - (distance / max_distance) # Size score (normalized by frame area) area = fw * fh max_area = frame_width * frame_height size_score = area / max_area # Combined score: 60% centrality, 40% size # (artist is usually central, even if not largest) return 0.6 * centrality_score + 0.4 * size_score # Sort by combined score faces = sorted(faces, key=face_score, reverse=True) # Use the best-scoring face x, y, w, h = faces[0] # Return center of face, biased slightly upward (eyes level) center_x = x + w // 2 center_y = y + int(h * 0.4) # Upper portion of face return center_x, center_y def _detect_edges_saliency(self, frame: np.ndarray) -> tuple[int | None, int | None]: """ Detect salient regions using edge detection and spatial weighting. Returns: Tuple of (center_x, center_y) or (None, None) if detection fails """ gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) frame_height, frame_width = gray.shape # Apply Canny edge detection edges = cv2.Canny(gray, 50, 150) # Blur edges to create regions blurred = cv2.GaussianBlur(edges, (21, 21), 0) # Create spatial weight map (center is more important) y_coords, x_coords = np.ogrid[:frame_height, :frame_width] center_y, center_x = frame_height // 2, frame_width // 2 # Gaussian weight centered on frame (resolution-aware) # Using standard deviation of ~1/3 of frame diagonal for consistent behavior frame_diagonal = (frame_width**2 + frame_height**2) ** 0.5 sigma_squared = (frame_diagonal / 3) ** 2 spatial_weight = np.exp( -((x_coords - center_x) ** 2 + (y_coords - center_y) ** 2) / (2 * sigma_squared) ) # Combine edges with spatial weighting weighted_saliency = blurred.astype(float) * spatial_weight # Find the peak of weighted saliency # Only accept if saliency is strong enough (avoid spurious detections) max_saliency = weighted_saliency.max() # Require at least 10% of theoretical maximum to avoid false positives # Theoretical max is 255 (from blurred edges) * 1.0 (from spatial weight) min_threshold = 255 * 0.10 if max_saliency > min_threshold: peak_y, peak_x = np.unravel_index(weighted_saliency.argmax(), weighted_saliency.shape) return int(peak_x), int(peak_y) return None, None def _segment_by_scene_changes( self, keyframes: list[CropKeyframe] ) -> list[list[CropKeyframe]]: """Split keyframes into segments at scene change boundaries.""" if not keyframes: return [] segments = [] current_segment = [] for kf in keyframes: if kf.scene_change and current_segment: segments.append(current_segment) current_segment = [kf] else: current_segment.append(kf) if current_segment: segments.append(current_segment) return segments def _smooth_keyframes(self, keyframes: list[CropKeyframe]) -> list[CropKeyframe]: """Apply scene-aware temporal smoothing to keyframes.""" if len(keyframes) < 2: return keyframes # If scene detection disabled, use original smoothing if not settings.enable_scene_detection: return self._smooth_segment(keyframes) # Segment keyframes by scene boundaries scene_segments = self._segment_by_scene_changes(keyframes) # Apply smoothing to each scene independently all_smoothed = [] for segment_keyframes in scene_segments: if len(segment_keyframes) >= 2: smoothed = self._smooth_segment(segment_keyframes) all_smoothed.extend(smoothed) else: all_smoothed.extend(segment_keyframes) # Note: Keyframe cap removed - we use frame-by-frame processing with scipy interpolation # Frame-by-frame processing can handle unlimited keyframes efficiently # More keyframes = smoother motion and better scene boundary preservation return all_smoothed def _smooth_segment(self, keyframes: list[CropKeyframe]) -> list[CropKeyframe]: """ Apply aggressive temporal smoothing to a single scene segment. Multi-pass smoothing: 1. Exponential moving average (removes high-frequency noise) 2. Gaussian filter (smooths overall trajectory) 3. Cubic spline interpolation (creates smooth sub-frame motion) Args: keyframes: List of raw keyframes Returns: List of smoothed keyframes """ if len(keyframes) < 2: return keyframes # Extract time series times = np.array([kf.time for kf in keyframes]) x_coords = np.array([kf.x for kf in keyframes]) y_coords = np.array([kf.y for kf in keyframes]) confidences = np.array([kf.confidence for kf in keyframes]) # Pass 1: Exponential moving average for noise reduction alpha = 0.3 # Smoothing factor (lower = more smoothing) x_ema = np.zeros_like(x_coords, dtype=float) y_ema = np.zeros_like(y_coords, dtype=float) x_ema[0] = x_coords[0] y_ema[0] = y_coords[0] for i in range(1, len(x_coords)): x_ema[i] = alpha * x_coords[i] + (1 - alpha) * x_ema[i - 1] y_ema[i] = alpha * y_coords[i] + (1 - alpha) * y_ema[i - 1] # Pass 2: Gaussian smoothing for overall trajectory sigma = settings.smoothing_window / (times[1] - times[0]) if len(times) > 1 else 1.0 x_smooth = ndimage.gaussian_filter1d(x_ema, sigma=sigma, mode="nearest") y_smooth = ndimage.gaussian_filter1d(y_ema, sigma=sigma, mode="nearest") conf_smooth = ndimage.gaussian_filter1d(confidences, sigma=sigma, mode="nearest") # Find most common detection method methods = [kf.detection_method for kf in keyframes] most_common_method = max(set(methods), key=methods.count) # Preserve scene_change flag from first keyframe (important for segment boundaries) first_keyframe_is_scene_change = keyframes[0].scene_change # Create cubic spline interpolation for sub-frame precision if len(times) >= 4: interp_x = interp1d(times, x_smooth, kind="cubic", fill_value="extrapolate") interp_y = interp1d(times, y_smooth, kind="cubic", fill_value="extrapolate") interp_conf = interp1d(times, conf_smooth, kind="linear", fill_value="extrapolate") # Generate more keyframes for smoother transitions # Cap to max 8 keyframes to prevent FFmpeg nested expression issues # FFmpeg can reliably handle 8-10 levels of nested if() expressions max_keyframes = 8 target_keyframes = min(len(times) * 2, max_keyframes) # Reduced from 4x to 2x dense_times = np.linspace(times[0], times[-1], target_keyframes) x_dense = interp_x(dense_times) y_dense = interp_y(dense_times) conf_dense = interp_conf(dense_times) # Create new keyframes smooth_keyframes = [ CropKeyframe( time=float(t), x=int(x), y=int(y), width=keyframes[0].width, height=keyframes[0].height, confidence=float(np.clip(conf, 0.0, 1.0)), detection_method=f"{most_common_method}_smoothed", scene_change=first_keyframe_is_scene_change if i == 0 else False, ) for i, (t, x, y, conf) in enumerate( zip(dense_times, x_dense, y_dense, conf_dense, strict=True) ) ] else: # Not enough points for cubic interpolation smooth_keyframes = [ CropKeyframe( time=kf.time, x=int(x_smooth[i]), y=int(y_smooth[i]), width=kf.width, height=kf.height, confidence=float(conf_smooth[i]), detection_method=kf.detection_method, scene_change=kf.scene_change, ) for i, kf in enumerate(keyframes) ] return smooth_keyframes def _save_debug_frames_from_keyframes( self, video_path: Path, keyframes: list[CropKeyframe], ) -> None: """Save debug frames for each keyframe (used in multi-scene path). The multi-scene code path collects detections first and computes keyframes afterward, so it has no frame data at keyframe-generation time. This method makes a single video pass over the keyframe timestamps and delegates to _save_debug_frame for each one. Args: video_path: Source video file. keyframes: Smoothed keyframes whose crop positions to visualise. """ if not keyframes: return cap = cv2.VideoCapture(str(video_path)) try: for kf in keyframes: cap.set(cv2.CAP_PROP_POS_MSEC, kf.time * 1000) ret, frame = cap.read() if not ret: continue # Mirror the instance state that _process_segment_unified sets # so _save_debug_frame labels are accurate. self._last_detection_method = kf.detection_method self._last_confidence = kf.confidence self._is_scene_change = kf.scene_change self._save_debug_frame( frame, kf.x, kf.y, kf.width, kf.height, kf.time ) finally: cap.release() def _get_video_dimensions(self, video_path: Path) -> tuple[int, int]: """Get video dimensions (height, width).""" cap = cv2.VideoCapture(str(video_path)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) cap.release() return height, width def _save_debug_frame( self, frame: np.ndarray, x: int, y: int, crop_width: int, crop_height: int, time: float, equilibrium_x: int | None = None, equilibrium_y: int | None = None, ) -> None: """Save debug visualization showing crop window position and equilibrium.""" debug_frame = frame.copy() # If scene change, draw red border if hasattr(self, "_is_scene_change") and self._is_scene_change: # Thick red border to indicate scene change cv2.rectangle( debug_frame, (0, 0), (frame.shape[1] - 1, frame.shape[0] - 1), (0, 0, 255), # Red 15, ) # "SCENE CHANGE" label (shifted to y=180 to avoid segment metadata lines) cv2.putText( debug_frame, "SCENE CHANGE", (10, 180), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 0, 255), 3, ) # Draw equilibrium position (BLUE, dashed) - only in equilibrium mode if equilibrium_x is not None: # Use dotted line for equilibrium for i in range(0, crop_height, 20): cv2.line( debug_frame, (equilibrium_x, equilibrium_y + i), (equilibrium_x, equilibrium_y + min(i + 10, crop_height)), (255, 0, 0), # Blue 2, ) cv2.line( debug_frame, (equilibrium_x + crop_width, equilibrium_y + i), (equilibrium_x + crop_width, equilibrium_y + min(i + 10, crop_height)), (255, 0, 0), # Blue 2, ) for i in range(0, crop_width, 20): cv2.line( debug_frame, (equilibrium_x + i, equilibrium_y), (equilibrium_x + min(i + 10, crop_width), equilibrium_y), (255, 0, 0), # Blue 2, ) cv2.line( debug_frame, (equilibrium_x + i, equilibrium_y + crop_height), (equilibrium_x + min(i + 10, crop_width), equilibrium_y + crop_height), (255, 0, 0), # Blue 2, ) # Draw current crop rectangle (GREEN) cv2.rectangle(debug_frame, (x, y), (x + crop_width, y + crop_height), (0, 255, 0), 3) # Draw safe zone (YELLOW, inner rectangle) - only in equilibrium mode if equilibrium_x is not None: safety_margin = settings.equilibrium_safety_margin safe_left = x + int(crop_width * safety_margin) safe_right = x + crop_width - int(crop_width * safety_margin) safe_top = y + int(crop_height * safety_margin) safe_bottom = y + crop_height - int(crop_height * safety_margin) cv2.rectangle( debug_frame, (safe_left, safe_top), (safe_right, safe_bottom), (0, 255, 255), # Yellow 1, ) # Add crosshair at crop center center_x = x + crop_width // 2 center_y = y + crop_height // 2 cv2.line(debug_frame, (center_x - 20, center_y), (center_x + 20, center_y), (0, 0, 255), 2) cv2.line(debug_frame, (center_x, center_y - 20), (center_x, center_y + 20), (0, 0, 255), 2) # Add timestamp and detection info cv2.putText( debug_frame, f"t={time:.2f}s crop=({x},{y})", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2, ) cv2.putText( debug_frame, f"method: {self._last_detection_method}", (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255, 255, 0), 2, ) cv2.putText( debug_frame, f"confidence: {self._last_confidence:.2f}", (10, 90), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255, 255, 0), 2, ) # Segment metadata from Gemini (type, score, description) — y=120/150 if hasattr(self, "_current_segment") and self._current_segment is not None: cv2.putText( debug_frame, f"[{self._current_segment.segment_type}] score={self._current_segment.score:.2f}", (10, 120), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), # Cyan 2, ) desc = self._current_segment.description desc_short = (desc[:69] + "...") if len(desc) > 72 else desc cv2.putText( debug_frame, desc_short, (10, 150), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), # White 1, ) # Show equilibrium info in equilibrium mode (shifted down to avoid segment label) if equilibrium_x is not None: cv2.putText( debug_frame, f"Equilibrium: ({equilibrium_x}, {equilibrium_y})", (10, 210), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 0, 0), # Blue 2, ) # Show offset from equilibrium dx = x - equilibrium_x dy = y - equilibrium_y cv2.putText( debug_frame, f"Offset: ({dx:+d}, {dy:+d})px", (10, 240), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 0), 2, ) # Save to root debug folder (for testing convenience) debug_dir = Path("debug_frames") debug_dir.mkdir(exist_ok=True) root_output_path = debug_dir / f"frame_{time:.2f}s.jpg" cv2.imwrite(str(root_output_path), debug_frame) # Also save to output directory if provided if self.output_dir is not None: output_debug_dir = self.output_dir / "debug_frames" output_debug_dir.mkdir(parents=True, exist_ok=True) output_path = output_debug_dir / f"frame_{time:.2f}s.jpg" cv2.imwrite(str(output_path), debug_frame)