"""Lipsync processing wrapper for OutofLipSync""" import os import traceback from ffmpy import FFmpeg from lipsync import apply_lipsync def get_video_info(video_path: str) -> dict: """Get video information: resolution, duration, fps Args: video_path: Path to video Returns: Dict with keys: width, height, duration, fps """ import json import subprocess from fractions import Fraction cmd = [ "ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,r_frame_rate", "-show_entries", "format=duration", "-of", "json", video_path, ] result = subprocess.run(cmd, capture_output=True, text=True, check=True) data = json.loads(result.stdout) width = data["streams"][0]["width"] height = data["streams"][0]["height"] fps = float(Fraction(data["streams"][0]["r_frame_rate"])) duration = float(data["format"]["duration"]) return {"width": width, "height": height, "fps": fps, "duration": duration} def apply_lipsync_to_video( video_path: str, audio_16k_path: str, output_dir: str, model_type: str = "LatentSync v1.6", quality_level: str = "Normal", ) -> tuple: """Apply lipsync to video using clean 16k audio Args: video_path: Path to input video audio_16k_path: Path to 16kHz audio output_dir: Directory to save output model_type: Model type for lipsync ("LatentSync v1.6" or "MuseTalk v1.5") quality_level: Quality level ("Fast", "Normal", "Medium", "Best", "Super Best") Returns: Tuple of (lipsynced_video_path, video_info) """ try: lipsynced_video = os.path.join(output_dir, "output_with_lipsync.mp4") if model_type == "LatentSync v1.6": crop_size = 512 print( f"Using LatentSync: video={video_path}, audio={audio_16k_path}, crop_size={crop_size}, quality={quality_level}" ) apply_lipsync( video_path, audio_16k_path, lipsynced_video, crop_size, quality_level ) else: raise ValueError(f"Unknown model_type: {model_type}") video_info = get_video_info(lipsynced_video) print( f"Lipsynced video: {lipsynced_video}, size: {video_info['width']}x{video_info['height']}" ) return lipsynced_video, video_info except RuntimeError as e: if "out of memory" in str(e).lower(): print("GPU OOM Error in lipsync processing!") raise RuntimeError( "GPU out of memory during lipsync. Try a shorter video or lower resolution." ) if "face not detected" in str(e).lower(): raise RuntimeError( "Face detection failed in lipsync pipeline. Please upload a video with a clear, visible face." ) print(f"Runtime Error in lipsync processing: {e}") traceback.print_exc() raise