from __future__ import annotations import json import os fr...

Prompt

from __future__ import annotations import json import os from pathlib import Path import modal APP_NAME = "vsfx-training" IMAGE_NAME = "vsfx-train" REMOTE_ROOT = Path("/root/vsfx") DATASETS_REMOTE = Path("/root/vsfx/datasets") RUNS_REMOTE = Path("/root/vsfx/runs") WEIGHTS_REMOTE = Path("/root/vsfx/weights") train_image = ( modal.Image.debian_slim(python_version="3.11") .apt_install("git", "unzip", "zip", "p7zip-full", "ffmpeg", "sox") .pip_install( "torch==2.14.0", "torchaudio==2.11.0", "numpy==2.4.6", "soundfile==0.14.0", "soxr==1.1.0", "librosa==1.0.0", "pydantic==2.13.5", "pydantic-settings==2.15.0", "structlog==26.1.0", "safetensors==0.8.0", "open-clip-torch==3.3.0", "yt-dlp==2026.8.19", "alembic==1.20.0", "asyncpg==0.31.0", "redis==8.1.0", "httpx==0.28.1", "python-multipart==0.0.32", "uvicorn==0.54.0", ) .add_local_dir( local_path=Path(__file__).resolve().parent.parent / "app", remote_path=str(REMOTE_ROOT / "app"), copy=False, ) .add_local_dir( local_path=Path(__file__).resolve().parent.parent / "training", remote_path=str(REMOTE_ROOT / "training"), copy=False, ) .add_local_dir( local_path=Path(__file__).resolve().parent.parent / "scripts", remote_path=str(REMOTE_ROOT / "scripts"), copy=False, ) .add_local_dir( local_path=Path(__file__).resolve().parent.parent / "config", remote_path=str(REMOTE_ROOT / "config"), copy=False, ) .workdir(str(REMOTE_ROOT)) .env( { "VSFX_TRAIN_DATA_DIR": str(DATASETS_REMOTE), "VSFX_TRAIN_OUTPUT_DIR": str(RUNS_REMOTE), "VSFX_TRAIN_WEIGHTS_DIR": str(WEIGHTS_REMOTE), "PYTHONPATH": str(REMOTE_ROOT), } ) ) datasets_volume = modal.Volume.from_name(f"{APP_NAME}-datasets", create_if_missing=True) runs_volume = modal.Volume.from_name(f"{APP_NAME}-runs", create_if_missing=True) weights_volume = modal.Volume.from_name(f"{APP_NAME}-weights", create_if_missing=True) app = modal.App(name=APP_NAME, image=train_image) GPU_CONFIG = os.environ.get("VSFX_MODAL_GPU", "a10g") VARIANT = os.environ.get("VSFX_MODAL_VARIANT", "small_16k") def _train_settings_env(epochs_override: str) -> dict[str, str]: """Build the stage environment from CLI knobs. Parameters: epochs_override: Optional epoch override value. Returns: Env overrides consumed by training.config.get_train_settings. """ env = { "VSFX_TRAIN_VARIANT": VARIANT, "VSFX_TRAIN_DEVICE": "cuda", } if epochs_override: env["VSFX_TRAIN_EPOCHS_OVERRIDE"] = epochs_override return env @app.function( image=train_image, gpu=GPU_CONFIG, timeout=60 * 60 * 24, volumes={ str(DATASETS_REMOTE): datasets_volume, str(RUNS_REMOTE): runs_volume, str(WEIGHTS_REMOTE): weights_volume, }, ) def run_stage(stage: str, epochs_override: str = "") -> str: """Execute one training stage on a Modal GPU. Parameters: stage: One of weights, fsd50k, vggsound, vae, vocoder, sync, shards, generator, export. epochs_override: Optional VSFX_TRAIN_EPOCHS_OVERRIDE value. Returns: Short human-readable result line. Raises: ValueError: On unknown stages. """ os.environ.update(_train_settings_env(epochs_override)) import subprocess import sys commands = { "weights": [sys.executable, "scripts/download_weights.py"], "fsd50k": [sys.executable, "-m", "training.data.fsd50k"], "vggsound": [sys.executable, "-m", "training.data.vggsound"], "vae": [sys.executable, "-m", "training.train_vae"], "vocoder": [sys.executable, "-m", "training.train_vocoder"], "sync": [sys.executable, "-m", "training.train_sync"], "shards": [sys.executable, "-m", "training.data.shards"], "generator": [sys.executable, "-m", "training.train_gen"], "export": [sys.executable, "-m", "training.export_weights"], } if stage not in commands: raise ValueError(f"unknown stage {stage!r}; expected one of {sorted(commands)}") result = subprocess.run(commands[stage], check=False, cwd=str(REMOTE_ROOT)) for volume in (datasets_volume, runs_volume, weights_volume): volume.commit() if result.returncode != 0: raise RuntimeError(f"stage {stage} failed with exit code {result.returncode}") return f"stage {stage} completed" @app.function(image=train_image, timeout=60 * 60 * 2) def stage_plan() -> str: """Return the ordered pipeline plan as JSON (no GPU required). Returns: JSON list of (stage, requires_gpu) pairs. """ plan = [ ("weights", False), ("fsd50k", False), ("vggsound", False), ("vae", True), ("vocoder", True), ("sync", True), ("shards", True), ("generator", True), ("export", False), ] return json.dumps(plan) @app.local_entrypoint() def main(stages: str = "all", epochs_override: str = "") -> None: """Run the training pipeline on Modal. GPU and variant are fixed at import time through VSFX_MODAL_GPU and VSFX_MODAL_VARIANT (e.g. `VSFX_MODAL_GPU=a100-40g modal run ...`). Parameters: stages: Comma-separated stage list, or `all` for the full pipeline. epochs_override: Optional epoch override (per-stage settings keep their defaults when empty). """ full_order = [ "weights", "fsd50k", "vggsound", "vae", "vocoder", "sync", "shards", "generator", "export", ] selected = full_order if stages.strip().lower() in ("", "all") else [ item.strip() for item in stages.split(",") if item.strip() ] for stage in selected: print(f"=== vsfx-training stage: {stage} (gpu={GPU_CONFIG}) ===", flush=True) print(run_stage.remote(stage, epochs_override=epochs_override), flush=True) print("all requested stages finished") Send back the complete code with all the fixes. Fix each of the listed errors one by one, making sure to actually correct them so that there are 0 errors remaining. Keep the original imports, since the files exist. Write out every single character; do not abbreviate anything. Fix every error. There must be exactly one file. Do not write anything else; just output the complete code, and it must not contain any comments. Never, under any circumstances, use simplified, substitute, dummy, simulated, or fake code. Write the entire file as complete, unabridged, production-ready code in a single code block. It must be 100% error-free, a complete, error-free file, and must be submitted as a downloadable file. These requirements are mandatory and must be strictly adhered to. If no list of errors is provided, you must find all the errors and fix them. If there were comments in the original code, delete them. And most importantly: YOU MUST NEVER SIMPLIFY!