# Expected usage # Inside of suno_utils directory run # modal run suno_utils/worker/shader_buffer_presets.py import os import modal import argparse # ────────────────────────────────────────────────────────── # 1) Model App Setup # ────────────────────────────────────────────────────────── app = modal.App("suno-shader-buffer-presets") # This name is the one you will see as the folder on Modal under VOLUME_NAME = "shader-outputs" outputs = modal.Volume.from_name(VOLUME_NAME, create_if_missing=True) OUTPUTS_PATH = "/outputs" PRESETS_DIR_NAME = "presets" RESOURCES_DIR_NAME = "resources" PRERENDERED_VIDEO_DIR_NAME = "prerendered_video" ASSET_CONFIG_DIR_NAME = "asset_config" BUFFER_CONFIG_DIR_NAME = "buffer_config" SHADER_DIR_NAME = "shaders" 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([]) ) # ────────────────────────────────────────────────────────── # 2) Preset Mapping # ────────────────────────────────────────────────────────── # This function handles remote resource downloads (unchanged) @app.function(image=image, volumes={OUTPUTS_PATH: outputs}, timeout=600) def upload_to_volume_resources(item: dict): import os import urllib.request resource_url = item["resource_url"] resource_filename = item["filename"] resource_subdir_path = item["resource_subdir_path"] resource_dir = os.path.join(OUTPUTS_PATH, RESOURCES_DIR_NAME) if resource_subdir_path is not None: resource_dir = os.path.join(resource_dir, resource_subdir_path) os.makedirs(resource_dir, exist_ok=True) file_path = os.path.join(resource_dir, resource_filename) try: urllib.request.urlretrieve(resource_url, file_path) except Exception as e: print(f"Error downloading resource file: {e}") raise # New remote function to write file bytes to the volume. @app.function(image=image, volumes={OUTPUTS_PATH: outputs}, timeout=600) def upload_local_file_to_volume(file_bytes: bytes, filename: str, toplevelfolder: str, subfolder: str): import os # Construct the destination directory inside the volume. dest_dir = os.path.join(OUTPUTS_PATH, toplevelfolder) if subfolder is not None: dest_dir = os.path.join(dest_dir, subfolder) os.makedirs(dest_dir, exist_ok=True) dest_file = os.path.join(dest_dir, filename) # Write the file bytes to the destination file. with open(dest_file, "wb") as dst_file: dst_file.write(file_bytes) print(f"Uploaded {filename} to {dest_file}") # ────────────────────────────────────────────────────────── # 4) Main Local Entrypoint with Flags # ────────────────────────────────────────────────────────── @app.local_entrypoint() def main(*arglist): parser = argparse.ArgumentParser() # Use "-r" for skip resources and "-p" for skip presets. parser.add_argument("--skip-resources", action="store_true", help="Skip uploading resource files.") parser.add_argument("--skip-presets", action="store_true", help="Skip uploading preset files.") parser.add_argument( "--only-prerendered-video", action="store_true", help="Only upload videos from the prerendered videos folder", ) args = parser.parse_args(args=arglist) # Define the local directories for uploads. local_uploads_dir = os.path.join(os.getcwd(), "suno_utils", "worker", "shader_resources") local_presets_dir = os.path.join(local_uploads_dir, PRESETS_DIR_NAME) local_resources_dir = os.path.join(local_uploads_dir, RESOURCES_DIR_NAME) local_prerendered_video_dir = os.path.join(local_resources_dir, PRERENDERED_VIDEO_DIR_NAME) if args.only_prerendered_video: # DO NOT COMMIT VIDEOS TO GIT # ONLY UPLOAD TO MODAL VOLUME video_paths = [ os.path.join(local_prerendered_video_dir, item) for item in os.listdir(local_prerendered_video_dir) ] for path in video_paths: with open(path, "rb") as file: file_bytes = file.read() upload_local_file_to_volume.remote( file_bytes, os.path.basename(path), RESOURCES_DIR_NAME, PRERENDERED_VIDEO_DIR_NAME ) return # Upload preset files if not skipped. if not args.skip_presets: # Asset Config: JSON files asset_config_dir = os.path.join(local_presets_dir, ASSET_CONFIG_DIR_NAME) asset_config_paths = [ os.path.join(asset_config_dir, item) for item in os.listdir(asset_config_dir) if item.endswith(".json") ] # Buffer Config: JSON files buffer_config_dir = os.path.join(local_presets_dir, BUFFER_CONFIG_DIR_NAME) buffer_config_paths = [ os.path.join(buffer_config_dir, item) for item in os.listdir(buffer_config_dir) if item.endswith(".json") ] # Shaders: GLSL files shaders_dir = os.path.join(local_presets_dir, SHADER_DIR_NAME) shader_paths = [ os.path.join(shaders_dir, item) for item in os.listdir(shaders_dir) if item.endswith(".glsl") ] for path in asset_config_paths: with open(path, "rb") as file: file_bytes = file.read() upload_local_file_to_volume.remote( file_bytes, os.path.basename(path), PRESETS_DIR_NAME, ASSET_CONFIG_DIR_NAME ) for path in buffer_config_paths: with open(path, "rb") as file: file_bytes = file.read() upload_local_file_to_volume.remote( file_bytes, os.path.basename(path), PRESETS_DIR_NAME, BUFFER_CONFIG_DIR_NAME ) for path in shader_paths: with open(path, "rb") as file: file_bytes = file.read() upload_local_file_to_volume.remote( file_bytes, os.path.basename(path), PRESETS_DIR_NAME, SHADER_DIR_NAME ) else: print("Skipping preset uploads (--skip-presets).") # Upload resource files if not skipped. if not args.skip_resources: resource_paths = [ os.path.join(local_resources_dir, item) for item in os.listdir(local_resources_dir) ] for path in resource_paths: with open(path, "rb") as file: file_bytes = file.read() upload_local_file_to_volume.remote( file_bytes, os.path.basename(path), RESOURCES_DIR_NAME, None ) else: print("Skipping resource uploads (--skip-resources).")