| |
| """ |
| Debug Enhanced Positioning System for Video Background Replacement |
| |
| Key debugging areas: |
| 1. UI parameter flow validation |
| 2. Coordinate system debugging |
| 3. Alpha/video alignment verification |
| 4. Canvas placement mathematics |
| 5. Temporal synchronization checks |
| """ |
|
|
| import cv2 |
| import numpy as np |
| import os |
| import time |
| import random |
| from pathlib import Path |
| from moviepy.editor import VideoFileClip |
| import logging |
|
|
| |
| logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s") |
| debug_logger = logging.getLogger("positioning_debug") |
|
|
| class PositioningDebugger: |
| """Comprehensive debugging for positioning system""" |
| |
| def __init__(self, debug_dir="debug_positioning"): |
| self.debug_dir = Path(debug_dir) |
| self.debug_dir.mkdir(exist_ok=True, parents=True) |
| self.frame_count = 0 |
| |
| def log_parameters(self, placement_dict): |
| """Log all positioning parameters""" |
| debug_logger.info("=== POSITIONING PARAMETERS ===") |
| debug_logger.info(f"Raw placement dict: {placement_dict}") |
| |
| px = float(placement_dict.get("x", 0.5)) |
| py = float(placement_dict.get("y", 0.75)) |
| ps = float(placement_dict.get("scale", 1.0)) |
| feather_px = int(placement_dict.get("feather", 3)) |
| |
| debug_logger.info(f"Parsed - px: {px}, py: {py}, ps: {ps}, feather: {feather_px}") |
| debug_logger.info(f"Clamped - px: {max(0.0, min(1.0, px))}, py: {max(0.0, min(1.0, py))}") |
| debug_logger.info(f"Scale range - ps: {max(0.3, min(2.0, ps))}") |
| |
| return px, py, ps, feather_px |
| |
| def debug_coordinate_calculation(self, frame_dims, alpha_dims, placement): |
| """Debug coordinate calculations step by step""" |
| hh, ww = frame_dims |
| alpha_h, alpha_w = alpha_dims |
| |
| px, py, ps, feather_px = self.log_parameters(placement) |
| |
| debug_logger.info("=== COORDINATE CALCULATIONS ===") |
| debug_logger.info(f"Original video dims: {ww}x{hh}") |
| debug_logger.info(f"Alpha video dims: {alpha_w}x{alpha_h}") |
| |
| |
| sw = max(1, int(ww * ps)) |
| sh = max(1, int(hh * ps)) |
| debug_logger.info(f"Scaled subject size: {sw}x{sh} (scale factor: {ps})") |
| |
| |
| cx = int(px * ww) |
| cy = int(py * hh) |
| debug_logger.info(f"Target center: ({cx}, {cy})") |
| |
| |
| x0 = int(cx - sw // 2) |
| y0 = int(cy - sh // 2) |
| debug_logger.info(f"Top-left corner: ({x0}, {y0})") |
| |
| |
| xs0, ys0 = max(0, x0), max(0, y0) |
| xs1, ys1 = min(ww, x0 + sw), min(hh, y0 + sh) |
| debug_logger.info(f"Clipped bounds: ({xs0}, {ys0}) to ({xs1}, {ys1})") |
| |
| if xs1 <= xs0 or ys1 <= ys0: |
| debug_logger.warning("INVALID BOUNDS - Subject will be clipped completely!") |
| |
| |
| src_x0 = xs0 - x0 |
| src_y0 = ys0 - y0 |
| src_x1 = src_x0 + (xs1 - xs0) |
| src_y1 = src_y0 + (ys1 - ys0) |
| debug_logger.info(f"Source region: ({src_x0}, {src_y0}) to ({src_x1}, {src_y1})") |
| |
| return { |
| 'scaled_size': (sw, sh), |
| 'center': (cx, cy), |
| 'top_left': (x0, y0), |
| 'clipped_dest': (xs0, ys0, xs1, ys1), |
| 'source_region': (src_x0, src_y0, src_x1, src_y1) |
| } |
| |
| def save_debug_frame(self, frame, alpha, composite, frame_idx, coords): |
| """Save debug visualization of frame processing""" |
| debug_path = self.debug_dir / f"frame_{frame_idx:04d}_debug.png" |
| |
| |
| h, w = frame.shape[:2] |
| debug_viz = np.zeros((h * 2, w * 2, 3), dtype=np.uint8) |
| |
| |
| debug_viz[0:h, 0:w] = cv2.cvtColor((frame * 255).astype(np.uint8), cv2.COLOR_RGB2BGR) |
| |
| |
| alpha_viz = cv2.cvtColor((alpha * 255).astype(np.uint8), cv2.COLOR_GRAY2BGR) |
| debug_viz[0:h, w:w*2] = alpha_viz |
| |
| |
| debug_viz[h:h*2, 0:w] = cv2.cvtColor((composite * 255).astype(np.uint8), cv2.COLOR_RGB2BGR) |
| |
| |
| coord_viz = np.zeros((h, w, 3), dtype=np.uint8) |
| |
| |
| cx, cy = coords['center'] |
| sw, sh = coords['scaled_size'] |
| x0, y0 = coords['top_left'] |
| |
| |
| cv2.line(coord_viz, (cx-10, cy), (cx+10, cy), (0, 255, 0), 2) |
| cv2.line(coord_viz, (cx, cy-10), (cx, cy+10), (0, 255, 0), 2) |
| |
| |
| cv2.rectangle(coord_viz, (x0, y0), (x0 + sw, y0 + sh), (255, 0, 0), 2) |
| |
| |
| cv2.rectangle(coord_viz, (0, 0), (w-1, h-1), (255, 255, 255), 1) |
| |
| debug_viz[h:h*2, w:w*2] = coord_viz |
| |
| |
| font = cv2.FONT_HERSHEY_SIMPLEX |
| cv2.putText(debug_viz, "Original", (10, 30), font, 0.7, (255, 255, 255), 2) |
| cv2.putText(debug_viz, "Alpha", (w + 10, 30), font, 0.7, (255, 255, 255), 2) |
| cv2.putText(debug_viz, "Composite", (10, h + 30), font, 0.7, (255, 255, 255), 2) |
| cv2.putText(debug_viz, "Coordinates", (w + 10, h + 30), font, 0.7, (255, 255, 255), 2) |
| |
| cv2.imwrite(str(debug_path), debug_viz) |
| debug_logger.info(f"Debug frame saved: {debug_path}") |
|
|
| def create_enhanced_composite_function(debugger=None): |
| """Create composite function with enhanced debugging""" |
| |
| def composite_frame_debug(get_frame, t, original_clip, alpha_clip, bg_rgb, placement, feather_px): |
| """Enhanced composite function with comprehensive debugging""" |
| |
| if debugger: |
| debugger.frame_count += 1 |
| debug_logger.info(f"=== PROCESSING FRAME {debugger.frame_count} at t={t:.3f}s ===") |
| |
| |
| frame = get_frame(t).astype(np.float32) / 255.0 |
| hh, ww = frame.shape[:2] |
| |
| |
| alpha_duration = alpha_clip.duration or 0 |
| if alpha_duration > 0: |
| alpha_t = min(t, max(0.0, alpha_duration - 0.01)) |
| else: |
| alpha_t = 0.0 |
| |
| debug_logger.info(f"Alpha lookup: t={t:.3f}, alpha_t={alpha_t:.3f}, alpha_duration={alpha_duration:.3f}") |
| |
| try: |
| a = alpha_clip.get_frame(alpha_t) |
| if a.ndim == 3: |
| a = a[:, :, 0] |
| a = a.astype(np.float32) / 255.0 |
| debug_logger.info(f"Alpha frame shape: {a.shape}, range: [{a.min():.3f}, {a.max():.3f}]") |
| except Exception as e: |
| debug_logger.error(f"Alpha frame error: {e}") |
| return (bg_rgb * 255).astype(np.uint8) |
| |
| |
| px = max(0.0, min(1.0, float(placement.get("x", 0.5)))) |
| py = max(0.0, min(1.0, float(placement.get("y", 0.75)))) |
| ps = max(0.3, min(2.0, float(placement.get("scale", 1.0)))) |
| |
| |
| coords = None |
| if debugger: |
| coords = debugger.debug_coordinate_calculation((hh, ww), a.shape, placement) |
| |
| |
| sw = max(1, int(ww * ps)) |
| sh = max(1, int(hh * ps)) |
| |
| debug_logger.info(f"Scaling: {ww}x{hh} -> {sw}x{sh} (factor: {ps})") |
| |
| fg_scaled = cv2.resize(frame, (sw, sh), interpolation=cv2.INTER_LINEAR) |
| a_scaled = cv2.resize(a, (sw, sh), interpolation=cv2.INTER_LINEAR) |
| |
| |
| fg_canvas = np.zeros_like(frame, dtype=np.float32) |
| a_canvas = np.zeros((hh, ww), dtype=np.float32) |
| |
| |
| cx = int(px * ww) |
| cy = int(py * hh) |
| x0 = int(cx - sw // 2) |
| y0 = int(cy - sh // 2) |
| |
| |
| xs0, ys0 = max(0, x0), max(0, y0) |
| xs1, ys1 = min(ww, x0 + sw), min(hh, y0 + sh) |
| |
| if xs1 <= xs0 or ys1 <= ys0: |
| debug_logger.warning("Subject completely outside frame bounds!") |
| if debugger: |
| debugger.save_debug_frame(frame, a, bg_rgb, debugger.frame_count, coords or {}) |
| return (bg_rgb * 255).astype(np.uint8) |
| |
| |
| src_x0 = xs0 - x0 |
| src_y0 = ys0 - y0 |
| src_x1 = src_x0 + (xs1 - xs0) |
| src_y1 = src_y0 + (ys1 - ys0) |
| |
| try: |
| fg_canvas[ys0:ys1, xs0:xs1, :] = fg_scaled[src_y0:src_y1, src_x0:src_x1, :] |
| a_canvas[ys0:ys1, xs0:xs1] = a_scaled[src_y0:src_y1, src_x0:src_x1] |
| except Exception as e: |
| debug_logger.error(f"Canvas placement error: {e}") |
| debug_logger.error(f"Canvas region: [{ys0}:{ys1}, {xs0}:{xs1}]") |
| debug_logger.error(f"Source region: [{src_y0}:{src_y1}, {src_x0}:{src_x1}]") |
| if debugger: |
| debugger.save_debug_frame(frame, a, bg_rgb, debugger.frame_count, coords or {}) |
| return (bg_rgb * 255).astype(np.uint8) |
| |
| |
| if feather_px > 0: |
| k = (feather_px * 2 + 1) |
| a_canvas = cv2.GaussianBlur(a_canvas, (k, k), feather_px) |
| |
| |
| a3 = a_canvas[:, :, None] |
| comp = a3 * fg_canvas + (1.0 - a3) * bg_rgb |
| result = np.clip(comp * 255, 0, 255).astype(np.uint8) |
| |
| |
| if debugger and debugger.frame_count % 30 == 1: |
| debugger.save_debug_frame(frame, a_canvas, comp, debugger.frame_count, coords or {}) |
| |
| return result |
| |
| return composite_frame_debug |
|
|
| def validate_ui_parameter_flow(): |
| """Test function to validate UI parameters reach processing correctly""" |
| |
| |
| test_placements = [ |
| {"x": 0.5, "y": 0.5, "scale": 1.0, "feather": 3}, |
| {"x": 0.2, "y": 0.8, "scale": 0.7, "feather": 5}, |
| {"x": 0.8, "y": 0.3, "scale": 1.5, "feather": 1}, |
| ] |
| |
| debugger = PositioningDebugger() |
| |
| for i, placement in enumerate(test_placements): |
| print(f"\n=== TEST CASE {i+1} ===") |
| |
| |
| frame_dims = (720, 1280) |
| alpha_dims = (720, 1280) |
| |
| coords = debugger.debug_coordinate_calculation(frame_dims, alpha_dims, placement) |
| |
| print(f"Expected center: {coords['center']}") |
| print(f"Scaled size: {coords['scaled_size']}") |
| print(f"Placement bounds: {coords['clipped_dest']}") |
|
|
| def create_debug_enhanced_process_video_main(): |
| """Enhanced version of process_video_main with debugging""" |
| |
| def process_video_main_debug( |
| video_path: str, |
| background_path: str = None, |
| trim_duration: float = None, |
| crf: int = 18, |
| preserve_audio_flag: bool = True, |
| placement: dict = None, |
| use_chunked_processing: bool = False, |
| enable_debug: bool = True, |
| progress=None, |
| ): |
| """ |
| Enhanced process_video_main with comprehensive positioning debugging |
| """ |
| |
| debugger = PositioningDebugger() if enable_debug else None |
| messages = [] |
| |
| try: |
| if debugger: |
| debug_logger.info("=== STARTING DEBUG SESSION ===") |
| debug_logger.info(f"Video: {video_path}") |
| debug_logger.info(f"Background: {background_path}") |
| debug_logger.info(f"Placement: {placement}") |
| |
| |
| placement = placement or {} |
| if debugger: |
| debugger.log_parameters(placement) |
| |
| |
| |
| |
| def composite_frame(get_frame, t): |
| return create_enhanced_composite_function(debugger)( |
| get_frame, t, original_clip, alpha_clip, bg_rgb, placement, |
| int(placement.get("feather", 3)) |
| ) |
| |
| |
| |
| if debugger: |
| debug_logger.info("=== DEBUG SESSION COMPLETE ===") |
| debug_logger.info(f"Debug files saved to: {debugger.debug_dir}") |
| |
| return result_path, "\n".join(messages) |
| |
| except Exception as e: |
| if debugger: |
| debug_logger.error(f"Processing failed: {e}") |
| raise |
| |
| return process_video_main_debug |
|
|
| |
| def quick_positioning_fixes(): |
| """ |
| Quick fixes to test for common positioning issues |
| """ |
| |
| fixes = { |
| "coordinate_system_flip": { |
| "description": "Test if Y coordinate should be flipped", |
| "change": "cy = int((1.0 - py) * hh) # Flip Y coordinate", |
| "original": "cy = int(py * hh)" |
| }, |
| |
| "anchor_point_adjustment": { |
| "description": "Test different anchor points for scaling", |
| "change": """ |
| # Try bottom-center anchor instead of center-center |
| x0 = int(cx - sw // 2) # Keep X centered |
| y0 = int(cy - sh) # Anchor at bottom |
| """, |
| "original": """ |
| x0 = int(cx - sw // 2) |
| y0 = int(cy - sh // 2) |
| """ |
| }, |
| |
| "alpha_scaling_sync": { |
| "description": "Ensure alpha and frame scaling are synchronized", |
| "change": """ |
| # Force exact same dimensions |
| if a.shape != (hh, ww): |
| a = cv2.resize(a, (ww, hh), interpolation=cv2.INTER_LINEAR) |
| """, |
| "original": "# No explicit resize check" |
| }, |
| |
| "ui_parameter_validation": { |
| "description": "Add parameter validation and logging", |
| "change": """ |
| px = float(placement.get("x", 0.5)) |
| py = float(placement.get("y", 0.75)) |
| ps = float(placement.get("scale", 1.0)) |
| |
| print(f"DEBUG: px={px}, py={py}, ps={ps}") |
| print(f"DEBUG: cx={px*ww}, cy={py*hh}") |
| """, |
| "original": "# No debug logging" |
| } |
| } |
| |
| return fixes |
|
|
| if __name__ == "__main__": |
| print("=== Video Background Replacement - Positioning Debug ===") |
| print("\n1. Running UI parameter validation...") |
| validate_ui_parameter_flow() |
| |
| print("\n2. Available quick fixes:") |
| fixes = quick_positioning_fixes() |
| for fix_name, fix_info in fixes.items(): |
| print(f"\n{fix_name.upper()}:") |
| print(f" Description: {fix_info['description']}") |
| print(f" Change: {fix_info['change']}") |
| |
| print("\n3. Debug system ready!") |
| print(" - Use PositioningDebugger class in your main pipeline") |
| print(" - Enable debug mode: enable_debug=True") |
| print(" - Check debug_positioning/ folder for output") |