Jesse_ChenandClaude Opus 5.5
c74bfcc06b
Goldens regenerated by the capture scripts (PYTHONHASHSEED=0, cache TTL 0; two runs byte-identical). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N4f2nya58RoRu4yEmJgRGE
1019 lines
46 KiB
Python
1019 lines
46 KiB
Python
#!/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.
|
||
|
||
TASK-consult-card-affliction-data-20261001 adds three outputs: the whole-sign
|
||
graha drishti and combustion (`graha_drishti`, from `aspects.calc_house_aspects`
|
||
and the full reading's `jyotish_engine.is_combust`), the domain tag of each
|
||
sieved yoga (`yoga_domains`, a fixed packet table) and transit Saturn's house
|
||
from the natal Moon (`slow_transits.saturn_from_moon_house`).
|
||
|
||
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"
|
||
)
|
||
# Graha drishti casters (TASK-consult-card-affliction-data-20261001 D2): the
|
||
# seven grahas; Rahu / Ketu receive aspects but cast none here (schools
|
||
# differ), their influence reaches the card through house co-tenancy.
|
||
DRISHTI_CASTERS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn")
|
||
DRISHTI_RULE = (
|
||
"Parashari whole-sign graha drishti (aspects.calc_house_aspects): every graha the 7th; "
|
||
"Mars also 4 / 8, Jupiter 5 / 9, Saturn 3 / 10; Rahu / Ketu cast none"
|
||
)
|
||
# Domain tag of each Raman support packet (scripts/raman_support_observations.py).
|
||
# A fixed table: a yoga's domain is the packet that governs it, never a guess
|
||
# from its name (BUG-1157). The Nabhasa / Chandra batch describes the whole
|
||
# chart, so it is `general`.
|
||
YOGA_PACKET_DOMAINS = {
|
||
"raman_bvr_nabhasa_chandra_sunapha_batch_closure_packet_2026_08_20.json": "general",
|
||
"raman_bvr_wealth_core_batch_closure_packet_2026_08_20.json": "wealth",
|
||
"raman_bvr_relationship_spouse_batch_closure_packet_2026_08_20.json": "marriage",
|
||
"raman_bvr_relationship_children_batch_closure_packet_2026_08_20.json": "children",
|
||
"raman_bvr_relationship_mother_batch_closure_packet_2026_08_20.json": "parents",
|
||
"raman_bvr_relationship_sibling_core_batch_closure_packet_2026_08_20.json": "siblings",
|
||
}
|
||
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
|
||
natal_moon_sign = _record(_record(chart.get("planets")).get("Moon")).get("sign")
|
||
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,
|
||
# BUG-1156: the house of transit Saturn's sign counted from the natal
|
||
# Moon's sign (1 = Saturn over the natal Moon), whole-sign.
|
||
"saturn_from_moon_house": _whole_sign_house(
|
||
_record(current.get("Saturn")).get("sign"), _sign_index(natal_moon_sign),
|
||
),
|
||
"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",
|
||
}
|
||
|
||
|
||
# ── Graha drishti and combustion ────────────────────────────────────────────
|
||
|
||
def build_graha_drishti(engine, chart: dict) -> dict:
|
||
"""Who aspects whom (whole-sign) and which planets are combust.
|
||
|
||
Aspected houses: `aspects.calc_house_aspects` for the seven grahas (the
|
||
full reading's own Step 5 producer). `aspected_by` of a house is every
|
||
caster whose aspected houses include it; `aspected_by` of a planet is that
|
||
of the house it occupies. Combustion: `jyotish_engine.is_combust`, the
|
||
full reading's rule and orbs; written only for the seven grahas.
|
||
"""
|
||
try:
|
||
from aspects import calc_house_aspects
|
||
except ModuleNotFoundError: # pragma: no cover
|
||
from scripts.aspects import calc_house_aspects
|
||
planets = _record(chart.get("planets"))
|
||
asc_lon = _planet_lon(chart.get("ascendant"))
|
||
lagna_idx = _lagna_idx(chart)
|
||
if asc_lon is None or lagna_idx is None or not planets:
|
||
return _blocked("natal_longitudes_missing")
|
||
rows: dict[str, dict] = {}
|
||
houses: dict[str, dict] = {str(number): {"aspected_by": []} for number in range(1, 13)}
|
||
for name in DRISHTI_CASTERS:
|
||
lon = _planet_lon(planets.get(name))
|
||
if lon is None:
|
||
continue
|
||
result = calc_house_aspects(lon, name, asc_lon)
|
||
targets = sorted(int(house) for house in _record(result.get("aspects_to")))
|
||
rows[name] = {"house": result.get("house"), "aspects_houses": targets}
|
||
for house in targets:
|
||
houses[str(house)]["aspected_by"].append(name)
|
||
for name, pdata in planets.items():
|
||
if not isinstance(pdata, dict):
|
||
continue
|
||
house = _record(rows.get(name)).get("house") or _whole_sign_house(pdata.get("sign"), lagna_idx)
|
||
if not isinstance(house, int):
|
||
continue
|
||
row = rows.setdefault(name, {"house": house})
|
||
row["aspected_by"] = list(houses[str(house)]["aspected_by"])
|
||
sun_lon = _planet_lon(planets.get("Sun"))
|
||
if sun_lon is not None:
|
||
for name in DRISHTI_CASTERS:
|
||
lon = _planet_lon(planets.get(name))
|
||
if lon is not None and name in rows:
|
||
rows[name]["combust"] = bool(engine.is_combust(name, lon, sun_lon))
|
||
for row in rows.values():
|
||
row.pop("house", None)
|
||
return {
|
||
"status": "executed",
|
||
"source": "aspects.calc_house_aspects + jyotish_engine.is_combust",
|
||
"rule": DRISHTI_RULE,
|
||
"planets": rows,
|
||
"houses": houses,
|
||
}
|
||
|
||
|
||
def _yoga_packet_tags() -> dict[str, str]:
|
||
try:
|
||
from raman_support_observations import packet_members
|
||
except ModuleNotFoundError: # pragma: no cover
|
||
from scripts.raman_support_observations import packet_members
|
||
tags: dict[str, str] = {}
|
||
for filename, domain in YOGA_PACKET_DOMAINS.items():
|
||
for member in packet_members(filename):
|
||
if member.get("rule_id"):
|
||
tags.setdefault(member["rule_id"], domain)
|
||
return tags
|
||
|
||
|
||
def build_yoga_domains(modules: dict) -> dict:
|
||
"""The domain tag of each yoga in `modules.yogas` (BUG-1157).
|
||
|
||
The tag is the Raman packet that governs the yoga's rule id
|
||
(`YOGA_PACKET_DOMAINS`); `hit` is the sieve's own value. A yoga whose rule
|
||
id is in no packet carries no tag.
|
||
"""
|
||
entries = _record(modules.get("yogas")).get("yogas")
|
||
if not isinstance(entries, list):
|
||
return _blocked("yogas_missing")
|
||
tags = _yoga_packet_tags()
|
||
rows: list[dict] = []
|
||
for entry in entries:
|
||
row = _record(entry)
|
||
if not isinstance(row.get("name"), str):
|
||
continue
|
||
out: dict[str, Any] = {"name": row["name"], "hit": bool(row.get("hit"))}
|
||
tag = tags.get(str(row.get("rule_id") or ""))
|
||
if tag:
|
||
out["domain"] = tag
|
||
rows.append(out)
|
||
return {
|
||
"status": "executed",
|
||
"source": "modules.yogas (Raman sieve) + YOGA_PACKET_DOMAINS",
|
||
"rows": rows,
|
||
}
|
||
|
||
|
||
def _yoga_key(entry: dict) -> str:
|
||
name = "".join(ch for ch in str(entry.get("name") or "").lower() if ch.isalnum())
|
||
return name.replace("yoga", "") or name
|
||
|
||
|
||
def merge_detected_yogas(*groups: list) -> list[dict]:
|
||
"""Merge yoga lists in order, dropping repeats by normalised name (BUG-1174)."""
|
||
merged: list[dict] = []
|
||
seen: set[str] = set()
|
||
for group in groups:
|
||
for entry in group or []:
|
||
if not isinstance(entry, dict) or not entry.get("name"):
|
||
continue
|
||
key = _yoga_key(entry)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
merged.append(entry)
|
||
return merged
|
||
|
||
|
||
def collect_consultation_yogas(handler, chart: dict, birth_payload: dict, planets: dict,
|
||
ascendant: Any, compact_entry) -> list[dict]:
|
||
"""Every yoga the consultation sieve sees: the chart's own list plus the rule engine.
|
||
|
||
BUG-1174 (TASK-consult-career-yoga-functional-20261002 T2): the rule engine
|
||
used to run only when `chart.yogas` was empty, so a chart that already
|
||
carried one or two yogas never had the Raman packet members checked. It now
|
||
always runs; the two lists are merged and de-duplicated by name. No
|
||
detection rule or sieve boundary changes.
|
||
"""
|
||
raw = chart.get("yogas") if isinstance(chart.get("yogas"), list) else []
|
||
own = [entry for entry in (compact_entry(item) for item in raw) if entry]
|
||
try:
|
||
api = handler._compute_yogas_api({**birth_payload, "planets": planets, "ascendant": ascendant})
|
||
except Exception:
|
||
if not own:
|
||
raise
|
||
return merge_detected_yogas(own)
|
||
result = api.get("result") if isinstance(api, dict) else {}
|
||
result = result if isinstance(result, dict) else {}
|
||
engine = [
|
||
entry for entry in (
|
||
compact_entry(item)
|
||
for item in (result.get("extended_yogas") or []) + (result.get("rule_engine_yogas") or [])
|
||
)
|
||
if entry
|
||
]
|
||
return merge_detected_yogas(own, engine)
|
||
|
||
|
||
# Closed list for `traditional_yogas` (TASK-consult-career-yoga-functional-20261002
|
||
# D3, BUG-1175): canonical name -> name-token prefixes the detectors emit
|
||
# (yoga_expansion / yoga_engine / chart.yogas / Raman packet hits). Nothing
|
||
# outside this list is shown, and no detector is added here.
|
||
TRADITIONAL_YOGA_LABEL = "传统格局，本站未验证，只作参考"
|
||
TRADITIONAL_YOGAS = (
|
||
("Ruchaka Yoga", ("ruchaka",)),
|
||
("Bhadra Yoga", ("bhadra",)),
|
||
("Hamsa Yoga", ("hamsa",)),
|
||
("Malavya Yoga", ("malavya",)),
|
||
("Sasa Yoga", ("sasa", "shasha")),
|
||
("Gajakesari Yoga", ("gajakesari",)),
|
||
("Neecha Bhanga Raja Yoga", ("neechabhanga", "neechbhanga")),
|
||
# Budha-Aditya: the engine also emits the same Sun-Mercury conjunction as
|
||
# "Surya-Budha Yoga"; both map to one row.
|
||
("Budha-Aditya Yoga", ("budhaaditya", "budhaditya", "suryabudha")),
|
||
)
|
||
|
||
|
||
def _traditional_name(name: Any) -> str | None:
|
||
token = "".join(ch for ch in str(name or "").lower() if ch.isalnum())
|
||
for canonical, prefixes in TRADITIONAL_YOGAS:
|
||
if any(token.startswith(prefix) for prefix in prefixes):
|
||
return canonical
|
||
return None
|
||
|
||
|
||
def build_traditional_yogas(modules: dict) -> dict:
|
||
"""Closed-list classical yogas the detectors found formed (BUG-1175).
|
||
|
||
Reads what the consultation sieve was handed (`raman_support.yogas`: packet
|
||
hits plus `ungoverned_detected`). Only formed yogas are listed; the label is
|
||
fixed. Reading rule (card): strength only, no stand-alone conclusion, never
|
||
counted as affliction or support against affliction. Raman packets,
|
||
`modules.yogas`, confidence caps and claim boundaries are unchanged.
|
||
"""
|
||
sieve = _record(_record(modules.get("raman_support")).get("yogas"))
|
||
if sieve.get("status") != "executed":
|
||
return _blocked("raman_support_yogas_missing")
|
||
found: dict[str, list[str]] = {}
|
||
for entry in list(sieve.get("hits") or []) + list(sieve.get("ungoverned_detected") or []):
|
||
row = _record(entry)
|
||
if row.get("hit") is False:
|
||
continue
|
||
canonical = _traditional_name(row.get("name"))
|
||
if not canonical:
|
||
continue
|
||
planets = found.setdefault(canonical, [])
|
||
for planet in row.get("planets") or []:
|
||
if isinstance(planet, str) and planet not in planets:
|
||
planets.append(planet)
|
||
names = [
|
||
f"{canonical}（{'、'.join(found[canonical])}）" if found[canonical] else canonical
|
||
for canonical, _prefixes in TRADITIONAL_YOGAS
|
||
if canonical in found
|
||
]
|
||
return {
|
||
"status": "executed",
|
||
"source": "raman_support.yogas (hits + ungoverned_detected), closed list TRADITIONAL_YOGAS",
|
||
"label": TRADITIONAL_YOGA_LABEL,
|
||
"names": names,
|
||
}
|
||
|
||
|
||
# ── 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)
|
||
layers["graha_drishti"] = _guard(build_graha_drishti, engine, chart)
|
||
layers["yoga_domains"] = _guard(build_yoga_domains, modules)
|
||
layers["traditional_yogas"] = _guard(build_traditional_yogas, 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
|