feat(engine): expose native technique layers in the consultation output
TASK-consult-evidence-card-v2-20260927 T1. New module scripts/consultation_native_layers.py, thinly registered after the merged engine fields in _attach_local_consultation_layers (no handler method, no new forgery site). It writes one new top-level chart key, chart.consultation_native_layers, kept out of chart.modules so the thematic report's full_reading_module_count does not move: - d9_summary: D9 lagna, each planet's D9 sign and dignity (jyotish_engine._get_dignity_level), Vargottama (_calc_vargottama), D1<->D9 reversals - punarphoo (punarphoo.detect_punarphoo, observation_only) - vivah_saham (jyotish_engine._calc_vivah_saham), day_night (gulika) - slow_transits: Jupiter / Saturn / Rahu / Ketu now with natal house and the Jupiter / Saturn SAV / BAV, and the next twelve months' ingress / station dates (ephemeris_events; nodes by the same daily-noon sampling) - double_transit: cmd_double_transit_pac for houses 1-12, conclusions only, plus DK / UL targets by the same PAC rule - karakamsha, argala, dispositor_chains, inter_chart_linkage, moon_transit (lookup-only layers) - annual_tajika on the annual route: build_annual_tajika_pack (annual lagna, Varshesha, Muntha, Mudda Dasha with dates, Sun / Moon / year-lord Tajika aspects), parameter_sensitive; blocked when the pack is not reliable, never natal data A/B on 3 public AA charts x 4 routes: every existing output is identical, only the new key is added; about 40 ms per workflow. Golden regenerated with annual and timing routes (family workflow byte-identical apart from the new key). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017eEAG8HD3mm8gsKXgk8uU8
This commit is contained in:
Jesse_Chen
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
5b061e4361
commit
d5616c347e
File diff suppressed because one or more lines are too long
@@ -0,0 +1,771 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Expose existing native technique modules in the web consultation output.
|
||||
|
||||
TASK-consult-evidence-card-v2-20260927 T1. The consultation workflow already
|
||||
computes the chart, the formal vargas, Ashtakavarga, Arudha padas and the chara
|
||||
karakas, but several techniques the repository implements never reached its
|
||||
output: the D9 dignity layer and Vargottama, Punarphoo, Double Transit, the
|
||||
Rahu/Ketu and slow-planet ingress and station dates, Vivah Saham, Karakamsha,
|
||||
Argala, the dispositor chain, the D1/D9/D10 linkage, the Moon transit and the
|
||||
annual Tajika / Varshaphala pack.
|
||||
|
||||
This module only adds output. Every value comes from a native producer
|
||||
(`jyotish_engine`, `punarphoo`, `ephemeris_events`, `jaimini`, `argala`,
|
||||
`annual_tajika_pack`); what this module adds on top is selection, the
|
||||
whole-sign house of a sign that a producer already returned, and the calendar
|
||||
dates of the Mudda Dasha segments (see `_mudda_segments`). It does not change
|
||||
any existing module, score, candidate or frozen scoring file, and it writes a
|
||||
single new top-level chart key, `chart.consultation_native_layers`, which no
|
||||
existing reader consumes. It is kept out of `chart.modules` on purpose: the
|
||||
thematic report counts the modules it was handed (`full_reading_module_count`)
|
||||
and that existing value must not move.
|
||||
|
||||
The annual pack is attached only when the declared route is `annual`. When the
|
||||
pack cannot be produced reliably it is reported as blocked with the reason;
|
||||
natal data is never substituted for the annual chart.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
SCHEMA = "jyotish.consultation_native_layers.v1"
|
||||
SIGNS = (
|
||||
"Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo",
|
||||
"Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
|
||||
)
|
||||
TRANSIT_WINDOW_DAYS = 365
|
||||
MAX_DOUBLE_TRANSIT_CONCLUSIONS = 24
|
||||
_STRONG_DIGNITIES = frozenset({"EXALTED", "MOOLATRIKONA", "OWN_SIGN"})
|
||||
_WEAK_DIGNITIES = frozenset({"DEBILITATED", "NEECHA_BHANGA"})
|
||||
REVERSAL_RULE = (
|
||||
"D1 in EXALTED / MOOLATRIKONA / OWN_SIGN and D9 in DEBILITATED / NEECHA_BHANGA, or the reverse; "
|
||||
"levels from jyotish_engine._get_dignity_level"
|
||||
)
|
||||
TAJIKA_BOUNDARY = (
|
||||
"parameter_sensitive: Tajika / Varshaphala is verified on the first target set only, "
|
||||
"not a global closure; annual structure, not an event verdict"
|
||||
)
|
||||
|
||||
|
||||
def _scripts_on_path() -> None:
|
||||
scripts_dir = str(Path(__file__).resolve().parent)
|
||||
if scripts_dir not in sys.path:
|
||||
sys.path.insert(0, scripts_dir)
|
||||
|
||||
|
||||
def _load_engine():
|
||||
# Same loader as consultation_engine_field_bridge: reuse the engine module
|
||||
# the consultation path already imported instead of loading another copy.
|
||||
_scripts_on_path()
|
||||
try:
|
||||
from scripts import jyotish_engine as engine
|
||||
except ModuleNotFoundError: # pragma: no cover - script execution
|
||||
import jyotish_engine as engine
|
||||
return engine
|
||||
|
||||
|
||||
def _record(value: Any) -> dict:
|
||||
return value if isinstance(value, dict) else {}
|
||||
|
||||
|
||||
def _sign_index(sign: Any) -> int | None:
|
||||
return SIGNS.index(sign) if sign in SIGNS else None
|
||||
|
||||
|
||||
def _whole_sign_house(sign: Any, lagna_sign_idx: int | None) -> int | None:
|
||||
idx = _sign_index(sign)
|
||||
if idx is None or lagna_sign_idx is None:
|
||||
return None
|
||||
return ((idx - lagna_sign_idx) % 12) + 1
|
||||
|
||||
|
||||
def _blocked(exc: Exception | str) -> dict:
|
||||
reason = exc if isinstance(exc, str) else f"{exc.__class__.__name__}"
|
||||
return {"status": "blocked", "reason": reason}
|
||||
|
||||
|
||||
def _planet_lon(pdata: Any) -> float | None:
|
||||
row = _record(pdata)
|
||||
for key in ("lon", "degree_raw"):
|
||||
value = row.get(key)
|
||||
if isinstance(value, (int, float)):
|
||||
return float(value)
|
||||
return None
|
||||
|
||||
|
||||
def _lagna_idx(chart: dict) -> int | None:
|
||||
asc = _record(chart.get("ascendant"))
|
||||
if isinstance(asc.get("sign_idx"), int):
|
||||
return int(asc["sign_idx"]) % 12
|
||||
return _sign_index(asc.get("sign"))
|
||||
|
||||
|
||||
def _d9_chart(modules: dict) -> dict:
|
||||
return _record(_record(modules.get("varga_full")).get("D9_Navamsa"))
|
||||
|
||||
|
||||
# ── D9 summary ──────────────────────────────────────────────────────────────
|
||||
|
||||
def build_d9_summary(engine, chart: dict, modules: dict) -> dict:
|
||||
"""D9 lagna, each planet's D9 sign and dignity, Vargottama, D1↔D9 reversals.
|
||||
|
||||
D9 dignity: `jyotish_engine._get_dignity_level` with the context rule the
|
||||
full reading's D9 step uses (`_build_dignity_context` of the D9 positions).
|
||||
D1 dignity for the reversal check is the chart's own `status` label. The
|
||||
D1 label itself is already in the base section, so it is not repeated.
|
||||
Vargottama comes from `jyotish_engine._calc_vargottama`.
|
||||
"""
|
||||
d9 = _d9_chart(modules)
|
||||
d9_planets = _record(d9.get("planets"))
|
||||
planets = _record(chart.get("planets"))
|
||||
if not d9_planets or not planets:
|
||||
return _blocked("d9_or_d1_positions_missing")
|
||||
d9_context = engine._build_dignity_context(d9_planets)
|
||||
vargottama = engine._calc_vargottama(planets, {"D9_Navamsa": d9_planets})
|
||||
labels_to_level = {label: level for level, label in engine.DIGNITY_LABELS.items()}
|
||||
rows: dict[str, dict] = {}
|
||||
reversals: list[dict] = []
|
||||
for name, pdata in planets.items():
|
||||
d9_row = _record(d9_planets.get(name))
|
||||
if not isinstance(pdata, dict) or not d9_row.get("sign") or not pdata.get("sign"):
|
||||
continue
|
||||
d9_level = engine._get_dignity_level(name, d9_row["sign"], d9_row.get("degree"), d9_context)
|
||||
# D1: the chart's own status label (what the base section shows); its
|
||||
# level is read back from the same label table.
|
||||
d1_label = pdata.get("status")
|
||||
d1_level = labels_to_level.get(d1_label) if isinstance(d1_label, str) else None
|
||||
if d1_level is None:
|
||||
d1_level = engine._get_dignity_level(name, pdata["sign"], pdata.get("degree_in_sign"), planets)
|
||||
d1_label = engine.DIGNITY_LABELS.get(d1_level, d1_level)
|
||||
is_vargottama = bool(_record(vargottama.get(name)).get("is_vargottama"))
|
||||
d9_label = engine.DIGNITY_LABELS.get(d9_level, d9_level)
|
||||
rows[name] = {
|
||||
"d9_sign": d9_row["sign"],
|
||||
"d9_dignity": d9_label,
|
||||
"vargottama": is_vargottama,
|
||||
}
|
||||
if (d1_level in _STRONG_DIGNITIES and d9_level in _WEAK_DIGNITIES) or (
|
||||
d1_level in _WEAK_DIGNITIES and d9_level in _STRONG_DIGNITIES
|
||||
):
|
||||
reversals.append({"planet": name, "d1": d1_label, "d9": d9_label})
|
||||
d9_lagna = _record(d9.get("ascendant")).get("sign")
|
||||
d1_lagna = _record(chart.get("ascendant")).get("sign")
|
||||
return {
|
||||
"status": "executed",
|
||||
"source": "jyotish_engine._get_dignity_level / _calc_vargottama on modules.varga_full.D9_Navamsa",
|
||||
"lagna": d9_lagna,
|
||||
"lagna_vargottama": bool(d9_lagna and d9_lagna == d1_lagna),
|
||||
"planets": rows,
|
||||
"vargottama": [name for name, row in rows.items() if row["vargottama"]],
|
||||
"reversals": reversals,
|
||||
"reversal_rule": REVERSAL_RULE,
|
||||
}
|
||||
|
||||
|
||||
# ── transit positions at the reference date ─────────────────────────────────
|
||||
|
||||
def _noon_local_jd(swe, day: datetime, tz: float) -> float:
|
||||
return swe.julday(day.year, day.month, day.day, 12.0 - float(tz))
|
||||
|
||||
|
||||
def _transit_positions(engine, day: datetime, tz: float, node_mode: str) -> dict:
|
||||
positions, _ = engine._calc_sidereal_planets_for_jd(
|
||||
_noon_local_jd(engine.swe, day, tz), node_mode=node_mode, include_ketu=True,
|
||||
)
|
||||
return positions if isinstance(positions, dict) else {}
|
||||
|
||||
|
||||
def _node_signs(engine, day: datetime, node_mode: str) -> dict[str, str]:
|
||||
jd = engine.swe.julday(day.year, day.month, day.day, 12.0)
|
||||
positions, _ = engine._calc_sidereal_planets_for_jd(jd, node_mode=node_mode, include_ketu=True)
|
||||
return {
|
||||
body: _record(_record(positions).get(body)).get("sign")
|
||||
for body in ("Rahu", "Ketu")
|
||||
if _record(_record(positions).get(body)).get("sign")
|
||||
}
|
||||
|
||||
|
||||
def _node_ingresses(engine, start: datetime, days: int, node_mode: str, lagna_idx: int | None) -> list[dict]:
|
||||
"""Rahu / Ketu sign changes, sampled like ephemeris_events (daily noon UT).
|
||||
|
||||
The mean node moves steadily about 0.05 degrees a day, so for it the scan
|
||||
steps a week at a time and resolves the day only inside a week whose sign
|
||||
changed; the reported date is the same first daily-noon sample in the new
|
||||
sign. The true node wobbles and can cross a boundary and back within a
|
||||
week, so it is sampled every day.
|
||||
"""
|
||||
step = 7 if node_mode == "mean" else 1
|
||||
events: list[dict] = []
|
||||
previous_day = start
|
||||
previous = _node_signs(engine, start, node_mode)
|
||||
offset = 0
|
||||
while offset < days:
|
||||
offset = min(offset + step, days)
|
||||
day = start + timedelta(days=offset)
|
||||
signs = _node_signs(engine, day, node_mode)
|
||||
if signs != previous:
|
||||
prior = previous
|
||||
probe = previous_day
|
||||
while probe < day:
|
||||
probe += timedelta(days=1)
|
||||
current = _node_signs(engine, probe, node_mode)
|
||||
for body in ("Rahu", "Ketu"):
|
||||
if prior.get(body) and current.get(body) and current[body] != prior[body]:
|
||||
events.append({
|
||||
"kind": "ingress",
|
||||
"date": probe.strftime("%Y-%m-%d"),
|
||||
"body": body.lower(),
|
||||
"from_sign": prior[body],
|
||||
"to_sign": current[body],
|
||||
"to_natal_house": _whole_sign_house(current[body], lagna_idx),
|
||||
})
|
||||
prior = current
|
||||
previous, previous_day = signs, day
|
||||
return events
|
||||
|
||||
|
||||
def build_slow_transits(engine, chart: dict, modules: dict, *, reference: datetime, tz: float,
|
||||
ayanamsa: str, node_mode: str) -> dict:
|
||||
"""Jupiter / Saturn / Rahu / Ketu now, and the key dates of the next twelve months.
|
||||
|
||||
Positions: `jyotish_engine._calc_sidereal_planets_for_jd` at local noon of
|
||||
the reference date. Jupiter / Saturn ingress and station dates:
|
||||
`ephemeris_events.build_ephemeris_events`. Rahu / Ketu ingress dates: the
|
||||
same daily-noon sampling over the engine's node positions (the ephemeris
|
||||
event scanner has no nodes). SAV / BAV of the Jupiter and Saturn signs are
|
||||
read from modules.ashtakavarga (sav.scores, bav.<planet>.bindus).
|
||||
"""
|
||||
try:
|
||||
from ephemeris_events import build_ephemeris_events
|
||||
except ModuleNotFoundError: # pragma: no cover
|
||||
from scripts.ephemeris_events import build_ephemeris_events
|
||||
lagna_idx = _lagna_idx(chart)
|
||||
positions = _transit_positions(engine, reference, tz, node_mode)
|
||||
ashtakavarga = _record(modules.get("ashtakavarga"))
|
||||
sav_scores = _record(_record(ashtakavarga.get("sav")).get("scores"))
|
||||
bav = _record(ashtakavarga.get("bav"))
|
||||
current: dict[str, dict] = {}
|
||||
for body in ("Jupiter", "Saturn", "Rahu", "Ketu"):
|
||||
row = _record(positions.get(body))
|
||||
if not row.get("sign"):
|
||||
continue
|
||||
entry = {
|
||||
"sign": row["sign"],
|
||||
"degree_in_sign": row.get("degree_in_sign"),
|
||||
"retrograde": row.get("retrograde"),
|
||||
"natal_house": _whole_sign_house(row["sign"], lagna_idx),
|
||||
}
|
||||
if body in ("Jupiter", "Saturn"):
|
||||
idx = _sign_index(row["sign"])
|
||||
if row["sign"] in sav_scores:
|
||||
entry["sav"] = sav_scores[row["sign"]]
|
||||
bindus = _record(bav.get(body)).get("bindus")
|
||||
if isinstance(bindus, list) and idx is not None and idx < len(bindus):
|
||||
entry["bav"] = bindus[idx]
|
||||
current[body] = entry
|
||||
start = reference.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=None)
|
||||
end = start + timedelta(days=TRANSIT_WINDOW_DAYS)
|
||||
scan = build_ephemeris_events({
|
||||
"start_date": start.strftime("%Y-%m-%d"),
|
||||
"end_date": end.strftime("%Y-%m-%d"),
|
||||
"ayanamsa": ayanamsa,
|
||||
"node_mode": node_mode,
|
||||
"bodies": ["jupiter", "saturn"],
|
||||
})
|
||||
events: list[dict] = []
|
||||
for item in scan.get("events") or []:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
row = {"kind": item.get("kind"), "date": item.get("date"), "body": item.get("body")}
|
||||
if item.get("kind") == "ingress":
|
||||
row["from_sign"] = item.get("from_sign")
|
||||
row["to_sign"] = item.get("to_sign")
|
||||
row["to_natal_house"] = _whole_sign_house(item.get("to_sign"), lagna_idx)
|
||||
else:
|
||||
row["motion"] = item.get("motion")
|
||||
events.append(row)
|
||||
events.extend(_node_ingresses(engine, start, TRANSIT_WINDOW_DAYS, node_mode, lagna_idx))
|
||||
events.sort(key=lambda row: (str(row.get("date")), str(row.get("body")), str(row.get("kind"))))
|
||||
houses_in_window: dict[str, list[int]] = {}
|
||||
for body in ("Jupiter", "Saturn"):
|
||||
houses: list[int] = []
|
||||
first = _record(current.get(body)).get("natal_house")
|
||||
if isinstance(first, int):
|
||||
houses.append(first)
|
||||
for row in events:
|
||||
if row.get("body") == body.lower() and row.get("kind") == "ingress":
|
||||
house = row.get("to_natal_house")
|
||||
if isinstance(house, int) and house not in houses:
|
||||
houses.append(house)
|
||||
houses_in_window[body] = houses
|
||||
return {
|
||||
"status": "executed",
|
||||
"source": "jyotish_engine._calc_sidereal_planets_for_jd + ephemeris_events.build_ephemeris_events",
|
||||
"reference_date": reference.strftime("%Y-%m-%d"),
|
||||
"window": {"start": start.strftime("%Y-%m-%d"), "end": end.strftime("%Y-%m-%d")},
|
||||
"current": current,
|
||||
"key_dates": events,
|
||||
"houses_in_window": houses_in_window,
|
||||
"boundary": scan.get("boundary") or "sign ingress and station scan, not a natal trigger",
|
||||
}
|
||||
|
||||
|
||||
def build_moon_transit(engine, chart: dict, *, reference: datetime, tz: float, node_mode: str) -> dict:
|
||||
lagna_idx = _lagna_idx(chart)
|
||||
moon = _record(_transit_positions(engine, reference, tz, node_mode).get("Moon"))
|
||||
natal_moon = _record(_record(chart.get("planets")).get("Moon"))
|
||||
if not moon.get("sign"):
|
||||
return _blocked("moon_transit_unavailable")
|
||||
return {
|
||||
"status": "executed",
|
||||
"source": "jyotish_engine._calc_sidereal_planets_for_jd (local noon)",
|
||||
"date": reference.strftime("%Y-%m-%d"),
|
||||
"sign": moon["sign"],
|
||||
"degree_in_sign": moon.get("degree_in_sign"),
|
||||
"natal_house": _whole_sign_house(moon["sign"], lagna_idx),
|
||||
"house_from_natal_moon": _whole_sign_house(moon["sign"], _sign_index(natal_moon.get("sign"))),
|
||||
"boundary": "month / day-level trigger only; changes sign about every 2.5 days",
|
||||
}
|
||||
|
||||
|
||||
# ── Punarphoo, Vivah Saham, day / night ─────────────────────────────────────
|
||||
|
||||
def build_punarphoo(chart: dict, transit_saturn_lon: float | None) -> dict:
|
||||
try:
|
||||
from punarphoo import detect_punarphoo
|
||||
except ModuleNotFoundError: # pragma: no cover
|
||||
from scripts.punarphoo import detect_punarphoo
|
||||
result = detect_punarphoo(_record(chart.get("planets")), transit_saturn_longitude=transit_saturn_lon)
|
||||
transit = _record(result.get("saturn_return_to_natal_moon"))
|
||||
return {
|
||||
"status": result.get("status"),
|
||||
"source": "punarphoo.detect_punarphoo",
|
||||
"natal_present": result.get("natal_present"),
|
||||
"orb_deg": result.get("orb_deg"),
|
||||
"orb_limit_deg": result.get("orb_limit_deg"),
|
||||
"same_house": result.get("same_house"),
|
||||
"saturn_return_to_natal_moon": {
|
||||
"active": transit.get("active"),
|
||||
"orb_deg": transit.get("orb_deg"),
|
||||
"orb_limit_deg": transit.get("orb_limit_deg"),
|
||||
},
|
||||
"claim_boundary": result.get("claim_boundary"),
|
||||
}
|
||||
|
||||
|
||||
def build_vivah_saham(engine, chart: dict) -> dict:
|
||||
"""Natal Vivah Saham from the full reading's own producer (Step 16)."""
|
||||
planets = {
|
||||
name: {"degree": lon}
|
||||
for name, lon in ((name, _planet_lon(row)) for name, row in _record(chart.get("planets")).items())
|
||||
if lon is not None
|
||||
}
|
||||
asc_lon = _planet_lon(chart.get("ascendant"))
|
||||
if not planets or asc_lon is None:
|
||||
return _blocked("natal_longitudes_missing")
|
||||
result = engine._calc_vivah_saham(planets, asc_lon)
|
||||
if not isinstance(result, dict) or result.get("error"):
|
||||
return _blocked(str(_record(result).get("error") or "vivah_saham_unavailable"))
|
||||
return {
|
||||
"status": "executed",
|
||||
"source": "jyotish_engine._calc_vivah_saham",
|
||||
"longitude": result.get("saham_lon"),
|
||||
"sign": result.get("saham_sign"),
|
||||
"degree_in_sign": result.get("saham_deg_in_sign"),
|
||||
"house": result.get("saham_house"),
|
||||
"natal_conjuncts": result.get("natal_conjuncts") or [],
|
||||
"boundary": result.get("note"),
|
||||
}
|
||||
|
||||
|
||||
def build_day_night(modules: dict) -> dict:
|
||||
evidence = _record(_record(modules.get("gulika")).get("daynight_evidence"))
|
||||
if not isinstance(evidence.get("is_daytime"), bool):
|
||||
return _blocked("daynight_evidence_missing")
|
||||
return {
|
||||
"status": "executed",
|
||||
"source": "chart.modules.gulika.daynight_evidence (swisseph.rise_trans)",
|
||||
"is_day": evidence["is_daytime"],
|
||||
}
|
||||
|
||||
|
||||
# ── Double Transit ──────────────────────────────────────────────────────────
|
||||
|
||||
def _pac_types(rows: Any) -> list[str]:
|
||||
return [str(row.get("type")) for row in rows if isinstance(row, dict) and row.get("type")] if isinstance(rows, list) else []
|
||||
|
||||
|
||||
def _double_transit_args(birth_payload: dict, reference: datetime, house: int):
|
||||
from types import SimpleNamespace
|
||||
|
||||
return SimpleNamespace(
|
||||
year=int(birth_payload["year"]), month=int(birth_payload["month"]), day=int(birth_payload["day"]),
|
||||
hour=int(birth_payload["hour"]), minute=int(birth_payload["minute"]),
|
||||
second=birth_payload.get("second") or 0,
|
||||
lat=float(birth_payload["lat"]), lon=float(birth_payload["lon"]), tz=float(birth_payload["tz"]),
|
||||
node_mode=str(birth_payload.get("node_mode") or "mean"),
|
||||
ayanamsa=birth_payload.get("ayanamsa"),
|
||||
date=reference.strftime("%Y-%m-%d"), house=house,
|
||||
)
|
||||
|
||||
|
||||
def build_double_transit(engine, chart: dict, modules: dict, *, birth_payload: dict, reference: datetime,
|
||||
transit_positions: dict) -> dict:
|
||||
"""Jupiter + Saturn activation per house from `jyotish_engine.cmd_double_transit_pac`.
|
||||
|
||||
The engine checks one event house per call (house, its lord, the lagna
|
||||
lord, the opposite lord; the same in D9 and from the Moon); all twelve are
|
||||
run. DK and UL are not targets of that producer, so the same PAC rule
|
||||
(`jyotish_engine._check_pac`) is applied to the DK longitude and the UL
|
||||
sign for the marriage card; those rows say so.
|
||||
"""
|
||||
conclusions: list[dict] = []
|
||||
summaries: dict[str, str] = {}
|
||||
for house in range(1, 13):
|
||||
result = engine.cmd_double_transit_pac(_double_transit_args(birth_payload, reference, house))
|
||||
if not isinstance(result, dict) or result.get("error"):
|
||||
return _blocked(str(_record(result).get("error") or "double_transit_unavailable"))
|
||||
summaries[str(house)] = str(result.get("summary") or "")
|
||||
for row in result.get("double_transit") or []:
|
||||
if not isinstance(row, dict):
|
||||
continue
|
||||
conclusions.append({
|
||||
"house": house,
|
||||
"layer": row.get("layer"),
|
||||
"target": row.get("target"),
|
||||
"strength": row.get("strength"),
|
||||
"jupiter_pac": _pac_types(row.get("jupiter_pac")),
|
||||
"saturn_pac": _pac_types(row.get("saturn_pac")),
|
||||
})
|
||||
active_houses = sorted({row["house"] for row in conclusions})
|
||||
lagna_idx = _lagna_idx(chart)
|
||||
jupiter_lon = _planet_lon({"lon": _record(transit_positions.get("Jupiter")).get("degree_raw")})
|
||||
saturn_lon = _planet_lon({"lon": _record(transit_positions.get("Saturn")).get("degree_raw")})
|
||||
karaka_targets: dict[str, dict] = {}
|
||||
if lagna_idx is not None and jupiter_lon is not None and saturn_lon is not None:
|
||||
dk = _record(_record(_record(modules.get("jaimini")).get("chara_karakas")).get("DK"))
|
||||
dk_planet = dk.get("planet")
|
||||
dk_lon = _planet_lon(_record(chart.get("planets")).get(dk_planet)) if dk_planet else None
|
||||
ul_sign = _record(_record(_record(modules.get("arudha_padas")).get("padas")).get("UL")).get("sign")
|
||||
ul_idx = _sign_index(ul_sign)
|
||||
targets = []
|
||||
if dk_lon is not None:
|
||||
targets.append(("DK", f"DK({dk_planet})", dk_lon))
|
||||
if ul_idx is not None:
|
||||
targets.append(("UL", f"UL({ul_sign})", ul_idx * 30 + 15))
|
||||
for key, label, target_lon in targets:
|
||||
jupiter = _pac_types(engine._check_pac("Jupiter", jupiter_lon, target_lon, lagna_idx))
|
||||
saturn = _pac_types(engine._check_pac("Saturn", saturn_lon, target_lon, lagna_idx))
|
||||
karaka_targets[key] = {
|
||||
"target": label,
|
||||
"jupiter_pac": jupiter,
|
||||
"saturn_pac": saturn,
|
||||
"double_transit": bool(jupiter and saturn),
|
||||
"method": "jyotish_engine._check_pac (same PAC rule as cmd_double_transit_pac; not a target of that producer)",
|
||||
}
|
||||
return {
|
||||
"status": "executed",
|
||||
"source": "jyotish_engine.cmd_double_transit_pac (houses 1-12)",
|
||||
"date": reference.strftime("%Y-%m-%d"),
|
||||
"active_houses": active_houses,
|
||||
"conclusions": conclusions[:MAX_DOUBLE_TRANSIT_CONCLUSIONS],
|
||||
"conclusion_count": len(conclusions),
|
||||
"house_7_summary": summaries.get("7"),
|
||||
"karaka_targets": karaka_targets,
|
||||
"boundary": "Jupiter and Saturn activation of the same target on one date; an activation window, not an event",
|
||||
}
|
||||
|
||||
|
||||
# ── Jaimini / lookup layers ─────────────────────────────────────────────────
|
||||
|
||||
def build_karakamsha(chart: dict, modules: dict) -> dict:
|
||||
try:
|
||||
from jaimini import calc_karakamsha
|
||||
except ModuleNotFoundError: # pragma: no cover
|
||||
from scripts.jaimini import calc_karakamsha
|
||||
ak = _record(_record(_record(modules.get("jaimini")).get("chara_karakas")).get("AK"))
|
||||
ak_planet = ak.get("planet")
|
||||
ak_d9 = _record(_record(_d9_chart(modules).get("planets")).get(ak_planet)) if ak_planet else {}
|
||||
if not ak_d9.get("sign"):
|
||||
return _blocked("atmakaraka_d9_missing")
|
||||
result = calc_karakamsha(ak_d9["sign"], ak_d9.get("degree"))
|
||||
return {
|
||||
"status": "executed",
|
||||
"source": "jaimini.calc_karakamsha (AK in D9)",
|
||||
"atmakaraka": ak_planet,
|
||||
"karakamsha_sign": result.get("karakamsha_sign"),
|
||||
"karakamsha_degree": result.get("karakamsha_degree"),
|
||||
"karakamsha_lord": result.get("karakamsha_lord"),
|
||||
"karakamsha_natal_house": _whole_sign_house(result.get("karakamsha_sign"), _lagna_idx(chart)),
|
||||
}
|
||||
|
||||
|
||||
def build_argala(chart: dict) -> dict:
|
||||
try:
|
||||
from argala import calc_argala
|
||||
except ModuleNotFoundError: # pragma: no cover
|
||||
from scripts.argala import calc_argala
|
||||
lagna_idx = _lagna_idx(chart)
|
||||
indices = {
|
||||
name: _sign_index(row.get("sign"))
|
||||
for name, row in _record(chart.get("planets")).items()
|
||||
if isinstance(row, dict) and _sign_index(row.get("sign")) is not None
|
||||
}
|
||||
if lagna_idx is None or not indices:
|
||||
return _blocked("argala_inputs_missing")
|
||||
result = calc_argala(indices, lagna_idx)
|
||||
return {"status": "executed", "source": "argala.calc_argala", **result}
|
||||
|
||||
|
||||
def build_dispositor_chains(engine, chart: dict) -> dict:
|
||||
chains = engine.calc_all_dispositor_chains(_record(chart.get("planets")))
|
||||
compact = {
|
||||
name: [
|
||||
{"planet": step.get("planet"), "sign": step.get("sign"), "dispositor": step.get("dispositor")}
|
||||
for step in chain if isinstance(step, dict)
|
||||
]
|
||||
for name, chain in _record(chains).items() if isinstance(chain, list)
|
||||
}
|
||||
return {
|
||||
"status": "executed" if compact else "blocked",
|
||||
"source": "jyotish_engine.calc_all_dispositor_chains",
|
||||
"chains": compact,
|
||||
"boundary": "D1 dispositor chains; parameter_sensitive",
|
||||
}
|
||||
|
||||
|
||||
def build_inter_chart_linkage(engine, chart: dict, modules: dict) -> dict:
|
||||
varga_full = _record(modules.get("varga_full"))
|
||||
|
||||
def flat(code: str) -> tuple[dict, int | None]:
|
||||
data = _record(varga_full.get(code))
|
||||
return _record(data.get("planets")), _sign_index(_record(data.get("ascendant")).get("sign"))
|
||||
|
||||
d9, d9_idx = flat("D9_Navamsa")
|
||||
d10, d10_idx = flat("D10_Dasamsa")
|
||||
if not d9 or not d10:
|
||||
return _blocked("d9_or_d10_missing")
|
||||
linkage = engine.calc_all_inter_chart_linkages(
|
||||
_record(chart.get("planets")), d9, d10, None, _lagna_idx(chart), d9_idx, d10_idx,
|
||||
)
|
||||
compact = {
|
||||
name: {
|
||||
code: {"sign": point.get("sign"), "house": point.get("house"), "lord": point.get("lord")}
|
||||
for code, point in _record(row).items() if isinstance(point, dict)
|
||||
}
|
||||
for name, row in _record(linkage).items()
|
||||
}
|
||||
return {
|
||||
"status": "executed",
|
||||
"source": "jyotish_engine.calc_all_inter_chart_linkages (D1 / D9 / D10)",
|
||||
"planets": compact,
|
||||
"boundary": "placement linkage only; parameter_sensitive",
|
||||
}
|
||||
|
||||
|
||||
# ── Annual Tajika / Varshaphala ─────────────────────────────────────────────
|
||||
|
||||
def _annual_payload(birth_payload: dict, target_year: int) -> dict:
|
||||
tz = float(birth_payload["tz"])
|
||||
sign = "-" if tz < 0 else "+"
|
||||
minutes = int(round(abs(tz) * 60))
|
||||
second = int(float(birth_payload.get("second") or 0))
|
||||
return {
|
||||
"birth": {
|
||||
"date": f"{int(birth_payload['year']):04d}-{int(birth_payload['month']):02d}-{int(birth_payload['day']):02d}",
|
||||
"time": f"{int(birth_payload['hour']):02d}:{int(birth_payload['minute']):02d}:{second:02d}",
|
||||
"latitude": float(birth_payload["lat"]),
|
||||
"longitude": float(birth_payload["lon"]),
|
||||
"utc_offset": f"{sign}{minutes // 60:02d}:{minutes % 60:02d}",
|
||||
},
|
||||
"settings": {
|
||||
"ayanamsa": birth_payload.get("ayanamsa") or "raman",
|
||||
"node_mode": birth_payload.get("node_mode") or "mean",
|
||||
},
|
||||
"target_year": int(target_year),
|
||||
}
|
||||
|
||||
|
||||
def _solar_return_local(pack: dict, tz: float) -> datetime | None:
|
||||
data = _record(_record(pack.get("solar_return")).get("data"))
|
||||
raw = data.get("dt_ut")
|
||||
try:
|
||||
moment = raw if isinstance(raw, datetime) else datetime.strptime(str(raw)[:19], "%Y-%m-%d %H:%M:%S")
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return moment + timedelta(hours=float(tz))
|
||||
|
||||
|
||||
def _mudda_segments(periods: list, start: datetime, end: datetime) -> list[dict]:
|
||||
"""Calendar dates of the native Mudda Dasha segments.
|
||||
|
||||
`tajika.calc_mudda_dasha` gives each lord's share of the year in months
|
||||
(Vimshottari proportion of 12). The year runs from this solar return to the
|
||||
next one; each segment's length is its months / 12 of that span.
|
||||
"""
|
||||
span = end - start
|
||||
cursor = start
|
||||
rows: list[dict] = []
|
||||
for period in periods:
|
||||
if not isinstance(period, dict) or not isinstance(period.get("months"), (int, float)):
|
||||
continue
|
||||
finish = cursor + span * (float(period["months"]) / 12.0)
|
||||
rows.append({
|
||||
"lord": period.get("lord"),
|
||||
"months": period.get("months"),
|
||||
"start": cursor.strftime("%Y-%m-%d"),
|
||||
"end": finish.strftime("%Y-%m-%d"),
|
||||
})
|
||||
cursor = finish
|
||||
return rows
|
||||
|
||||
|
||||
def build_annual_tajika(birth_payload: dict, chart: dict, *, reference: datetime) -> dict:
|
||||
try:
|
||||
from annual_tajika_pack import build_annual_tajika_pack
|
||||
except ModuleNotFoundError: # pragma: no cover
|
||||
from scripts.annual_tajika_pack import build_annual_tajika_pack
|
||||
tz = float(birth_payload["tz"])
|
||||
ref = reference.replace(tzinfo=None)
|
||||
target = ref.year
|
||||
pack = build_annual_tajika_pack(_annual_payload(birth_payload, target))
|
||||
start = _solar_return_local(pack, tz)
|
||||
if start is not None and start > ref:
|
||||
target -= 1
|
||||
pack = build_annual_tajika_pack(_annual_payload(birth_payload, target))
|
||||
start = _solar_return_local(pack, tz)
|
||||
following = build_annual_tajika_pack(_annual_payload(birth_payload, target + 1))
|
||||
end = _solar_return_local(following, tz)
|
||||
fields = {key: _record(pack.get(key)) for key in ("solar_return", "annual_chart", "muntha", "year_lord", "tajika_yogas", "mudda_dasha")}
|
||||
producers = {key: field.get("status") for key, field in fields.items()}
|
||||
periods = fields["mudda_dasha"].get("periods")
|
||||
required = ("solar_return", "annual_chart", "muntha", "year_lord", "mudda_dasha")
|
||||
unreliable = [key for key in required if producers.get(key) != "partial_verified"]
|
||||
if start is None or end is None or end <= start:
|
||||
unreliable.append("solar_return_dates")
|
||||
if not isinstance(periods, list) or not periods:
|
||||
unreliable.append("mudda_dasha_periods")
|
||||
if unreliable:
|
||||
return {
|
||||
"status": "blocked",
|
||||
"reason": "annual_pack_not_reliable:" + ",".join(dict.fromkeys(unreliable)),
|
||||
"source": "annual_tajika_pack.build_annual_tajika_pack",
|
||||
"producers": producers,
|
||||
"boundary": "annual chart not attached; natal data is not a substitute",
|
||||
}
|
||||
annual_chart = _record(fields["annual_chart"].get("data"))
|
||||
muntha = _record(fields["muntha"].get("data"))
|
||||
year_lord = _record(fields["year_lord"].get("data")).get("year_lord")
|
||||
annual_lagna_idx = _sign_index(annual_chart.get("asc_sign"))
|
||||
focus = {"Sun", "Moon", year_lord}
|
||||
aspects = []
|
||||
for row in _record(fields["tajika_yogas"].get("data")).get("candidate_yogas") or []:
|
||||
if not isinstance(row, dict):
|
||||
continue
|
||||
pair = row.get("planets") if isinstance(row.get("planets"), list) else []
|
||||
if not focus.intersection(pair):
|
||||
continue
|
||||
aspects.append({
|
||||
"name": row.get("name"),
|
||||
"planets": pair,
|
||||
"aspect": row.get("aspect"),
|
||||
"motion": row.get("motion"),
|
||||
})
|
||||
return {
|
||||
"status": "parameter_sensitive",
|
||||
"source": "annual_tajika_pack.build_annual_tajika_pack (solar_return_full_report + cmd_tajika)",
|
||||
"target_year": target,
|
||||
"age": annual_chart.get("age"),
|
||||
"varsha_year": {"start": start.strftime("%Y-%m-%d"), "end": end.strftime("%Y-%m-%d")},
|
||||
"annual_lagna": annual_chart.get("asc_sign"),
|
||||
"varshesha": year_lord,
|
||||
"muntha": {
|
||||
"sign": muntha.get("muntha_sign"),
|
||||
"lord": muntha.get("muntha_lord"),
|
||||
"house_in_annual_chart": _whole_sign_house(muntha.get("muntha_sign"), annual_lagna_idx),
|
||||
"house_in_natal_chart": _whole_sign_house(muntha.get("muntha_sign"), _lagna_idx(chart)),
|
||||
},
|
||||
"mudda_dasha": _mudda_segments(periods, start, end),
|
||||
"tajika_aspects": aspects,
|
||||
"tajika_aspects_status": producers.get("tajika_yogas"),
|
||||
"producers": producers,
|
||||
"boundary": TAJIKA_BOUNDARY,
|
||||
}
|
||||
|
||||
|
||||
# ── entry point ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _declared_routes(body: dict) -> set[str]:
|
||||
values: list[str] = []
|
||||
for key in ("strict_workflow_route", "theme", "themes"):
|
||||
raw = body.get(key)
|
||||
if isinstance(raw, str) and raw.strip():
|
||||
values.append(raw.strip().lower())
|
||||
elif isinstance(raw, list):
|
||||
values.extend(str(item).strip().lower() for item in raw if item)
|
||||
return {item for item in values if item}
|
||||
|
||||
|
||||
def _guard(builder, *args, **kwargs) -> dict:
|
||||
try:
|
||||
result = builder(*args, **kwargs)
|
||||
return result if isinstance(result, dict) else _blocked("producer_returned_non_dict")
|
||||
except Exception as exc: # native producers stay isolated from the workflow
|
||||
return _blocked(exc)
|
||||
|
||||
|
||||
def attach_native_consultation_layers(chart: dict, *, birth_payload: dict, body: dict,
|
||||
reference_dt: datetime) -> dict:
|
||||
"""Write `chart.consultation_native_layers`. Safe on empty charts."""
|
||||
if not isinstance(chart, dict) or not chart.get("success", True):
|
||||
return chart
|
||||
modules = _record(chart.get("modules"))
|
||||
if not _record(chart.get("planets")) or not _record(chart.get("ascendant")):
|
||||
chart["consultation_native_layers"] = {"schema": SCHEMA, **_blocked("chart_positions_missing")}
|
||||
return chart
|
||||
reference = reference_dt.replace(tzinfo=None)
|
||||
routes = _declared_routes(body)
|
||||
layers: dict[str, Any] = {"schema": SCHEMA, "status": "executed"}
|
||||
try:
|
||||
engine = _load_engine()
|
||||
_scripts_on_path()
|
||||
from ayanamsa_utils import temporary_ayanamsa
|
||||
from domain_calculation_service import swiss_ephemeris_lock
|
||||
tz = float(birth_payload.get("tz") or 0)
|
||||
ayanamsa = str(birth_payload.get("ayanamsa") or "raman")
|
||||
node_mode = str(birth_payload.get("node_mode") or "mean")
|
||||
with swiss_ephemeris_lock(), temporary_ayanamsa(ayanamsa, engine.swe):
|
||||
positions = _guard(_transit_positions, engine, reference, tz, node_mode)
|
||||
saturn_lon = _record(positions.get("Saturn")).get("degree_raw")
|
||||
layers["d9_summary"] = _guard(build_d9_summary, engine, chart, modules)
|
||||
layers["punarphoo"] = _guard(build_punarphoo, chart, saturn_lon if isinstance(saturn_lon, (int, float)) else None)
|
||||
layers["vivah_saham"] = _guard(build_vivah_saham, engine, chart)
|
||||
layers["day_night"] = _guard(build_day_night, modules)
|
||||
layers["slow_transits"] = _guard(
|
||||
build_slow_transits, engine, chart, modules,
|
||||
reference=reference, tz=tz, ayanamsa=ayanamsa, node_mode=node_mode,
|
||||
)
|
||||
layers["moon_transit"] = _guard(build_moon_transit, engine, chart, reference=reference, tz=tz, node_mode=node_mode)
|
||||
layers["double_transit"] = _guard(
|
||||
build_double_transit, engine, chart, modules,
|
||||
birth_payload=birth_payload, reference=reference, transit_positions=positions,
|
||||
)
|
||||
layers["karakamsha"] = _guard(build_karakamsha, chart, modules)
|
||||
layers["argala"] = _guard(build_argala, chart)
|
||||
layers["dispositor_chains"] = _guard(build_dispositor_chains, engine, chart)
|
||||
layers["inter_chart_linkage"] = _guard(build_inter_chart_linkage, engine, chart, modules)
|
||||
if "annual" in routes:
|
||||
layers["annual_tajika"] = _guard(build_annual_tajika, birth_payload, chart, reference=reference)
|
||||
else:
|
||||
layers["annual_tajika"] = {
|
||||
"status": "not_applicable",
|
||||
"reason": "annual pack is attached for the annual route only",
|
||||
}
|
||||
except Exception as exc:
|
||||
layers = {"schema": SCHEMA, **_blocked(exc)}
|
||||
chart["consultation_native_layers"] = layers
|
||||
return chart
|
||||
@@ -1842,8 +1842,12 @@ def _attach_local_consultation_layers(handler, chart: dict, birth_payload: dict,
|
||||
diagnostics.append({'layer': 'transits', 'status': 'unavailable', 'reason': exc.__class__.__name__})
|
||||
|
||||
from scripts.consultation_engine_field_bridge import attach_merged_engine_fields
|
||||
from scripts.consultation_native_layers import attach_native_consultation_layers
|
||||
|
||||
attach_merged_engine_fields(chart, reference_dt=_consultation_reference_date(body).replace(tzinfo=None))
|
||||
attach_native_consultation_layers(
|
||||
chart, birth_payload=birth_payload, body=body, reference_dt=_consultation_reference_date(body),
|
||||
)
|
||||
chart['local_consultation_layers'] = {
|
||||
'status': 'ready' if not diagnostics else 'partial',
|
||||
'source': 'repository_local_engines',
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
"""Capture real engine consultation responses for the evidence-card tests.
|
||||
|
||||
Three public AA charts (Steve Jobs, Barack Obama, Elizabeth Taylor), the
|
||||
research reference date, raman ayanamsa, mean nodes, family route. External
|
||||
VedAstro is not called (same stand-in as the research runner). The response is
|
||||
trimmed by key only: every kept value is the engine's own value, unchanged.
|
||||
research reference date, raman ayanamsa, mean nodes. Each chart carries the
|
||||
family route (`workflow`) and, for evidence card v2, the annual
|
||||
(`annual_workflow`) and timing (`timing_workflow`) routes with the frontend's
|
||||
default twelve-month horizon. External VedAstro is not called (same stand-in
|
||||
as the research runner). Each response is trimmed by key only: every kept
|
||||
value is the engine's own value, unchanged.
|
||||
|
||||
PYTHONHASHSEED=0 python scripts/research/capture_consult_evidence_card_golden.py
|
||||
"""
|
||||
@@ -36,7 +39,15 @@ TOP_KEEP = (
|
||||
"range_boundary_contexts",
|
||||
)
|
||||
RECTIFICATION_KEEP = ("effective_accuracy", "lagna_boundary", "summary", "enabled_vargas")
|
||||
CHART_KEEP = ("success", "birth", "ascendant", "planets", "houses", "shadbala", "dasha", "yogas")
|
||||
CHART_KEEP = (
|
||||
"success", "birth", "ascendant", "planets", "houses", "shadbala", "dasha", "yogas",
|
||||
# Evidence card v2: native technique layers (scripts/consultation_native_layers.py).
|
||||
"consultation_native_layers",
|
||||
)
|
||||
EXTRA_ROUTES = (
|
||||
("annual_workflow", {"id": "annual", "domain": "annual", "question": "未来一年重点是什么？"}),
|
||||
("timing_workflow", {"id": "timing", "domain": "timing", "question": "接下来一年哪些时间点重要？"}),
|
||||
)
|
||||
MODULE_KEEP = (
|
||||
"varga_spectrum", "shadbala", "arudha_padas", "jaimini", "narayana_dasha",
|
||||
"ashtakavarga", "dasha_sub_periods", "kp_cusps", "gulika",
|
||||
@@ -70,14 +81,17 @@ def main() -> int:
|
||||
charts = []
|
||||
for chart in load_public_charts(ROOT):
|
||||
workflow = runner._run_workflow(runner._workflow_body(chart, question))
|
||||
charts.append({
|
||||
entry = {
|
||||
"id": chart["id"],
|
||||
"label": chart["label"],
|
||||
"source": chart["source"],
|
||||
"case_id": chart["case_id"],
|
||||
"rodden_rating": chart["rodden_rating"],
|
||||
"workflow": trim(workflow),
|
||||
})
|
||||
}
|
||||
for key, extra in EXTRA_ROUTES:
|
||||
entry[key] = trim(runner._run_workflow(runner._workflow_body(chart, extra)))
|
||||
charts.append(entry)
|
||||
payload = {
|
||||
"source": "scripts/research/capture_consult_evidence_card_golden.py",
|
||||
"note": "Real engine consultation_workflow responses for three public AA charts, trimmed by key only.",
|
||||
@@ -86,6 +100,7 @@ def main() -> int:
|
||||
"node_mode": NODE_MODE,
|
||||
"route": question["domain"],
|
||||
"question": question["question"],
|
||||
"extra_routes": {key: extra for key, extra in EXTRA_ROUTES},
|
||||
"external_vedastro": "not_called",
|
||||
"charts": charts,
|
||||
}
|
||||
|
||||
@@ -74,7 +74,9 @@ def _workflow_body(chart: dict[str, Any], question: dict[str, str]) -> dict[str,
|
||||
"claim_boundary": contract.claim_boundary,
|
||||
"plan_depth": "standard",
|
||||
"requested_domains": list(contract.requested_domains),
|
||||
"timing_horizon": None,
|
||||
# The frontend plan defaults timing / annual to the next twelve months;
|
||||
# the engine refuses those two routes without a horizon.
|
||||
"timing_horizon": "next_12_months" if domain in ("timing", "annual") else None,
|
||||
"precision_boundary": "server_evidence_required",
|
||||
"required_evidence_categories": list(contract.required_evidence_categories),
|
||||
"surface": "api_web",
|
||||
|
||||
@@ -100,6 +100,8 @@ CORE_PYTEST_TARGETS = [
|
||||
"tests/test_mcp_strict_workflow_finance.py",
|
||||
"tests/test_mcp_strict_workflow_relationship.py",
|
||||
"tests/test_punarphoo_observation.py",
|
||||
# Native technique layers exposed to the web consultation (evidence card v2).
|
||||
"tests/test_consultation_native_layers.py",
|
||||
"tests/test_relationship_event_class_evidence.py",
|
||||
"tests/test_thematic_dasha_info_source.py",
|
||||
"tests/test_report_orchestrator_marriage_timing.py",
|
||||
|
||||
@@ -0,0 +1,240 @@
|
||||
"""Native technique layers in the consultation output (TASK-consult-evidence-card-v2-20260927 T1).
|
||||
|
||||
Real engine runs on a public AA chart (the evidence-card research charts).
|
||||
Each layer is compared with its native producer called directly, the annual
|
||||
pack with `build_annual_tajika_pack`, and the rest of the response with a run
|
||||
where the new layer is switched off: only the new key may differ.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
for path in (ROOT, ROOT / "scripts", ROOT / "scripts" / "research"):
|
||||
if str(path) not in sys.path:
|
||||
sys.path.insert(0, str(path))
|
||||
|
||||
import consult_evidence_card_run as runner # noqa: E402
|
||||
from consult_evidence_card_lib import load_public_charts # noqa: E402
|
||||
import scripts.consultation_native_layers as native # noqa: E402
|
||||
|
||||
QUESTIONS = {
|
||||
"marriage": "我的关系模式是什么？",
|
||||
"annual": "未来一年重点是什么？",
|
||||
"timing": "接下来一年哪些时间点重要？",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
def _no_external_vedastro():
|
||||
runner._block_external_vedastro()
|
||||
|
||||
|
||||
def _chart(index: int = 0) -> dict:
|
||||
return load_public_charts(ROOT)[index]
|
||||
|
||||
|
||||
def _run(domain: str, index: int = 0) -> dict:
|
||||
return runner._run_workflow(runner._workflow_body(_chart(index), {
|
||||
"id": domain, "domain": domain, "question": QUESTIONS[domain],
|
||||
}))
|
||||
|
||||
|
||||
def _layers(workflow: dict) -> dict:
|
||||
layers = workflow["chart"]["consultation_native_layers"]
|
||||
assert layers["schema"] == native.SCHEMA
|
||||
assert layers["status"] == "executed"
|
||||
return layers
|
||||
|
||||
|
||||
def _birth_args(workflow: dict, house: int) -> SimpleNamespace:
|
||||
birth = workflow["chart"]["birth"]
|
||||
body = _chart()["body"]
|
||||
return SimpleNamespace(
|
||||
year=body["year"], month=body["month"], day=body["day"], hour=body["hour"], minute=body["minute"],
|
||||
second=0, lat=body["lat"], lon=body["lon"], tz=float(body["tz"]), node_mode=birth["node_mode"],
|
||||
ayanamsa=birth["ayanamsa_name"], date=body["today"], house=house,
|
||||
)
|
||||
|
||||
|
||||
def test_quick_gate_runs_this_file() -> None:
|
||||
import run_quality_gate
|
||||
|
||||
assert "tests/test_consultation_native_layers.py" in run_quality_gate.CORE_PYTEST_TARGETS
|
||||
|
||||
|
||||
def test_every_layer_is_the_native_producers_own_output() -> None:
|
||||
workflow = _run("marriage")
|
||||
layers = _layers(workflow)
|
||||
chart = workflow["chart"]
|
||||
modules = chart["modules"]
|
||||
engine = native._load_engine()
|
||||
|
||||
d9 = modules["varga_full"]["D9_Navamsa"]
|
||||
summary = layers["d9_summary"]
|
||||
assert summary["lagna"] == d9["ascendant"]["sign"]
|
||||
expected_vargottama = engine._calc_vargottama(chart["planets"], {"D9_Navamsa": d9["planets"]})
|
||||
context = engine._build_dignity_context(d9["planets"])
|
||||
for name, row in summary["planets"].items():
|
||||
assert row["d9_sign"] == d9["planets"][name]["sign"]
|
||||
assert row["vargottama"] is expected_vargottama[name]["is_vargottama"]
|
||||
level = engine._get_dignity_level(name, d9["planets"][name]["sign"], d9["planets"][name]["degree"], context)
|
||||
assert row["d9_dignity"] == engine.DIGNITY_LABELS[level]
|
||||
assert summary["vargottama"] == [name for name, row in summary["planets"].items() if row["vargottama"]]
|
||||
assert len(summary["planets"]) == 9
|
||||
|
||||
from punarphoo import detect_natal_punarphoo
|
||||
|
||||
natal = detect_natal_punarphoo(chart["planets"])
|
||||
assert layers["punarphoo"]["status"] == "observation_only"
|
||||
assert layers["punarphoo"]["natal_present"] == natal["natal_present"]
|
||||
assert layers["punarphoo"]["orb_deg"] == natal["orb_deg"]
|
||||
assert "not marriage" in layers["punarphoo"]["claim_boundary"]
|
||||
|
||||
vivah = engine._calc_vivah_saham(
|
||||
{name: {"degree": row["lon"]} for name, row in chart["planets"].items()}, chart["ascendant"]["lon"],
|
||||
)
|
||||
assert layers["vivah_saham"]["longitude"] == vivah["saham_lon"]
|
||||
assert layers["vivah_saham"]["house"] == vivah["saham_house"]
|
||||
assert layers["vivah_saham"]["sign"] == vivah["saham_sign"]
|
||||
|
||||
assert layers["day_night"]["is_day"] is modules["gulika"]["daynight_evidence"]["is_daytime"]
|
||||
|
||||
seventh = engine.cmd_double_transit_pac(_birth_args(workflow, 7))
|
||||
double = layers["double_transit"]
|
||||
assert double["house_7_summary"] == seventh["summary"]
|
||||
assert [row["target"] for row in double["conclusions"] if row["house"] == 7] == [
|
||||
row["target"] for row in seventh["double_transit"]
|
||||
]
|
||||
assert set(double["karaka_targets"]) == {"DK", "UL"}
|
||||
assert double["karaka_targets"]["DK"]["target"] == f"DK({modules['jaimini']['chara_karakas']['DK']['planet']})"
|
||||
assert double["karaka_targets"]["UL"]["target"] == f"UL({modules['arudha_padas']['padas']['UL']['sign']})"
|
||||
# Conclusions, not raw degrees.
|
||||
assert not re.search(r"\d+\.\d+°", json.dumps(double["conclusions"], ensure_ascii=False))
|
||||
|
||||
slow = layers["slow_transits"]
|
||||
assert set(slow["current"]) == {"Jupiter", "Saturn", "Rahu", "Ketu"}
|
||||
sav = modules["ashtakavarga"]["sav"]["scores"]
|
||||
for body in ("Jupiter", "Saturn"):
|
||||
row = slow["current"][body]
|
||||
assert row["sav"] == sav[row["sign"]]
|
||||
assert row["bav"] == modules["ashtakavarga"]["bav"][body]["bindus"][native.SIGNS.index(row["sign"])]
|
||||
from ephemeris_events import build_ephemeris_events
|
||||
|
||||
scan = build_ephemeris_events({
|
||||
"start_date": slow["window"]["start"], "end_date": slow["window"]["end"],
|
||||
"ayanamsa": chart["birth"]["ayanamsa_name"], "bodies": ["jupiter", "saturn"],
|
||||
})
|
||||
expected = [(row["kind"], row["date"], row["body"]) for row in scan["events"]]
|
||||
got = [(row["kind"], row["date"], row["body"]) for row in slow["key_dates"] if row["body"] in ("jupiter", "saturn")]
|
||||
assert got == expected
|
||||
assert slow["houses_in_window"]["Jupiter"][0] == slow["current"]["Jupiter"]["natal_house"]
|
||||
|
||||
ak = modules["jaimini"]["chara_karakas"]["AK"]["planet"]
|
||||
assert layers["karakamsha"]["atmakaraka"] == ak
|
||||
assert layers["karakamsha"]["karakamsha_sign"] == d9["planets"][ak]["sign"]
|
||||
assert layers["argala"]["summary"]["supported_count"] == len(layers["argala"]["summary"]["supported_houses"])
|
||||
assert set(layers["dispositor_chains"]["chains"]) >= {"Sun", "Moon"}
|
||||
assert layers["inter_chart_linkage"]["planets"]["Sun"]["D9"]["sign"] == d9["planets"]["Sun"]["sign"]
|
||||
assert layers["moon_transit"]["sign"] in native.SIGNS
|
||||
assert layers["annual_tajika"]["status"] == "not_applicable"
|
||||
|
||||
|
||||
def test_annual_route_attaches_the_annual_pack_verbatim() -> None:
|
||||
workflow = _run("annual")
|
||||
annual = _layers(workflow)["annual_tajika"]
|
||||
assert annual["status"] == "parameter_sensitive"
|
||||
assert "parameter_sensitive" in annual["boundary"]
|
||||
from annual_tajika_pack import build_annual_tajika_pack
|
||||
|
||||
body = _chart()["body"]
|
||||
payload = native._annual_payload({**body, "tz": float(body["tz"]), "ayanamsa": "raman", "node_mode": "mean"}, annual["target_year"])
|
||||
pack = build_annual_tajika_pack(payload)
|
||||
assert annual["annual_lagna"] == pack["annual_chart"]["data"]["asc_sign"]
|
||||
assert annual["varshesha"] == pack["year_lord"]["data"]["year_lord"]
|
||||
assert annual["muntha"]["sign"] == pack["muntha"]["data"]["muntha_sign"]
|
||||
assert annual["muntha"]["lord"] == pack["muntha"]["data"]["muntha_lord"]
|
||||
periods = pack["mudda_dasha"]["periods"]
|
||||
assert [(row["lord"], row["months"]) for row in annual["mudda_dasha"]] == [(row["lord"], row["months"]) for row in periods]
|
||||
segments = annual["mudda_dasha"]
|
||||
assert segments[0]["start"] == annual["varsha_year"]["start"]
|
||||
assert segments[-1]["end"] == annual["varsha_year"]["end"]
|
||||
for before, after in zip(segments, segments[1:]):
|
||||
assert before["end"] == after["start"]
|
||||
assert annual["varsha_year"]["start"] <= body["today"] < annual["varsha_year"]["end"]
|
||||
focus = {"Sun", "Moon", annual["varshesha"]}
|
||||
assert annual["tajika_aspects"], "the annual chart has Tajika aspects for the Sun, Moon or year lord"
|
||||
for row in annual["tajika_aspects"]:
|
||||
assert focus.intersection(row["planets"])
|
||||
assert row["name"] in {"Ithasala_candidate", "Easarapha_candidate"}
|
||||
|
||||
|
||||
def test_an_unreliable_annual_pack_is_blocked_and_never_filled_with_natal_data(monkeypatch) -> None:
|
||||
import annual_tajika_pack
|
||||
|
||||
real = annual_tajika_pack.build_annual_tajika_pack
|
||||
|
||||
def conflicted(payload, **kwargs):
|
||||
pack = real(payload, **kwargs)
|
||||
pack["muntha"] = {"status": "conflict", "field": "muntha", "reason": "same_profile_annual_producers_disagree"}
|
||||
return pack
|
||||
|
||||
monkeypatch.setattr(annual_tajika_pack, "build_annual_tajika_pack", conflicted)
|
||||
annual = _layers(_run("annual"))["annual_tajika"]
|
||||
assert annual["status"] == "blocked"
|
||||
assert "muntha" in annual["reason"]
|
||||
for key in ("annual_lagna", "varshesha", "muntha", "mudda_dasha", "tajika_aspects"):
|
||||
assert key not in annual
|
||||
|
||||
|
||||
def test_a_failing_producer_blocks_its_own_layer_only(monkeypatch) -> None:
|
||||
engine = native._load_engine()
|
||||
|
||||
def broken(*args, **kwargs):
|
||||
raise RuntimeError("fixture")
|
||||
|
||||
monkeypatch.setattr(engine, "cmd_double_transit_pac", broken)
|
||||
workflow = _run("timing")
|
||||
layers = _layers(workflow)
|
||||
assert workflow["success"] is True
|
||||
assert layers["double_transit"] == {"status": "blocked", "reason": "RuntimeError"}
|
||||
assert layers["d9_summary"]["status"] == "executed"
|
||||
assert layers["slow_transits"]["status"] == "executed"
|
||||
|
||||
|
||||
def test_the_new_layer_leaves_every_existing_output_unchanged(monkeypatch) -> None:
|
||||
# The first run of a chart fills the chart cache; a cache hit re-reads the
|
||||
# stored chart, whose VedAstro request manifest and key order differ from a
|
||||
# fresh compute. Compare two cache-hit runs.
|
||||
_run("marriage", index=1)
|
||||
with_layers = _run("marriage", index=1)
|
||||
monkeypatch.setattr(native, "attach_native_consultation_layers", lambda chart, **_: chart)
|
||||
without = _run("marriage", index=1)
|
||||
assert "consultation_native_layers" not in without["chart"]
|
||||
volatile = {"runtime_cache", "vedastro_gateway"}
|
||||
|
||||
def strip(value):
|
||||
if isinstance(value, dict):
|
||||
return {key: strip(item) for key, item in value.items() if key not in volatile and key != "consultation_native_layers"}
|
||||
if isinstance(value, list):
|
||||
return [strip(item) for item in value]
|
||||
return value
|
||||
|
||||
assert json.dumps(strip(with_layers), sort_keys=True, default=str) == json.dumps(strip(without), sort_keys=True, default=str)
|
||||
|
||||
|
||||
def test_mudda_segment_dates_split_the_solar_year_by_the_native_months() -> None:
|
||||
start, end = datetime(2026, 2, 25), datetime(2027, 2, 25)
|
||||
rows = native._mudda_segments([{"lord": "Sun", "months": 6.0}, {"lord": "Moon", "months": 6.0}], start, end)
|
||||
assert rows == [
|
||||
{"lord": "Sun", "months": 6.0, "start": "2026-02-25", "end": "2026-08-26"},
|
||||
{"lord": "Moon", "months": 6.0, "start": "2026-08-26", "end": "2027-02-25"},
|
||||
]
|
||||
Reference in New Issue
Block a user