"""Modal runner for generating shareable assets.""" import json import pathlib import modal import time import os import uuid import datetime import traceback from datetime import timedelta try: import skia except ImportError: skia = None import boto3 import subprocess import warnings import shutil from suno_utils.utils.share_asset_utils import ( RenderedVideoOverlay, SpectrogramVisualizer, download_and_trim_audio, download_image, download_video, load_aligned_lyrics, load_preset_buffer, RenderConfig, PRESET_ID_PRERENDERED_VIDEO, PRESET_ID_PRERENDERED_BLACK_FIFTY, draw_timer_on_canvas, render_static_overlays_to_canvas, parse_sticker_overlays, ) from suno_utils.utils.lyrics_renderer import LyricsRenderer # Import the new shader rendering function from suno_utils.utils.share_asset_shader import render_shader_background_video from suno_utils.worker.schema import QueueItem import queue import threading DEPLOYMENT_TYPE = "dev" APP_NAME = f"share-asset-runner-{DEPLOYMENT_TYPE}" EVENTS_QUEUE_NAME = f"share-asset-events-{DEPLOYMENT_TYPE}" events_queue = modal.Queue.from_name(EVENTS_QUEUE_NAME, create_if_missing=True) SECRETS = [ modal.Secret.from_name("studio-aws"), # For S3 upload modal.Secret.from_name("api-callback-token"), ] base_image = ( modal.Image.debian_slim(python_version="3.10") .apt_install( "curl", "unzip", ) .run_commands( [ 'curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"', "unzip -q awscliv2.zip", "./aws/install", ] ) .apt_install( "ffmpeg", ) .pip_install( "boto3", "skia-python", ) .pip_install_from_pyproject( str(pathlib.Path(__file__).parent.parent.parent / "pyproject.toml"), ) .dockerfile_commands( [ "COPY --from=datadog/serverless-init /datadog-init /app/datadog-init", 'ENTRYPOINT ["/app/datadog-init"]', ] ) .add_local_python_source("suno_utils", copy=False) ) worker_image = ( modal.Image.from_registry("ghcr.io/selkies-project/nvidia-egl-desktop:latest", add_python="3.11") .apt_install( "ffmpeg", "fontconfig", ) .run_commands( [ "fc-cache -f -v", ] ) # Install moderngl with headless support and ensure dependencies are installed .pip_install( "moderngl[headless]==5.8.2", # Specify version to ensure compatibility "skia-python==87.7", "requests", "tqdm", "numpy>=1.20.0", # Ensure numpy is installed for arrays "pydub", # For audio processing "scipy", # For FFT and other signal processing "Pillow", "boto3", ) .entrypoint([]) .add_local_python_source("suno_utils", copy=False) ) app = modal.App(APP_NAME) # Set up shared volume for intermediate files VOLUME_NAME = "shader-outputs" outputs = modal.Volume.from_name(VOLUME_NAME, create_if_missing=True) OUTPUTS_PATH = "/outputs" SHARE_ASSET_WORK_DIR = os.path.join(OUTPUTS_PATH, "share_asset_work") S3_BUCKET_NAME = "suno-data-uploads" S3_PREFIX = "studio/uploads" # Filter warnings for specific messages warnings.filterwarnings("ignore", message=".*unknown element.*reset-dirs.*") @app.cls( secrets=SECRETS, cpu=4, # Start lean, can increase if needed gpu="T4", memory=4096, timeout=300, scaledown_window=300, min_containers=1 if DEPLOYMENT_TYPE == "dev" else 1, max_containers=50, volumes={OUTPUTS_PATH: outputs}, # Read-only access to shared resources image=worker_image, region="us-east", ) @modal.concurrent(max_inputs=4) class RenderVideoStub: """ Self-contained video rendering stub that handles the complete process from downloading assets to rendering and uploading the final video. This class uses the shared volume in read-only mode for accessing shared resources like presets, prerendered videos, and configuration files, but all dynamic processing is done in local filesystem to avoid volume synchronization issues. """ def __init__(self): print("RenderVideoStub initialized.") # Define persistent cache location self.persistent_cache_dir = "/tmp/suno_render_cache" os.makedirs(self.persistent_cache_dir, exist_ok=True) # Copy all required resources from Modal volume to local filesystem cache self._initialize_resource_cache() def _initialize_resource_cache(self): """ Copy all necessary resources from Modal volume to local filesystem. This is done once per container at initialization time. """ # Track volumes paths volume_resources_dir = os.path.join(OUTPUTS_PATH, "resources") volume_presets_dir = os.path.join(OUTPUTS_PATH, "presets") # Target paths in the persistent cache persistent_resources_dir = os.path.join(self.persistent_cache_dir, "resources") persistent_presets_dir = os.path.join(self.persistent_cache_dir, "presets") # Create cache init marker file cache_init_marker = os.path.join(self.persistent_cache_dir, ".cache_initialized") # Only initialize cache once per container if os.path.exists(cache_init_marker): print(f"Resource cache already initialized at {self.persistent_cache_dir}") return print(f"Initializing resource cache at {self.persistent_cache_dir}") # Ensure parent directory exists os.makedirs(self.persistent_cache_dir, exist_ok=True) # 1. Copy resources directory recursively if os.path.exists(volume_resources_dir): print(f"Recursively copying resources from {volume_resources_dir}") try: # Remove if exists to ensure clean copy if os.path.exists(persistent_resources_dir): shutil.rmtree(persistent_resources_dir) # Copy entire directory tree shutil.copytree(volume_resources_dir, persistent_resources_dir) print("Resources directory copied successfully") # Check for fonts and stickers fonts = [f for f in os.listdir(persistent_resources_dir) if f.endswith((".ttf", ".otf"))] stickers = [ f for f in os.listdir(persistent_resources_dir) if f.endswith((".png", ".jpg", ".jpeg")) ] print(f"Found {len(fonts)} font files in resources/") print(f"Found {len(stickers)} sticker files in resources/") # Add fallback fonts if no fonts found if not fonts: print("No fonts found, creating fallback fonts") try: # Try to find a system font to use as fallback system_fonts = [ "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", "/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf", "/usr/share/fonts/TTF/Arial.ttf", "/System/Library/Fonts/Helvetica.ttc", ] for font_path in system_fonts: if os.path.exists(font_path): print(f"Using system font as fallback: {font_path}") shutil.copy( font_path, os.path.join(persistent_resources_dir, "PPNeueMontreal-Bold.otf"), ) shutil.copy( font_path, os.path.join(persistent_resources_dir, "InputSans-Medium.otf"), ) break else: # If no system fonts found, create empty placeholder files print("No system fonts found, creating empty placeholder files") open( os.path.join(persistent_resources_dir, "PPNeueMontreal-Bold.otf"), "wb" ).close() open( os.path.join(persistent_resources_dir, "InputSans-Medium.otf"), "wb" ).close() except Exception as e: print(f"Error creating fallback fonts: {e}") # Create empty files to pass validation open( os.path.join(persistent_resources_dir, "PPNeueMontreal-Bold.otf"), "wb" ).close() open( os.path.join(persistent_resources_dir, "InputSans-Medium.otf"), "wb" ).close() except Exception as e: print(f"Error copying resources directory: {e}") # Create resources dir if it failed os.makedirs(persistent_resources_dir, exist_ok=True) # 2. Copy presets directory recursively if os.path.exists(volume_presets_dir): print(f"Recursively copying presets from {volume_presets_dir}") try: # Remove if exists to ensure clean copy if os.path.exists(persistent_presets_dir): shutil.rmtree(persistent_presets_dir) # Copy entire directory tree shutil.copytree(volume_presets_dir, persistent_presets_dir) print("Presets directory copied successfully") except Exception as e: print(f"Error copying presets directory: {e}") # Create presets dir if it failed os.makedirs(persistent_presets_dir, exist_ok=True) # Print summary of copied files print("\n=== RESOURCE CACHE SUMMARY ===") # Print flat file listing for resources dir if os.path.exists(persistent_resources_dir): print(f"\nResources directory ({persistent_resources_dir}):") for file in os.listdir(persistent_resources_dir): if os.path.isfile(os.path.join(persistent_resources_dir, file)): file_size = os.path.getsize(os.path.join(persistent_resources_dir, file)) print(f" {file} ({file_size/1024:.1f} KB)") elif os.path.isdir(os.path.join(persistent_resources_dir, file)): print(f" {file}/ (directory)") # Print directory structure for dir_name, subdir_list, file_list in os.walk(self.persistent_cache_dir): rel_path = os.path.relpath(dir_name, self.persistent_cache_dir) if rel_path == ".": continue print(f"Directory: {rel_path} - {len(file_list)} files") print("==============================\n") # Create marker file to indicate successful initialization with open(cache_init_marker, "w") as f: f.write(f"Cache initialized at {datetime.datetime.now().isoformat()}") print("Resource cache initialization complete") def _publish_event(self, asset_id, event_type, data={}): """Simple helper for progress reporting within the renderer""" print(f"{event_type} {data=}") events_queue.put( { "type": event_type, "data": data | {"timestamp": time.time()}, }, partition=asset_id, partition_ttl=300, # Keep partition alive ) def render_dynamic_overlay_iter(self, config: RenderConfig): """ Generator function yielding raw RGBA bytes for each frame of the dynamic overlay. Draws lyrics, timer, and other animated elements using Skia. """ self._publish_event(config.asset_id, "starting_dynamic_overlay_generation") assert skia is not None, "Skia is not installed" # Initialization - Use CANVAS dimensions surface = skia.Surface(config.canvas_width, config.canvas_height) canvas = surface.getCanvas() pixmap = skia.Pixmap() if not canvas.peekPixels(pixmap): raise RuntimeError("Could not peek pixels from canvas") # Load typefaces needed for dynamic elements lyrics_typeface = config.get_typeface(config.lyrics_font_path) timer_typeface = config.get_typeface(config.timer_font_path) # Load sticker configuration if available sticker_config = None timer_overlay = None spectrogram_overlay = None if ( hasattr(config, "sticker_config_path") and config.sticker_config_path and os.path.exists(config.sticker_config_path) ): try: with open(config.sticker_config_path, "r") as f: sticker_config = json.load(f) # Parse overlay configurations if sticker_config: # Get the appropriate overlay configuration for the current sticker style overlays = parse_sticker_overlays(sticker_config, config.sticker_style) # Look for a timer overlay in rendered_video_overlays for overlay in overlays.rendered_video_overlays: if overlay.video_source_id == "timer": timer_overlay = overlay elif overlay.video_source_id == "five_band_wave": spectrogram_overlay = overlay except Exception as e: print(f"WARNING: Failed to load or parse sticker configuration: {e}") # Timer setup (moved outside the frame loop) show_timer = hasattr(config, "show_timer") and config.show_timer temp_timer = timer_overlay if show_timer or timer_overlay: print( f"\nTIMER CONFIG: show_timer={getattr(config, 'show_timer', False)}, total_time={config.original_duration}" ) if timer_overlay: pos_info = getattr(timer_overlay, "ffmpeg_pos", "None") print(f"TIMER OVERLAY: {timer_overlay.video_source_id}, ffmpeg_pos={pos_info}") # Validate if the ffmpeg_pos is in a supported format if hasattr(timer_overlay, "ffmpeg_pos") and timer_overlay.ffmpeg_pos: ffmpeg_pos = timer_overlay.ffmpeg_pos # Check for common recognized formats has_numeric_y = ( any(c.isdigit() for c in ffmpeg_pos.split(":")[-1]) if ":" in ffmpeg_pos else False ) has_bottom = "bottom" in ffmpeg_pos has_formula = ("W" in ffmpeg_pos or "H" in ffmpeg_pos) and ( "(" in ffmpeg_pos or ")" in ffmpeg_pos ) # Keep the original configuration if it looks valid if has_numeric_y or has_bottom or has_formula: print(f"TIMER: Using original position format: {ffmpeg_pos}") else: # If overlay has a position property that's not in a recognizable format, fix it print(f"TIMER WARNING: Unrecognized position format: {ffmpeg_pos}") temp_timer = RenderedVideoOverlay( sticker_id="timer", video_source_id="timer", ffmpeg_pos="center:400" ) print("TIMER: Created replacement timer overlay with position 'center:400'") else: # If no ffmpeg_pos is provided, create a default one print("TIMER WARNING: No position specified in timer overlay") temp_timer = RenderedVideoOverlay( sticker_id="timer", video_source_id="timer", ffmpeg_pos="center:400" ) print("TIMER: Created default timer overlay with position 'center:400'") # Load lyrics data if available and initialize the LyricsRenderer lyrics_renderer = None if ( config.lyrics_style != "none" and config.local_lyrics_path and os.path.exists(config.local_lyrics_path) ): try: # Load lyrics using existing function lyric_lines = load_aligned_lyrics(config.local_lyrics_path) print( f"\nLYRICS CONFIG: style={config.lyrics_style}, lyrics_path={config.local_lyrics_path}" ) print(f"LYRICS LINES LOADED: {len(lyric_lines)}") # Initialize the lyrics renderer with the loaded lyrics and configuration lyrics_renderer = LyricsRenderer(lyrics=lyric_lines, typeface=lyrics_typeface) except Exception as e: print(f"WARNING: Failed to load or parse lyrics: {e}") traceback.print_exc() # Print full stack trace for debugging else: if config.lyrics_style == "none": print("LYRICS ERROR: Lyrics style is set to 'none', disabling lyrics display") elif not config.local_lyrics_path: print("LYRICS ERROR: No lyrics path provided in config") elif not os.path.exists(config.local_lyrics_path): print(f"LYRICS ERROR: Lyrics file not found at {config.local_lyrics_path}") else: print("LYRICS ERROR: Unknown error preventing lyrics initialization") # Initialize the spectrogram visualizer if config was found spectrogram_visualizer = None if spectrogram_overlay: try: # Use assets directory instead of job directory spectrogram_path = os.path.join(config.asset_dir, "spectrogram.png") if os.path.exists(spectrogram_path): spectrogram_visualizer = SpectrogramVisualizer( spectrogram_path=spectrogram_path, bands=5, # Use 5 bands as specified spectrogram_overlay=spectrogram_overlay, canvas_width=config.canvas_width, canvas_height=config.canvas_height, ) print(f"Initialized spectrogram visualizer from: {spectrogram_path}") else: print(f"WARNING: Spectrogram file not found at: {spectrogram_path}") except Exception as e: print(f"WARNING: Failed to initialize spectrogram visualizer: {e}") spectrogram_visualizer = None # Frame Generation Loop for frame_idx in range(config.frame_count): current_time = config.start_time + (frame_idx / config.fps) # 1. Clear Canvas (transparent) canvas.clear(skia.Color4f(0, 0, 0, 0)) # 2. Draw Lyrics if enabled and lyrics renderer is initialized if config.lyrics_style != "none" and lyrics_renderer: # Print lyrics rendering info for first frame and every 90 frames (3 seconds at 30fps) if frame_idx == 0: print( f"\nRendering lyrics with style '{config.lyrics_style}', time={current_time:.2f}s" ) # Render lyrics directly using the lyrics renderer lyrics_renderer.render( canvas=canvas, current_time=current_time, style=config.lyrics_style, ) # 3. Draw Spectrogram visualization if initialized if spectrogram_visualizer and spectrogram_visualizer.is_valid: spectrogram_visualizer.draw( canvas=canvas, current_time=current_time, total_time=config.duration, frame_idx=frame_idx, ) # 4. Draw Timer if enabled - Use CANVAS dimensions if show_timer or timer_overlay: draw_timer_on_canvas( canvas=canvas, typeface=timer_typeface, current_time=current_time, total_time=config.original_duration, width=config.canvas_width, # Use canvas width height=config.canvas_height, # Use canvas height timer_overlay=temp_timer, # Use our fixed position if needed font_size=32.0, ) # 5. Yield Frame Bytes (from the canvas-sized surface) yield pixmap # Clean up resources if lyrics_renderer: lyrics_renderer.clear_caches() self._publish_event(config.asset_id, "dynamic_overlay_generation_finished") def render_static_overlay_image(self, config: RenderConfig): """ Renders non-animated elements (logo, static text) to a single transparent PNG image using Skia. Output: config.static_overlay_path (PNG file) """ self._publish_event(config.asset_id, "rendering_static_overlay") assert skia is not None, "Skia is not installed" # Create a surface and canvas - Use CANVAS dimensions surface = skia.Surface(config.canvas_width, config.canvas_height) canvas = surface.getCanvas() bg_alpha = 0.0 if config.preset_id == "cover": bg_alpha = 0.5 if config.preset_id == PRESET_ID_PRERENDERED_BLACK_FIFTY: bg_alpha = 0.5 canvas.clear(skia.Color4f(0, 0, 0, bg_alpha)) # Transparent background # Parse sticker configuration sticker_overlays = None if ( hasattr(config, "sticker_config_path") and config.sticker_config_path and os.path.exists(config.sticker_config_path) ): try: with open(config.sticker_config_path, "r") as f: sticker_config = json.load(f) if sticker_config: sticker_overlays = parse_sticker_overlays(sticker_config, config.sticker_style) except Exception as e: print(f"WARNING: Failed to load or parse sticker configuration: {e}") # CRITICAL FIX: Override the resources_dir to use the persistent cache config.resources_dir = self.persistent_cache_dir + "/resources" # CRITICAL FIX: Also set absolute font paths using the persistent cache config.lyrics_font_path = os.path.join( self.persistent_cache_dir, "resources", "PPNeueMontreal-Bold.otf" ) config.title_font_path = os.path.join( self.persistent_cache_dir, "resources", "EditorialNew-Regular.otf" ) config.author_font_path = os.path.join( self.persistent_cache_dir, "resources", "PPNeueMontreal-Medium.otf" ) config.timer_font_path = os.path.join( self.persistent_cache_dir, "resources", "InputSans-Medium.otf" ) render_static_overlays_to_canvas(canvas, config, sticker_overlays) # Save the canvas as a PNG image = surface.makeImageSnapshot() png_data = image.encodeToData() if png_data is None: raise RuntimeError("Failed to encode static overlay image to PNG.") with open(config.static_overlay_path, "wb") as f: f.write(png_data.bytes()) self._publish_event(config.asset_id, "static_overlay_rendered") return config.static_overlay_path def render_shader_video(self, config: RenderConfig): """Renders shader base to config.shader_output_path using the extracted function.""" self._publish_event(config.asset_id, "rendering_shader_base") # Load shader buffer config required by the rendering function graphic_buffer_config = load_preset_buffer(OUTPUTS_PATH, "v1", config.preset_id, config.style_id) # Call the extracted shader rendering function rendered_path = render_shader_background_video( buffers=graphic_buffer_config, width=config.canvas_width, # Use canvas width for shader rendering height=config.canvas_height, # Use canvas height for shader rendering duration=config.duration, fps=config.fps, output_path=config.shader_output_path, # Use the designated path in config use_gpu=True, # Assuming GPU is desired/available # outputs_volume=outputs # Pass volume if commit is needed inside shader func ) self._publish_event(config.asset_id, "shader_base_rendered") # Set the base_layer_path for the composite step config.base_layer_path = rendered_path return rendered_path # Return the actual path returned by the function def prepare_media_base(self, config: RenderConfig): """ Downloads and prepares image/video covers if needed. Sets config.base_layer_path to the path of the prepared media. NOTE: This does NOT resize the media yet. Resizing happens in the composite step. """ if config.preset_id == "cover": target_filename_base = os.path.join(config.asset_dir, f"base_cover_media_{config.asset_id}") if config.style_id == "video" and config.video_cover_url: self._publish_event(config.asset_id, "downloading_video_cover") ext = os.path.splitext(config.video_cover_url)[1] or ".mp4" local_path = target_filename_base + ext download_video(url=config.video_cover_url, download_path=local_path) config.base_layer_path = local_path elif config.style_id == "image" and config.image_cover_url: self._publish_event(config.asset_id, "downloading_image_cover") ext = os.path.splitext(config.image_cover_url)[1] or ".png" local_path = target_filename_base + ext download_image(url=config.image_cover_url, save_path=local_path) config.base_layer_path = local_path elif config.image_cover_url: # Fallback for video style with only image url self._publish_event(config.asset_id, "downloading_image_cover_fallback") ext = os.path.splitext(config.image_cover_url)[1] or ".png" local_path = target_filename_base + ext download_image(url=config.image_cover_url, save_path=local_path) config.base_layer_path = local_path else: raise ValueError("Cover preset requires video_cover_url or image_cover_url") elif config.preset_id in [PRESET_ID_PRERENDERED_VIDEO, PRESET_ID_PRERENDERED_BLACK_FIFTY]: # Use the cached prerendered video prerendered_file = f"{config.style_id}" if not os.path.splitext(prerendered_file)[1]: prerendered_file += ".mp4" # Prerendered videos are in resources/prerendered_video cached_video_path = os.path.join( self.persistent_cache_dir, "resources", "prerendered_video", prerendered_file ) # If the specific file doesn't exist, use the fallback if not os.path.exists(cached_video_path): fallback_style = "aura_block_v1.mp4" cached_video_path = os.path.join( self.persistent_cache_dir, "resources", "prerendered_video", fallback_style ) # Copy from cache to local job directory local_base_media_path = os.path.join(config.asset_dir, os.path.basename(cached_video_path)) shutil.copy(cached_video_path, local_base_media_path) config.base_layer_path = local_base_media_path base_is_image = False else: # This is a shader-based preset self._publish_event( config.asset_id, "using_shader_for_preset", {"preset": config.preset_id, "style": config.style_id}, ) # Shader output will be handled by render_shader_video # No local_base_media_path needed as it will be set by render_shader_video return config.base_layer_path def composite(self, config: RenderConfig, dynamic_overlay_iter): """ Uses FFmpeg to composite the base layer, static overlay, and dynamic overlay (from stdin). This implementation uses an asynchronous producer-consumer pattern for optimal performance: 1. A producer thread generates frames as fast as possible from the dynamic_overlay_iter 2. Frames are stored in a memory buffer queue (up to 300 frames, ~10s at 30fps) 3. A consumer thread feeds frames to FFmpeg at the rate it can accept them 4. This decouples the frame generation from encoding, allowing both to run at optimal speeds The method also includes detailed performance timing to identify bottlenecks. """ func_start_time = time.perf_counter() iter_wait_time = 0.0 ffmpeg_feed_time = 0.0 # Time spent writing to ffmpeg stdin ffmpeg_wait_time = 0.0 setup_time = 0.0 cleanup_commit_time = 0.0 verification_time = 0.0 self._publish_event(config.asset_id, "starting_composition") # --- Setup Start --- cmd = ["ffmpeg", "-y"] # --- Input Definitions --- # Input 0: Base Layer (Video or Looped Image) base_is_image = config.base_layer_path.lower().endswith((".png", ".jpg", ".jpeg")) if base_is_image: cmd.extend(["-loop", "1", "-i", config.base_layer_path, "-t", str(config.duration)]) else: # Assumes video. Handle potential looping for short cover videos if necessary if config.preset_id == "cover" and config.style_id == "video": cmd.extend(["-stream_loop", "-1"]) # Loop video base if it's a cover video # Add buffer sizes for video input cmd.extend( [ "-thread_queue_size", "4096", # Increase thread queue size for video "-i", config.base_layer_path, ] ) # Input 1: Static Overlay (PNG) cmd.extend( [ "-thread_queue_size", "1024", # Smaller buffer for static image "-i", config.static_overlay_path, ] ) # Input 2: Dynamic Overlay (from stdin) cmd.extend( [ "-f", "rawvideo", "-pixel_format", "rgba", # Dynamic overlay is generated at the target canvas size (1080x1920) "-video_size", "1080x1920", "-framerate", str(config.fps), "-thread_queue_size", "8192", # Largest buffer for raw video input "-i", "pipe:0", ] ) # Input 3: Audio cmd.extend( [ "-thread_queue_size", "4096", # Audio buffer "-i", config.local_trimmed_audio_path, ] ) # --- Filter Complex --- filters = [] # Determine if we need to directly resize to the final dimensions needs_resize = (config.width != 1080) or (config.height != 1920) target_width = config.width target_height = config.height # 1. Scale base layer AND convert it to RGBA filters.append( "[0:v]scale=1080:1920:force_original_aspect_ratio=increase:flags=fast_bilinear,crop=1080:1920,format=rgba[scaled_base_rgba]" ) # 2. Overlay static elements (Input 1: PNG = RGBA) onto scaled RGBA base filters.append("[scaled_base_rgba][1:v]overlay=x=0:y=0[base_plus_static]") # 3. Extract Alpha from Dynamic Overlay (Input 2: Skia RGBA) filters.append("[2:v]alphaextract[dynamic_alpha]") # 4. Create a constant white source with the correct duration filters.append(f"color=white:s=1080x1920:d={config.duration}[white_src]") # 5. Merge white color with extracted dynamic alpha filters.append("[white_src][dynamic_alpha]alphamerge[dynamic_overlay_rgba]") # 6. Overlay the merged dynamic layer onto the static base result filters.append("[base_plus_static][dynamic_overlay_rgba]overlay=x=0:y=0[composited_rgba]") # 7. Scale the final RGBA result if needed if needs_resize: filters.append( f"[composited_rgba]scale={target_width}:{target_height}:flags=fast_bilinear[final_rgba]" ) pre_yuv_stream_name = "[final_rgba]" else: pre_yuv_stream_name = "[composited_rgba]" # 8. Convert the final composited stream to YUV420p for encoding filters.append(f"{pre_yuv_stream_name}format=pix_fmts=yuv420p[final_video]") final_stream_name = "[final_video]" filter_complex = ";".join(filters) cmd.extend(["-filter_complex", filter_complex]) # --- Output Mapping and Encoding --- cmd.extend( [ "-map", final_stream_name, # Map the correct final video stream from filter_complex "-map", "3:a", # Map the audio stream (ensure input 3 is the correct audio source) # Global muxing parameters # Frame rate control "-fps_mode", "cfr", # Ensure Constant Frame Rate output "-r", str( config.fps ), # Set the output frame rate (redundant with cfr sometimes, but good practice) # --- Video Encoder Settings (NVENC H.264) --- "-c:v", "h264_nvenc", # Optimized NVENC parameters "-preset", "p1", # Fastest preset "-tune", "ll", # Low latency tuning "-rc", "cbr", # Constant bitrate - faster than constqp for most content "-b:v", "2M", # 1 Mbps bitrate - good enough for small 412x732 video "-bufsize", "2M", # Match bitrate for CBR "-g", "60", # Fixed GOP size (2 seconds at 30fps) "-bf", "0", # Disable B-frames completely for faster encode # Output Format "-pix_fmt", "yuv420p", # Standard pixel format for web/mobile compatibility # --- Audio Settings --- "-c:a", "aac", # Audio Codec: AAC (Advanced Audio Coding) "-b:a", "192k", # Audio Bitrate: 192 kbps # --- Container/Muxing Settings --- "-shortest", # Finish encoding when the shortest input stream ends "-movflags", "+faststart", # Optimize MP4 for streaming (move metadata to the beginning) config.final_output_path, # The final output file path ] ) print("\n=== FFMPEG COMMAND ===") print(" ".join(cmd)) print("======================\n") # --- Run FFmpeg and Pipe Data --- ffmpeg_process = None try: # --- Setup End / Process Start --- setup_end_time = time.perf_counter() setup_time = setup_end_time - func_start_time # Create FFmpeg process with non-blocking stderr ffmpeg_process = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if not ffmpeg_process.stdin: raise RuntimeError("FFmpeg stdin pipe is not available.") # --- ASYNCHRONOUS FRAME PROCESSING --- # This section uses a producer-consumer pattern to decouple frame generation # from frame encoding. The producer thread can run ahead of FFmpeg, filling # a buffer, while the consumer feeds frames to FFmpeg at the rate it accepts them. frame_count = 0 loop_start_time = time.perf_counter() last_iter_end_time = loop_start_time # Initialize # Create a queue to buffer frames between the iterator and FFmpeg buffer_size = 300 # ~10 seconds at 30fps frame_queue = queue.Queue(maxsize=buffer_size) self._publish_event( config.asset_id, "async_buffer_initialized", {"buffer_size": buffer_size} ) # Flag to signal when iteration is complete iteration_done = threading.Event() # Flag to signal if an error occurred in any thread error_event = threading.Event() error_message = [""] # List to hold error message (mutable) # Producer function to run in a separate thread def frame_producer(): nonlocal frame_count, iter_wait_time, last_iter_end_time producer_start_time = time.perf_counter() try: frame_idx = 0 for pixmap in dynamic_overlay_iter: frame_idx += 1 iter_end_time = time.perf_counter() # Time spent waiting FOR the iterator to yield the next item iter_wait_time += iter_end_time - last_iter_end_time # Put frame in queue - will block if queue is full frame_bytes = pixmap.tobytes() frame_queue.put((frame_idx, frame_bytes)) frame_count += 1 # Log progress periodically if frame_idx % 300 == 0: # Every ~10 seconds at 30fps buffer_fullness = frame_queue.qsize() / buffer_size * 100 elapsed = time.perf_counter() - producer_start_time fps = frame_idx / elapsed if elapsed > 0 else 0 print( f"PRODUCER: Frames={frame_idx} | Buffer={buffer_fullness:.1f}% | FPS={fps:.1f}" ) # Record time after processing this frame last_iter_end_time = time.perf_counter() # Check if an error occurred in the consumer thread if error_event.is_set(): print("ERROR: Producer stopping due to consumer error") break except Exception as e: error_event.set() error_message[0] = f"Producer thread error: {str(e)}" print(f"ERROR: Producer thread failed: {str(e)}") finally: # Signal that all frames have been produced iteration_done.set() producer_end_time = time.perf_counter() total_time = producer_end_time - producer_start_time print( f"PRODUCER COMPLETE: Frames={frame_count} | Time={total_time:.2f}s | FPS={frame_count/total_time:.1f}" ) # Consumer function (runs in main thread) def frame_consumer(): nonlocal ffmpeg_feed_time frame_processed = 0 consumer_start_time = time.perf_counter() try: # Continue until producer is done AND queue is empty while not (iteration_done.is_set() and frame_queue.empty()): try: # Get frame with timeout to check flags periodically frame_idx, frame_bytes = frame_queue.get(timeout=0.1) # Write frame to FFmpeg feed_start_time = time.perf_counter() if ffmpeg_process and ffmpeg_process.stdin: ffmpeg_process.stdin.write(frame_bytes) feed_end_time = time.perf_counter() # Time spent writing to FFmpeg ffmpeg_feed_time += feed_end_time - feed_start_time frame_processed += 1 # Progress reporting if frame_processed % 60 == 0: # Every ~2 seconds at 30fps queue_size = frame_queue.qsize() buffer_fullness = queue_size / buffer_size * 100 elapsed = time.perf_counter() - consumer_start_time fps = frame_processed / elapsed if elapsed > 0 else 0 print( f"CONSUMER: Frames={frame_processed} | Buffer={buffer_fullness:.1f}% | FPS={fps:.1f}" ) except queue.Empty: # Queue is empty but iteration might not be done if iteration_done.is_set(): # Just a final check - producer is done, make sure we've processed all frames if frame_queue.empty(): break continue except (BrokenPipeError, IOError) as e: error_event.set() error_message[0] = f"Broken pipe in consumer: {str(e)}" print(f"ERROR: Writing to ffmpeg stdin failed: {e}") except Exception as e: error_event.set() error_message[0] = f"Consumer thread error: {str(e)}" print(f"ERROR: Consumer thread failed: {str(e)}") finally: consumer_end_time = time.perf_counter() total_time = consumer_end_time - consumer_start_time print( f"CONSUMER COMPLETE: Frames={frame_processed} | Time={total_time:.2f}s | FPS={frame_processed/total_time:.1f}" ) # Start producer thread and log the event self._publish_event(config.asset_id, "async_producer_starting") producer_thread = threading.Thread(target=frame_producer) producer_thread.daemon = True # Allow program to exit if thread is still running producer_thread.start() # Run consumer in main thread and log the event self._publish_event(config.asset_id, "async_consumer_starting") frame_consumer() # Wait for producer to finish and log completion producer_thread.join(timeout=60) # Wait up to a minute for thread to finish if producer_thread.is_alive(): print("Warning: Producer thread did not complete within timeout period") # Check for errors if error_event.is_set() and error_message[0]: self._publish_event( config.asset_id, "async_processing_error", {"message": error_message[0]} ) raise RuntimeError(f"Error during frame processing: {error_message[0]}") self._publish_event( config.asset_id, "async_processing_complete", {"frames_generated": frame_count, "buffer_size": buffer_size}, ) # --- End Async Processing --- # --- Cleanup/Wait Start --- pre_close_time = time.perf_counter() try: if ffmpeg_process.stdin: ffmpeg_process.stdin.close() except Exception as close_err: # Log error but proceed to wait, pipe might already be closed print(f"WARNING: Error closing stdin pipe: {close_err}") pre_wait_time = time.perf_counter() ffmpeg_process.wait() # Wait for FFmpeg process to finish post_wait_time = time.perf_counter() ffmpeg_wait_time = post_wait_time - pre_wait_time # Time spent blocked on wait() return_code = ffmpeg_process.returncode # Read stderr *after* waiting stderr_output = "" if ffmpeg_process.stderr: try: stderr_bytes = ffmpeg_process.stderr.read() stderr_output = stderr_bytes.decode(errors="ignore") except Exception as read_err: print(f"WARNING: Error reading stderr after wait: {read_err}") if return_code != 0: print("ERROR: FFmpeg failed with exit code", return_code) print("\n=== FFMPEG STDERR ===") print(stderr_output) print("=====================\n") raise RuntimeError(f"FFmpeg composition failed with code {return_code}") # Pre-verify time pre_verify_time = time.perf_counter() if ( not os.path.exists(config.final_output_path) or os.path.getsize(config.final_output_path) == 0 ): print("ERROR: Output file missing or empty:", config.final_output_path) raise RuntimeError("FFmpeg output file is missing or empty") # Verify integrity try: verify_cmd = ["ffprobe", "-v", "error", config.final_output_path] # capture_output=False to avoid excessive memory usage if ffprobe has output subprocess.run(verify_cmd, check=True, capture_output=False, text=True) except subprocess.CalledProcessError as verify_error: verify_error_msg = ( verify_error.stderr if hasattr(verify_error, "stderr") and verify_error.stderr else "Unknown error" ) print(f"ERROR: FFprobe validation failed: {verify_error_msg}") raise RuntimeError(f"Generated video file is corrupt: {verify_error_msg}") except Exception as verify_error: print(f"WARNING: Error running ffprobe verification: {verify_error}") # Decide if verification failure should be fatal post_verify_time = time.perf_counter() verification_time = post_verify_time - pre_verify_time # --- Verification End --- # --- Final Cleanup/Commit Start --- pre_commit_time = time.perf_counter() self._publish_event(config.asset_id, "composition_finished") outputs.commit() # Ensure final video is written post_commit_time = time.perf_counter() # Cleanup time includes closing pipe, reading stderr, and final commit cleanup_commit_time = ( (pre_wait_time - pre_close_time) + (pre_verify_time - post_wait_time) + (post_commit_time - pre_commit_time) ) # --- Log Performance --- func_end_time = time.perf_counter() total_time = func_end_time - func_start_time # Calculate the frame rate for generation and processing generation_fps = frame_count / iter_wait_time if iter_wait_time > 0 else 0 processing_fps = frame_count / ffmpeg_feed_time if ffmpeg_feed_time > 0 else 0 overall_fps = frame_count / total_time print("\n=== PERFORMANCE SUMMARY ===") print(f"Total Time: {total_time:.2f}s ({overall_fps:.1f} fps)") print(f" Setup: {setup_time:.2f}s") print(f" Generation: {iter_wait_time:.2f}s ({generation_fps:.1f} fps)") print(f" Encoding: {ffmpeg_feed_time:.2f}s ({processing_fps:.1f} fps)") print(f" FFmpeg: {ffmpeg_wait_time:.2f}s") print(f" Verify: {verification_time:.2f}s") print(f" Cleanup: {cleanup_commit_time:.2f}s") print("\n=== BUFFER ANALYSIS ===") print(f" Generator Speed: {generation_fps:.1f} fps") print(f" Encoder Speed: {processing_fps:.1f} fps") ratio = generation_fps / processing_fps if processing_fps > 0 else 0 ratio_note = "generator faster" if ratio > 1 else "encoder faster" print(f" Speed Ratio: {ratio:.2f}x ({ratio_note})") print(f" Buffer Size: {buffer_size} frames ({buffer_size/30:.1f}s)") print(f" Total Frames: {frame_count}") print("==========================") self._publish_event( config.asset_id, "performance_stats", { "total_time": total_time, "overall_fps": overall_fps, "generator_fps": generation_fps, "consumer_fps": processing_fps, "frames": frame_count, }, ) return config.final_output_path except Exception as e: self._publish_event(config.asset_id, "composition_error", {"message": str(e)}) print(f"ERROR: FFmpeg composition failed: {str(e)}") # --- ADDED: Print stderr on error --- stderr_output = "" if ffmpeg_process and ffmpeg_process.stderr: try: # Try to read whatever stderr is available stderr_bytes = ffmpeg_process.stderr.read() stderr_output = stderr_bytes.decode(errors="ignore") print("\n=== FFMPEG STDERR (on error) ===") print(stderr_output) print("=============================\n") except Exception as read_err: print(f"WARNING: Error reading stderr during exception handling: {read_err}") # --- END ADDED --- # Attempt cleanup if process exists and is running if ffmpeg_process and ffmpeg_process.poll() is None: print("WARNING: Killing FFmpeg process due to error") ffmpeg_process.kill() # --- Log Performance on Error --- func_end_time = time.perf_counter() total_time = func_end_time - func_start_time print("\n=== ERROR PERFORMANCE DATA ===") print(f"Total Time: {total_time:.2f}s (interrupted)") print(f" Setup: {setup_time:.2f}s") print(f" Generation: {iter_wait_time:.2f}s") print(f" Encoding: {ffmpeg_feed_time:.2f}s") print(f" FFmpeg: {ffmpeg_wait_time:.2f}s") print("=============================\n") raise @modal.method() def full_render(self, render_config_dict: dict): """ Completely self-contained video rendering pipeline for a shareable asset. This method handles the entire workflow: 1. Downloads all required assets directly from S3 2. Processes everything locally in a temp directory 3. Renders the complete video 4. Uploads the final result to S3 5. Returns a status dictionary with all relevant information No file-system synchronization with other Modal functions is required. """ import tempfile # Initialize S3 client for downloading/uploading s3_client = boto3.client("s3") # Use the persistent cache initialized in __init__ # Keep directory structure simple persistent_cache_dir = self.persistent_cache_dir persistent_resources_dir = os.path.join(persistent_cache_dir, "resources") persistent_presets_dir = os.path.join(persistent_cache_dir, "presets") # Create a local temp directory for job-specific processing local_temp_dir = tempfile.mkdtemp(prefix="suno_render_job_") local_asset_dir = os.path.join(local_temp_dir, "assets") os.makedirs(local_asset_dir, exist_ok=True) # Extract key information from config and ensure non-None values asset_id = render_config_dict.get("asset_id", "unknown_asset") clip_id = render_config_dict.get("clip_id", "unknown_clip") start_time = render_config_dict.get("start_time", 0.0) end_time = render_config_dict.get("end_time", 15.0) default_video_only = render_config_dict.get("default_video_only", False) # Ensure style values have safe defaults for path joining preset_id = render_config_dict.get("preset_id", "unknown_preset") style_id = render_config_dict.get("preset_style", "unknown_style") sticker_style = render_config_dict.get("sticker_style", "none") presets_version = render_config_dict.get("render_pipeline_version", "v1") try: self._publish_event(asset_id, "started_processing") # ---------- STEP 1: SETUP AND VALIDATE PERSISTENT CACHE ---------- # The cache is now initialized in __init__, so we just need to verify it exists if ( not os.path.exists(persistent_cache_dir) or not os.path.exists(persistent_resources_dir) or not os.path.exists(persistent_presets_dir) ): raise RuntimeError( f"Persistent cache not properly initialized at {persistent_cache_dir}" ) self._publish_event(asset_id, "using_resource_cache", {"cache_dir": persistent_cache_dir}) # ---------- STEP 2: DOWNLOAD AND PREPARE AUDIO ---------- self._publish_event(asset_id, "downloading_audio") audio_s3_key = f"studio/uploads/{clip_id}.mp3" # Use the original download_and_trim_audio function # which handles S3 download and trimming in one step audio_result = download_and_trim_audio( S3_BUCKET_NAME, audio_s3_key, start_time, end_time, local_asset_dir ) # ---------- STEP 4: DOWNLOAD AND PREPARE LYRICS ---------- lyrics_path = None lyrics_style_actual = "none" lyrics_style_requested = render_config_dict.get("lyrics_style", "omni") if lyrics_style_requested != "none": self._publish_event(asset_id, "downloading_lyrics") lyrics_s3_key = f"studio/uploads/{clip_id}_hoot.json" local_lyrics_path = os.path.join(local_asset_dir, "lyrics.json") try: s3_client.download_file(S3_BUCKET_NAME, lyrics_s3_key, local_lyrics_path) # Validate lyrics content with open(local_lyrics_path, "r", encoding="utf-8") as f: content = json.load(f) if isinstance(content, list) and len(content) > 0: lyrics_path = local_lyrics_path lyrics_style_actual = lyrics_style_requested except Exception as e: print(f"WARNING: Failed to download or parse lyrics: {e}") lyrics_path = None # ---------- STEP 5: DETERMINE AND PREPARE BASE MEDIA ---------- local_base_media_path = None base_is_image = False base_is_shader = False if preset_id == "cover": video_url = render_config_dict.get("video_cover_url") image_url = render_config_dict.get("image_cover_url") if style_id == "video" and video_url: self._publish_event(asset_id, "downloading_video_cover", {"url": video_url}) ext = os.path.splitext(video_url)[1] or ".mp4" local_base_media_path = os.path.join(local_asset_dir, f"base_cover_media{ext}") download_video(url=video_url, download_path=local_base_media_path) base_is_image = False elif style_id == "image" and image_url: self._publish_event(asset_id, "downloading_image_cover", {"url": image_url}) ext = os.path.splitext(image_url)[1] or ".png" local_base_media_path = os.path.join(local_asset_dir, f"base_cover_media{ext}") download_image(url=image_url, save_path=local_base_media_path) base_is_image = True elif image_url: self._publish_event( asset_id, "downloading_image_cover", {"url": image_url, "reason": "Fallback"} ) ext = os.path.splitext(image_url)[1] or ".png" local_base_media_path = os.path.join(local_asset_dir, f"base_cover_media{ext}") download_image(url=image_url, save_path=local_base_media_path) base_is_image = True else: raise ValueError("Cover preset selected but no valid image/video URL found") elif preset_id in [PRESET_ID_PRERENDERED_VIDEO, PRESET_ID_PRERENDERED_BLACK_FIFTY]: # Use the cached prerendered video prerendered_file = f"{style_id}" if not os.path.splitext(prerendered_file)[1]: prerendered_file += ".mp4" # Prerendered videos are in resources/prerendered_video cached_video_path = os.path.join( persistent_resources_dir, "prerendered_video", prerendered_file ) # If the specific file doesn't exist, use the fallback if not os.path.exists(cached_video_path): fallback_style = "aura_block_v1.mp4" cached_video_path = os.path.join( persistent_resources_dir, "prerendered_video", fallback_style ) # Copy from cache to local job directory local_base_media_path = os.path.join( local_asset_dir, os.path.basename(cached_video_path) ) shutil.copy(cached_video_path, local_base_media_path) base_is_image = False else: # This is a shader-based preset self._publish_event( asset_id, "using_shader_for_preset", {"preset": preset_id, "style": style_id} ) base_is_shader = True # Shader output will be determined later in the render process # No local_base_media_path needed as it will be set by render_shader_video # ---------- STEP 6: PREPARE STICKER CONFIGURATION ---------- sticker_config_path = None # Use the cached sticker config file cached_config_file = f"asset_config_list_{presets_version}.json" persistent_asset_config_dir = os.path.join(persistent_presets_dir, "asset_config") cached_config_path = os.path.join(persistent_asset_config_dir, cached_config_file) if os.path.exists(cached_config_path): # Copy to job-specific directory local_asset_config_dir = os.path.join(local_temp_dir, "presets", "asset_config") os.makedirs(local_asset_config_dir, exist_ok=True) local_config_path = os.path.join(local_asset_config_dir, cached_config_file) shutil.copy(cached_config_path, local_config_path) sticker_config_path = local_config_path self._publish_event(asset_id, "using_sticker_config", {"sticker_style": sticker_style}) else: self._publish_event( asset_id, "sticker_config_not_found", {"sticker_style": sticker_style, "path_checked": cached_config_path}, ) # ---------- STEP 7: BUILD COMPLETE RENDERING CONFIG ---------- config = RenderConfig( { # Copy original config values **render_config_dict, # Override with locally prepared paths "job_output_dir": local_temp_dir, "asset_id": asset_id, "local_trimmed_audio_path": audio_result["trimmed_audio_path"], "original_duration": audio_result["original_duration"], "local_lyrics_path": lyrics_path, "lyrics_style": lyrics_style_actual, "local_base_media_path": local_base_media_path, "base_media_is_static_image": base_is_image, "base_is_shader": base_is_shader, "sticker_config_path": sticker_config_path, # Just use persistent resources directory directly for all resources "resources_dir": persistent_resources_dir, }, local_temp_dir, ) # Configure additional paths config.asset_dir = local_asset_dir # Set default font paths if needed (but not overriding if already set) if not hasattr(config, "title_font_path") or not config.title_font_path: config.title_font_path = "EditorialNew-Regular.otf" if not hasattr(config, "author_font_path") or not config.author_font_path: config.author_font_path = "PPNeueMontreal-Medium.otf" if not hasattr(config, "lyrics_font_path") or not config.lyrics_font_path: config.lyrics_font_path = "PPNeueMontreal-Bold.otf" if not hasattr(config, "timer_font_path") or not config.timer_font_path: config.timer_font_path = "InputSans-Medium.otf" # No need to set suno_logo_path - all images are handled through stickers # ---------- STEP 7.5: VALIDATE RESOURCES ---------- missing_resources = [] warnings = [] available_resources = [] # 1. Verify audio is available - critical if not os.path.exists(audio_result["trimmed_audio_path"]): missing_resources.append(f"Audio file: {audio_result['trimmed_audio_path']}") else: available_resources.append( f"Audio file: {audio_result['trimmed_audio_path']} ({os.path.getsize(audio_result['trimmed_audio_path'])/1024:.1f} KB)" ) # 2. Verify base media if needed - critical if not base_is_shader: if local_base_media_path is None or not os.path.exists(local_base_media_path): missing_resources.append(f"Base media: {local_base_media_path}") else: available_resources.append( f"Base media: {local_base_media_path} ({os.path.getsize(local_base_media_path)/1024/1024:.1f} MB)" ) else: available_resources.append( f"Base media: Using shader renderer for '{preset_id}/{style_id}'" ) # 3. Verify fonts needed for rendering - warnings only, not critical required_fonts = [] if ( hasattr(config, "lyrics_font_path") and config.lyrics_font_path and lyrics_style_actual != "none" ): font_filename = os.path.basename(config.lyrics_font_path) # Look directly in resources/ directory font_path = os.path.join(persistent_resources_dir, font_filename) required_fonts.append(("Lyrics font", font_path)) if ( hasattr(config, "timer_font_path") and config.timer_font_path and hasattr(config, "show_timer") and config.show_timer ): font_filename = os.path.basename(config.timer_font_path) # Look directly in resources/ directory font_path = os.path.join(persistent_resources_dir, font_filename) required_fonts.append(("Timer font", font_path)) for font_desc, font_path in required_fonts: if not os.path.exists(font_path): warnings.append(f"{font_desc}: {font_path}") else: available_resources.append( f"{font_desc}: {font_path} ({os.path.getsize(font_path)/1024:.1f} KB)" ) # 4. Verify sticker config if non-default sticker style is used - warning only if sticker_style != "none": if sticker_config_path is None or not os.path.exists(sticker_config_path): warnings.append(f"Sticker config: {sticker_config_path}") else: available_resources.append( f"Sticker config: {sticker_config_path} ({os.path.getsize(sticker_config_path)/1024:.1f} KB)" ) # Check stickers directly in resources directory sticker_files = [ f for f in os.listdir(persistent_resources_dir) if f.endswith((".png", ".jpg", ".jpeg")) ] if sticker_files: sticker_count = len(sticker_files) available_resources.append( f"Stickers: {persistent_resources_dir} ({sticker_count} sticker files)" ) else: warnings.append(f"No sticker files found in: {persistent_resources_dir}") # 5. Check spectrogram if needed for visualization - not critical if "spectrogram_path" in audio_result and audio_result["spectrogram_path"]: spectrogram_path = audio_result["spectrogram_path"] if not os.path.exists(spectrogram_path): warnings.append(f"Spectrogram: {spectrogram_path}") else: available_resources.append( f"Spectrogram: {spectrogram_path} ({os.path.getsize(spectrogram_path)/1024:.1f} KB)" ) # 6. List persistent cache info font_files = [ f for f in os.listdir(persistent_resources_dir) if f.endswith((".ttf", ".otf")) ] sticker_files = [ f for f in os.listdir(persistent_resources_dir) if f.endswith((".png", ".jpg", ".jpeg")) ] prerendered_path = os.path.join(persistent_resources_dir, "prerendered_video") prerendered_count = ( len(os.listdir(prerendered_path)) if os.path.exists(prerendered_path) else 0 ) cache_info = [ f"Persistent cache: {persistent_cache_dir}", f"Font files: {len(font_files)}", f"Sticker files: {len(sticker_files)}", f"Prerendered videos: {prerendered_count}", f"Config files: {len(os.listdir(persistent_presets_dir)) if os.path.exists(persistent_presets_dir) else 0}", ] # Print all resource info for debugging print("\n===== RESOURCE VALIDATION REPORT =====") if available_resources: print("\nAVAILABLE RESOURCES:") for resource in available_resources: print(f"✓ {resource}") if warnings: print("\nWARNINGS (non-critical):") for resource in warnings: print(f"⚠ {resource}") if missing_resources: print("\nMISSING RESOURCES (critical):") for resource in missing_resources: print(f"✗ {resource}") print("\nCACHE INFO:") for info in cache_info: print(f"• {info}") print("\nCONFIG SUMMARY:") print(f"• Asset ID: {asset_id}") print(f"• Preset/Style: {preset_id}/{style_id}") print(f"• Sticker Style: {sticker_style}") print(f"• Lyrics Style: {lyrics_style_actual}") print( f"• Duration: {start_time:.1f}s to {end_time:.1f}s (total: {end_time-start_time:.1f}s)" ) print("=====================================\n") # Publish resource validation data self._publish_event( asset_id, "resource_validation_report", { "available": available_resources, "warnings": warnings, "missing": missing_resources, "cache_info": cache_info, }, ) # Raise error with detailed report if any CRITICAL resources are missing if missing_resources: error_message = "Missing required resources for rendering:\n" + "\n".join( missing_resources ) self._publish_event( asset_id, "resource_validation_failed", {"missing": missing_resources} ) raise FileNotFoundError(error_message) # Validation successful, but log warnings if warnings: self._publish_event(asset_id, "resource_validation_warnings", {"warnings": warnings}) else: self._publish_event(asset_id, "resource_validation_passed") # ---------- STEP 8: RENDER THE VIDEO ---------- self._publish_event(asset_id, "starting_render_process") # Prepare Base Layer if config.base_is_shader: # Call the shader rendering method with proper configuration self._publish_event( asset_id, "rendering_shader_base", {"preset": preset_id, "style": style_id} ) # Set up shader output path shader_output_path = os.path.join(local_asset_dir, f"shader_base_{asset_id}.mp4") config.shader_output_path = shader_output_path # Render the shader video renderer = RenderVideoStub() rendered_path = renderer.render_shader_video(config) # Set the base layer path for the composite step config.base_layer_path = rendered_path else: # Use the existing media base self.prepare_media_base(config) # Render Static Overlay self.render_static_overlay_image(config) # Prepare Dynamic Overlay Iterator dynamic_iter = self.render_dynamic_overlay_iter(config) # Composite Final Video final_video_path = self.composite(config, dynamic_iter) # ---------- STEP 9: UPLOAD FINAL VIDEO TO S3 ---------- self._publish_event(asset_id, "uploading_to_s3") # Generate S3 key asset_s3_id = clip_id if default_video_only else f"share_{clip_id}_{asset_id}" s3_key = f"{S3_PREFIX}/{asset_s3_id}.mp4" # Upload with appropriate metadata try: if default_video_only: s3_client.upload_file(final_video_path, S3_BUCKET_NAME, s3_key) else: expiration_date = datetime.datetime.utcnow() + timedelta(days=1) formatted_date = expiration_date.strftime("%Y-%m-%d") metadata = {"expiration_date": formatted_date} s3_client.upload_file( final_video_path, S3_BUCKET_NAME, s3_key, ExtraArgs={ "Metadata": metadata, "Expires": expiration_date, }, ) # Try to set lifecycle tag as well try: s3_client.put_object_tagging( Bucket=S3_BUCKET_NAME, Key=s3_key, Tagging={"TagSet": [{"Key": "expiration_date", "Value": formatted_date}]}, ) except Exception as tag_error: print(f"WARNING: Failed to set expiration tag: {tag_error}") asset_url = f"https://cdn1.suno.ai/{asset_s3_id}.mp4" self._publish_event(asset_id, "upload_complete", {"asset_url": asset_url}) return { "status": "completed", "asset_url": asset_url, "asset_s3_id": asset_s3_id, "render_id": str(uuid.uuid4()), } except Exception as upload_error: raise RuntimeError(f"Failed to upload final video to S3: {upload_error}") except Exception as e: print(traceback.format_exc()) self._publish_event(asset_id, "render_error", {"message": str(e)}) return { "status": "error", "error_message": str(e), "error_type": type(e).__name__, } finally: # Clean up temporary job directory, but keep the persistent cache try: shutil.rmtree(local_temp_dir) except Exception as cleanup_error: print(f"WARNING: Error during local temp cleanup: {cleanup_error}") @app.cls( secrets=SECRETS, cpu=4, memory=1024, timeout=360, scaledown_window=300, min_containers=1 if DEPLOYMENT_TYPE == "dev" else 1, # Scale down to 0 in dev max_containers=50, image=base_image, region="us-east", ) @modal.concurrent(max_inputs=4) class ShareAssetStub: """ Simplified stub that delegates all processing to RenderVideoStub. This class doesn't need access to the shared volume since all file operations are handled by RenderVideoStub. """ def __init__(self): print("ShareAssetStub initialized.") def _publish_event(self, partition, event_type, data={}, partition_ttl=300): """Simple helper for progress reporting and event tracking""" print(f"{event_type} {data=}") events_queue.put( { "type": event_type, "data": data | {"timestamp": time.time()}, }, partition=partition, partition_ttl=partition_ttl, ) @modal.method() def generate_share_asset(self, queue_item_json: str): """ Simplified handler for share asset requests that delegates all processing to RenderVideoStub. This method: 1. Parses the queue item 2. Creates the render configuration 3. Calls RenderVideoStub.full_render to handle all processing 4. Processes the result to notify the caller No file system operations or synchronization required. """ process_start_time = time.time() item = QueueItem(**json.loads(queue_item_json)) asset_id = item.id try: # Extract the render configuration render_config = item.metadata.get("render_config") if not render_config: raise ValueError(f"Missing render_config for asset {asset_id}") # Validate clip ID clip_id = render_config.get("clip_id") if not clip_id: raise ValueError(f"Missing clip_id for asset {asset_id}") # Add asset_id and other required fields renderer_config = render_config.copy() renderer_config["asset_id"] = asset_id renderer_config["default_video_only"] = item.metadata.get("default_video_only", False) # Publish event so the caller knows processing has started self._publish_event(asset_id, "started_processing") # IMPORTANT: Call RenderVideoStub.full_render to handle everything self._publish_event(asset_id, "invoking_renderer") render_result = RenderVideoStub().full_render.remote(renderer_config) # Process the result if render_result and render_result.get("status") == "completed": # Successful render asset_url = render_result.get("asset_url") asset_s3_id = render_result.get("asset_s3_id") render_id = render_result.get("render_id", str(uuid.uuid4())) self._publish_event( asset_id, "share_asset_finished", {"asset_url": asset_url, "asset_s3_id": asset_s3_id}, ) # Notify the caller item.notify_progress( { "type": "share_asset_finished", "id": asset_id, "asset_url": asset_url, "asset_s3_id": asset_s3_id, "render_id": render_id, "status": "complete", } ) # Return the asset URL for use in the local entrypoint return asset_url else: # Failed render error_msg = render_result.get("error_message", "Unknown error during rendering") error_type = render_result.get("error_type", "RenderError") raise RuntimeError(f"Rendering failed: {error_msg}") except Exception as e: print(f"ERROR: In generate_share_asset for asset {asset_id}: {e}") print(traceback.format_exc()) self._publish_event(asset_id, "error", {"message": str(e)}) # Notify the caller about the error item.notify_progress( { "type": "error", "id": asset_id, "error_type": type(e).__name__, "error_message": str(e), "status": "error", } ) # Return None or error info in case of failure return None finally: # Report completion time self._publish_event(asset_id, "completed", {"duration": time.time() - process_start_time}) @app.local_entrypoint() def main(): test_items = { "meta_data_cover_image": { # Visual configuration "render_pipeline_version": "v1", "preset_id": "cover", "preset_style": "image", "width": 1080, "height": 1920, "sticker_style": "core_lyrics_standard", "image_cover_url": "https://cdn2.suno.ai/5962bc75-c5e0-458c-848c-512eae911f64_7c0c4ce4.jpeg", "video_cover_url": "https://cdn1.suno.ai/video_upload_0748ec6f-112f-490f-bfaf-d8547e2906a7.mp4", # Timing configuration "start_time": 30.0, "end_time": 60.0, # Content identifiers "clip_id": "95dc0c01-636a-424c-82df-8af739d0fda9", "clip_title": "Jamaican Ginger", "clip_author": "sunnysuno", # Styling preferences "lyrics_style": "lyrics_box", }, "meta_data_cover_video": { # Visual configuration "render_pipeline_version": "v1", "preset_id": "cover", "preset_style": "video", "width": 1080, "height": 1920, "sticker_style": "core_info_standard", "image_cover_url": "https://cdn2.suno.ai/5962bc75-c5e0-458c-848c-512eae911f64_7c0c4ce4.jpeg", "video_cover_url": "https://cdn1.suno.ai/video_upload_0748ec6f-112f-490f-bfaf-d8547e2906a7.mp4", # Timing configuration "start_time": 30.0, "end_time": 60.0, # Content identifiers "clip_id": "d9d0fd10-eedc-4bc7-951c-31c179599a95", "clip_title": "Party at Suno", "clip_author": "sunnysuno", # Styling preferences "lyrics_style": "lyrics_three_line", }, "meta_data_prerendered_as_is": { # Visual configuration "render_pipeline_version": "v1", "preset_id": "prerendered_video", "preset_style": "aura_video_daylight_v2.mp4", "width": 1080, "height": 1920, "sticker_style": "core_info_standard", "image_cover_url": "https://cdn2.suno.ai/5962bc75-c5e0-458c-848c-512eae911f64_7c0c4ce4.jpeg", "video_cover_url": "https://cdn1.suno.ai/video_upload_0748ec6f-112f-490f-bfaf-d8547e2906a7.mp4", # Timing configuration "start_time": 30.0, "end_time": 60.0, # Content identifiers "clip_id": "d9d0fd10-eedc-4bc7-951c-31c179599a95", "clip_title": "Party at Suno", "clip_author": "sunnysuno", # Styling preferences "lyrics_style": "lyrics_three_line", }, "meta_data_prerendered_overlay": { "render_pipeline_version": "v1", "preset_id": "prerendered_video_with_black_overlay_fifty", "preset_style": "aura_video_daylight_v2.mp4", "width": 1080, "height": 1920, "sticker_style": "core_info_standard", "image_cover_url": "https://cdn2.suno.ai/5962bc75-c5e0-458c-848c-512eae911f64_7c0c4ce4.jpeg", "video_cover_url": "https://cdn1.suno.ai/video_upload_0748ec6f-112f-490f-bfaf-d8547e2906a7.mp4", # Timing configuration "start_time": 30.0, "end_time": 60.0, # Content identifiers "clip_id": "e6b7263f-d9bf-4218-be41-3778d00b0a58", "clip_title": "Breeze", "clip_author": "sunnysuno", # Styling preferences "lyrics_style": "lyrics_three_line", }, "meta_data_leading_no_lyrics": { # Visual configuration "render_pipeline_version": "v1", "preset_id": "aura", "preset_style": "blue", "width": 1080, "height": 1920, "sticker_style": "leading_info_standard", "image_cover_url": "https://cdn2.suno.ai/5962bc75-c5e0-458c-848c-512eae911f64_7c0c4ce4.jpeg", "video_cover_url": "https://cdn1.suno.ai/video_upload_0748ec6f-112f-490f-bfaf-d8547e2906a7.mp4", # Timing configuration "start_time": 30.0, "end_time": 60.0, # Content identifiers "clip_id": "95dc0c01-636a-424c-82df-8af739d0fda9", "clip_title": "Jamaican Ginger", "clip_author": "sunnysuno", # Styling preferences "lyrics_style": "none", }, "meta_data_framed_sky": { # Visual configuration "render_pipeline_version": "v1", "preset_id": "clouds", "preset_style": "standard", "width": 1080, "height": 1920, "sticker_style": "framed_core_center", "image_cover_url": "https://cdn2.suno.ai/5962bc75-c5e0-458c-848c-512eae911f64_7c0c4ce4.jpeg", "video_cover_url": "https://cdn1.suno.ai/video_upload_0748ec6f-112f-490f-bfaf-d8547e2906a7.mp4", # Timing configuration "start_time": 15.0, "end_time": 75.0, # Content identifiers "clip_id": "95dc0c01-636a-424c-82df-8af739d0fda9", "clip_title": "Jamaican Ginger", "clip_author": "sunnysuno", # Styling preferences "lyrics_style": "none", }, "meta_data_framed_fire_sparks": { # Visual configuration "render_pipeline_version": "v1", "preset_id": "fire", "preset_style": "sparks_v1", "width": 1080, "height": 1920, "sticker_style": "framed_core_top", "image_cover_url": "https://cdn2.suno.ai/5962bc75-c5e0-458c-848c-512eae911f64_7c0c4ce4.jpeg", "video_cover_url": "https://cdn1.suno.ai/video_upload_0748ec6f-112f-490f-bfaf-d8547e2906a7.mp4", # Timing configuration "start_time": 15.0, "end_time": 75.0, # Content identifiers "clip_id": "95dc0c01-636a-424c-82df-8af739d0fda9", "clip_title": "Jamaican Ginger", "clip_author": "sunnysuno", # Styling preferences "lyrics_style": "none", }, } # Initialize test run with unique identifier run_id = 49 jobs = [] # Create and spawn test jobs for each configuration for config_name, config_data in test_items.items(): test_job = QueueItem( id=f"chi-test-{config_name}-{run_id}", metadata={"render_config": config_data}, callback_url="https://httpbin.org/post", # Test endpoint ) jobs.append(test_job.model_dump_json()) results = list(ShareAssetStub().generate_share_asset.map(jobs)) print("Generated Asset URLs:") for i, url in enumerate(results): if url: print(f"Asset {i+1}: {url}") else: print(f"Asset {i+1}: Failed to generate URL")