#!/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", "") 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())