sots-re/tools/dashboard.py

493 lines
21 KiB
Python
Executable file
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/python3
"""Generate campaign/DASHBOARD.md — coverage dashboard for the SotS RE + reimplementation campaign.
Pure stdlib. Idempotent: re-run after every integration.
tools/dashboard.py write campaign/DASHBOARD.md
tools/dashboard.py --print also echo the markdown to stdout
tools/dashboard.py --no-write render only (implies --print)
tools/dashboard.py --engine P sots-engine checkout (default ~/sots-engine or $SOTS_ENGINE)
Every number's source and heuristic is documented in tools/DASHBOARD_README.md.
Parsing is tolerant: malformed inputs are reported in the footer "warnings" line, never fatal.
"""
import argparse
import datetime
import json
import os
import re
import subprocess
import sys
RE_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
WARN = []
FUNCTIONS_FALLBACK = 41411 # findings/01-fingerprint.md, in case the number cannot be parsed
STATUSES = ["verified", "mapped", "in-progress", "backlog", "blocked"]
TYPE_ROWS = [ # (label, board Type values)
("objects", ("object",)), ("control-flow", ("control-flow",)), ("subsystems", ("subsystem",)),
("engine", ("engine",)), ("verify", ("verify",)), ("phase2", ("phase2",)), ("meta", ("meta",)),
]
def warn(msg):
WARN.append(msg)
def rp(*parts):
return os.path.join(RE_ROOT, *parts)
def read(path):
try:
with open(path, encoding="utf-8", errors="replace") as f:
return f.read()
except OSError as e:
warn(f"cannot read {os.path.relpath(path, RE_ROOT)}: {e.strerror}")
return ""
def count_lines(path):
txt = read(path)
return sum(1 for ln in txt.splitlines() if ln.strip())
def git(repo, *args):
try:
return subprocess.check_output(["git", "-C", repo] + list(args), stderr=subprocess.DEVNULL,
text=True).strip()
except (subprocess.CalledProcessError, OSError):
warn(f"git {' '.join(args)} failed in {repo}")
return ""
def bar(num, den, width=10):
pct = 0 if not den else 100.0 * num / den
filled = 0 if not den else int(round(width * num / den))
return f"`[{'█' * filled}{'░' * (width - filled)}] {pct:.0f}%`"
def n(x):
return f"{x:,}"
# ---------------------------------------------------------------- board
def parse_board(path):
rows = []
for i, ln in enumerate(read(path).splitlines(), 1):
if not ln.startswith("|"):
continue
cells = ln.split("|", 7) # ['', target, type, status, conf, coverage, updated, 'notes |']
if len(cells) < 8:
if not re.match(r"^\|\s*(Target|-+)\s*\|", ln):
warn(f"board.md line {i}: fewer than 7 cells, skipped")
continue
target, typ, status, conf, cov, updated = (c.strip() for c in cells[1:7])
notes = cells[7].strip().rstrip("|").strip()
if target == "Target" or set(target) <= {"-"}:
continue
if status not in STATUSES:
warn(f"board.md line {i}: unknown status '{status}' for '{target}'")
rows.append(dict(target=target.strip("`"), type=typ, status=status, conf=conf, cov=cov,
updated=updated, notes=notes))
if not rows:
warn("board.md: no rows parsed")
return rows
# ---------------------------------------------------------------- findings
def parse_functions(path):
m = re.search(r"\*{0,2}([\d,]{5,})\s+functions", read(path))
if m:
return int(m.group(1).replace(",", "")), "parsed"
warn("01-fingerprint.md: function count not found, using hardcoded 41,411")
return FUNCTIONS_FALLBACK, "hardcoded"
def recovered_layouts(paths):
"""Heuristic: unique `Game::X` / `Mars::X` names appearing in ##/###/#### headings of the
struct-recovery docs, excluding purely descriptive headings (no backticked class)."""
names = set()
for p in paths:
for ln in read(p).splitlines():
if re.match(r"^#{2,4} ", ln):
for m in re.finditer(r"`((?:Game|Mars)::[A-Za-z_][\w:]*)`", ln):
names.add(m.group(1))
return sorted(names)
def parse_addresses(path):
try:
d = json.loads(read(path) or "{}")
entries = d.get("entries", d if isinstance(d, list) else [])
except json.JSONDecodeError as e:
warn(f"addresses.json: {e}")
entries = []
total = len(entries)
verified = sum(1 for e in entries if str(e.get("status", "")).startswith("verified"))
return total, verified
# ---------------------------------------------------------------- verify
def parse_catalogs():
txt = read(rp("findings", "subsystems", "data-parsers.md"))
m = re.search(r"Total:\s*([\d,]+)\s+parsed,\s*([\d,]+)\s+failed", txt)
parsed = failed = None
if m:
parsed, failed = (int(x.replace(",", "")) for x in m.groups())
else:
warn("data-parsers.md: 'Total: N parsed, M failed' line not found")
kinds = None
try:
kinds = len(json.loads(read(rp("verify", "results", "data-catalogs", "schema_stats.json")) or "{}"))
except json.JSONDecodeError as e:
warn(f"schema_stats.json: {e}")
dangling = None
try:
x = json.loads(read(rp("verify", "results", "data-catalogs", "crosslink.json")) or "{}")
dangling = sum(len(v) for k, v in x.items() if k.endswith("_dangling") and isinstance(v, list))
except json.JSONDecodeError as e:
warn(f"crosslink.json: {e}")
return parsed, failed, kinds, dangling
def parse_oracles(engine):
"""docs/mars-*.md: the `| **total** |` row of a table whose header has a files column and an
agree/match column."""
out = []
for name in sorted(os.listdir(os.path.join(engine, "docs")) if os.path.isdir(os.path.join(engine, "docs")) else []):
if not (name.startswith("mars-") and name.endswith(".md")):
continue
header = None
found = False
in_table = False
for ln in read(os.path.join(engine, "docs", name)).splitlines():
if not ln.startswith("|"):
in_table = False
continue
cells = [c.strip().strip("*").strip().lower() for c in ln.strip().strip("|").split("|")]
if set("".join(cells)) <= {"-", ":", " "}:
continue
if not in_table: # first row of a table is its header
header, in_table = cells, True
continue
if cells and cells[0] == "total" and header:
def col(pred):
for h, c in zip(header, cells):
if pred(h):
m = re.search(r"[\d,]+", c)
if m:
return int(m.group(0).replace(",", ""))
return None
files = col(lambda h: "files" in h)
agree = col(lambda h: "agree" in h or "match" in h)
if files is not None and agree is not None:
out.append((name[:-3], agree, files))
found = True
break
if not found:
warn(f"{name}: no oracle total row parsed")
return out
def parse_saves():
d = rp("verify", "results", "saves")
saves = [f for f in (os.listdir(d) if os.path.isdir(d) else []) if f.endswith(".sav")]
head = "\n".join(read(os.path.join(d, "turn2-strict-issues.txt")).splitlines()[:3])
m = re.search(r"strict exit (\d+)\s*--\s*(\d+) errors,\s*(\d+) warnings", head)
if m:
ok = m.group(1) == "0" and m.group(2) == "0"
detail = f"strict exit {m.group(1)}, {m.group(2)} errors, {m.group(3)} warnings"
else:
ok = "0 resyncs" in head and "0 hint-failures" in head
detail = head.splitlines()[0][:80] if head else "issues file missing"
if not head:
warn("turn2-strict-issues.txt missing")
return len(saves), ok, detail
def parse_design_rules():
txt = read(rp("verify", "design-rules", "SHIP_DESIGN_RULES.md"))
m = re.search(r"(\d+)/(\d+) designs", txt)
if m:
return int(m.group(1)), int(m.group(2))
warn("SHIP_DESIGN_RULES.md: 'N/N designs' not found")
return None, None
# ---------------------------------------------------------------- engine
SRC_EXT = (".cpp", ".h", ".c")
def loc(path):
total = 0
for root, dirs, files in os.walk(path):
dirs[:] = [d for d in dirs if d != "third_party"]
for f in files:
if f.endswith(SRC_EXT):
with open(os.path.join(root, f), "rb") as fh:
total += sum(1 for _ in fh)
return total
CHECK_RE = re.compile(r"\b(?:CHECK|REQUIRE|ASSERT|EXPECT)\w*\s*\(|\bcheck\(|\bassert\(")
def engine_modules(engine):
src = os.path.join(engine, "src")
root_cmake = read(os.path.join(engine, "CMakeLists.txt"))
mods = set()
for root, dirs, files in os.walk(src):
dirs[:] = [d for d in dirs if d != "third_party"]
if "CMakeLists.txt" in files:
mods.add(os.path.relpath(root, engine))
# dirs whose sources are compiled directly from the root CMakeLists (e.g. src/shim/main.cpp)
for m in re.finditer(r"(?<![\w/])(src/[\w/]+)/[\w.]+\.(?:cpp|c|def)\b", root_cmake):
mods.add(m.group(1))
out = []
for mod in sorted(mods):
name = mod[len("src/"):]
tname = name.replace("/", "_")
tdir = os.path.join(engine, "tests", tname)
tfiles, checks = 0, 0
if os.path.isdir(tdir):
for root, dirs, files in os.walk(tdir):
dirs[:] = [d for d in dirs if d not in ("build",)]
for f in files:
tfiles += 1
if f.endswith(".cpp"):
checks += sum(1 for ln in read(os.path.join(root, f)).splitlines() if CHECK_RE.search(ln))
if f"add_subdirectory({mod})" in root_cmake:
wired = "yes"
elif re.search(rf"\b{re.escape(mod)}/", root_cmake):
wired = "direct (WIN32)" if "if(WIN32)" in root_cmake else "direct"
else:
wired = "no"
doc = ""
cand = os.path.join(engine, "docs", name.replace("/", "-") + ".md")
if os.path.exists(cand):
doc = os.path.basename(cand)
else:
leaf = name.split("/")[-1]
ddir = os.path.join(engine, "docs")
for d in sorted(os.listdir(ddir)) if os.path.isdir(ddir) else []:
if d.endswith(".md") and re.search(rf"\b{leaf}\b", "\n".join(read(os.path.join(ddir, d)).splitlines()[:5]), re.I):
doc = d
break
out.append(dict(name=name, loc=loc(os.path.join(engine, mod)), tfiles=tfiles, checks=checks,
wired=wired, doc=doc or "—"))
if not out:
warn("sots-engine: no modules found under src/")
return out
# ---------------------------------------------------------------- questions / backlog
def parse_questions(path):
open_, closed = [], []
for ln in read(path).splitlines():
m = re.match(r"^- \*\*(.*)", ln)
if not m:
continue
text = m.group(1)
if re.match(r"(RESOLVED|Resolved|\(parked\))", text):
closed.append(text)
else:
open_.append(text)
return open_, closed
def parse_backlog(path):
sec, counts = None, {}
for ln in read(path).splitlines():
m = re.match(r"^## (.+)", ln)
if m:
sec = m.group(1).split("(")[0].strip()
counts[sec] = 0
elif sec and re.match(r"^\s*(\d+\.|[-*]) ", ln):
counts[sec] += 1
return counts
def north_star(path):
txt = read(path)
m = re.search(r"## North star.*?\n\*\*(.+?)\*\*", txt, re.S)
return " ".join(m.group(1).split()) if m else "functional reimplementation, behavior-equivalent (see findings/00-strategy.md)"
def previous_metrics(path):
m = re.search(r"<!-- dashboard-metrics (\{.*?\}) -->", read(path) if os.path.exists(path) else "")
if not m:
return None
try:
return json.loads(m.group(1))
except json.JSONDecodeError:
warn("previous DASHBOARD.md: metrics comment unreadable")
return None
# ---------------------------------------------------------------- render
GLYPH = {"verified": "✅", "mapped": "✅", "in-progress": "🔄", "backlog": "⬜", "blocked": "⛔"}
def render(engine, prev):
L = []
now = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
re_commit = git(RE_ROOT, "log", "-1", "--format=%h,%cs")
eng_commit = git(engine, "log", "-1", "--format=%h,%cs")
eng_commits = git(engine, "rev-list", "--count", "HEAD") or "?"
rows = parse_board(rp("campaign", "board.md"))
by = {s: sum(1 for r in rows if r["status"] == s) for s in STATUSES}
total = len(rows)
mapped_plus = by["verified"] + by["mapped"]
# 1 header
L += ["# SotS RE campaign — coverage dashboard", "",
f"Generated {now} · `sots-re` @ {re_commit or '?'} · `sots-engine` @ {eng_commit or '?'} "
f"({eng_commits} commits) · regenerate with `tools/dashboard.py`", "",
f"> **North star:** {north_star(rp('findings', '00-strategy.md'))}", ""]
# 2 map coverage
L += ["## 1. Map coverage (campaign/board.md)", "",
f"{total} targets · mapped-or-better **{mapped_plus}/{total}** {bar(mapped_plus, total)} · "
f"verified **{by['verified']}/{total}** {bar(by['verified'], total)}", "",
"| Status | Count | % |", "|---|---:|---:|"]
for s in STATUSES:
L.append(f"| {s} | {by[s]} | {0 if not total else round(100 * by[s] / total)}% |")
L += ["", "| Type | verified | mapped | in-progress | backlog | blocked | total |", "|---|---:|---:|---:|---:|---:|---:|"]
known = set()
for label, types in TYPE_ROWS:
known.update(types)
sub = [r for r in rows if r["type"] in types]
c = {s: sum(1 for r in sub if r["status"] == s) for s in STATUSES}
L.append(f"| {label} | {c['verified']} | {c['mapped']} | {c['in-progress']} | {c['backlog']} | {c['blocked']} | {len(sub)} |")
other = [r for r in rows if r["type"] not in known]
if other:
warn("board.md: unknown types " + ", ".join(sorted({r['type'] for r in other})))
c = {s: sum(1 for r in other if r["status"] == s) for s in STATUSES}
L.append(f"| other | {c['verified']} | {c['mapped']} | {c['in-progress']} | {c['backlog']} | {c['blocked']} | {len(other)} |")
L.append("")
# 3 binary understanding
rtti = count_lines(rp("findings", "objects", "rtti-raw.txt"))
cg = count_lines(rp("findings", "objects", "classes-game.txt"))
cm = count_lines(rp("findings", "objects", "classes-mars.txt"))
ser = count_lines(rp("findings", "objects", "serializable-types.txt"))
layouts = recovered_layouts([rp("findings", "objects", "struct-recovery.md"), rp("findings", "objects", "schema-gaps-resolved.md")])
funcs, fsrc = parse_functions(rp("findings", "01-fingerprint.md"))
a_total, a_ver = parse_addresses(rp("ghidra", "addresses.json"))
L += ["## 2. Binary understanding", "",
f"- RTTI type descriptors: **{n(rtti)}** (`Game::` {n(cg)}, `Mars::` {n(cm)}; serializable types {n(ser)})",
f"- Classes with recovered member layouts: **{len(layouts)}** / {ser} serializable types "
f"{bar(len(layouts), ser)} — heuristic: distinct `Game::X`/`Mars::X` in `##`–`####` headings of "
f"`struct-recovery.md` + `schema-gaps-resolved.md`",
f"- Functions: **{n(funcs)}** ({fsrc} from `01-fingerprint.md`); named/annotated in the **address contract** "
f"(`ghidra/addresses.json`, not Ghidra's full rename count): **{a_total}**, verified **{a_ver}** "
f"{bar(a_ver, a_total)}", ""]
# 4 data layer
parsed, failed, kinds, dangling = parse_catalogs()
oracles = parse_oracles(engine)
nsaves, s_ok, s_detail = parse_saves()
d_ok, d_tot = parse_design_rules()
cat = f"**{n(parsed)}/{n(parsed + failed)}** files parsed" if parsed is not None else "**?** files parsed"
L += ["## 3. Data layer", "",
f"- Catalogs: {cat} ({kinds if kinds is not None else '?'} block kinds in `schema_stats.json`), "
f"dangling cross-refs **{dangling if dangling is not None else '?'}** (`crosslink.json`)"]
for name, agree, files in oracles:
L.append(f"- Oracle `{name}`: **{n(agree)}/{n(files)}** files agree {bar(agree, files)}")
if not oracles:
L.append("- Oracle agreements: none parsed from `sots-engine/docs/mars-*.md`")
L += [f"- Saves: **{nsaves}/{nsaves}** real saves strict-clean — {s_detail}" if s_ok else f"- Saves: {nsaves} real saves, strict **NOT clean** — {s_detail}",
f"- Design rules: **{d_ok}/{d_tot}** stock designs pass {bar(d_ok or 0, d_tot or 1)}" if d_ok is not None else "- Design rules: ? (unparsed)", ""]
# 5 engine
mods = engine_modules(engine)
eng_rows = [r for r in rows if r["type"] == "engine"]
e_ver = sum(1 for r in eng_rows if r["status"] == "verified")
e_map = sum(1 for r in eng_rows if r["status"] == "mapped")
e_fly = sum(1 for r in eng_rows if r["status"] == "in-progress")
tot_loc = sum(m["loc"] for m in mods)
tot_tf = sum(m["tfiles"] for m in mods)
tot_ck = sum(m["checks"] for m in mods)
L += ["## 4. Engine accrual (sots-engine)", "",
"| Module | LOC | Test files | Checks | Wired | Doc |", "|---|---:|---:|---:|---|---|"]
for m in mods:
L.append(f"| `{m['name']}` | {n(m['loc'])} | {m['tfiles']} | {m['checks']} | {m['wired']} | {m['doc']} |")
L += [f"| **total** | **{n(tot_loc)}** | **{tot_tf}** | **{tot_ck}** | | |", "",
f"Board `engine:` rows: verified **{e_ver}**, mapped {e_map}, in flight {e_fly} (of {len(eng_rows)}) — "
f"verified & merged {bar(e_ver, len(eng_rows))}", ""]
# 6 phase 2
L += ["## 5. Phase 2 milestones", "", "| Milestone | Status | Coverage | Notes |", "|---|---|---:|---|"]
p2 = [r for r in rows if r["target"].startswith("P2-M")]
for r in p2:
L.append(f"| {r['target']} | {GLYPH.get(r['status'], '?')} {r['status']} | {r['cov']} | {r['notes'][:90]} |")
if not p2:
warn("board.md: no P2-M* rows")
L.append("")
# 7 verification ledger
def status_of(target_re):
for r in rows:
if re.search(target_re, r["target"]):
return r["status"]
return "not on board"
m0 = os.path.exists(rp("verify", "results", "shim", "m0.log"))
harness = os.path.isdir(rp("verify", "harness", "compare"))
L += ["## 6. Verification ledger", "",
f"- {'✅' if s_ok else '❌'} Saves strict: {nsaves}/{nsaves} ({s_detail})",
f"- {'✅' if d_ok == d_tot and d_ok else '❌'} Design rules: {d_ok}/{d_tot}",
"- " + (" · ".join(f"{'✅' if a == f else '❌'} oracle {nm} {a}/{f}" for nm, a, f in oracles) if oracles else "❌ oracle parsers: none parsed"),
f"- {'✅' if harness else '❌'} Compare harness present (`verify/harness/compare/`)",
f"- {'✅' if m0 else '❌'} M0 evidence present (`verify/results/shim/m0.log`)",
f"- {GLYPH.get(status_of(r'^determinism oracle'), '⬜')} Determinism oracle: {status_of(r'^determinism oracle')}", ""]
# 8 open questions
oq, closed = parse_questions(rp("campaign", "open-questions.md"))
bl = parse_backlog(rp("campaign", "backlog.md"))
L += ["## 7. Open questions", "",
f"Open **{len(oq)}** · resolved/parked {len(closed)} · backlog items: " +
(", ".join(f"{k} {v}" for k, v in bl.items()) if bl else "?"), "", "Most recent open:", ""]
for t in oq[-5:][::-1]:
L.append(f"- {t[:100].replace('**', '')}{'…' if len(t) > 100 else ''}")
L.append("")
# 9 delta
cur = dict(verified=by["verified"], mapped_plus=mapped_plus, targets=total, loc=tot_loc, tests=tot_tf,
checks=tot_ck, addr_verified=a_ver, addr_total=a_total, layouts=len(layouts), open_q=len(oq))
L += ["## 8. Delta since previous dashboard", ""]
if prev:
def d(k):
if k not in prev:
return "n/a"
diff = cur[k] - prev[k]
return f"{prev[k]:,} → {cur[k]:,} ({diff:+,})"
L += [f"- verified targets: {d('verified')} · mapped-or-better: {d('mapped_plus')}",
f"- engine LOC: {d('loc')} · test files: {d('tests')} · checks: {d('checks')}",
f"- addresses verified: {d('addr_verified')} · recovered layouts: {d('layouts')} · open questions: {d('open_q')}", ""]
else:
L += ["- first run (no previous `DASHBOARD.md` metrics found)", ""]
L += ["---", f"warnings: {'; '.join(WARN) if WARN else 'none'}",
f"<!-- dashboard-metrics {json.dumps(cur)} -->", ""]
return "\n".join(L)
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--print", action="store_true", help="echo the markdown to stdout")
ap.add_argument("--no-write", action="store_true", help="do not write campaign/DASHBOARD.md (implies --print)")
ap.add_argument("--engine", default=os.environ.get("SOTS_ENGINE", os.path.expanduser("~/sots-engine")))
a = ap.parse_args()
if not os.path.isdir(a.engine):
warn(f"sots-engine not found at {a.engine}")
out_path = rp("campaign", "DASHBOARD.md")
md = render(a.engine, previous_metrics(out_path))
if a.print or a.no_write:
sys.stdout.write(md)
if not a.no_write:
with open(out_path, "w", encoding="utf-8") as f:
f.write(md)
print(f"wrote {os.path.relpath(out_path, RE_ROOT)} ({len(WARN)} warnings)", file=sys.stderr)
if __name__ == "__main__":
main()