VoxelForge/Standalone/Tools/webviewer/export_world.py

#!/usr/bin/env python3
"""Export phone-sized LOD0-2 arrival views of the owner world as viewer GLBs."""
from __future__ import annotations

import argparse
import json
import struct
import subprocess
import sys
import tempfile
import time
from pathlib import Path


ROOT = Path(__file__).resolve().parents[3]
WEBVIEWER = Path(__file__).resolve().parent
DATA = ROOT / "Standalone/data/owner_world.json"
DEFAULT_BINARY = ROOT / "Saved/Linux/l101/bin/VoxelForgeCoreHeadless-l101-clang"
LOG_DIR = ROOT / "Saved/Linux/w2"
MESH_DIR = WEBVIEWER / "meshes"
CONVERTER = WEBVIEWER / "convert.py"

# Arrival positions used by the existing L70-L80 headless arrival/census runners.
VIEWS = (
    ("sw-arrival", (8.0, 16.0, -372.2626953125)),
    ("fi-arrival", (25.8369140625, 3.19921875, -737.0)),
    ("tn-arrival", (19.166015625, -14.2001953125, -753.3740234375)),
    ("vs-arrival", (-20.0625, -5.068359375, -900.353515625)),
    ("fi-passage", (0.0, 0.0, -600.0)),
)


def mesh_stats(path: Path) -> dict:
    triangles = tiles = vertices = 0
    lod_counts: dict[int, int] = {}
    with path.open("rb") as stream:
        header = stream.read(60)
        if len(header) != 60 or header[:8] not in (b"VFMSH001", b"VFMSH002"):
            raise RuntimeError(f"invalid VFMSH output: {path}")
        version, _, tiles = struct.unpack_from("<3I", header, 8)
        expected = 1 if header[:8] == b"VFMSH001" else 2
        if version != expected:
            raise RuntimeError(f"VFMSH version mismatch: {path}")
        for _ in range(tiles):
            record = stream.read(36)
            if len(record) != 36:
                raise RuntimeError(f"truncated tile header: {path}")
            key = struct.unpack_from("<4i", record, 0)
            _, _, vertex_count, index_count, mask_count = struct.unpack_from("<5I", record, 16)
            triangles += index_count // 3
            vertices += vertex_count
            lod_counts[key[3]] = lod_counts.get(key[3], 0) + 1
            stream.seek(vertex_count * (34 if version == 2 else 24)
                        + index_count * 4 + mask_count, 1)
        if stream.read(1):
            raise RuntimeError(f"trailing bytes after VFMSH tiles: {path}")
    return {"tiles": tiles, "vertices": vertices, "triangles": triangles,
            "lod_counts": lod_counts}


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--headless-binary", type=Path, default=DEFAULT_BINARY,
                        help=f"headless exporter binary (default: {DEFAULT_BINARY})")
    parser.add_argument("--clip-radius", type=int, default=3,
                        help="headless neighborhood radius (default: 3)")
    parser.add_argument("--max-lod", type=int, default=2,
                        help="highest exported LOD (default: 2)")
    parser.add_argument("--timeout-seconds", type=int, default=1740,
                        help="per-view limit; maximum is 1740 seconds")
    args = parser.parse_args()

    binary = args.headless_binary if args.headless_binary.is_absolute() else ROOT / args.headless_binary
    if args.timeout_seconds < 1 or args.timeout_seconds > 1740:
        parser.error("--timeout-seconds must be between 1 and 1740")
    if args.clip_radius < 1 or args.max_lod < 0:
        parser.error("--clip-radius must be positive and --max-lod cannot be negative")
    for required in (binary, DATA, CONVERTER):
        if not required.is_file():
            parser.error(f"required file not found: {required}")

    LOG_DIR.mkdir(parents=True, exist_ok=True)
    MESH_DIR.mkdir(parents=True, exist_ok=True)
    rows = []
    with tempfile.TemporaryDirectory(prefix=".w2-export-", dir=WEBVIEWER) as scratch:
        scratch_dir = Path(scratch)
        for name, camera in VIEWS:
            source = scratch_dir / f"{name}.vfmesh"
            camera_arg = ",".join(f"{value:.12f}" for value in (*camera, 0.0, 0.0))
            headless_log = LOG_DIR / f"{name}-headless.log"
            command = [
                "nice", "-n", "19", "taskset", "-c", "0-11",
                "prlimit", "--as=6442450944", "--",
                "timeout", f"{args.timeout_seconds}s", str(binary), str(DATA),
                "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
                "--workers=8", "--force-generate", f"--clip-radius={args.clip_radius}",
                f"--max-clip-level={args.max_lod}", f"--max-level={args.max_lod}",
                "--render-distance-chunks=0", f"--camera={camera_arg}",
                f"--mesh-export={source}",
            ]
            started = time.monotonic()
            with headless_log.open("w", encoding="utf-8") as log:
                result = subprocess.run(command, cwd=ROOT, stdout=log,
                                        stderr=subprocess.STDOUT, check=False,
                                        timeout=args.timeout_seconds + 15)
            if result.returncode != 0 or not source.is_file() or source.stat().st_size == 0:
                raise RuntimeError(
                    f"headless export failed for {name} (exit={result.returncode}); see {headless_log}")

            stats = mesh_stats(source)
            convert_log = LOG_DIR / f"{name}-convert.log"
            with convert_log.open("w", encoding="utf-8") as log:
                converted = subprocess.run(
                    [sys.executable, str(CONVERTER), str(source), "--out-dir", str(MESH_DIR)],
                    cwd=ROOT, stdout=log, stderr=subprocess.STDOUT, check=False)
            destination = MESH_DIR / f"{name}.glb"
            if converted.returncode != 0 or not destination.is_file() or destination.stat().st_size == 0:
                raise RuntimeError(f"GLB conversion failed for {name}; see {convert_log}")

            row = {
                "name": name,
                "glb": str(destination.relative_to(ROOT)),
                "bytes": destination.stat().st_size,
                "triangles": stats["triangles"],
                "tiles": stats["tiles"],
                "vertices": stats["vertices"],
                "lod_counts": stats["lod_counts"],
                "wall_s": round(time.monotonic() - started, 3),
            }
            rows.append(row)
            print(f"{name}: {destination.stat().st_size / 1_000_000:.2f} MB, "
                  f"{stats['triangles']:,} triangles, {stats['tiles']:,} tiles, "
                  f"LOD tiles {json.dumps(stats['lod_counts'], sort_keys=True)}")

    summary_path = LOG_DIR / "export-summary.json"
    summary_path.write_text(json.dumps(rows, indent=2) + "\n", encoding="utf-8")
    print(f"Summary: {summary_path.relative_to(ROOT)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())