Files
Jyotisha/scripts/three_engine_mismatch_arbitrator.py
T
2026-07-17 22:19:09 +08:00

101 lines
3.8 KiB
Python

#!/usr/bin/env python3
"""Classify parity mismatches by evidence shape without choosing truth by vote."""
from __future__ import annotations
import argparse
import json
from collections import Counter
from pathlib import Path
from typing import Any
VARGA_SECTIONS = {"D2", "D4", "D9", "D10"}
def _category(section: str, differing: list[str]) -> tuple[str, str]:
if section in VARGA_SECTIONS and differing == ["VedAstro"]:
return (
"endpoint_or_varga_semantics",
"Confirm VedAstro endpoint returns the requested varga under the same ayanamsa/node/method contract.",
)
if section.startswith("ashtakavarga"):
return (
"ashtakavarga_table_or_contributor_variant",
"Compare contributor tables, Lagna inclusion, shodhana state, and BAV/SAV row semantics.",
)
if section == "shadbala_components":
return (
"shadbala_formula_variant",
"Compare component formula, units, local solar context, aspect model, and Chesta lineage before totals.",
)
if section == "shadbala_total":
return (
"derived_total_from_component_variants",
"Do not arbitrate totals until all six component variants and Virupa/Rupa units are aligned.",
)
if differing == ["VedAstro"]:
return (
"vedastro_deployment_or_method_drift",
"Replay against an identified VedAstro build and method; hosted anonymous output cannot decide truth.",
)
return (
"cross_engine_numeric_or_schema_difference",
"Normalize versions, schema paths, units, and formula variants; require an external worked example.",
)
def arbitrate_manifest(path: str | Path) -> dict[str, Any]:
manifest_path = Path(path)
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
rows: list[dict[str, Any]] = []
categories: Counter[str] = Counter()
for source in manifest.get("comparison_rows") or []:
if source.get("status") != "mismatch":
continue
local = source.get("local_value")
differing = sorted(
engine for engine, value in (source.get("oracle_values") or {}).items() if value != local
)
category, closure = _category(str(source.get("section")), differing)
categories[category] += 1
rows.append({
"section": source.get("section"),
"field": source.get("field"),
"local_value": local,
"oracle_values": source.get("oracle_values"),
"differing_engines": differing,
"category": category,
"closure_requirement": closure,
"truth_status": "unresolved",
})
return {
"scope": "three_engine_field_level_mismatch_arbitration",
"manifest_path": str(manifest_path),
"truth_policy": "no_majority_vote",
"mismatch_count": len(rows),
"classified_count": sum(categories.values()),
"unclassified_count": len(rows) - sum(categories.values()),
"category_counts": dict(sorted(categories.items())),
"rows": rows,
"status": "classified_unresolved" if rows else "no_mismatches",
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("manifest", nargs="?", default="references/oracle/three_engine_parity_replay_manifest.json")
parser.add_argument("--output", type=Path)
args = parser.parse_args()
report = arbitrate_manifest(args.manifest)
text = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
if args.output:
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(text, encoding="utf-8")
print(text, end="")
return 0
if __name__ == "__main__":
raise SystemExit(main())