Files
Jyotisha/tests/test_dasha_api_engine_parity.py
T
jesse-uxandGrok (agent loop) 5ac88e58f5 让大运接口走完整解读的同一套算法,不再返回编出来的周期。
Ashtottari 在适用时把出生前已经走过的部分算进时间线。

Co-Authored-By: Grok (agent loop)
2026-10-16 00:52:29 +08:00

424 lines
15 KiB
Python

"""API dasha periods must match the full-reading modules (BUG-1341, BUG-1344).
fixtureProvenance:
fictional chart from docs/tasks/TASK-dasha-unify-20261013.md section 1
public charts already stored in tests/smoke_test_runner.py (Elvis Presley)
and tests/verify_jyotish_v6_1.py (Britney Spears, UT clock, tz 0)
PyJHora 4.8.7: drik.set_ayanamsa_mode("LAHIRI");
graha get_dhasa_bhukthi(dob, tob, place, dhasa_level_index=1);
ashtottari get_ashtottari_dhasa_bhukthi(julian_day_number(dob, tob), place, dhasa_level_index=1)
"""
from __future__ import annotations
import importlib.util
import os
import random
import re
import sys
from datetime import date, datetime
from functools import lru_cache
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / "scripts"
os.environ.setdefault("VEDASTRO_ENABLE_NETWORK", "0")
FICTION = {
"year": 1990,
"month": 1,
"day": 1,
"hour": 12,
"minute": 0,
"second": 0,
"lat": 28.6139,
"lon": 77.2090,
"tz": 5.5,
"ayanamsa": "lahiri",
"node_mode": "mean",
"today": "2026-01-01",
}
# Registry key -> full-reading module id. These are the systems T1 compares.
READING_KEYS = (
("panchottari", "panchottari_dasha"),
("satabdika", "satabdika_dasha"),
("chaturaaseeti", "chaturshitisama_dasha"),
("shasti_hayani", "shastihayani_dasha"),
("shodasottari", "shodashottari_dasha"),
("dwadashottari", "dwadashottari_dasha"),
("dwisaptati", "dwisaptatisama_dasha"),
("shattrimsa", "shattrimshatsama_dasha"),
("yogini", "yogini_dasha"),
("ashtottari", "ashtottari_dasha"),
("kalachakra", "kalachakra_dasha"),
("narayana", "narayana_dasha"),
)
# Anchors copied from the six modules before they shared one helper.
BALANCE_ANCHORS = (
("panchottari_dasha", "ANURADHA_INDEX"),
("satabdika_dasha", "REVATI_INDEX"),
("chaturshitisama_dasha", "SWATI_INDEX"),
("dwisaptatisama_dasha", "MULA_INDEX"),
("shattrimshatsama_dasha", "SRAVANA_INDEX"),
("shodashottari_dasha", "PUSHYA_INDEX"),
)
FROZEN_YEAR_TABLES = {"dasha_analyzer.py", "divisional_charts_extended.py"}
YEAR_TABLE = re.compile(
r"""['"]Ketu['"]\s*:\s*7\b.*?['"]Venus['"]\s*:\s*20\b.*?['"]Mercury['"]\s*:\s*17\b""",
re.DOTALL,
)
PYJHORA_LORDS = {
0: "Sun",
1: "Moon",
2: "Mars",
3: "Mercury",
4: "Jupiter",
5: "Venus",
6: "Saturn",
7: "Rahu",
8: "Ketu",
}
def _handler():
path = ROOT / "tests" / "test_api_server_security.py"
spec = importlib.util.spec_from_file_location("dasha_parity_security_helpers", path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module._handler()
def _drop_injected_repo_paths() -> None:
"""PyJHora inserts the main checkout onto sys.path. Keep this worktree."""
banned = {
os.path.normcase(str(ROOT.parent)),
os.path.normcase(str(ROOT.parent.parent)),
}
sys.path[:] = [
item
for item in sys.path
if not item or os.path.normcase(os.path.abspath(item)) not in banned
]
scripts = str(SCRIPTS)
if scripts not in sys.path:
sys.path.insert(0, scripts)
def _major_rows(payload):
if not isinstance(payload, dict):
return []
if isinstance(payload.get("major"), list):
return payload["major"]
if isinstance(payload.get("mahadasha_sequence"), list):
return payload["mahadasha_sequence"]
if isinstance(payload.get("periods"), list):
return payload["periods"]
return []
def _chart_body(report: dict, dasha: str) -> dict:
chart = report["chart"]
planets = {
name: {"lon": row["lon"]}
for name, row in chart["planets"].items()
if isinstance(row, dict) and isinstance(row.get("lon"), (int, float))
}
return {
"dasha": dasha,
"year": FICTION["year"],
"month": FICTION["month"],
"day": FICTION["day"],
"hour": FICTION["hour"],
"minute": FICTION["minute"],
"lat": FICTION["lat"],
"lon": FICTION["lon"],
"tz": FICTION["tz"],
"today": FICTION["today"],
"planets": planets,
"ascendant": {"lon": chart["ascendant"]["lon"]},
}
@lru_cache(maxsize=1)
def _fiction_reading() -> dict:
from scripts.jyotish_engine import cmd_full_reading
from scripts.professional_report_reference import _export_args
return cmd_full_reading(_export_args(dict(FICTION)))
def _legacy_birth_balance(moon_longitude, anchor_index, sequence):
"""Source copy of the six helpers before they called dasha_common.
Operation order matches the deleted bodies: modulo, nakshatra index,
offset, lord from the sequence, fraction, then years times (1 - fraction).
"""
moon_longitude = float(moon_longitude) % 360.0
span = 360.0 / 27.0
nakshatra_index = int(moon_longitude // span)
offset = (nakshatra_index - anchor_index) % 27
starting_lord = sequence[offset % len(sequence)][0]
years_by_lord = dict(sequence)
fraction_elapsed = (moon_longitude % span) / span
remaining_years = years_by_lord[starting_lord] * (1.0 - fraction_elapsed)
return starting_lord, remaining_years, nakshatra_index
def _as_date(value) -> date:
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
text = str(value)
return datetime.fromisoformat(text[:19]).date()
def _period_covering(majors, moment: datetime):
first = datetime.fromisoformat(majors[0]["start_date"])
last = datetime.fromisoformat(majors[-1]["end_date"])
span = last - first
if span.total_seconds() <= 0 or moment < first:
return None
cycles = int((moment - first).total_seconds() // span.total_seconds())
for period in majors:
start = datetime.fromisoformat(period["start_date"]) + cycles * span
end = datetime.fromisoformat(period["end_date"]) + cycles * span
if start <= moment < end:
return period["planet"], start, end
return None
def test_api_major_periods_match_full_reading_modules() -> None:
report = _fiction_reading()
modules = report["modules"]
handler = _handler()
for key, module_id in READING_KEYS:
engine_rows = _major_rows(modules[module_id])
result = handler._compute_dasha_system(_chart_body(report, key))
assert result["precision"] == "engine_module", key
assert result["precision"] not in {"generic", "calculator"}
assert result["periods"] == engine_rows, key
def test_unimplemented_dasha_keys_are_rejected() -> None:
handler = _handler()
body = _chart_body(_fiction_reading(), "tara")
from scripts.jyotish_api_server import BadRequest
with pytest.raises(BadRequest, match="dasha system not implemented"):
handler._compute_dasha_system(body)
body["dasha"] = "chara"
with pytest.raises(BadRequest, match=r"/api/dasha/chara"):
handler._compute_dasha_system(body)
source = (SCRIPTS / "jyotish_api_server.py").read_text(encoding="utf-8")
assert "except BadRequest as e:" in source
assert "_error_json(str(e), 400, 'ERR_BAD_REQUEST')" in source
def test_narayana_keeps_the_legacy_label() -> None:
result = _handler()._compute_dasha_system(_chart_body(_fiction_reading(), "narayana"))
assert "旧算法" in result["algorithm_label"]
assert result["profile"] == "legacy_reference_only"
assert "Sanjay Rath" not in result["algorithm_label"]
def test_shoola_uses_the_native_rashi_section() -> None:
report = _fiction_reading()
from scripts.rashi_dashas_native import calculate_rashi_dashas
chart = report["chart"]
planet_lons = {name: row["lon"] for name, row in chart["planets"].items()}
expected = calculate_rashi_dashas(
date(1990, 1, 1),
chart["ascendant"]["lon"],
planet_lons,
)["shoola"]["periods"]
result = _handler()._compute_dasha_system(_chart_body(report, "shoola"))
assert result["precision"] == "engine_module"
assert result["periods"] == expected
def test_birth_balance_matches_the_old_formula() -> None:
from scripts.dasha_common import nakshatra_seeded_birth_balance
rng = random.Random(1341)
samples = [rng.uniform(-40.0, 400.0) for _ in range(2000)]
for module_name, anchor_name in BALANCE_ANCHORS:
module = importlib.import_module(f"scripts.{module_name}")
sequence = module.DASHA_SEQUENCE
anchor = getattr(module, anchor_name)
for moon in samples:
assert module._birth_lord_and_balance(moon) == nakshatra_seeded_birth_balance(
moon, anchor_index=anchor, sequence=sequence
)
assert nakshatra_seeded_birth_balance(
moon, anchor_index=anchor, sequence=sequence
) == _legacy_birth_balance(moon, anchor, sequence)
def test_vimshottari_year_literals_stay_at_the_shared_table() -> None:
matches = sorted(
path.name
for path in SCRIPTS.glob("*.py")
if path.name not in FROZEN_YEAR_TABLES and YEAR_TABLE.search(path.read_text(encoding="utf-8"))
)
assert matches == ["dasha_common.py"]
frozen = SCRIPTS / "dasha_analyzer.py"
assert YEAR_TABLE.search(frozen.read_text(encoding="utf-8"))
def _pyjhora_rows(module_name: str, birth: dict):
pytest.importorskip("jhora")
_drop_injected_repo_paths()
from jhora import utils
from jhora.panchanga import drik
drik.set_ayanamsa_mode("LAHIRI")
place = drik.Place("f", birth["lat"], birth["lon"], birth["tz"])
dob = drik.Date(birth["year"], birth["month"], birth["day"])
tob = (birth["hour"], birth["minute"], int(birth.get("second", 0)))
if module_name == "ashtottari":
from jhora.horoscope.dhasa.graha.ashtottari import get_ashtottari_dhasa_bhukthi
jd = utils.julian_day_number((birth["year"], birth["month"], birth["day"]), tob)
return get_ashtottari_dhasa_bhukthi(jd, place, dhasa_level_index=1)
module = importlib.import_module(f"jhora.horoscope.dhasa.graha.{module_name}")
return module.get_dhasa_bhukthi(dob, tob, place, dhasa_level_index=1)
def _assert_same_boundaries(engine_rows, py_rows, birth: datetime, slack_days: int) -> None:
engine = [(_row_lord(row), _as_date(row.get("start_date"))) for row in engine_rows]
oracle = [(PYJHORA_LORDS[row[0][0]], date(*row[1][:3])) for row in py_rows]
assert engine[0][0] == oracle[0][0]
birth_day = birth.date()
if engine[0][1] == birth_day and oracle[0][1] < birth_day:
pairs = zip(engine[1:], oracle[1:])
limit = 0
else:
pairs = zip(engine, oracle)
limit = slack_days
compared = 0
for (engine_lord, engine_start), (oracle_lord, oracle_start) in pairs:
assert engine_lord == oracle_lord
assert abs((engine_start - oracle_start).days) <= limit
compared += 1
assert compared >= 3
def _row_lord(row: dict) -> str:
return row.get("lord") or row.get("planet")
def test_conditional_tables_match_pyjhora_start_days() -> None:
from scripts.chaturshitisama_dasha import calculate_chaturshitisama_dasha
from scripts.dasha_registry import build_reading_birth_info
from scripts.domain_calculation_service import compute_chart
from scripts.panchottari_dasha import calculate_panchottari_dasha
from scripts.satabdika_dasha import calculate_satabdika_dasha
from scripts.source_bounded_child_periods import PYJHORA_PARENT_SEEDED_EQUAL_SPLIT_PROFILE
chart = compute_chart({**FICTION, "ayanamsa": "lahiri"})
birth = datetime(1990, 1, 1, 12, 0)
_alt, conditional, _lons = build_reading_birth_info(
birth,
chart["planets"],
chart["ascendant"],
{"lat": FICTION["lat"], "lon": FICTION["lon"], "tz": FICTION["tz"], "year": 1990},
)
info = {
**conditional,
"child_period_profile": PYJHORA_PARENT_SEEDED_EQUAL_SPLIT_PROFILE,
"standard_table_profile": "pl9_reports_all_dasha_tables_v1",
}
calls = (
("panchottari", calculate_panchottari_dasha),
("sataatbika", calculate_satabdika_dasha),
("chathuraaseethi_sama", calculate_chaturshitisama_dasha),
)
for module_name, calculator in calls:
rows = calculator(info)["major"]
assert rows
_assert_same_boundaries(rows, _pyjhora_rows(module_name, FICTION), birth, 0)
def _public_case(name: str, birth: dict, family: str) -> None:
from scripts.ashtottari_dasha import calculate_ashtottari_dasha
from scripts.ashtottari_rule_profiles import (
RULE_FAMILY_MOON_NAKSHATRA_PLUS_PAKSHA,
RULE_FAMILY_RAHU_FROM_LAGNA_LORD,
)
from scripts.dasha_registry import build_reading_birth_info
from scripts.domain_calculation_service import compute_chart
families = {
"moon": RULE_FAMILY_MOON_NAKSHATRA_PLUS_PAKSHA,
"rahu": RULE_FAMILY_RAHU_FROM_LAGNA_LORD,
}
chart = compute_chart({**birth, "second": 0, "node_mode": "mean", "ayanamsa": "lahiri"})
born = datetime(birth["year"], birth["month"], birth["day"], birth["hour"], birth["minute"])
alt, conditional, _lons = build_reading_birth_info(
born,
chart["planets"],
chart["ascendant"],
{"lat": birth["lat"], "lon": birth["lon"], "tz": birth["tz"], "year": birth["year"]},
)
result = calculate_ashtottari_dasha({
**alt,
"moon_longitude": conditional["moon_longitude"],
"birth_is_daytime": conditional["birth_is_daytime"],
"rahu_sign_index": conditional["rahu_sign_index"],
"lagna_lord_sign_index": conditional["lagna_lord_sign_index"],
"applicability_rule_family": families[family],
})
assert result["applicable"] is True, name
majors = result["major"]
start = datetime.fromisoformat(majors[0]["start_date"])
end = datetime.fromisoformat(majors[0]["end_date"])
assert start <= born < end
now = datetime.now()
covered = _period_covering(majors, now)
assert covered is not None, name
current = result["current"]
assert current["planet"] == covered[0]
assert datetime.fromisoformat(current["start_date"]) == covered[1]
assert datetime.fromisoformat(current["end_date"]) == covered[2]
_assert_same_boundaries(majors, _pyjhora_rows("ashtottari", birth), born, 1)
def test_ashtottari_birth_balance_matches_pyjhora_within_one_day() -> None:
fiction = {key: FICTION[key] for key in ("year", "month", "day", "hour", "minute", "lat", "lon", "tz")}
presley = {
"year": 1935,
"month": 1,
"day": 8,
"hour": 4,
"minute": 35,
"lat": 34.3,
"lon": -88.4,
"tz": -6.0,
}
spears = {
"year": 1981,
"month": 12,
"day": 2,
"hour": 19,
"minute": 55,
"lat": 31.17,
"lon": -89.18,
"tz": 0.0,
}
_public_case("fiction", fiction, "moon")
_public_case("presley", presley, "rahu")
_public_case("spears", spears, "moon")
def test_placeholder_modules_are_gone() -> None:
assert not (SCRIPTS / "extended_dashas.py").exists()
assert not (SCRIPTS / "conditional_dashas.py").exists()
assert not (ROOT / "skills" / "jyotish-vedic-astrology" / "scripts" / "extended_dashas.py").exists()