app.py

import os

os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True")

import spaces  # noqa: E402  (must precede torch)
import torch  # noqa: E402
import gradio as gr  # noqa: E402

import shutil
import subprocess
import sys


def _diagnostics() -> str:
    lines = [f"python: {sys.version}"]
    for name in (
        "torch",
        "triton",
        "diffusers",
        "transformers",
        "peft",
        "accelerate",
        "numpy",
        "timm",
        "kernels",
        "imageio",
        "huggingface_hub",
        "gradio",
        "spaces",
    ):
        try:
            mod = __import__(name)
            lines.append(f"{name}: {getattr(mod, '__version__', '?')}")
        except Exception as exc:  # noqa: BLE001
            lines.append(f"{name}: IMPORT FAILED {exc!r}")

    lines.append("")
    lines.append(f"cpu_count: {os.cpu_count()}")
    lines.append(f"HF_HOME={os.environ.get('HF_HOME')}")
    lines.append(f"HF_HUB_CACHE={os.environ.get('HF_HUB_CACHE')}")
    lines.append(f"torch.cuda.is_available(): {torch.cuda.is_available()}")

    lines.append("")
    for path in ("/", "/tmp", "/home/user", "/data", os.getcwd()):
        try:
            total, used, free = shutil.disk_usage(path)
            lines.append(
                f"disk {path}: total={total / 2**30:.1f}G "
                f"used={used / 2**30:.1f}G free={free / 2**30:.1f}G"
            )
        except Exception as exc:  # noqa: BLE001
            lines.append(f"disk {path}: {exc!r}")

    for cmd in (["df", "-h"], ["free", "-g"]):
        try:
            out = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
            lines.append("")
            lines.append(f"$ {' '.join(cmd)}\n{out.stdout}{out.stderr}")
        except Exception as exc:  # noqa: BLE001
            lines.append(f"{' '.join(cmd)}: {exc!r}")

    lines.append("")
    try:
        with open("/sys/fs/cgroup/memory.max") as fh:
            lines.append(f"cgroup memory.max: {fh.read().strip()}")
    except Exception as exc:  # noqa: BLE001
        lines.append(f"cgroup memory.max: {exc!r}")

    lines.append("")
    try:
        from worldcrafter import WorldCrafter  # noqa: F401

        lines.append("import worldcrafter: OK")
    except Exception as exc:  # noqa: BLE001
        import traceback

        lines.append(f"import worldcrafter: FAILED {exc!r}\n{traceback.format_exc()}")

    try:
        from worldcrafter.fast.resident import ResidentBranches  # noqa: F401

        lines.append("import worldcrafter.fast.resident (triton): OK")
    except Exception as exc:  # noqa: BLE001
        lines.append(f"import worldcrafter.fast.resident: FAILED {exc!r}")

    try:
        from worldcrafter.camera import build_trajectory, parse_trajectory

        events, options = parse_trajectory("forward1x2\nyaw_left30")
        camera, records = build_trajectory(events, **options)
        lines.append(
            f"camera smoke: events={events} camera={camera.shape} chunks={len(records)}"
        )
    except Exception as exc:  # noqa: BLE001
        import traceback

        lines.append(f"camera smoke: FAILED {exc!r}\n{traceback.format_exc()}")

    return "\n".join(lines)


REPORT = _diagnostics()
print("==== WORLDCRAFTER SPACE DIAGNOSTICS ====", flush=True)
print(REPORT, flush=True)
print("==== END DIAGNOSTICS ====", flush=True)


def report() -> str:
    """Return the environment diagnostics collected at startup."""
    return REPORT


with gr.Blocks(theme=gr.themes.Citrus(), title="WorldCrafter (provisioning)") as demo:
    gr.Markdown("# WorldCrafter — provisioning\nEnvironment diagnostics:")
    out = gr.Textbox(value=REPORT, lines=40, label="diagnostics")
    gr.Button("Refresh").click(fn=report, inputs=None, outputs=out)

if __name__ == "__main__":
    demo.launch()