Jyotisha/scripts/research/d60_two_way_run.py

#!/usr/bin/env python3
"""G1: compare the production D60 ascendant with an external engine (issue #149).

For every case of the v5 minute-rectification holdout this prints the D60
(rashi) sign of the ascendant from the production mapping and from PyJHora's
traditional Parasara shashtyamsa chart. The gate in
docs/tasks/TASK-rectification-d60-two-way-20261028.md asks for >=95% agreement
with at least one external engine; after BUG-1500 both count from the sign
itself and the expected result is 77/77.

The output is deterministic: no timestamps, sorted keys, fixed formatting, so
a rerun on another machine can be compared byte for byte.

Usage:
  python scripts/research/d60_two_way_run.py            # writes the results JSON
  python scripts/research/d60_two_way_run.py --json      # prints it to stdout

PyJHora is AGPL; it is imported here only, never copied into production code.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))
SCRIPTS = ROOT / "scripts"
if str(SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SCRIPTS))

from scripts.jyotish_engine import compute_chart_data  # noqa: E402

DATASET = ROOT / "references/real_case_calibration/minute_rectification_holdout_v5.json"
DEFAULT_OUT = ROOT / "docs/research/d60_two_way_results_2026_10_29.json"
GATE_RATE = 0.95


def _cases() -> list[dict]:
    return json.loads(DATASET.read_text(encoding="utf-8"))["cases"]


def our_d60_ascendant(case: dict) -> tuple[float, int]:
    birth = case["birth"]
    year, month, day = (int(part) for part in birth["date"].split("-"))
    hour, minute = (int(part) for part in birth["time"].split(":"))
    chart, _asc_idx, _jd, _ayanamsa = compute_chart_data(
        year,
        month,
        day,
        hour,
        minute,
        birth["latitude"],
        birth["longitude"],
        birth["timezone_offset"],
        node_mode="mean",
        ayanamsa_name="raman",
    )
    import varga

    longitude = float((chart.get("ascendant") or {})["degree_raw"])
    return longitude, int(varga.calc_varga(longitude, 60)["sign_idx"])


def pyjhora_d60_ascendant(case: dict) -> int:
    from jhora import utils
    from jhora.horoscope.chart import charts
    from jhora.panchanga import drik

    birth = case["birth"]
    year, month, day = (int(part) for part in birth["date"].split("-"))
    hour, minute = (int(part) for part in birth["time"].split(":"))
    drik.set_ayanamsa_mode("Raman")
    jd = utils.julian_day_number((year, month, day), (hour, minute, 0))
    place = drik.Place("case", birth["latitude"], birth["longitude"], birth["timezone_offset"])
    chart = charts.divisional_chart(jd, place, divisional_chart_factor=60)
    lagna = next(item for item in chart if str(item[0]) == "L")
    return int(lagna[1][0])


def run() -> dict:
    rows = []
    for case in _cases():
        longitude, ours = our_d60_ascendant(case)
        theirs = pyjhora_d60_ascendant(case)
        rows.append(
            {
                "case_id": case["case_id"],
                "ascendant_longitude": round(longitude, 6),
                "d60_sign_idx_production": ours,
                "d60_sign_idx_pyjhora": theirs,
                "matches_pyjhora": ours == theirs,
            }
        )
    matches = sum(row["matches_pyjhora"] for row in rows)
    count = len(rows)
    return {
        "scope": "d60_gate_g1_engine_reconciliation",
        "task": "docs/tasks/TASK-rectification-d60-two-way-20261028.md",
        "brief": "docs/tasks/TASK-rectification-d60-fix-20261029.md",
        "issue": 149,
        "bug": "BUG-1500",
        "dataset": {
            "path": DATASET.relative_to(ROOT).as_posix(),
            "sha256": hashlib.sha256(DATASET.read_bytes()).hexdigest(),
            "cases": count,
        },
        "profile": {
            "ayanamsa": "raman",
            "node_mode": "mean",
            "pyjhora_ayanamsa_mode": "Raman",
            "pyjhora_chart": "divisional_chart(divisional_chart_factor=60), traditional Parasara from sign",
        },
        "gate": {
            "threshold": GATE_RATE,
            "matches": matches,
            "rate": round(matches / count, 4) if count else None,
            "passed": bool(count) and matches / count >= GATE_RATE,
        },
        "cases": rows,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--json", action="store_true", help="print instead of writing the default path")
    parser.add_argument("--out", default=None)
    args = parser.parse_args()
    payload = run()
    text = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
    if args.json:
        print(text, end="")
        return 0
    out = Path(args.out) if args.out else DEFAULT_OUT
    out.write_text(text, encoding="utf-8", newline="\n")
    print(out.relative_to(ROOT).as_posix())
    return 0


if __name__ == "__main__":
    raise SystemExit(main())