sots-re/tools/dashboard.py
alex c504729341 lane S2: the standalone scaffold, and the measured distance to the byte-match
tools/standalone_report.py drives sots-engine's sots_turn over each
consecutive-turn save pair and diffs the result against the game's own
post-turn save with state_checksum.py, which localises to named leaves and
proves its own coverage by re-serialisation.

  turn1-state -> turn2-state   baseline 209 diverging, after 204, closed 5
  turn2-state -> turn3-state   baseline 108 diverging, after 103, closed 5
  regressed 0 on both

`regressed` is reported next to `closed` and never netted off. It earned its
place immediately: committing the phase-31 player-status restore turned two
agreeing leaves into disagreeing ones, because the phase writes 1 and the file
carries 4.

The stable-system stand-in feeding the colony pass is a labelled hypothesis and
it survived a changed workload -- the same 3 ntdev leaves closed on both pairs,
six agreements, zero disagreements.

Two things deliberately NOT implemented: the TShn/ltis counters (18 leaves, a
`+1` would close them, but "+1 across one observed turn" is a hypothesis, not a
reading), and the RNG state write-back (an advanced-but-incomplete generator is
wrong in a different way from an untouched one).

dashboard.py gains section 6, reading verify/results/standalone/status.json:
phases modelled/committed per driver, baseline vs after, closed vs regressed,
the subsystem breakdown of what still differs, and the RNG gap. Sections 6-8
renumbered to 7-9; the delta footer tracks the two new counts.
DASHBOARD_README.md documents every number.

findings/control-flow/standalone-scaffold.md has the ranked blocker list.
2026-09-08 10:35:56 -04:00

576 lines
26 KiB
Python
Executable file

#!/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 glob
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", "objects")), ("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):
"""Classes with a recovered member layout.
Primary source is objects/layouts.json (the serializer-recovery output — machine-generated,
hundreds of classes). The heading heuristic over the hand-written struct-recovery docs is kept
as a supplement: it catches classes recovered by hand that the serializer pass never saw."""
names = set()
try:
d = json.loads(read(rp("objects", "layouts.json")) or "{}")
names.update(k for k in d if "::" in k)
except json.JSONDecodeError as e:
warn(f"layouts.json: {e}")
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):
# addresses.json plus every per-lane ghidra/addresses.d/*.json fragment — the same set
# gen_addresses.py merges. Counting only the base file undercounts while lanes are in flight.
sources = [path] + sorted(glob.glob(os.path.join(os.path.dirname(path), "addresses.d", "*.json")))
entries = []
for src in sources:
try:
d = json.loads(read(src) or "{}")
entries += d.get("entries", d if isinstance(d, list) else [])
except json.JSONDecodeError as e:
warn(f"{os.path.basename(src)}: {e}")
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
# ---------------------------------------------------------------- standalone
def parse_standalone(path):
"""verify/results/standalone/status.json, written by tools/standalone_report.py.
Absent or malformed -> None, so the section renders as 'not measured' rather than
breaking the dashboard."""
if not os.path.exists(path):
return None
try:
with open(path, encoding="utf-8") as f:
d = json.load(f)
except (OSError, ValueError) as e:
warn(f"standalone status.json unreadable: {e}")
return None
if d.get("schema") != "sots-standalone-status/1":
warn(f"standalone status.json: unexpected schema {d.get('schema')!r}")
return None
return d
# ---------------------------------------------------------------- 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)}** / {n(cg + cm)} named classes "
f"{bar(len(layouts), cg + cm)} — `objects/layouts.json` (serializer recovery) plus classes "
f"recovered by hand in `struct-recovery.md` + `schema-gaps-resolved.md`. Note {ser} types are "
f"*serializable*; the recovery also reaches non-serializable ones, so this is not a subset of that",
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("")
# 6 standalone progress -- the north star, measured
sa = parse_standalone(rp("verify", "results", "standalone", "status.json"))
L += ["## 6. Standalone (`src/app`) — distance to the byte-match", ""]
sa_closed = sa_left = 0
if not sa:
L += ["Not measured. Build `sots-engine`'s host preset and run "
"`tools/standalone_report.py`.", ""]
else:
ph, tp = sa.get("phases") or {}, sa.get("tailPhases") or {}
ref = sa.get("reference") or {}
sa_closed = ref.get("closed") or 0
sa_left = ref.get("divergingAfterTurn") or 0
base = ref.get("baselineDiverging") or 0
if ph:
L += [f"Turn-driver phases: **{ph['modelled']}/{ph['total']}** modelled "
f"({ph['committed']} committed) {bar(ph['modelled'], ph['total'])}",
"",
f"| | verified | implemented | partial | blocked | stub |",
"|---|---:|---:|---:|---:|---:|",
f"| turn drivers (44) | {ph['verified']} | {ph['implemented']} | "
f"{ph['partial']} | {ph['blocked']} | {ph['stub']} |"]
if tp:
L.append(f"| post-combat tail ({tp['total']}) | {tp['verified']} | "
f"{tp['implemented']} | {tp['partial']} | {tp['blocked']} | "
f"{tp['stub']} |")
L.append("")
L += [f"Reference pair `{ref.get('input')}` → `{ref.get('oracle')}`, leaves localised by "
"`state_checksum.py` (coverage proved by re-serialisation):", "",
f"- baseline (a standalone that does nothing): **{base}** leaves diverge",
f"- after one standalone turn: **{sa_left}** leaves diverge "
f"— closed {sa_closed}, regressed {ref.get('regressed')}",
f"- byte match: {'✅ YES' if ref.get('byteMatch') else '❌ not yet'} "
f"{bar(sa_closed, base) if base else ''}", ""]
subs = ref.get("subsystems") or {}
if subs:
L += ["Where the remaining divergence lives:", "",
"| Subsystem | Leaves |", "|---|---:|"]
for k, v in list(subs.items())[:8]:
L.append(f"| `{k}` | {v} |")
L.append("")
rng = sa.get("rng") or {}
L += [f"Generator: {rng.get('wordsModelled')} word(s) modelled per turn; "
f"unattributed per turn: {rng.get('wordsPerTurnUnattributed')}. "
"A byte-match is impossible until that closes — the generator state is saved "
"state.", "",
"Detail: `verify/results/standalone/report.txt`.", ""]
# 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 += ["## 7. 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 += ["## 8. 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),
sa_closed=sa_closed, sa_left=sa_left)
L += ["## 9. 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')}",
f"- standalone leaves closed: {d('sa_closed')} · leaves still diverging: {d('sa_left')}", ""]
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()