SilverSight/python/build_corpus250.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = []
# ///
"""
Build formal/SilverSight/RRC/Corpus250.lean from
archive/experimental-shim-probes/rrc_equation_classifier_receipt.json,
merged with 8×8 braid adjacency matrices from
shared-data/rrc_pist_predictions_250_v1.json.

Python's role:
  - read raw features from the classifier receipt
  - merge matrices by invariant_receipt.object_id
  - emit deterministic Lean source

Lean's role:
  - PIST classification (classifyProxy/classifyExact) from the matrix
  - alignment gate via determineAlignment
  - receipt stamping and all admissibility/promotion decisions

Usage:
    python3 python/build_corpus250.py
    python3 python/build_corpus250.py \
        --receipt /path/to/rrc_equation_classifier_receipt.json \
        --predictions /path/to/rrc_pist_predictions_250_v1.json \
        --out-lean formal/SilverSight/RRC/Corpus250.lean
"""
from __future__ import annotations
import argparse, hashlib, json, sys
from pathlib import Path


# classifier JSON shape name → SilverSight.RRCLogogramProjection.RRCShape constructor
SHAPE_MAP = {
    "CognitiveLoadField":                   ".cognitiveLoadField",
    "SignalShapedRouteCompiler":             ".signalShapedRouteCompiler",
    "ProjectableGeometryTopology":          ".projectableGeometryTopology",
    "CadForceProbeReceipt":                 ".cadForceProbeReceipt",
    "LogogramProjection":                   ".logogramProjection",
    "HoldForUnlawfulOrUnderspecifiedShape": ".holdForUnlawfulOrUnderspecifiedShape",
}


def template_key(rrc_kind: str, status: str) -> str:
    """Map rrc_kind + classifier status to a page-generator template key."""
    if status == "HOLD":
        return "hold"
    kind_map = {
        "cognitive_field_receipt":  "definition",
        "compression_route_prior":  "master_equation",
        "geometry_topology_receipt": "definition",
        "cad_force_receipt":        "gate",
        "logogram_projection":      "receipt",
        "negative_control":         "hold",
    }
    return kind_map.get(rrc_kind, "definition")


def operator_tokens(er: dict) -> list[str]:
    """Derive operator/domain tokens from route_hint, rrc_kind, and equation text."""
    tokens = []
    rh = (er.get("route_hint_non_authoritative") or "").strip()
    rk = (er.get("rrc_kind") or "").strip()
    if rh and rh != "unclassified_equation":
        tokens.append(rh)
    if rk:
        tokens.append(rk)
    eq_text = (er.get("equation") or "").lower()
    for op in ["exp(", "log(", "max(", "min(", "sum(", "integral", "derivative",
               "laplacian", "nabla", "div(", "curl(", "sigmoid", "softmax",
               "tanh(", "relu(", "norm(", "dot(", "cross("]:
        if op in eq_text:
            tokens.append(op.rstrip("("))
    return list(dict.fromkeys(tokens))


def lean_str(s: str) -> str:
    s = s.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{s}"'


def lean_opt(s: str | None) -> str:
    return "none" if s is None else f"some {lean_str(s)}"


def lean_str_list(xs: list[str]) -> str:
    return "[" + ", ".join(lean_str(x) for x in xs) + "]"


def load_matrices(path: Path) -> dict[str, list[list[int]]]:
    """Load predictions JSON into equation_id → matrix lookup."""
    data = json.loads(path.read_text())
    return {
        p.get("equation_id", ""): p.get("matrix_8x8", [])
        for p in data.get("predictions", [])
        if p.get("equation_id")
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Generate SilverSight RRC Corpus250.lean")
    parser.add_argument(
        "--receipt",
        type=Path,
        default=Path("/home/allaun/Research Stack/archive/experimental-shim-probes/rrc_equation_classifier_receipt.json"),
        help="Path to rrc_equation_classifier_receipt.json",
    )
    parser.add_argument(
        "--predictions",
        type=Path,
        default=Path("/home/allaun/Research Stack/shared-data/rrc_pist_predictions_250_v1.json"),
        help="Path to rrc_pist_predictions_250_v1.json",
    )
    parser.add_argument(
        "--out-lean",
        type=Path,
        default=Path("formal/SilverSight/RRC/Corpus250.lean"),
        help="Output Lean module path",
    )
    args = parser.parse_args()

    receipt = json.loads(args.receipt.read_text())
    eqs = receipt.get("compiled_equations", [])
    matrices = load_matrices(args.predictions)
    print(f"Loaded {len(eqs)} equations and {len(matrices)} matrices", file=sys.stderr)

    # Deterministic order by equation_id.
    eqs_sorted = sorted(eqs, key=lambda eq: eq.get("invariant_receipt", {}).get("object_id", ""))

    # Content hash over the raw corpus data for reproducibility.
    content_blob = json.dumps(
        [
            {
                "object_id": eq.get("invariant_receipt", {}).get("object_id"),
                "name": eq.get("equation_record", {}).get("name"),
                "shape": eq.get("invariant_receipt", {}).get("shape"),
                "status": eq.get("invariant_receipt", {}).get("status"),
                "matrix": matrices.get(eq.get("invariant_receipt", {}).get("object_id", "")),
            }
            for eq in eqs_sorted
        ],
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    content_hash = hashlib.sha256(content_blob).hexdigest()

    rows: list[str] = []
    for eq in eqs_sorted:
        er = eq["equation_record"]
        ir = eq["invariant_receipt"]
        tw = eq["type_witness"]

        eq_id = ir.get("object_id", "")
        name = er.get("name", "")
        shape_str = ir.get("shape", "HoldForUnlawfulOrUnderspecifiedShape")
        lean_shape = SHAPE_MAP.get(shape_str, ".holdForUnlawfulOrUnderspecifiedShape")
        status_str = ir.get("status", "HOLD")
        lean_status = ".candidate" if status_str == "CANDIDATE" else ".hold"
        rrc_kind = er.get("rrc_kind", "")
        weak_cnt = len(tw.get("missing_or_weak_axes") or [])

        op_tokens = operator_tokens(er)
        inv_declared = (er.get("domain_type") or "unknown").strip() or "unknown"
        bound_conds = (er.get("bind_class") or "unknown").strip() or "unknown"
        t_key = template_key(rrc_kind, status_str)
        route_hint = er.get("route_hint_non_authoritative") or "unclassified_equation"
        t_params = f"route={route_hint};shape={shape_str}"
        arxiv_pid = (er.get("arxiv_paper_id") or "").strip() or None

        rows.append(
            f"  {{ equationId          := {lean_str(eq_id)}\n"
            f"    name                := {lean_str(name)}\n"
            f"    shape               := {lean_shape}\n"
            f"    status              := {lean_status}\n"
            f"    rrcKind             := {lean_str(rrc_kind)}\n"
            f"    weakAxesCnt         := {weak_cnt}\n"
            f"    pistProxyLabel      := Option.bind (findMatrix {lean_str(eq_id)}) SilverSight.PIST.Classify.classifyProxy\n"
            f"    pistExactLabel      := Option.bind (findMatrix {lean_str(eq_id)}) SilverSight.PIST.Classify.classifyExact\n"
            f"    arxivPaperId        := {lean_opt(arxiv_pid)}\n"
            f"    operatorTokens      := {lean_str_list(op_tokens)}\n"
            f"    invariantsDeclared  := {lean_str(inv_declared)}\n"
            f"    boundaryConds       := {lean_str(bound_conds)}\n"
            f"    templateKey         := {lean_str(t_key)}\n"
            f"    templateParams      := {lean_str(t_params)} }}"
        )

    lines = [
        "-- SilverSight.RRC.Corpus250 — AUTO-GENERATED by python/build_corpus250.py",
        "-- DO NOT EDIT BY HAND. Regenerate with:",
        "--   python3 python/build_corpus250.py",
        "--",
        "-- Python role: raw feature extraction + matrix merge.",
        "-- Lean role: PIST classification, alignment gate (determineAlignment),",
        "--            receipt stamping, and all admissibility/promotion decisions.",
        "--",
        "-- Source: rrc_equation_classifier_receipt.json",
        "-- Matrices: rrc_pist_predictions_250_v1.json",
        f"-- Content hash (SHA-256): {content_hash}",
        f"-- Equation count: {len(rows)}",
        "--",
        "import SilverSight.RRC.Emit",
        "import SilverSight.PIST.Classify",
        "import SilverSight.PIST.Matrices250",
        "",
        "namespace SilverSight.RRC.Corpus250",
        "",
        "open SilverSight.RRC.Emit",
        "open SilverSight.RRCLogogramProjection",
        "open SilverSight.ReceiptCore",
        "open SilverSight.PIST.Matrices250",
        "",
        "/-- Full 250-equation corpus from rrc_equation_classifier_receipt.json,",
        "    merged with 8×8 braid adjacency matrices from",
        "    rrc_pist_predictions_250_v1.json.",
        "    Each row carries raw features only; the alignment gate in",
        "    SilverSight.RRC.Emit.emitCorpus makes all admissibility decisions. -/",
        "def corpus250 : List FixtureRow := [",
        ",\n".join(rows),
        "]",
        "",
        "end SilverSight.RRC.Corpus250",
        "",
    ]

    args.out_lean.parent.mkdir(parents=True, exist_ok=True)
    args.out_lean.write_text("\n".join(lines))
    print(f"Wrote {args.out_lean} ({len(rows)} rows)", file=sys.stderr)
    return 0


if __name__ == "__main__":
    sys.exit(main())