#!/usr/bin/env python3 """DWCT Generative Sound Studio — Space entry point. Launcher hardening plus the UI. All pipeline logic lives in dwct/. """ from __future__ import annotations import argparse import asyncio import atexit import gc import hashlib import importlib.util import json import logging import os import shutil import signal import subprocess import sys import threading import time import traceback import warnings from dataclasses import dataclass, field from pathlib import Path from typing import Any, Optional # --- suppress known-benign noise before anything can emit it ------------- warnings.filterwarnings("ignore", message=r".*Can't initialize NVML.*") warnings.filterwarnings("ignore", message=r".*reduce_op.*is deprecated.*") warnings.filterwarnings("ignore", message=r".*unclosed event loop.*") warnings.filterwarnings("ignore", message=r".*Invalid file descriptor.*") _BENIGN = ("Invalid file descriptor: -1", "unclosed event loop") def _quiet_unraisable(u: Any) -> None: exc = getattr(u, "exc_value", None) text = f"{type(exc).__name__}: {exc}" if exc is not None else "" if isinstance(exc, (ValueError, KeyError, ResourceWarning)) and any( m in text for m in _BENIGN ): return sys.__unraisablehook__(u) sys.unraisablehook = _quiet_unraisable logging.basicConfig( level=os.environ.get("DWCT_LOG_LEVEL", "INFO").upper(), format="%(asctime)s %(levelname)-7s %(message)s", datefmt="%H:%M:%S", stream=sys.stdout, ) LOG = logging.getLogger("dwct") __version__ = "5.0" GB = 1024 ** 3 MODEL_ROOT = Path(os.environ.get("DWCT_MODEL_ROOT", "/models")) # --- spaces MUST precede gradio and any CUDA touch ----------------------- _HAS_SPACES = False try: import spaces # noqa: F401 _HAS_SPACES = True GPU = spaces.GPU except Exception: def GPU(*a, **kw): # no-op off ZeroGPU def deco(fn): return fn return deco import gradio as gr # noqa: E402 def _install_bucket_mamba_wheels() -> None: """Install the prebuilt CUDA extensions from the mounted Space bucket.""" if importlib.util.find_spec("mamba_ssm") is not None: return wheel_dir = Path("/data/wheels") wheel_names = ( "causal_conv1d-1.7.0-cp312-cp312-linux_x86_64.whl", "mamba_ssm-2.3.2.post1-cp312-cp312-linux_x86_64.whl", ) wheels = [wheel_dir / name for name in wheel_names] missing = [str(path) for path in wheels if not path.is_file()] if missing: raise RuntimeError( "Nemotron requires mamba-ssm, but its bucket wheels are missing: " + ", ".join(missing) ) LOG.info("Installing Nemotron CUDA extensions from mounted bucket %s", wheel_dir) subprocess.run( [sys.executable, "-m", "pip", "install", "--no-deps", *map(str, wheels)], check=True, ) _install_bucket_mamba_wheels() from dwct import config # noqa: E402 from dwct.pipeline import run as run_pipeline # noqa: E402 def _fingerprint() -> str: try: return hashlib.sha256(Path(__file__).resolve().read_bytes()).hexdigest()[:12] except Exception: return "unknown" def _sh(cmd: list[str]) -> str: try: return subprocess.run(cmd, capture_output=True, text=True, timeout=10, check=True).stdout.strip() except Exception: return "" # ---------------------------------------------------------------- probe @dataclass class GpuInfo: index: int name: str total_vram: int free_vram: int native_fp4: bool @dataclass class AccelProbe: gpus: list[GpuInfo] = field(default_factory=list) details: dict[str, Any] = field(default_factory=dict) hints: list[str] = field(default_factory=list) @property def total_free_vram(self) -> int: return sum(g.free_vram for g in self.gpus) @property def total_vram(self) -> int: return sum(g.total_vram for g in self.gpus) @property def all_native_fp4(self) -> bool: return bool(self.gpus) and all(g.native_fp4 for g in self.gpus) def probe_via_smi() -> Optional[AccelProbe]: if not shutil.which("nvidia-smi"): return None out = _sh(["nvidia-smi", "--query-gpu=index,name,memory.total,memory.free,compute_cap", "--format=csv,noheader,nounits"]) if not out: return None gpus, cap_known = [], True for line in out.splitlines(): parts = [p.strip() for p in line.split(",")] if len(parts) < 4: continue major = 0 if len(parts) >= 5: try: major = int(float(parts[4])) except ValueError: cap_known = False else: cap_known = False try: gpus.append(GpuInfo(int(parts[0]), parts[1], int(parts[2]) * 2**20, int(parts[3]) * 2**20, major >= 10)) except ValueError: continue if not gpus: return None d: dict[str, Any] = {"probe": "nvidia-smi", "device_count": len(gpus)} if not cap_known: d["compute_capability"] = "unavailable from nvidia-smi" return AccelProbe(gpus, d, []) def diagnose() -> AccelProbe: d: dict[str, Any] = {} hints: list[str] = [] d["CUDA_VISIBLE_DEVICES"] = os.environ.get("CUDA_VISIBLE_DEVICES", "") d["NVIDIA_VISIBLE_DEVICES"] = os.environ.get("NVIDIA_VISIBLE_DEVICES", "") nodes = sorted(p.name for p in Path("/dev").glob("nvidia*")) d["device_nodes"] = ", ".join(nodes) if nodes else "none" d["nvidia_smi"] = shutil.which("nvidia-smi") or "not found" d["libcuda"] = _sh(["sh", "-c", "ldconfig -p 2>/dev/null | grep -m1 'libcuda\\.so'"]) or "not found" if not nodes: hints.append("no /dev/nvidia* nodes - no GPU passthrough") if os.environ.get("CUDA_VISIBLE_DEVICES", "x") == "": hints.append("CUDA_VISIBLE_DEVICES is empty - CUDA is masked off") probe = probe_via_smi() or AccelProbe([], {}, []) probe.details.update(d) probe.hints.extend(hints) if len(probe.gpus) > 1: probe.hints.append(f"{len(probe.gpus)} host GPUs visible - on ZeroGPU you get " "a slice per call, not the whole pool") return probe def print_report(probe: AccelProbe) -> None: bar = "=" * 64 LOG.info(bar) LOG.info("DWCT GENERATIVE SOUND STUDIO - ENVIRONMENT REPORT") LOG.info("app.py v%s sha256:%s", __version__, _fingerprint()) LOG.info(bar) if not probe.gpus: LOG.info("GPU : NONE") else: LOG.info("GPU : %d host device(s), %.1f GB free / %.1f GB total", len(probe.gpus), probe.total_free_vram / GB, probe.total_vram / GB) for g in probe.gpus: LOG.info(" [%d] %-44s %6.1f / %6.1f GB", g.index, g.name, g.free_vram / GB, g.total_vram / GB) LOG.info("Python : %s", sys.version.split()[0]) LOG.info("gradio : %s spaces: %s", gr.__version__, _HAS_SPACES) LOG.info("-" * 64) for k, v in probe.details.items(): LOG.info(" %-20s %s", k, v) for h in probe.hints: LOG.warning("HINT: %s", h) LOG.info(bar) # ----------------------------------------------------------- loop hygiene _owned: list[asyncio.AbstractEventLoop] = [] def make_loop() -> asyncio.AbstractEventLoop: loop = asyncio.new_event_loop() _owned.append(loop) asyncio.set_event_loop(loop) return loop def close_all_loops() -> None: for loop in list(_owned): try: if not loop.is_closed() and not loop.is_running(): loop.close() except Exception: pass _owned.clear() with warnings.catch_warnings(): warnings.simplefilter("ignore") for obj in gc.get_objects(): if isinstance(obj, asyncio.AbstractEventLoop): try: if not obj.is_closed() and not obj.is_running(): obj.close() except Exception: pass gc.collect() atexit.register(close_all_loops) _stop = threading.Event() _started = time.monotonic() def install_signals() -> None: def handler(signum, _f): try: name = signal.Signals(signum).name except ValueError: name = str(signum) LOG.info("received %s - shutting down", name) _stop.set() for s in (signal.SIGINT, signal.SIGTERM): try: signal.signal(s, handler) except ValueError: pass # ------------------------------------------------------------------- UI EXTS = {".wav", ".mp3", ".flac", ".m4a", ".ogg"} def _library_choices() -> list[str]: if not config.LIBRARY.is_dir(): return [] return sorted(p.name for p in config.LIBRARY.iterdir() if p.suffix.lower() in EXTS) def _resolve(upload: str | None, folder: str, track: str | None) -> Path | None: if upload: return Path(upload) if folder: p = Path(folder) if p.is_file(): return p if p.is_dir(): hits = sorted(q for q in p.iterdir() if q.suffix.lower() in EXTS) if hits: return hits[0] if track: return config.LIBRARY / track return None # @GPU lives inside pipeline.run now - CPU stages must not hold a slice. def generate(upload, folder, track, brief, chain, progress=gr.Progress()): audio = _resolve(upload, folder, track) if audio is None or not audio.exists(): raise gr.Error("no audio resolved - upload a file, give a path, or pick a track") progress(0.02, desc=f"analysing {audio.name}") final = run_pipeline(audio, brief or "", chain=bool(chain), progress=lambda f, m: progress(f, desc=m)) manifest = Path(final).parent / "manifest.json" return str(final), str(manifest) if manifest.exists() else None def run_diagnostics() -> str: import platform import shutil as _sh out = [ f"platform : {platform.platform()}", f"python : {sys.version.split()[0]}", f"cwd : {os.getcwd()}", f"ffmpeg : {_sh.which('ffmpeg') or 'NOT ON PATH'}", ] try: import imageio_ffmpeg out.append(f"bundled : {imageio_ffmpeg.get_ffmpeg_exe()}") except Exception as exc: out.append(f"bundled : MISSING ({exc})") out.append("") for mod in ("numpy", "soundfile", "librosa", "torch", "transformers", "diffusers", "PIL", "imageio", "gradio", "spaces"): try: m = __import__(mod) out.append(f"{mod:<14} {getattr(m, '__version__', 'ok')}") except Exception as exc: out.append(f"{mod:<14} MISSING ({type(exc).__name__})") out.append("") out.append(f"gpus visible: {len(PROBE.gpus)}") for g in PROBE.gpus[:2]: out.append(f" {g.name} {g.free_vram / GB:.1f} GB free") return "\n".join(out) def run_director() -> str: """GPU smoke test: lease, load Nemotron, read semantics, plan shots. Exercises the path that used to die on allow_all_kernels. No render.""" import tempfile import time as _t import numpy as np import soundfile as sf from dwct.audio_analysis import analyze from dwct.pipeline import _stage_direct sr = 22050 t = np.linspace(0, 12, sr * 12) x = 0.25 * np.sin(2 * np.pi * 220 * t) + 0.15 * np.sin(2 * np.pi * 330 * t) for b in np.arange(0, 12, 0.5): # 120 BPM kicks m = (t >= b) & (t < b + 0.2) x[m] += 0.5 * np.sin(2 * np.pi * 55 * t[m]) * np.exp(-(t[m] - b) * 25) path = Path(tempfile.gettempdir()) / "dwct_director.wav" sf.write(str(path), x.astype("float32"), sr) tl = analyze(path) t0 = _t.monotonic() sem, cfg, shots = _stage_direct(str(path), tl, "neon rain-soaked city, cinematic") dt = _t.monotonic() - t0 return "\n".join([ f"OK director stage {dt:.1f}s", f"tempo={tl.tempo:.1f} beats={len(tl.beat_times)} sections={len(tl.sections)}", f"shots planned: {len(shots)}", "", str(sem)[:600], ]) def run_stage1() -> str: import tempfile import numpy as np import soundfile as sf from dwct.audio_analysis import analyze sr = 22050 t = np.linspace(0, 20, sr * 20) x = 0.3 * np.sin(2 * np.pi * 110 * t) + 0.2 * np.sin(2 * np.pi * 220 * t) path = Path(tempfile.gettempdir()) / "dwct_stage1.wav" sf.write(str(path), x.astype("float32"), sr) tl = analyze(path) head = [f"OK {tl.duration:.1f}s {tl.tempo:.1f} BPM {tl.key} {tl.mode}", f"{len(tl.beat_times)} beats {len(tl.sections)} sections", ""] return "\n".join(head) + tl.digest() def build_ui() -> gr.Blocks: with gr.Blocks(title="DWCT Generative Sound Studio") as ui: gr.Markdown("# DWCT Generative Sound Studio") gr.Markdown(f"**v{__version__}** · `sha256:{_fingerprint()}` \n" "Audio in, audio-reactive video out. Timing from DSP, " "meaning from the director model.") with gr.Row(): with gr.Column(): upload = gr.Audio(label="Upload audio", type="filepath") folder = gr.Textbox(label="Or a path", placeholder="/music/track.wav or /music/album") track = gr.Dropdown(label="Or pick from library", choices=_library_choices(), value=None) brief = gr.Textbox(label="Brief", lines=4, placeholder="neon city, rain, slow reveal") chain = gr.Checkbox(label="Continuous take (chain last frame)", value=False) go = gr.Button("Generate", variant="primary") with gr.Column(): video = gr.Video(label="Result") manifest = gr.File(label="Manifest") with gr.Tab("Diagnostics"): with gr.Row(): b_diag = gr.Button("Run diagnostics") b_s1 = gr.Button("Run stage 1 self-test") b_dir = gr.Button("Run director (GPU) smoke test") diag_out = gr.Code(label="Environment", language="shell") s1_out = gr.Code(label="Stage 1", language="shell") b_diag.click(run_diagnostics, None, diag_out) b_s1.click(run_stage1, None, s1_out) dir_out = gr.Code(label="Director (GPU)", language="shell") b_dir.click(run_director, None, dir_out) go.click(generate, [upload, folder, track, brief, chain], [video, manifest]) return ui if gr.NO_RELOAD: PROBE = diagnose() print_report(PROBE) else: PROBE = globals().get("PROBE") or diagnose() demo = build_ui() demo.queue() # ---------------------------------------------------------------- entry def main(argv: Optional[list[str]] = None) -> int: ap = argparse.ArgumentParser(prog="app.py") ap.add_argument("--check", action="store_true") ap.add_argument("--json", action="store_true") ap.add_argument("--heartbeat", type=int, default=int(os.environ.get("DWCT_HEARTBEAT", "0"))) ap.add_argument("--host", default=os.environ.get("DWCT_HOST", "0.0.0.0")) ap.add_argument("--port", type=int, default=int(os.environ.get("DWCT_PORT", "7860"))) ap.add_argument("--share", action="store_true", default=os.environ.get("DWCT_SHARE", "0") == "1") ap.add_argument("--log-level", default=os.environ.get("DWCT_LOG_LEVEL", "INFO")) args = ap.parse_args(argv) logging.getLogger().setLevel(args.log_level.upper()) make_loop() LOG.info("=" * 64) LOG.info("app.py v%s sha256:%s", __version__, _fingerprint()) LOG.info("path: %s", Path(__file__).resolve()) LOG.info("=" * 64) if args.json: print(json.dumps({ "version": __version__, "gpus": [{"index": g.index, "name": g.name, "free_gb": round(g.free_vram / GB, 2)} for g in PROBE.gpus], "details": PROBE.details, "hints": PROBE.hints, }, indent=2)) if args.check: close_all_loops() return 0 install_signals() demo.launch(server_name=args.host, server_port=None, share=args.share, mcp_server=True, prevent_thread_lock=True, show_error=True) LOG.info("Ready: http://127.0.0.1:%d (bind %s:%d)", args.port, args.host, args.port) LOG.info("main loop entered; waiting for SIGINT/SIGTERM") last = time.monotonic() try: while not _stop.wait(0.5): if args.heartbeat and time.monotonic() - last >= args.heartbeat: LOG.info("alive - uptime %.0fs", time.monotonic() - _started) last = time.monotonic() except KeyboardInterrupt: pass finally: LOG.info("closing server...") try: demo.close() except Exception: pass close_all_loops() LOG.info("clean shutdown complete - uptime %.0fs", time.monotonic() - _started) return 0 if __name__ == "__main__": sys.exit(main())