Spaces:

gulle1155
/

studio

Paused

studio / app.py

Avoid duplicate GPU leases on generate

b271239 15 days ago

17.3 kB

#!/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", "<unset>")
    d["NVIDIA_VISIBLE_DEVICES"] = os.environ.get("NVIDIA_VISIBLE_DEVICES", "<unset>")
    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())

	#!/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", "<unset>")
	d["NVIDIA_VISIBLE_DEVICES"] = os.environ.get("NVIDIA_VISIBLE_DEVICES", "<unset>")
	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())