507 lines
22 KiB
Python
Executable file
507 lines
22 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
|
|
|
|
|
|
# ---------------------------------------------------------------- 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("")
|
|
|
|
# 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()
|