"""p06_report.py: render the Phase 2 report from report_template.md and the values file. Two render modes share one code path: normal every {{token|format}} is replaced by the formatted value, every generated block is built; skeleton every token becomes "§" and every verbatim source string (quotes, citations) becomes "¤", so the check script can prove that no digit in the report was typed by hand. Authored strings that carry tokens (score reasons, vet reasons) are rendered token by token in both modes. """ import json, re, sys, csv sys.path.insert(0, "/home/claude/mill/code") from vals import fmt_value, TOKEN INP = "/home/claude/mill/inputs" OUT = "/home/claude/mill/outputs" CODE = "/home/claude/mill/code" class R: def __init__(self, values, mode="normal", extra=None): self.v = dict(values) if extra: self.v.update(extra) self.mode = mode self.used = set() self.missing = [] def tok(self, name, spec=""): if name not in self.v: self.missing.append(name) return "{{" + name + "}}" self.used.add(name) return "§" if self.mode == "skeleton" else fmt_value(self.v[name], spec) def text(self, s): """Authored text that may contain tokens.""" return TOKEN.sub(lambda m: self.tok(m.group(1), m.group(2) or ""), s) def quote(self, s): """Verbatim source text (quote, citation, URL): shown as retrieved.""" return "¤" if self.mode == "skeleton" else s def load(): return { "ideas": json.load(open(f"{INP}/ideas.json")), "scores": json.load(open(f"{INP}/scores.json")), "nov": json.load(open(f"{INP}/novelty.json")), "jr": json.load(open(f"{INP}/journals.json")), "vet": json.load(open(f"{INP}/vet.json")), "src": json.load(open(f"{INP}/sources.json")), "isc": json.load(open(f"{OUT}/idea_scores.json")), "qlog": list(csv.DictReader(open(f"{INP}/query_log.tsv"), delimiter="\t")), } CH_NAME = {"acc0": "Acc0 spindle accelerometer", "acc1": "Acc1 base-column accelerometer", "s0": "Sound0 microphone", "s1": "Sound1 stethoscope sensor near spindle", "s2": "Sound2 stethoscope sensor on base column", "load": "MTConnect spindle load", "smarms": "MTConnect spindle-motor vibration RMS", "watt": "MTConnect true power"} MACH = ("imi_vm20i", "imi_vmx30ui") CRIT = ["novelty", "feasibility", "effect", "audience", "desk_risk", "cost"] CRIT_SHORT = {"novelty": "Nov", "feasibility": "Feas", "effect": "Eff", "audience": "Aud", "desk_risk": "Desk", "cost": "Cost"} def cap(t): return t[:1].upper() + t[1:] if t else t def end(t): """Close a sentence with a period unless it already ends with terminal punctuation (inside a closing quote too).""" t = t.rstrip() return t if t.endswith((".", "?", "!", '."', '?"', '!"')) else t + "." def days(r, name): """A day count with the right noun.""" return r.tok(name, "int") + (" day" if r.v[name]["value"] == 1 else " days") def est_ci(r, name, spec_pt, spec_ci): if name not in r.v: return "not recorded" return f"{r.tok(name, spec_pt)} ({r.tok(name + '_ci', spec_ci)})" def blocks(r, D): B = {} # --- sources lines = [] for s in D["src"]["sources"]: lines.append(f"- {s['id']}: {r.quote(s['citation'].rstrip('.'))}. {r.quote(s['url'])}. Retrieved {r.tok('sources_retrieved', 'date')} by {s['method']}. {s['tag']}.") B["sources"] = "\n".join(lines) ka = next(s for s in D["src"]["sources"] if s["id"] == "S_kaggle_api") B["landing"] = f"Landing page: {r.quote(ka['landing_page'])} (DOI {r.quote(ka['doi_url'])})." # --- validity rows = ["| Component | Score | Evidence |", "|---|---|---|"] names = {"provenance": "Provenance and collection rigour", "documentation": "Documentation completeness", "licensing": "Licensing clarity", "size": "Sample size and representativeness", "errata": "Known errata"} for k, c in D["vet"]["components"].items(): rows.append(f"| {names[k]} | {r.tok('vet_' + k, 'int')} | {cap(r.text(c['why']))} |") B["validity"] = "\n".join(rows) # --- D1 by tool (laboratory machines) t2 = {row[0].replace("*", ""): row[1] for row in next(x for x in D["src"]["sources"] if x["id"] == "S_table2")["table"]["rows"]} rows = ["| Tool | Tool table entry (verbatim) | Paths | Runs | Label 1, share of paths | Label 1, share of cutting time | Label 2, share of paths |", "|---|---|---|---|---|---|---|"] for key in sorted(k[len("d1_paths_tool_"):] for k in r.v if k.startswith("d1_paths_tool_")): name = key.replace("star", "\\*").upper() desc = t2.get(key.replace("star", "").upper(), "") rows.append(f"| {name} | {r.quote(desc)} | {r.tok(f'd1_paths_tool_{key}', 'int')} | {r.tok(f'd1_runs_tool_{key}', 'int')} | {r.tok(f'd1_l1_share_tool_{key}', 'pct0')} | {r.tok(f'd1_l1_time_share_tool_{key}', 'pct0')} | {r.tok(f'd1_l2_share_tool_{key}', 'pct0')} |") B["d1_tools"] = "\n".join(rows) # --- D2 pairs rows = [f"| Machine | Design pairs | Up abnormal, down normal | Down abnormal, up normal | Both abnormal | Odds ratio, up against down ({r.tok('ci_level', 'pct0')} interval) |", "|---|---|---|---|---|---|"] for m in MACH: rows.append(f"| {m} | {r.tok(f'd2_pairs_{m}', 'int')} | {r.tok(f'd2_disc_up_{m}', 'int')} | {r.tok(f'd2_disc_down_{m}', 'int')} | {r.tok(f'd2_concord1_{m}', 'int')} | {est_ci(r, f'd2_or_updown_{m}', 'f0', 'ci_f0')} |") B["d2_pairs"] = "\n".join(rows) rows = [f"| Parameter | Probability that the label 1 path has the larger value ({r.tok('ci_level', 'pct0')} interval) |", "|---|---|"] pn = {"ap": "axial depth ap", "ae": "radial width ae", "n": "spindle speed N", "fz": "feed per tooth", "mrr": "material removal rate", "up": "up milling (scored higher) against down milling"} for c, lab in pn.items(): rows.append(f"| {lab} | {est_ci(r, f'd2_within_auc_{c}', 'f2', 'ci_f2')} |") B["d2_within"] = "\n".join(rows) # --- D3 and D4 rows = ["| Channel | imi_vm20i, all paths | imi_vm20i, within matched strata | imi_vmx30ui, all paths | imi_vmx30ui, within matched strata |", "|---|---|---|---|---|"] for ch, lab in CH_NAME.items(): cells = [] for m in MACH: cells.append(est_ci(r, f"d3_auc_{m}_{ch}", "f2", "ci_f2")) few = (f"d4_auc_{m}_{ch}" in r.v and r.v[f"d4_runs_{m}"]["value"] < r.v["min_runs_interpretable"]["value"]) cells.append(est_ci(r, f"d4_auc_{m}_{ch}", "f2", "ci_f2") + (" (few runs)" if few else "")) rows.append(f"| {lab} | " + " | ".join(cells) + " |") B["auc"] = "\n".join(rows) # --- D5 rows = ["| Machine and channel | Label 1 paths | Label 0 paths | Difference, percentage points | Paths |", "|---|---|---|---|---|"] for m in MACH: for ch in ("acc0", "acc1", "s1", "s2"): if f"d5_nhdom_{m}_{ch}_l1" not in r.v: continue rows.append(f"| {m}, {CH_NAME[ch]} | {est_ci(r, f'd5_nhdom_{m}_{ch}_l1', 'pct0', 'ci_pct0')} | {est_ci(r, f'd5_nhdom_{m}_{ch}_l0', 'pct0', 'ci_pct0')} | {est_ci(r, f'd5_nhdom_diff_{m}_{ch}', 'pp0', 'ci_pp0')} | {r.tok(f'd5_n_{m}_{ch}', 'int')} |") B["d5"] = "\n".join(rows) # --- D7 rows = [f"| Machine and workpiece | Specific cutting energy, J/mm3 ({r.tok('ci_level', 'pct0')} interval) | Paths | Runs | Paths at or below the pre-cut power |", "|---|---|---|---|---|"] for m in MACH: for wp in ("aa6061", "aisi1018", "aisi4140"): k = f"d7_u_{m}_{wp}" if k not in r.v: continue wpn = {"aa6061": "AA 6061", "aisi1018": "AISI 1018", "aisi4140": "AISI 4140"}[wp] few = " (few runs)" if r.v[f"d7_runs_{m}_{wp}"]["value"] < r.v["min_runs_interpretable"]["value"] else "" rows.append(f"| {m}, {wpn} | {est_ci(r, k, 'f2', 'ci_f2')}{few} | {r.tok(f'd7_n_{m}_{wp}', 'int')} | {r.tok(f'd7_runs_{m}_{wp}', 'int')} | {r.tok(f'd7_negative_share_{m}_{wp}', 'pct0')} |") B["d7"] = "\n".join(rows) # --- D10 rows = [f"| Channel | RMS ratio, defective over matched good ({r.tok('ci_level', 'pct0')} interval) | Matched label 2 paths | Runs |", "|---|---|---|---|"] for ch in ("acc0", "acc1", "s0", "s1", "s2"): if f"d10_ratio_{ch}" in r.v: few = " (few runs)" if r.v[f"d10_runs_{ch}"]["value"] < r.v["min_runs_interpretable"]["value"] else "" rows.append(f"| {CH_NAME[ch]} | {est_ci(r, f'd10_ratio_{ch}', 'f2', 'ci_f2')}{few} | {r.tok(f'd10_n_{ch}', 'int')} | {r.tok(f'd10_runs_{ch}', 'int')} |") B["d10"] = "\n".join(rows) # --- D9 rows = ["| Machine and channel | Samples at full scale during cutting | Paths above the clipping threshold | Pre-cut windows above the threshold |", "|---|---|---|---|"] for m in ("imi_vm20i", "imi_vmx30ui", "tmf_vf10"): for ch in ("s0", "s1", "s2"): k = f"d9_fs_share_{m}_{ch}" if k in r.v: rows.append(f"| {m}, {CH_NAME[ch]} | {r.tok(k, 'pct1')} | {r.tok(f'd9_paths_gt1pct_{m}_{ch}', 'pct0')} | {r.tok(f'd9_pre_paths_gt1pct_{m}_{ch}', 'pct0')} |") B["d9"] = "\n".join(rows) # --- screening S1 rows = ["| Machine | Features | Random path split: median AUC (range over repeats) | Run split: median AUC (range) | Gap |", "|---|---|---|---|---|"] fsn = {"highrate": "RMS of the high-rate channels", "mtc": "MTConnect spindle vibration, load and power", "s2only": "Sound2 RMS alone", "clip": "Sound0 full-scale share alone"} for m in MACH: for fs_, lab in fsn.items(): k = f"s1_auc_{m}_{fs_}" if f"{k}_random" not in r.v: continue rows.append(f"| {m} | {lab} | {r.tok(k + '_random', 'f3')} ({r.tok(k + '_random_min', 'f3')} to {r.tok(k + '_random_max', 'f3')}) | {r.tok(k + '_run', 'f3')} ({r.tok(k + '_run_min', 'f3')} to {r.tok(k + '_run_max', 'f3')}) | {r.tok(f's1_gap_{m}_{fs_}', 'f3')} |") B["s1"] = "\n".join(rows) # --- rubric rows = ["| Criterion | Weight |", "|---|---|"] for k in CRIT: rows.append(f"| {k.replace('_', ' ')} | {r.tok('w_' + k, 'f2')} |") B["rubric"] = "\n".join(rows) # --- ideas list lines = [] for i in D["ideas"]: lines.append(f"- **{i['id']}. {end(r.text(i['title']))}** {end(r.text(i['question']))} Data: {end(r.text(i['data']))} Method: {end(r.text(i['method']))}") B["ideas"] = "\n".join(lines) # --- scores matrix, sorted by weighted score thr = D["scores"]["rubric"]["threshold"] rows = ["| Idea | " + " | ".join(CRIT_SHORT[k] for k in CRIT) + " | Weighted | Status | Why |", "|---|" + "---|" * len(CRIT) + "---|---|---|"] for iid in D["isc"]["ranking"]: st = "survives" if iid in D["isc"]["survivors"] else ("near" if iid in D["isc"]["near"] else "drops") short = next(i["short"] for i in D["ideas"] if i["id"] == iid) rows.append(f"| {iid} {short} | " + " | ".join(r.tok(f"sc_{iid.lower()}_{k}", "int") for k in CRIT) + f" | {r.tok('tot_' + iid.lower(), 'f2')} | {st} | {r.text(D['scores']['scores'][iid]['why'])} |") B["scores"] = "\n".join(rows) # --- packages lines = [] for p, pk in D["scores"]["packages"].items(): mem = ", ".join(f"{i} ({r.tok('tot_' + i.lower(), 'f2')})" for i in pk["ideas"]) lines.append(f"- **{p}: {pk['title']}.** Members: {mem}. {r.text(pk['why'][0].upper() + pk['why'][1:])}.") B["packages"] = "\n".join(lines) # --- databases lines = [f"- {d['name']}: {d['route']}; {r.text(d['status'])}." for d in D["nov"]["databases"]] B["databases"] = "\n".join(lines) # --- query log summary by = {} for q in D["qlog"]: by.setdefault(q["database"], []).append(q) rows = ["| Database | Entries in the log |", "|---|---|"] for db, qs in by.items(): rows.append(f"| {db} | {r.tok('qlog_n_' + re.sub('[^a-z]', '', db.lower()), 'int')} |") B["qlog"] = "\n".join(rows) # --- dataset users lines = [] for u in D["nov"]["dataset_users"]: q = (" " + u.get("quote_lead", "Quoted:") + " " + " ".join(f"\"{r.quote(x)}\"" for x in u["quotes"])) if u.get("quotes") else "" pre = f" Preprint: {r.quote(u['preprint_citation'])}." if u.get("preprint_citation") else "" lines.append(f"- {u['id']}: {r.quote(u['citation'])} ({u['tag']}; read: {u['read']}).{pre} {end(cap(r.text(u['what'])))}{q}") B["users"] = "\n".join(lines) # --- closest works per package users = {u["id"]: u for u in D["nov"]["dataset_users"]} for p, lst in D["nov"]["closest"].items(): lines = [] for n, c in enumerate(lst, 1): if "id" in c: cit = r.quote(users[c["id"]]["citation"]); tag = users[c["id"]]["tag"] else: cit = r.quote(c["citation"]); tag = c["tag"] lines.append(f"{n}. {cit} ({tag}). How this differs: {end(r.text(c['differs']))}") B["closest_" + p] = "\n".join(lines) # --- missed-work probabilities rows = ["| Package or idea | Probability that close prior work was missed | Why | What would move it |", "|---|---|---|---|"] for p, rec in D["nov"]["missed_work_probability"].items(): rows.append(f"| {p} | {r.tok('pmiss_' + p.lower(), 'pct0')} | {cap(r.text(rec['reason']))} | {cap(r.text(rec.get('move', '')))} |") B["pmiss"] = "\n".join(rows) # --- journals: which journal for which paper, then one card per proposed journal J = {j["id"]: j for j in D["jr"]["journals"]} rows = ["| Paper | Candidate journals, in order of preference |", "|---|---|"] for p, ids in D["jr"]["assignment"].items(): rows.append(f"| {p} | " + "; ".join(J[jid]["name"] for jid in ids) + " |") B["journal_assign"] = "\n".join(rows) order = [] for ids in D["jr"]["assignment"].values(): order += [jid for jid in ids if jid not in order] cards = [] for jid in order: j = J[jid]; k = jid.lower() used_in = ", ".join(p for p, ids in D["jr"]["assignment"].items() if jid in ids) lines = [f"**{j['name']}** ({j['publisher']}; proposed for {used_in}; {j['tag']}, read {r.tok('sources_retrieved', 'date')})", ""] # impact factor with its release src_if = j["urls"].get("insights", j["urls"].get("home", "")) if j.get("if_release_quote"): rel = f"release named in the page's impact factor information box: \"{r.quote(j['if_release_quote'])}\"" elif f"j_{k}_if_year" in r.v: rel = f"printed as the {r.tok(f'j_{k}_if_year', 'raw')} Journal Impact Factor" else: rel = "release year not printed" extra = f"; five-year impact factor {r.tok(f'j_{k}_if5', 'f1')}" if f"j_{k}_if5" in r.v else "" extra += f"; CiteScore {r.tok(f'j_{k}_citescore', 'f1')}" if f"j_{k}_citescore" in r.v else "" lines.append(f"- Impact factor: {r.tok(f'j_{k}_if', 'f1')} ({rel}){extra}. Source: {r.quote(src_if)}") # fees fq = " ".join(f"\"{r.quote(q)}\"" for q in j["fee_quotes"]) lines.append(f"- Fees: {j['fee_model']}. {fq} Fee page: {r.quote(j['urls']['fees'])}") # scope sc = f"\"{r.quote(j['scope_quote'])}\"" if j.get("scope_quote_2"): sc += f" \"{r.quote(j['scope_quote_2'])}\"" lines.append(f"- Scope: {sc} Precedent: {end(r.text(j['precedent']))} Source: {r.quote(j['urls'].get('scope', j['urls'].get('home', '')))}") # time to first decision if f"j_{k}_first_decision_days" in r.v: fd = f"{days(r, f'j_{k}_first_decision_days')}, submission to first decision" + (" (median, as printed)" if j.get("first_decision_median") else "") else: fd = "not printed" if f"j_{k}_after_review_days" in r.v: fd += f"; {days(r, f'j_{k}_after_review_days')}, submission to decision after review" if f"j_{k}_acceptance_days" in r.v: fd += f"; {days(r, f'j_{k}_acceptance_days')}, submission to acceptance" if f"j_{k}_acceptance_rate" in r.v: fd += f"; acceptance rate {r.tok(f'j_{k}_acceptance_rate', 'pct0')}" if j.get("review_rule"): fd += f"; the guide says \"{r.quote(j['review_rule'])}\"" lines.append(f"- Time to first decision: {end(fd)}") # article type at = f"\"{r.quote(j['article_types'])}\"" if j.get("article_types_is_quote") else r.text(j["article_types"]) if j.get("article_types_label"): at = f"{j['article_types_label']}: {at}" if j.get("review_quote"): at += f"; the guide says \"{r.quote(j['review_quote'])}\"" if j.get("full_paper_quote"): at += f"; the guide adds \"{r.quote(j['full_paper_quote'])}\"" lines.append(f"- Article type: {end(at)} Guide: {r.quote(j['urls'].get('guide', ''))}") # policies that could block us pol = [f"{end(cap(r.text(j['blocking'])))}"] if j.get("data_quote"): pol.append(f"Data: \"{r.quote(j['data_quote'])}\"") if j.get("preprint_quote"): src = f" ({r.quote(j['preprint_url'])})" if j.get("preprint_url") else "" pol.append(f"Preprints: \"{r.quote(j['preprint_quote'])}\"{src}") if j.get("authorship_quote"): pol.append(f"Authorship: \"{r.quote(j['authorship_quote'])}\"; no restriction by career stage or institution type was seen on the pages read.") ai = f"\"{r.quote(j['ai_policy_quote'])}\"" for kq in ("ai_declaration_quote", "ai_policy_quote_2"): if j.get(kq): ai += f" \"{r.quote(j[kq])}\"" pol.append(f"Generative AI (flagged): {ai}") lines.append("- Policies that could block us: " + " ".join(pol)) lines.append(f"- Predatory signals: {end(cap(r.text(j['predatory_signals'])))}") if j.get("notes"): lines.append(f"- Gaps: {end(cap(r.text(j['notes'])))}") cards.append("\n".join(lines)) B["journal_cards"] = "\n\n".join(cards) # --- confidence texts B["conf"] = end(cap(r.text(D["scores"]["confidence"]["what_moves_it"]))) B["conf_gate"] = end(cap(r.text(D["scores"]["confidence"]["gate_what_moves_it"]))) # --- vet texts B["impact_ceiling"] = end(cap(r.text(D["vet"]["impact_ceiling"]))) B["bad_choice"] = end(cap(r.text(D["vet"]["bad_choice_reason"]))) B["access"] = r.text(D["vet"]["access"]["text"]) # --- appendix: the full query log, verbatim esc = lambda x: (x or "").replace("|", "\\|") rows = ["| Batch | Idea | Database | Route | Query, verbatim | Time (UTC) | Status | Hits | Identifiers or notes |", "|---|---|---|---|---|---|---|---|---|"] for q in D["qlog"]: rows.append("| " + " | ".join(r.quote(esc(q[c])) for c in ("batch", "idea", "database", "route", "query", "time_utc", "status", "count", "top_ids_seen")) + " |") B["qlog_full"] = "\n".join(rows) return B def render(values, mode="normal", extra=None): D = load() r = R(values, mode, extra) tpl = open(f"{CODE}/report_template.md").read() B = blocks(r, D) def sub_block(m): return B[m.group(1)] out = re.sub(r"\{\{block:([A-Za-z0-9_]+)\}\}", sub_block, tpl) if re.search(r"\{\{block:", out): raise ValueError("unresolved block marker") out = TOKEN.sub(lambda m: r.tok(m.group(1), m.group(2) or ""), out) return out, r def qlog_counts(values): """Entries per database in the query log, registered as derived values for the summary table.""" D = load() extra = {} for q in D["qlog"]: k = "qlog_n_" + re.sub("[^a-z]", "", q["database"].lower()) extra.setdefault(k, {"value": 0, "unit": "entries", "estimand": {"population": "inputs/query_log.tsv", "filter": q["database"], "statistic": "entries"}, "source": "p06_report.py", "kind": "value"}) extra[k]["value"] += 1 return extra if __name__ == "__main__": vals = json.load(open(f"{OUT}/values.json")) extra = qlog_counts(vals) extra["check_line"] = {"value": "check not yet run", "unit": "text", "estimand": "p08_check.py", "source": "p06_report.py", "kind": "text"} txt, r = render(vals, "normal", extra) if r.missing: print("MISSING TOKENS:", sorted(set(r.missing))) open(f"{OUT}/phase2_report_draft.md", "w").write(txt) print("rendered", len(txt), "chars;", len(r.used), "distinct values used")