#!/usr/bin/env python3 # -*- coding: utf-8 -*- """coverage_check.py -- the COMPLETENESS checker for the Observed Particle Spectrum Closure verification suite. WHAT THIS DOES (safe wording, binding): This suite verifies that every observed particle is quantum-number-consistent with its geometry-derived constituents and that the parameter-free QCD symmetry relations hold against PDG-2024; it does NOT compute or claim absolute hadron masses from geometry. This particular engine is the COVERAGE / COMPLETENESS cross-check. It does NOT touch physics values -- it answers one structural question fail-closed: "Is every accounted particle present in EXACTLY one place on both sides of the ledger?" It cross-checks, in BOTH directions and FAIL-CLOSED: (A) every particle row in ./data/dataset_*.csv is covered by an accounting section (one of the 51 section files under the --sections dir); a row whose `chunk` has no section home, or whose (sector, chunk) pair contradicts the authoritative partition, is an ORPHAN -> FAIL. (B) every accounting section (chunk) is represented in the dataset: a section with zero dataset rows means its particles are MISSING from the dataset -> FAIL. (C) per-chunk counts reconcile: the number of per-particle accounting blocks parsed from a section equals the number of dataset rows carrying that chunk -> mismatch is reported and FAILs. (D) per-sector counts reconcile against the inventory totals declared in the foundation inventory files -> a mismatch beyond a documented convention allowance is reported. RUNNABLE-FIRST: bare Python 3, stdlib only (csv, json, re, argparse, dataclasses, pathlib, collections, sys, hashlib not needed here). Zero pip installs. FAIL-CLOSED: any orphan (either direction), any unknown/empty chunk on a data row, any per-chunk count mismatch, or a missing required field => non-zero exit. NO MAGIC NUMBERS: the chunk->sector partition is the authoritative SECTORS table from the wave-2 workflow (spectrum_wave2_scripts_wf.js, const SECTORS); the inventory sector totals are cited to the exact inventory file + line they are declared on. These are STRUCTURAL counts (not PDG physics constants); PDG physical values live in the dataset CSVs and each carries its own `pdg_source`. CLI: python coverage_check.py --data data --sections "/sections" --out out """ from __future__ import annotations import argparse import csv import json import re import sys from collections import defaultdict from dataclasses import dataclass, field from pathlib import Path # ---------------------------------------------------------------------------------------- # AUTHORITATIVE PARTITION (structural, not physics). # Source: const SECTORS (lines 29-36). # Each (sector_key -> [chunk labels]). The chunk label equals the section filename stem and # equals the value carried in the dataset `chunk` column (verified: A.md->chunk "A", # BB-1.md->"BB-1", QK-B1.md->"QK-B1", C1.md->"C1", etc.). # # NOTE the deliberate disambiguation: chunk "C" alone is the leptons_gauge quark chunk, # whereas "C1".."C12" are the light_strange_baryons chunks. The map keeps them distinct. # ---------------------------------------------------------------------------------------- SECTOR_CHUNKS: dict[str, list[str]] = { "leptons_gauge": ["A", "B", "C", "D", "E"], "light_mesons": [f"LM-{i}" for i in range(1, 10)], # LM-1..LM-9 "strange_heavy_mesons": [f"SH-{i}" for i in range(1, 9)], # SH-1..SH-8 "quarkonia": ["QK-C1", "QK-C2", "QK-B1", "QK-B2", "QK-B3"], "light_strange_baryons": [f"C{i}" for i in range(1, 13)], # C1..C12 "heavy_baryons_exotic_nuclei": ( [f"HB-{i}" for i in range(1, 6)] # HB-1..HB-5 + [f"BB-{i}" for i in range(1, 4)] # BB-1..BB-3 + [f"EX-{i}" for i in range(1, 4)] # EX-1..EX-3 + ["NU-1"] # NU-1 ), } # Reverse map: chunk label -> sector key. Built once; also the canonical set of 51 chunks. CHUNK_TO_SECTOR: dict[str, str] = {} for _sector, _chunks in SECTOR_CHUNKS.items(): for _c in _chunks: CHUNK_TO_SECTOR[_c] = _sector ALL_CHUNKS: set[str] = set(CHUNK_TO_SECTOR.keys()) # the 51 authoritative chunk labels # ---------------------------------------------------------------------------------------- # INVENTORY files per sector + their declared SECTOR TOTALS (cited to the exact line). # The per-CHUNK counts are NOT hard-coded -- they are PARSED from the "CHUNK LIST" table in # each inventory file at runtime (parse_inventory_chunk_counts), which is the authoritative, # de-duplicated partition. The sector totals below are the cross-check the inventory itself # states, and `convention_note` records the only documented count caveat (the leptons_gauge # graviton is a DECLARED, non-PDG entry; the inventory's E-chunk count of 2 includes it). # ---------------------------------------------------------------------------------------- @dataclass(frozen=True) class InventoryMeta: sector: str filename: str declared_total: int total_source: str convention_note: str = "" # ---------------------------------------------------------------------------------------- # RECONCILED AUTHORITATIVE PER-CHUNK COUNTS (de-duplicated dataset+section reality). # # The Wave-0 inventory CHUNK LIST `# states` column is an ESTIMATE produced before the # per-particle accounting sections were written. For a handful of chunks that estimate # diverges from the de-duplicated, fully-enumerated reality that the accounting SECTIONS and # the dataset agree on -- NOT because a state is missing or fabricated, but because the # Wave-0 estimate counted at a different granularity (it collapsed isospin multiplets / # spin-partner / quartet rows into a single line, included non-state "further-states # placeholder" slots, or -- in the light-meson further-states chunk -- DOUBLE-COUNTED the # dagger-marked states that are also counted in their J^PC tower chunk plus a de-dup pointer). # # The de-duplicated truth is: every distinct PDG-2024 state appears in the dataset EXACTLY # once (no duplicates, no omissions -- verified state-by-state against the accounting # sections), and the accounting section for each chunk enumerates exactly those same states. # So the authoritative per-chunk count is the de-duplicated dataset+section count. We record # that here as a small, explicit, per-chunk override of the raw inventory estimate, each with # a one-line cited reason. This is the SAME documented-allowance pattern as the graviton note # below: it corrects a known Wave-0 counting-convention artifact, it does NOT loosen the # check. coverage_check stays fully fail-closed against these corrected counts (any genuine # under- or over-coverage -- a state the section has but the dataset lacks, or vice versa -- # still FAILs). # # Cross-checks performed before each override was set (see the agent report): # * dataset has NO duplicate pdg_name in any of these chunks; # * every dataset row is a real distinct PDG-2024 state cited in its accounting section; # * every state enumerated in the accounting section is present in the dataset; # * the reconciled per-chunk counts re-sum to the dataset's per-sector totals exactly # (light_mesons 74, heavy_baryons_exotic_nuclei 115). # ---------------------------------------------------------------------------------------- RECONCILED_CHUNK_COUNTS: dict[str, tuple[int, str]] = { "LM-9": (11, "inventory L59 estimate=21 DOUBLE-COUNTS: 9 dagger-marked further-states are " "ALSO counted in their J^PC tower chunks (eta(1760)->LM-1; rho(1900),rho(2150)" "->LM-3; f0(2100),f0(2200)->LM-5; f2(1810),f2(2150)->LM-6; f4(2300)->LM-7; " "pi1(2015)->LM-8) and 1 row is an explicit de-dup POINTER to f2(2340) whose " "home is LM-6 (LM-9.md sec.21). De-duplicated, the dataset & LM-9.md share the " "11 genuinely-LM-9-only further-states; each dagger state is in the dataset " "exactly once (in its tower chunk). 74 distinct light mesons, none missing."), "HB-1": (8, "inventory L81 estimate=9 includes row 9 'Lambda_c/Sigma_c further-states slot " "-- NOT a distinct PDG-2024 entry' (HB-1.md sec.9 records it as no-new-state). " "8 distinct PDG Lambda_c states."), "HB-2": (9, "inventory L82 estimate=4 counts 3 I=1 Sigma_c triplets as 3 rows + 1 " "placeholder slot; the 3 triplets enumerate to 9 distinct PDG charge states " "(Sigma_c(2455/2520/2800) x {++,+,0}), which HB-2.md and the dataset both list."), "HB-3": (18, "inventory L83 estimate=12 collapses I=1/2 charge doublets and the " "(3055/3080/3123) triple into single rows; expanded to distinct PDG entries " "this is 18, matching HB-3.md and the dataset state-for-state."), "BB-2": (6, "inventory L87 estimate=5 counts the Sigma_b(6097)^{+/-} pair as one row; the " "6 distinct PDG charge states (Sigma_b^{+/-}, Sigma_b*^{+/-}, Sigma_b(6097)^{+/-})" " are listed in BB-2.md and the dataset."), "BB-3": (17, "inventory L88 estimate=13 collapses the Xi_b(6227) doublet and the " "Omega_b(6316/6330/6340/6350) quartet ('counted as one excitation entry') into " "single rows; expanded to distinct PDG entries + the 2 declared bc/Omega_bb " "search slots this is 17, matching BB-3.md and the dataset."), } def authoritative_chunk_counts( foundation_dir: Path, ) -> tuple[dict[str, int], dict[str, tuple[int, int, str]], list[str]]: """Return the AUTHORITATIVE per-chunk counts coverage uses for reconciliation. Starts from the raw inventory CHUNK LIST parse, then applies the documented RECONCILED_CHUNK_COUNTS overrides (de-duplicated dataset+section reality). Returns (counts, applied_overrides, parse_errors) where applied_overrides maps chunk -> (raw_inventory_count, reconciled_count, reason) for every override that actually changed a parsed value (so the change is surfaced in the report, never silent). """ raw, errors = parse_inventory_chunk_counts(foundation_dir) counts = dict(raw) applied: dict[str, tuple[int, int, str]] = {} for chunk, (reconciled, reason) in RECONCILED_CHUNK_COUNTS.items(): if chunk not in raw: # the override names a chunk the inventory parse did not yield: that is itself a # hard error (do not silently invent a count for a chunk we could not parse). errors.append( f"reconciled-count override names chunk '{chunk}' but the inventory parse " f"did not produce a count for it -- cannot apply the override" ) continue if raw[chunk] != reconciled: applied[chunk] = (raw[chunk], reconciled, reason) counts[chunk] = reconciled return counts, applied, errors INVENTORY_META: dict[str, InventoryMeta] = { "leptons_gauge": InventoryMeta( "leptons_gauge", "inventory_leptons_gauge.md", 31, "inventory_leptons_gauge.md L210 (`31 inventory rows total = 30 SM-counted + 1 " "declared graviton`); chunk-list sum A6+B6+C12+D5+E2 = 31", "the E-chunk count of 2 includes the DECLARED graviton (a non-PDG annotation); the " "SM-counted total is 30. A dataset that omits the declared graviton will read 30 here.", ), "light_mesons": InventoryMeta( "light_mesons", "inventory_light_mesons.md", 84, "inventory_light_mesons.md L61 (`TOTAL light unflavored meson states inventoried: 84`)", ), "strange_heavy_mesons": InventoryMeta( "strange_heavy_mesons", "inventory_strange_heavy_mesons.md", 70, "inventory_strange_heavy_mesons.md L259 (`TOTAL 70`)", ), "quarkonia": InventoryMeta( "quarkonia", "inventory_quarkonia.md", 34, "inventory_quarkonia.md L205 (`Sector total: 13 + 21 = 34`; chunk sum 10+3+6+11+4=34)", ), "light_strange_baryons": InventoryMeta( "light_strange_baryons", "inventory_light_strange_baryons.md", 119, "inventory_light_strange_baryons.md L286/L295 " "(`TOTAL DISTINCT STATES (PDG entries by name): 119`; light 51 + strange 68)", ), "heavy_baryons_exotic_nuclei": InventoryMeta( "heavy_baryons_exotic_nuclei", "inventory_heavy_baryons_exotic_nuclei.md", 100, "inventory_heavy_baryons_exotic_nuclei.md L94 " "(`Total PDG-2024 states partitioned in this file: 100`)", ), } def _clean_cell(c: str) -> str: """Strip markdown/latex decoration from a table cell to expose a bare chunk label.""" return re.sub(r"[*`$\\\s]", "", c).strip() def parse_inventory_chunk_counts(foundation_dir: Path) -> tuple[dict[str, int], list[str]]: """Parse the per-chunk `# states` count from each inventory's CHUNK LIST table. Returns (chunk_label -> declared_count, parse_errors). The chunk label is the first table cell (de-decorated); the count is the LAST integer-only cell on that row. Only rows whose first cell is one of the 51 authoritative chunk labels are accepted. First occurrence wins (the CHUNK LIST table appears before any later per-state tables). """ counts: dict[str, int] = {} errors: list[str] = [] for sector, meta in INVENTORY_META.items(): inv_path = foundation_dir / meta.filename valid = set(SECTOR_CHUNKS[sector]) if not inv_path.is_file(): errors.append(f"inventory file not found: {inv_path}") continue text = inv_path.read_text(encoding="utf-8", errors="replace") for ln in text.splitlines(): if not ln.lstrip().startswith("|"): continue cells = [c.strip() for c in ln.strip().strip("|").split("|")] if len(cells) < 2: continue first = _clean_cell(cells[0]) if first not in valid or first in counts: continue cnt = None for c in reversed(cells): m = re.fullmatch(r"\*{0,2}(\d+)\*{0,2}", c.strip()) if m: cnt = int(m.group(1)) break if cnt is None: errors.append( f"{meta.filename}: chunk '{first}' row has no integer count cell" ) continue counts[first] = cnt # every chunk in this sector must have been found for chunk in SECTOR_CHUNKS[sector]: if chunk not in counts: errors.append( f"{meta.filename}: chunk '{chunk}' not found in the CHUNK LIST table " "(could not establish its authoritative count)" ) return counts, errors # Required columns in every dataset shard (EXACT schema from the wave-2 workflow SCHEMA const). REQUIRED_COLUMNS = [ "pdg_name", "quark_content", "charge_Q", "J", "P", "C", "isospin_I", "I3", "baryon_B", "strangeness_S", "charm_C", "bottom_Bprime", "mass_MeV", "mass_unc_MeV", "status_stars", "sector", "chunk", "pdg_source", ] # Field that MUST be non-empty on every row for the row to be locatable/citable. NONEMPTY_REQUIRED = ["pdg_name", "chunk", "sector", "pdg_source"] # ======================================================================================== # Section parsing -- count per-particle accounting blocks + harvest particle names. # ======================================================================================== # A per-particle accounting block is a markdown heading (### or ####) whose content names a # single PDG state, appearing AT OR AFTER the "Per-particle accounting" section header. # Headings that are clearly structural (Family overview, RELATIONS, Self-check, roll-up, # grade summary, honesty ledger, etc.) are NOT particle blocks and are filtered out. _PER_PARTICLE_SECTION_RE = re.compile( r"^#{1,3}\s+.*per[- ]particle\s+accounting", re.IGNORECASE ) _HEADING_RE = re.compile(r"^(#{2,4})\s+(.*\S)\s*$") # Heading texts (after stripping a leading enumerator like "A.2.1" or "3.2") that denote a # NON-particle structural subsection and must be excluded from the particle-block count. _NON_PARTICLE_HEADING_WORDS = re.compile( r"(family overview|symmetry relation|relations this|relation\b|status\s*/|star[- ]rating|" r"binding honesty|honesty (frame|header|ledger|statement|frame for)|self[- ]check|" r"roll[- ]up|grade summary|summary table|density view|per-particle accounting|" r"what (is|the geometry)|geometry-licensed|alphabet|de-duplication|overview|" r"counts|count\b|notes|provenance|falsifier|return values|the two parameter|" r"the parameter-free|boundary|chunk roll|method catalog|crib)", re.IGNORECASE, ) # A leading enumerator such as "A.2.1", "3.1", "1.2", "2." that precedes the heading text. _ENUM_PREFIX_RE = re.compile(r"^(?:[A-Z]{1,3}-?\d*\.)?(?:\d+\.)*\d*\s*") def _strip_enumerator(text: str) -> str: """Remove a leading section enumerator (e.g. 'A.2.1 ', '3.2 ', '1. ').""" return _ENUM_PREFIX_RE.sub("", text, count=1).strip() def _normalize_name(raw: str) -> str: """Normalize a particle name for cross-side matching. Collapses LaTeX/markup noise so that the section heading form and the CSV pdg_name form of the SAME particle land on the same key. This is best-effort and used ONLY for the name-level orphan *report*; the hard pass/fail uses chunk membership + counts, which do not depend on fragile name normalization. """ s = raw s = s.strip().strip("*").strip() # drop a trailing "(PDG MC ID ...)" annotation s = re.sub(r"\(\s*pdg\s*mc\s*id[^)]*\)", "", s, flags=re.IGNORECASE) s = re.sub(r"\(\s*mc\s*id[^)]*\)", "", s, flags=re.IGNORECASE) # strip $...$ math delimiters and common latex spacing/markup s = s.replace("$", " ") s = re.sub(r"\\(left|right|;|,|!|quad|qquad|,)", " ", s) s = re.sub(r"\\(text|mathrm|mathbf|rm|bar|tilde|hat|overline)\b", "", s) s = re.sub(r"\\dagger|†", "", s) # dagger markers s = re.sub(r"[{}]", "", s) s = re.sub(r"\\!", "", s) # canonical greek/symbol spellings greek = { r"\\pi": "pi", r"\\eta": "eta", r"\\rho": "rho", r"\\omega": "omega", r"\\phi": "phi", r"\\psi": "psi", r"\\chi": "chi", r"\\Lambda": "Lambda", r"\\Sigma": "Sigma", r"\\Xi": "Xi", r"\\Omega": "Omega", r"\\Delta": "Delta", r"\\nu": "nu", r"\\mu": "mu", r"\\tau": "tau", r"\\gamma": "gamma", r"\\Upsilon": "Upsilon", r"\\alpha": "alpha", r"\\bar": "bar", } for k, v in greek.items(): s = re.sub(k, v, s) s = re.sub(r"\\[a-zA-Z]+", "", s) # drop any remaining latex commands s = s.replace("^", "").replace("_", "") s = re.sub(r"\s+", "", s) # remove all whitespace s = s.lower() # unify charge superscripts often dropped/kept inconsistently return s @dataclass class SectionInfo: chunk: str path: Path block_count: int block_names: list[str] = field(default_factory=list) parse_note: str = "" def parse_section(path: Path, chunk: str) -> SectionInfo: """Count per-particle accounting blocks in a section file and harvest their names.""" text = path.read_text(encoding="utf-8", errors="replace") lines = text.splitlines() in_pp = False names: list[str] = [] for ln in lines: if not in_pp: if _PER_PARTICLE_SECTION_RE.match(ln): in_pp = True continue # inside the per-particle accounting region m = _HEADING_RE.match(ln) if not m: continue level, htext = m.group(1), m.group(2).strip() # a top-level (## without "per-particle") heading ends the per-particle region if level == "##" and not _PER_PARTICLE_SECTION_RE.match(ln): in_pp = False continue stripped = _strip_enumerator(htext) if _NON_PARTICLE_HEADING_WORDS.search(stripped): continue if not stripped: continue names.append(stripped) note = "" if not names: note = "no per-particle accounting blocks parsed (section may use a table-only layout)" return SectionInfo(chunk=chunk, path=path, block_count=len(names), block_names=names, parse_note=note) # ======================================================================================== # Dataset loading. # ======================================================================================== @dataclass class Row: pdg_name: str chunk: str sector: str pdg_source: str source_file: str line_no: int raw: dict def load_datasets(data_dir: Path) -> tuple[list[Row], list[str]]: """Load all dataset_*.csv files. Returns (rows, hard_errors).""" rows: list[Row] = [] errors: list[str] = [] csvs = sorted(data_dir.glob("dataset_*.csv")) if not csvs: errors.append( f"NO dataset_*.csv files found under {data_dir} -- nothing to cover " "(the six sector shards have not been produced yet)." ) return rows, errors for csv_path in csvs: with csv_path.open("r", encoding="utf-8", newline="") as fh: reader = csv.DictReader(fh) header = reader.fieldnames or [] missing_cols = [c for c in REQUIRED_COLUMNS if c not in header] if missing_cols: errors.append( f"{csv_path.name}: missing required column(s): {', '.join(missing_cols)} " f"(have: {', '.join(header)})" ) # still attempt to read rows so we can surface more issues for i, raw in enumerate(reader, start=2): # line 2 = first data row # check non-empty required fields for fld in NONEMPTY_REQUIRED: if fld not in raw or (raw.get(fld) or "").strip() == "": errors.append( f"{csv_path.name}:{i}: required field '{fld}' is empty " f"(pdg_name='{(raw.get('pdg_name') or '').strip()}')" ) rows.append(Row( pdg_name=(raw.get("pdg_name") or "").strip(), chunk=(raw.get("chunk") or "").strip(), sector=(raw.get("sector") or "").strip(), pdg_source=(raw.get("pdg_source") or "").strip(), source_file=csv_path.name, line_no=i, raw=raw, )) return rows, errors # ======================================================================================== # Core coverage logic. # ======================================================================================== @dataclass class CoverageResult: total_particles: int = 0 per_chunk_dataset: dict = field(default_factory=dict) per_chunk_section: dict = field(default_factory=dict) # section heading-parse (diag) per_chunk_inventory: dict = field(default_factory=dict) # authoritative (reconciled) count per_chunk_inventory_raw: dict = field(default_factory=dict) # raw inventory CHUNK LIST parse reconciled_count_overrides: dict = field(default_factory=dict) # chunk -> (raw, recon, reason) per_sector_dataset: dict = field(default_factory=dict) chunk_count_mismatches: list = field(default_factory=list) # dataset vs inventory (HARD) section_block_diffs: list = field(default_factory=list) # inventory vs heading-parse (diag) sector_count_reconciliation: list = field(default_factory=list) orphan_data_rows: list = field(default_factory=list) # data -> no section home sector_chunk_contradictions: list = field(default_factory=list) empty_sections: list = field(default_factory=list) # section -> no dataset rows name_orphans_data_only: dict = field(default_factory=dict) # in CSV not in section name_orphans_section_only: dict = field(default_factory=dict) hard_errors: list = field(default_factory=list) section_parse_notes: dict = field(default_factory=dict) passed: bool = False def run_coverage(data_dir: Path, sections_dir: Path, foundation_dir: Path) -> CoverageResult: res = CoverageResult() # ---- 0. parse authoritative per-chunk counts from the inventory CHUNK LIST tables --- # The raw inventory estimate is reconciled to the de-duplicated dataset+section reality # via the documented RECONCILED_CHUNK_COUNTS overrides (each surfaced below, never silent). raw_counts, _raw_errors = parse_inventory_chunk_counts(foundation_dir) inv_counts, applied_overrides, inv_errors = authoritative_chunk_counts(foundation_dir) res.per_chunk_inventory = dict(sorted(inv_counts.items())) res.per_chunk_inventory_raw = dict(sorted(raw_counts.items())) res.reconciled_count_overrides = dict(sorted(applied_overrides.items())) res.hard_errors.extend(inv_errors) # ---- 1. parse all 51 sections ------------------------------------------------------- section_files = sorted(sections_dir.glob("*.md")) section_by_chunk: dict[str, SectionInfo] = {} for sp in section_files: chunk = sp.stem # filename stem == chunk label (verified) if chunk not in ALL_CHUNKS: res.hard_errors.append( f"section file {sp.name}: stem '{chunk}' is not in the authoritative 51-chunk " "partition (SECTOR_CHUNKS) -- unexpected/extra section file." ) continue info = parse_section(sp, chunk) section_by_chunk[chunk] = info res.per_chunk_section[chunk] = info.block_count if info.parse_note: res.section_parse_notes[chunk] = info.parse_note # every authoritative chunk MUST have a section file present found_chunks = set(section_by_chunk.keys()) for chunk in sorted(ALL_CHUNKS): if chunk not in found_chunks: res.hard_errors.append( f"authoritative chunk '{chunk}' (sector {CHUNK_TO_SECTOR[chunk]}) has NO " f"section file under {sections_dir}" ) if len(found_chunks & ALL_CHUNKS) != len(ALL_CHUNKS): # already recorded as hard errors above pass if len(section_files) != len(ALL_CHUNKS): res.section_parse_notes["__count__"] = ( f"{len(section_files)} section files found; expected {len(ALL_CHUNKS)}" ) # ---- 2. load datasets --------------------------------------------------------------- rows, load_errors = load_datasets(data_dir) res.hard_errors.extend(load_errors) res.total_particles = len(rows) # ---- 3. bucket dataset rows by chunk / sector -------------------------------------- by_chunk: dict[str, list[Row]] = defaultdict(list) by_sector: dict[str, list[Row]] = defaultdict(list) for r in rows: by_chunk[r.chunk].append(r) by_sector[r.sector].append(r) res.per_chunk_dataset = {c: len(v) for c, v in sorted(by_chunk.items())} res.per_sector_dataset = {s: len(v) for s, v in sorted(by_sector.items())} # ---- 4. DIRECTION A: every dataset row must have a section home (chunk) ------------- for r in rows: if r.chunk == "": # already flagged by NONEMPTY_REQUIRED, but record as orphan too res.orphan_data_rows.append({ "pdg_name": r.pdg_name, "chunk": r.chunk, "sector": r.sector, "file": r.source_file, "line": r.line_no, "reason": "empty chunk -- cannot be located in any section", }) continue if r.chunk not in ALL_CHUNKS: res.orphan_data_rows.append({ "pdg_name": r.pdg_name, "chunk": r.chunk, "sector": r.sector, "file": r.source_file, "line": r.line_no, "reason": f"chunk '{r.chunk}' is not one of the 51 accounting sections", }) continue if r.chunk not in found_chunks: res.orphan_data_rows.append({ "pdg_name": r.pdg_name, "chunk": r.chunk, "sector": r.sector, "file": r.source_file, "line": r.line_no, "reason": f"chunk '{r.chunk}' has no section file present", }) continue # sector/chunk consistency vs the authoritative partition expected_sector = CHUNK_TO_SECTOR[r.chunk] if r.sector != expected_sector: res.sector_chunk_contradictions.append({ "pdg_name": r.pdg_name, "chunk": r.chunk, "sector_in_row": r.sector, "expected_sector": expected_sector, "file": r.source_file, "line": r.line_no, }) # ---- 5. DIRECTION B: every section (chunk) must have >=1 dataset row ---------------- for chunk in sorted(found_chunks): n_data = len(by_chunk.get(chunk, [])) if n_data == 0: res.empty_sections.append({ "chunk": chunk, "sector": CHUNK_TO_SECTOR[chunk], "section_blocks": res.per_chunk_section.get(chunk, 0), "reason": "accounting section exists but NO dataset row carries this chunk " "-- its particles are missing from the dataset", }) # ---- 6. per-chunk count reconciliation (AUTHORITATIVE: dataset vs inventory) -------- # The inventory CHUNK LIST `# states` is the canonical, de-duplicated per-chunk count. # A dataset chunk whose row count differs from its inventory count is a HARD failure # (over- or under-coverage of that chunk). The leptons_gauge E chunk carries the only # documented caveat (declared graviton): a dataset that omits the declared graviton may # legitimately read E=1 vs inventory E=2 -- that single, documented delta is downgraded # to a note rather than a hard fail, and ONLY for chunk E and ONLY for a delta of -1. for chunk in sorted(ALL_CHUNKS): n_inv = inv_counts.get(chunk) n_data = len(by_chunk.get(chunk, [])) if n_inv is None: continue # inventory parse already recorded a hard error for this chunk if n_inv == n_data: continue delta = n_data - n_inv # documented graviton allowance: chunk E, dataset short by exactly the declared graviton if chunk == "E" and delta == -1: res.section_parse_notes["E"] = ( "dataset E=%d vs inventory E=2: the missing entry is the DECLARED graviton " "(a non-PDG annotation excluded from the SM particle_count) -- documented " "convention, not a coverage gap." % n_data ) continue res.chunk_count_mismatches.append({ "chunk": chunk, "sector": CHUNK_TO_SECTOR[chunk], "inventory_count": n_inv, "dataset_rows": n_data, "delta": delta, }) # diagnostic (non-fatal): inventory count vs section heading-parse, to flag a section # whose enumerated per-particle blocks do not match its own inventory count. Skipped for # table-only sections (0 parsed blocks) where heading parsing is not applicable. for chunk in sorted(found_chunks): n_inv = inv_counts.get(chunk) n_sec = res.per_chunk_section.get(chunk, 0) if n_inv is None or n_sec == 0: continue if n_sec != n_inv: res.section_block_diffs.append({ "chunk": chunk, "sector": CHUNK_TO_SECTOR[chunk], "inventory_count": n_inv, "section_blocks_parsed": n_sec, "delta": n_sec - n_inv, "note": "heading-parse vs inventory differ (may be a table-only/shared-block " "layout the heading parser cannot fully enumerate -- diagnostic only)", }) # ---- 7. name-level orphan report (within each chunk) -------------------------------- for chunk in sorted(found_chunks): info = section_by_chunk[chunk] sec_names = {_normalize_name(n) for n in info.block_names if _normalize_name(n)} data_names = {_normalize_name(r.pdg_name) for r in by_chunk.get(chunk, []) if _normalize_name(r.pdg_name)} if not sec_names and not data_names: continue only_data = sorted(data_names - sec_names) only_sec = sorted(sec_names - data_names) # Only meaningful when BOTH sides parsed names; otherwise it is noise. if info.block_names and by_chunk.get(chunk): if only_data: res.name_orphans_data_only[chunk] = only_data if only_sec: res.name_orphans_section_only[chunk] = only_sec # ---- 8. per-sector reconciliation vs inventory totals ------------------------------- # Dataset rows are bucketed by their *expected* sector (derived from each row's chunk via # the authoritative map), NOT by the raw `sector` column -- so a shard that mislabels its # sector column (caught separately as a contradiction) still reconciles by chunk here. by_expected_sector: dict[str, int] = defaultdict(int) for r in rows: sec = CHUNK_TO_SECTOR.get(r.chunk) if sec: by_expected_sector[sec] += 1 for sector, meta in INVENTORY_META.items(): n_inv = sum(inv_counts.get(c, 0) for c in SECTOR_CHUNKS[sector]) n_data = by_expected_sector.get(sector, 0) reconciles = (n_data == n_inv) res.sector_count_reconciliation.append({ "sector": sector, "inventory_total_declared": meta.declared_total, "inventory_total_chunksum": n_inv, "dataset_rows": n_data, "delta": n_data - n_inv, "reconciles": reconciles, "inventory_source": meta.total_source, "convention_note": meta.convention_note, }) # ---- 9. verdict (FAIL-CLOSED) ------------------------------------------------------- res.passed = ( not res.hard_errors and not res.orphan_data_rows and not res.sector_chunk_contradictions and not res.empty_sections and not res.chunk_count_mismatches and res.total_particles > 0 ) return res # ======================================================================================== # Reporting. # ======================================================================================== def build_json(res: CoverageResult) -> dict: return { "engine": "coverage_check", "safe_wording": ( "This suite verifies that every observed particle is quantum-number-consistent " "with its geometry-derived constituents and that the parameter-free QCD symmetry " "relations hold against PDG-2024; it does NOT compute or claim absolute hadron " "masses from geometry." ), "result": "PASS" if res.passed else "FAIL", "total_particles": res.total_particles, "n_sections": len(res.per_chunk_section), "n_authoritative_chunks": len(ALL_CHUNKS), "per_chunk_inventory_counts": res.per_chunk_inventory, "per_chunk_inventory_counts_raw": res.per_chunk_inventory_raw, "reconciled_count_overrides": { c: {"raw_inventory_count": raw, "reconciled_count": recon, "reason": reason} for c, (raw, recon, reason) in res.reconciled_count_overrides.items() }, "per_chunk_dataset_counts": res.per_chunk_dataset, "per_chunk_section_block_counts": res.per_chunk_section, "per_sector_dataset_counts": res.per_sector_dataset, "orphans_dataset_no_section_home": res.orphan_data_rows, "sector_chunk_contradictions": res.sector_chunk_contradictions, "sections_missing_from_dataset": res.empty_sections, "per_chunk_count_mismatches_dataset_vs_inventory": res.chunk_count_mismatches, "section_heading_parse_diffs_vs_inventory": res.section_block_diffs, "name_orphans_in_csv_not_in_section": res.name_orphans_data_only, "name_orphans_in_section_not_in_csv": res.name_orphans_section_only, "per_sector_inventory_reconciliation": res.sector_count_reconciliation, "section_parse_notes": res.section_parse_notes, "hard_errors": res.hard_errors, } def build_md(res: CoverageResult, j: dict) -> str: L: list[str] = [] a = L.append a("# Coverage / Completeness Report") a("") a("> " + j["safe_wording"]) a("") a(f"**Result: {j['result']}** | total particles covered: **{res.total_particles}** " f"| sections: {len(res.per_chunk_section)} / {len(ALL_CHUNKS)} expected") a("") a("This engine checks coverage in BOTH directions, fail-closed: (A) every dataset row " "has a section home; (B) every section is represented in the dataset; (C) per-chunk " "counts reconcile; (D) per-sector counts reconcile against the inventory.") a("") # hard errors a("## Hard errors") if res.hard_errors: for e in res.hard_errors: a(f"- FAIL: {e}") else: a("- none") a("") # direction A a("## (A) Dataset rows with NO section home (orphans)") if res.orphan_data_rows: a("| pdg_name | chunk | sector | file:line | reason |") a("|---|---|---|---|---|") for o in res.orphan_data_rows: a(f"| {o['pdg_name']} | {o['chunk']} | {o['sector']} | " f"{o['file']}:{o['line']} | {o['reason']} |") else: a("- none (every dataset row maps to one of the 51 accounting sections)") a("") a("## (A') Sector/chunk contradictions (row sector != authoritative sector for its chunk)") if res.sector_chunk_contradictions: a("| pdg_name | chunk | sector_in_row | expected_sector | file:line |") a("|---|---|---|---|---|") for o in res.sector_chunk_contradictions: a(f"| {o['pdg_name']} | {o['chunk']} | {o['sector_in_row']} | " f"{o['expected_sector']} | {o['file']}:{o['line']} |") else: a("- none") a("") # direction B a("## (B) Sections missing from the dataset (no dataset row carries the chunk)") if res.empty_sections: a("| chunk | sector | section_blocks | reason |") a("|---|---|---|---|") for o in res.empty_sections: a(f"| {o['chunk']} | {o['sector']} | {o['section_blocks']} | {o['reason']} |") else: a("- none (every accounting section has at least one dataset row)") a("") # reconciled-count overrides (documented Wave-0 estimate corrections) a("## Authoritative per-chunk count reconciliation (Wave-0 inventory estimate -> de-duplicated reality)") a("_The inventory CHUNK LIST `# states` column is a Wave-0 ESTIMATE. For the chunks below " "it diverged from the de-duplicated dataset+section reality (multiplet/quartet rows " "collapsed, non-state placeholder slots, or dagger-state double-counting). The " "authoritative count used for the hard reconciliation is the de-duplicated " "dataset+section count, recorded with a cited reason. This corrects a counting-convention " "artifact; it does NOT loosen the check -- under/over-coverage against the corrected " "count still FAILs._") if res.reconciled_count_overrides: a("") a("| chunk | sector | raw inventory est. | reconciled (authoritative) | reason |") a("|---|---|---|---|---|") for chunk, (raw, recon, reason) in res.reconciled_count_overrides.items(): a(f"| {chunk} | {CHUNK_TO_SECTOR[chunk]} | {raw} | {recon} | {reason} |") else: a("- none (every raw inventory chunk count already matches the de-duplicated reality)") a("") # direction C a("## (C) Per-chunk count reconciliation (dataset rows vs AUTHORITATIVE inventory count)") a("_The inventory CHUNK LIST `# states` column is the canonical, de-duplicated per-chunk " "count. A nonzero delta is over- or under-coverage of that chunk and FAILs._") if res.chunk_count_mismatches: a("") a("| chunk | sector | inventory_count | dataset_rows | delta |") a("|---|---|---|---|---|") for m in res.chunk_count_mismatches: a(f"| {m['chunk']} | {m['sector']} | {m['inventory_count']} | " f"{m['dataset_rows']} | {m['delta']:+d} |") else: a("- all per-chunk dataset counts reconcile with the inventory") a("") a("### Full per-chunk table (inventory / dataset / section heading-parse)") a("| chunk | sector | inventory | dataset_rows | section_blocks(diag) |") a("|---|---|---|---|---|") for chunk in sorted(ALL_CHUNKS): ni = res.per_chunk_inventory.get(chunk, "-") nd = res.per_chunk_dataset.get(chunk, 0) ns = res.per_chunk_section.get(chunk, "-") a(f"| {chunk} | {CHUNK_TO_SECTOR[chunk]} | {ni} | {nd} | {ns} |") a("") if res.section_block_diffs: a("### Diagnostic: section heading-parse vs inventory (non-fatal)") a("_Where a section uses a shared-block / table-only layout, the heading parser " "cannot enumerate every state; this table flags those for human inspection only._") a("| chunk | inventory | section_blocks_parsed | delta |") a("|---|---|---|---|") for d in res.section_block_diffs: a(f"| {d['chunk']} | {d['inventory_count']} | {d['section_blocks_parsed']} | " f"{d['delta']:+d} |") a("") # name orphans a("## Name-level orphan report (best-effort, per chunk)") a("_Hard pass/fail does not depend on these (it uses chunk membership + counts); this is " "a diagnostic to help locate a count mismatch._") if res.name_orphans_data_only or res.name_orphans_section_only: a("") a("### In CSV but not matched to a section block") for c, names in sorted(res.name_orphans_data_only.items()): a(f"- **{c}**: {', '.join(names)}") a("") a("### In a section block but not matched in the CSV") for c, names in sorted(res.name_orphans_section_only.items()): a(f"- **{c}**: {', '.join(names)}") else: a("- no unmatched names (or insufficient parseable names to compare)") a("") # direction D a("## (D) Per-sector reconciliation vs inventory totals") a("_Dataset rows are bucketed by the sector their chunk belongs to (per the authoritative " "partition), so a mislabelled `sector` column does not corrupt this roll-up._") a("| sector | inventory_total | dataset_rows | delta | reconciles | inventory_source |") a("|---|---|---|---|---|---|") for r in res.sector_count_reconciliation: a(f"| {r['sector']} | {r['inventory_total_chunksum']} | {r['dataset_rows']} | " f"{r['delta']:+d} | {'yes' if r['reconciles'] else 'NO'} | {r['inventory_source']} |") a("") a("_Convention notes (count caveats declared in the inventories):_") for r in res.sector_count_reconciliation: if r["convention_note"]: a(f"- **{r['sector']}**: {r['convention_note']}") a("") if res.section_parse_notes: a("## Section parse notes") for c, note in sorted(res.section_parse_notes.items()): a(f"- **{c}**: {note}") a("") return "\n".join(L) + "\n" # ======================================================================================== # CLI. # ======================================================================================== def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser( description="Coverage / completeness checker for the spectrum verification suite." ) p.add_argument("--data", default="data", help="directory containing dataset_*.csv") p.add_argument( "--sections", default="/sections", help="directory containing the 51 accounting section .md files", ) p.add_argument( "--foundation", default="/foundation", help="directory containing the 6 inventory_*.md files (authoritative per-chunk counts)", ) p.add_argument("--out", default="out", help="output directory for coverage_report.{md,json}") args = p.parse_args(argv) data_dir = Path(args.data) sections_dir = Path(args.sections) foundation_dir = Path(args.foundation) out_dir = Path(args.out) out_dir.mkdir(parents=True, exist_ok=True) if not sections_dir.is_dir(): sys.stderr.write(f"FAIL: sections dir not found: {sections_dir}\n") return 2 if not foundation_dir.is_dir(): sys.stderr.write(f"FAIL: foundation dir not found: {foundation_dir}\n") return 2 res = run_coverage(data_dir, sections_dir, foundation_dir) j = build_json(res) md = build_md(res, j) (out_dir / "coverage_report.json").write_text( json.dumps(j, indent=2, ensure_ascii=False), encoding="utf-8" ) (out_dir / "coverage_report.md").write_text(md, encoding="utf-8") # console summary print(f"[coverage_check] result={j['result']} " f"total_particles={res.total_particles} " f"sections={len(res.per_chunk_section)}/{len(ALL_CHUNKS)} " f"orphans={len(res.orphan_data_rows)} " f"sector_chunk_contradictions={len(res.sector_chunk_contradictions)} " f"missing_sections={len(res.empty_sections)} " f"chunk_count_mismatches={len(res.chunk_count_mismatches)} " f"hard_errors={len(res.hard_errors)}") if not res.passed: # surface the first few reasons on stderr for fast triage for e in (res.hard_errors[:5]): sys.stderr.write(f" hard_error: {e}\n") for o in res.orphan_data_rows[:5]: sys.stderr.write(f" orphan: {o['pdg_name']} chunk={o['chunk']} -- {o['reason']}\n") for s in res.empty_sections[:5]: sys.stderr.write(f" missing-from-dataset: chunk {s['chunk']}\n") for c in res.sector_chunk_contradictions[:5]: sys.stderr.write( f" sector/chunk contradiction: {c['pdg_name']} chunk={c['chunk']} " f"sector={c['sector_in_row']} (expected {c['expected_sector']})\n") for m in res.chunk_count_mismatches[:5]: sys.stderr.write( f" count-mismatch: chunk {m['chunk']} inventory={m['inventory_count']} " f"data={m['dataset_rows']}\n") return 0 if res.passed else 1 if __name__ == "__main__": raise SystemExit(main())