VoxelForge/Standalone/Research/WorldAssemblerM3/Linux83/AnalyzeL83Census.py

#!/usr/bin/env python3
"""Symbolize L77 samples and group the passage work by source term."""
from __future__ import annotations

import collections
import json
import re
import subprocess
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l83/census"
SAMPLE = re.compile(
    r"^\[L77Sample\] pc=(0x\w+) lod=(\d+) .*?phase=(\S+) .*?weight=(\d+)$", re.M)
SYMBOLIZER = "llvm-symbolizer-19"
BINARIES = {
    "pre": ROOT / "Saved/Linux/l82/profile-bin/VoxelForgeCoreHeadless-clang19",
    "post": ROOT / "Saved/Linux/l83/profile-bin/VoxelForgeCoreHeadless-l83-profile-clang19",
}


def source_term(sample_phase: str, symbol: dict) -> str:
    function = symbol.get("FunctionName", "")
    file_name = symbol.get("FileName", "").rsplit("/", 1)[-1]
    line = int(symbol.get("Line", 0))
    if any(name in function for name in (
        "VF_FieldCachedValueNoise", "VF_FieldValueNoise", "VF_FieldSurfaceRelief",
        "VF_FieldPassageWallRelief", "VF_FieldNoiseValue", "VF_FieldNoiseHash",
        "VF_FieldMix", "VF_FieldLerp", "VF_FieldFade")):
        return "remaining relief"
    if file_name == "ShapeWorldInternal.h" and (
        1459 <= line <= 1486 or 1807 <= line <= 1843):
        return "smooth joins / Catmull-Rom"
    if file_name == "ShapeWorldInternal.h" and (
        1417 <= line <= 1426 or 1591 <= line <= 1778):
        return "mouth shaping"
    if any(name in function for name in (
        "PassageColumnSignedDistance", "PassageBridgeSignedDistance",
        "PassagePlayerCapsuleBlocksSupport", "PassageOtherCapsuleBlocksSupport",
        "PassageSupportMayBeCarvedByLaterBore")):
        return "support restoration / columns"
    if file_name == "ShapeWorldInternal.h" and (
        1275 <= line <= 1337 or 1781 <= line <= 2200):
        return "support restoration / columns"
    if sample_phase == "support":
        return "support restoration / columns"
    if any(name in function for name in (
        "PassagePotentialWithinNode", "VisitNearestPassageSpatialNode",
        "PassageNearestNodeLowerBound", "PassageMayAffectDensity",
        "BuildPassageSegmentRangeNodeMask", "segmentRangeNodeMask")):
        return "carve distance evaluation"
    if file_name == "ShapeWorldInternal.h" and (
        1049 <= line <= 1179 or 1344 <= line <= 1538):
        return "carve distance evaluation"
    if line == 0 and "ApplyResolvedStratePassageCarves" in function:
        return "passage body (line not resolved)"
    return "other passage work"


def analyze(label: str) -> dict:
    samples_by_pc: collections.Counter[tuple[str, str, str]] = collections.Counter()
    phase_ticks: collections.Counter[tuple[str, str]] = collections.Counter()
    files = sorted(OUT.glob(f"{label}-fi-r*.log")) + sorted(OUT.glob(f"{label}-tn-r*.log"))
    for path in files:
        world = "FI" if "-fi-" in path.name else "TN"
        output = path.read_text(encoding="utf-8", errors="replace")
        for pc, lod, phase, weight in SAMPLE.findall(output):
            if int(lod) > 2:
                continue
            ticks = int(weight)
            phase_ticks[(world, phase)] += ticks
            samples_by_pc[(world, pc, phase)] += ticks
    pcs = sorted({pc for _world, pc, _phase in samples_by_pc})
    command = [str(SYMBOLIZER), f"--obj={BINARIES[label]}",
               "--output-style=JSON", "--no-inlines"]
    result = subprocess.run(command, input="\n".join(pcs) + "\n", text=True,
                            capture_output=True, check=True)
    symbols = {}
    for line in result.stdout.splitlines():
        item = json.loads(line)
        symbols[item["Address"]] = item.get("Symbol", [{}])[0]

    worlds = {}
    for world in ("FI", "TN"):
        all_ticks = sum(value for (w, _phase), value in phase_ticks.items() if w == world)
        passage_ticks = phase_ticks[(world, "passages")]
        support_phase_ticks = phase_ticks[(world, "support")]
        terms: collections.Counter[str] = collections.Counter()
        top_symbols: collections.Counter[str] = collections.Counter()
        for (sample_world, pc, phase), ticks in samples_by_pc.items():
            if sample_world != world or phase != "passages":
                continue
            symbol = symbols.get(pc, {})
            terms[source_term(phase, symbol)] += ticks
            function = symbol.get("FunctionName", "<unresolved>")
            top_symbols[function] += ticks
        # Support is a separately tagged L77 phase; classify it here with relief kept
        # separate by symbol, to avoid folding relief calls into support CPU.
        for (sample_world, pc, phase), ticks in samples_by_pc.items():
            if sample_world != world or phase != "support":
                continue
            symbol = symbols.get(pc, {})
            term = source_term(phase, symbol)
            if term == "remaining relief":
                terms[term] += ticks
            else:
                terms["support restoration / columns"] += ticks
        passage_plus_support_ticks = passage_ticks + support_phase_ticks
        worlds[world] = {
            "near_lod_total_ticks": all_ticks,
            "passage_phase_ticks": passage_ticks,
            "passage_phase_share_pct": 100.0 * passage_ticks / all_ticks,
            "support_phase_ticks": support_phase_ticks,
            "support_within_passage_plus_support_pct":
                100.0 * support_phase_ticks / passage_plus_support_ticks,
            "term_ticks": dict(terms),
            "terms_pct_of_passage_and_support_phases": {
                term: 100.0 * value / passage_plus_support_ticks
                for term, value in terms.items()},
            "top_symbols": [
                {"symbol": symbol, "ticks": ticks,
                 "pct_of_passage_phase": 100.0 * ticks / passage_ticks}
                for symbol, ticks in top_symbols.most_common(14)],
        }
    return {"label": label, "worlds": worlds}


def main() -> int:
    result = {label: analyze(label) for label in ("pre", "post")}
    (OUT / "analysis.json").write_text(json.dumps(result, indent=2) + "\n",
                                        encoding="utf-8")
    for label in ("pre", "post"):
        for world, row in result[label]["worlds"].items():
            print(json.dumps({"label": label, "world": world,
                              "passage_share_pct": row["passage_phase_share_pct"],
                              "support_phase_ticks": row["support_phase_ticks"],
                              "terms_pct": row["terms_pct_of_passage_and_support_phases"],
                              "top_symbols": row["top_symbols"][:8]}, sort_keys=True))
    return 0


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