| """ |
| Simple Video Generation using SadTalker |
| Integrates with our Mindfull pipeline |
| """ |
| import os |
| import sys |
| import subprocess |
| import tempfile |
| import shutil |
| from pathlib import Path |
| import cv2 |
| import numpy as np |
|
|
| def generate_video_simple(image_path, audio_path, output_path): |
| """ |
| Generate video using SadTalker with fallback options |
| """ |
| try: |
| |
| print("Attempting SadTalker video generation...") |
| success = generate_with_sadtalker(image_path, audio_path, output_path) |
| |
| if success and os.path.exists(output_path): |
| size = os.path.getsize(output_path) |
| print(f"β Video generation successful: {output_path} (size: {size} bytes)") |
| return True |
| else: |
| print(f"β SadTalker failed to generate video") |
| return False |
| |
| except Exception as e: |
| print(f"β Video generation error: {e}") |
| return False |
|
|
| def generate_with_sadtalker(image_path, audio_path, output_path): |
| """ |
| Use enhanced SadTalker + Wav2Lip pipeline for video generation |
| """ |
| try: |
| |
| import python313_compat_patch |
| |
| |
| sadtalker_path = "sadtalker+wav2lip" |
| if os.path.exists(sadtalker_path): |
| |
| enhanced_script = os.path.join(sadtalker_path, "enhanced_pipeline.py") |
| |
| if os.path.exists(enhanced_script): |
| |
| cmd = [ |
| sys.executable, |
| enhanced_script, |
| "--source_image", image_path, |
| "--audio", audio_path, |
| "--output", output_path, |
| "--use_wav2lip" |
| ] |
| else: |
| |
| cmd = [ |
| sys.executable, |
| os.path.join(sadtalker_path, "simple_pipeline.py"), |
| "--source_image", image_path, |
| "--audio", audio_path, |
| "--output", output_path, |
| "--skip_wav2lip" |
| ] |
|
|
| result = subprocess.run(cmd, capture_output=True, text=True, timeout=600) |
| print(f"Enhanced Pipeline STDOUT:\n{result.stdout}") |
| if result.stderr: |
| print(f"Enhanced Pipeline STDERR:\n{result.stderr}") |
|
|
| |
| if result.returncode == 0 and os.path.exists(output_path): |
| size = os.path.getsize(output_path) |
| if size > 0: |
| print(f"β Enhanced video generated successfully: {output_path} (size: {size} bytes)") |
| return True |
| else: |
| print(f"β Enhanced pipeline output file is empty: {output_path}") |
| return False |
| else: |
| print(f"β SadTalker pipeline failed or output not created") |
| print(f"Return code: {result.returncode}") |
| return False |
| else: |
| print("β SadTalker directory not found") |
| return False |
| |
| except Exception as e: |
| print(f"β SadTalker generation failed: {e}") |
| return False |
|
|
| def generate_basic_video(image_path, audio_path, output_path): |
| """ |
| Generate basic video by combining static image with audio |
| """ |
| try: |
| |
| cmd = [ |
| "ffmpeg", "-y", |
| "-loop", "1", |
| "-i", image_path, |
| "-i", audio_path, |
| "-c:v", "libx264", |
| "-c:a", "aac", |
| "-b:a", "192k", |
| "-shortest", |
| "-pix_fmt", "yuv420p", |
| output_path |
| ] |
| |
| result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) |
| |
| if result.returncode == 0 and os.path.exists(output_path): |
| print(f"β Basic video generated: {output_path}") |
| return True |
| else: |
| print(f"β FFmpeg failed: {result.stderr}") |
| return False |
| |
| except subprocess.TimeoutExpired: |
| print("β FFmpeg generation timed out") |
| return False |
| except Exception as e: |
| print(f"β Basic video generation failed: {e}") |
| return False |
|
|
| def generate_placeholder_video(image_path, audio_path, output_path): |
| """ |
| Create a placeholder video with simple effects |
| """ |
| try: |
| import soundfile as sf |
| import moviepy.editor as mp |
| |
| |
| try: |
| audio_data, sample_rate = sf.read(audio_path) |
| duration = len(audio_data) / sample_rate |
| except: |
| duration = 5.0 |
| |
| |
| image = cv2.imread(image_path) |
| if image is None: |
| raise ValueError(f"Could not load image: {image_path}") |
| |
| |
| height, width = image.shape[:2] |
| target_width, target_height = 640, 480 |
| |
| if width != target_width or height != target_height: |
| image = cv2.resize(image, (target_width, target_height)) |
| |
| |
| fps = 25 |
| total_frames = int(duration * fps) |
| |
| |
| os.makedirs(os.path.dirname(output_path), exist_ok=True) |
| |
| |
| temp_video = output_path.replace('.mp4', '_temp.mp4') |
| fourcc = cv2.VideoWriter_fourcc(*'mp4v') |
| out = cv2.VideoWriter(temp_video, fourcc, fps, (target_width, target_height)) |
| |
| for frame_num in range(total_frames): |
| |
| zoom_factor = 1.0 + 0.05 * np.sin(frame_num * 0.1) |
| |
| if zoom_factor != 1.0: |
| center_x, center_y = target_width // 2, target_height // 2 |
| new_width = int(target_width / zoom_factor) |
| new_height = int(target_height / zoom_factor) |
| |
| x1 = max(0, center_x - new_width // 2) |
| y1 = max(0, center_y - new_height // 2) |
| x2 = min(target_width, x1 + new_width) |
| y2 = min(target_height, y1 + new_height) |
| |
| cropped = image[y1:y2, x1:x2] |
| frame = cv2.resize(cropped, (target_width, target_height)) |
| else: |
| frame = image.copy() |
| |
| out.write(frame) |
| |
| out.release() |
| |
| |
| cmd = [ |
| "ffmpeg", "-y", |
| "-i", temp_video, |
| "-i", audio_path, |
| "-c:v", "libx264", |
| "-c:a", "aac", |
| "-shortest", |
| output_path |
| ] |
| |
| result = subprocess.run(cmd, capture_output=True, text=True) |
| |
| |
| if os.path.exists(temp_video): |
| os.remove(temp_video) |
| |
| if result.returncode == 0 and os.path.exists(output_path): |
| print(f"β Placeholder video generated: {output_path}") |
| return True |
| else: |
| print(f"β Placeholder video generation failed") |
| return False |
| |
| except Exception as e: |
| print(f"β Placeholder video generation failed: {e}") |
| return False |
|
|
| if __name__ == "__main__": |
| |
| test_image = "avatar_assets/officer.png" |
| test_audio = "outputs/test_audio_simple.wav" |
| test_output = "outputs/test_video.mp4" |
| |
| print("Testing video generation...") |
| if os.path.exists(test_image) and os.path.exists(test_audio): |
| if generate_video_simple(test_image, test_audio, test_output): |
| print("β Video generation successful") |
| else: |
| print("β Video generation failed") |
| else: |
| print(f"β Missing test files:") |
| print(f" Image: {test_image} - {'β' if os.path.exists(test_image) else 'β'}") |
| print(f" Audio: {test_audio} - {'β' if os.path.exists(test_audio) else 'β'}") |
|
|