VoxelForge/Standalone/Research/WorldAssemblerM3/Linux101/RunL101Census.py

#!/usr/bin/env python3
"""Count measured off-grid and passage-composer work by tile and caller."""
from __future__ import annotations

import json
import re
import subprocess
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l101/census"
DATA = ROOT / "Standalone/data/owner_world.json"
HEADLESS = ROOT / "Saved/Linux/l101/bin/VoxelForgeCoreHeadless-l101-clang"
CAMERAS = {
    "SW": (8.0, 16.0, -372.2626953125),
    "FI": (25.8369140625, 3.19921875, -737.0),
    "TN": (19.166015625, -14.2001953125, -753.3740234375),
    "VS": (-20.0625, -5.068359375, -900.353515625),
}
CALLERS = (
    "unknown", "density_grid", "gradient_normal_fallback", "mc_midpoint_correction",
    "skirt_surface_vertex", "skirt_extruded_vertex", "skirt_clipped_vertex",
    "skirt_refinement_face", "skirt_coarse_centroid", "far_shell_surface_query",
    "far_shell_projection_normal", "far_shell_stratum_refinement",
)
SKIRT_CALLERS = (
    "skirt_surface_vertex", "skirt_extruded_vertex", "skirt_clipped_vertex",
    "skirt_refinement_face", "skirt_coarse_centroid",
)
CPU_LINE = re.compile(r"^\[CpuTileStages\] key=\([^,]+,[^,]+,[^,]+,lod(\d+)\) (.*)$", re.M)
COMPOSER_LINE = re.compile(
    r"^\[PassageComposerTile\] key=\([^,]+,[^,]+,[^,]+,lod(\d+)\) (.*)$", re.M)
FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")
ARRAY_FIELDS = (
    "sweep_missing_by_caller", "sweep_exact_fallback_by_caller",
    "sweep_bound_skip_by_caller", "sweep_tile_free_skip_by_caller",
    "sweep_vertical_bound_skip_by_caller", "sweep_negative_reach_skip_by_caller",
    "sweep_positive_reach_skip_by_caller", "density_offgrid_float_by_caller",
    "density_offgrid_unclamped_by_caller", "density_offgrid_detail_by_caller",
)
SCALAR_FIELDS = (
    "density_float_points", "density_lattice_points", "density_unclamped_points",
    "grad_calls", "grad_memo_hits", "grad_fallback_calls", "midpoint_attempts",
    "midpoint_memo_hits", "skirt_density_probes", "skirt_surface_vertex",
    "skirt_surface_vertex_offgrid", "skirt_extruded_vertex",
    "skirt_extruded_vertex_offgrid", "skirt_clipped_vertex",
    "skirt_clipped_vertex_offgrid", "skirt_refinement_face",
    "skirt_refinement_face_offgrid", "skirt_coarse_centroid",
    "skirt_coarse_centroid_offgrid",
)
COMPOSER_FIELDS = (
    "queries", "boundary_band_queries", "column_distance_calls",
    "carve_segment_visits", "support_segment_visits", "support_capsule_checks",
    "later_bore_checks", "other_capsule_checks", "mouth_support_calls",
    "mouth_route_segment_checks", "relief_field_calls", "sweep_segments",
    "sweep_pairs", "sweep_lookups", "sweep_cache_hits", "sweep_missing",
    "tile_bound_checks", "tile_bound_rejects", "tile_bound_skipped_iterations",
)


def counts(value: str | None) -> list[int]:
    result = [int(part) for part in value.split(",")] if value else []
    if len(result) != len(CALLERS):
        raise ValueError(f"expected {len(CALLERS)} caller counts, got {len(result)}")
    return result


def run(strate: str) -> dict:
    OUT.mkdir(parents=True, exist_ok=True)
    log = OUT / f"census-{strate.lower()}.log"
    camera = ",".join(f"{value:.12f}" for value in CAMERAS[strate]) + ",0,0"
    command = [
        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
        "timeout", "1740s", str(HEADLESS), str(DATA), "--world-assembler-prototype",
        "--stream-benchmark", "--stream-pipeline", "--workers=8", "--clip-radius=3",
        "--max-clip-level=8", "--max-level=8", f"--camera={camera}",
        "--profile-passage-composer", "--profile-lod-work",
    ]
    start = time.monotonic()
    with log.open("w", encoding="utf-8") as stream:
        result = subprocess.run(command, cwd=ROOT, stdout=stream,
                                stderr=subprocess.STDOUT, check=False, timeout=1740)
    output = log.read_text(encoding="utf-8", errors="replace")
    if result.returncode:
        raise RuntimeError(f"off-grid census failed: exit={result.returncode} {log}")
    cpu_rows, composer_rows = [], []
    for pattern, destination, scalar_fields in (
        (CPU_LINE, cpu_rows, SCALAR_FIELDS),
        (COMPOSER_LINE, composer_rows, COMPOSER_FIELDS),
    ):
        for level_text, tail in pattern.findall(output):
            values = dict(FIELDS.findall(tail))
            row = {"lod": int(level_text)}
            row.update({key: int(values.get(key, 0)) for key in scalar_fields})
            if pattern is CPU_LINE:
                row.update({key: counts(values.get(key)) for key in ARRAY_FIELDS})
            destination.append(row)
    if not cpu_rows or not composer_rows:
        raise RuntimeError(f"missing tile census output: cpu={len(cpu_rows)} composer={len(composer_rows)} {log}")

    def aggregate(rows: list[dict], fields: tuple[str, ...], maximum_lod: int | None) -> dict:
        selected = [row for row in rows if maximum_lod is None or row["lod"] <= maximum_lod]
        result = {field: sum(row[field] for row in selected) for field in fields}
        if selected and "midpoint_attempts" in fields:
            for key in ARRAY_FIELDS:
                result[key] = [sum(row[key][index] for row in selected)
                               for index in range(len(CALLERS))]
        result["tiles"] = len(selected)
        result["tiles_by_lod"] = {
            str(lod): sum(row["lod"] == lod for row in selected)
            for lod in sorted({row["lod"] for row in selected})
        }
        return result

    all_cpu = aggregate(cpu_rows, SCALAR_FIELDS, None)
    near_cpu = aggregate(cpu_rows, SCALAR_FIELDS, 2)
    all_composer = aggregate(composer_rows, COMPOSER_FIELDS, None)
    near_composer = aggregate(composer_rows, COMPOSER_FIELDS, 2)

    def caller_map(values: list[int]) -> dict[str, int]:
        return dict(zip(CALLERS, values))

    validation = {
        "near_midpoint_exact_queries_match_unclamped_offgrid":
            near_cpu["midpoint_attempts"] - near_cpu["midpoint_memo_hits"]
            == near_cpu["density_offgrid_unclamped_by_caller"][
                CALLERS.index("mc_midpoint_correction")],
        "all_sweep_misses_match_caller_totals":
            sum(all_cpu["sweep_missing_by_caller"])
            == all_composer["sweep_missing"],
        "all_sweep_misses_have_exact_disposition":
            all(sum(all_cpu[key][index] for key in (
                "sweep_exact_fallback_by_caller", "sweep_bound_skip_by_caller",
                "sweep_tile_free_skip_by_caller", "sweep_vertical_bound_skip_by_caller",
                "sweep_negative_reach_skip_by_caller", "sweep_positive_reach_skip_by_caller"))
                == all_cpu["sweep_missing_by_caller"][index]
                for index in range(len(CALLERS))),
    }
    # Validate each skirt caller independently against the generic density entry counter.
    skirt_pairs = {
        "skirt_surface_vertex": "skirt_surface_vertex_offgrid",
        "skirt_extruded_vertex": "skirt_extruded_vertex_offgrid",
        "skirt_clipped_vertex": "skirt_clipped_vertex_offgrid",
        "skirt_refinement_face": "skirt_refinement_face_offgrid",
        "skirt_coarse_centroid": "skirt_coarse_centroid_offgrid",
    }
    skirt_validation = {
        caller: near_cpu["density_offgrid_float_by_caller"][CALLERS.index(caller)]
        == near_cpu[probe_count]
        for caller, probe_count in skirt_pairs.items()
    }
    validation["near_skirt_callers_match"] = all(skirt_validation.values())
    validation["near_skirt_caller_checks"] = skirt_validation
    unresolved = [all_cpu["sweep_missing_by_caller"][index]
        - all_cpu["sweep_exact_fallback_by_caller"][index]
        - all_cpu["sweep_bound_skip_by_caller"][index]
        - all_cpu["sweep_tile_free_skip_by_caller"][index]
        - all_cpu["sweep_vertical_bound_skip_by_caller"][index]
        - all_cpu["sweep_negative_reach_skip_by_caller"][index]
        - all_cpu["sweep_positive_reach_skip_by_caller"][index]
        for index in range(len(CALLERS))]
    row = {
        "strate": strate, "workers": 8, "exit_code": result.returncode,
        "wall_s": round(time.monotonic() - start, 3),
        "caller_order": CALLERS, "near_lod0_2": near_cpu,
        "all_lods": all_cpu, "near_composer_lod0_2": near_composer,
        "all_composer_lods": all_composer,
        "offgrid_float_by_caller": caller_map(all_cpu["density_offgrid_float_by_caller"]),
        "offgrid_unclamped_by_caller": caller_map(
            all_cpu["density_offgrid_unclamped_by_caller"]),
        "offgrid_detail_by_caller": caller_map(all_cpu["density_offgrid_detail_by_caller"]),
        "sweep_misses_by_caller": caller_map(all_cpu["sweep_missing_by_caller"]),
        "exact_route_fallbacks_by_caller": caller_map(
            all_cpu["sweep_exact_fallback_by_caller"]),
        "feature_bound_skips_by_caller": caller_map(
            all_cpu["sweep_bound_skip_by_caller"]),
        "tile_free_skips_by_caller": caller_map(
            all_cpu["sweep_tile_free_skip_by_caller"]),
        "vertical_bound_skips_by_caller": caller_map(
            all_cpu["sweep_vertical_bound_skip_by_caller"]),
        "negative_reach_skips_by_caller": caller_map(
            all_cpu["sweep_negative_reach_skip_by_caller"]),
        "positive_reach_skips_by_caller": caller_map(
            all_cpu["sweep_positive_reach_skip_by_caller"]),
        "misses_without_recorded_disposition_by_caller":
            caller_map(unresolved),
        "validation": validation, "log": str(log.relative_to(ROOT)),
    }
    print("[L101Census] " + json.dumps(row, sort_keys=True), flush=True)
    return row


def main() -> int:
    if not HEADLESS.is_file() or not DATA.is_file():
        raise SystemExit("missing L101 headless binary or owner world")
    rows = [run(strate) for strate in CAMERAS]
    (OUT / "summary.json").write_text(json.dumps({"runs": rows}, indent=2) + "\n",
                                      encoding="utf-8")
    return 0 if all(all(row["validation"].values()) for row in rows) else 1


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