feat(reading): fill the five full-reading modules missing from upstream

The western comparison layer was invoked without its files, so it stayed blocked. Navamsha and Niryaana Shoola are filled from the native rashi dashas.

Co-Authored-By: Grok (agent loop)
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jesse-ux
2026-10-18 19:32:38 +08:00
co-authored by Grok (agent loop) Cursor
parent b3ef2d8158
commit 900581068e
11 changed files with 13869 additions and 1 deletions
+14
View File
@@ -0,0 +1,14 @@
"""Put the repository root on sys.path for scripts copied from upstream.
Those modules optionally import ``_script_root`` before their other imports.
This file only adjusts the import path. It does not calculate charts.
"""
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
root_text = str(ROOT)
if root_text not in sys.path:
sys.path.insert(0, root_text)
File diff suppressed because it is too large Load Diff
+81
View File
@@ -435,6 +435,52 @@ def _calc_pushkara_flags(planets):
return result
def _calc_lagna_vargottamamsa(report, varga_full):
chart = report.get('modules', {}).get('chart', {}) if isinstance(report, dict) else {}
ascendant = chart.get('ascendant', {}) if isinstance(chart, dict) else {}
d1_sign = ascendant.get('sign') if isinstance(ascendant, dict) else None
d9 = varga_full.get('D9_Navamsa', {}) if isinstance(varga_full, dict) else {}
d9_ascendant = d9.get('ascendant') if isinstance(d9, dict) else None
if not isinstance(d9_ascendant, dict) and isinstance(d9, dict):
d9_ascendant = d9.get('Ascendant', {})
d9_sign = d9_ascendant.get('sign') if isinstance(d9_ascendant, dict) else None
return {
'd1_sign': d1_sign,
'd9_sign': d9_sign,
'is_lagna_vargottamamsa': bool(d1_sign and d9_sign and d1_sign == d9_sign),
'source': 'raman_horoscope_v1_bounded_pp64',
}
def _summarize_pushkara_planets(flags):
"""Return the compact full-reading list without discarding raw flags."""
return [
f"{planet} ({data['sign']} {data['degree_in_sign']:.1f}°)"
for planet, data in flags.items()
if data.get('pushkara_navamsa') or data.get('pushkara_bhaga')
]
def _public_secondary_dasha_module(native):
"""Expose one native rashi dasha under the upstream full-reading key.
The period table and profile id stay ours. The outer status is what the
upstream synastry validator reads, and it is never a bare error object.
"""
if not isinstance(native, dict):
return {
'status': 'blocked',
'reason': 'native_rashi_dasha_missing',
'periods': [],
}
return {
'status': 'parameter_sensitive',
'profile_id': native.get('profile_id'),
'periods': list(native.get('periods') or []),
'native_result': native,
}
def _calc_sensitive_points(planets):
result = {}
try:
@@ -15551,7 +15597,9 @@ def cmd_full_reading(args):
report['errors'].append(f"yoga-context-injection: {yoga_ctx_e}")
report['modules']['vargottama'] = _calc_vargottama(planets, varga_result)
report['modules']['lagna_vargottamamsa'] = _calc_lagna_vargottamamsa(report, varga_result)
report['modules']['pushkara'] = _calc_pushkara_flags(planets)
report['modules']['pushkara_planets'] = _summarize_pushkara_planets(report['modules']['pushkara'])
report['modules']['sensitive_points'] = _calc_sensitive_points(planets)
except Exception as e:
report['errors'].append(f"varga-full: {e}")
@@ -16036,6 +16084,39 @@ def cmd_full_reading(args):
'status': 'blocked',
'reason': 'reference_date_missing',
}
try:
from punarphoo import detect_punarphoo
report['modules']['punarphoo'] = detect_punarphoo(planets)
except Exception as e:
report['modules']['punarphoo'] = {
'status': 'blocked',
'reason': f'punarphoo_producer_failed:{e}',
}
try:
from rashi_dashas_native import calculate_rashi_dashas
secondary_dashas = calculate_rashi_dashas(
_birth_datetime_from_args(args),
float(asc_deg),
planet_lons,
)
report['modules']['navamsha_dasha'] = _public_secondary_dasha_module(secondary_dashas.get('navamsha'))
report['modules']['niryaana_shoola_dasha'] = _public_secondary_dasha_module(
secondary_dashas.get('niryaana_shoola')
)
except Exception as e:
report['modules']['navamsha_dasha'] = {
'status': 'blocked',
'reason': f'navamsha_dasha_producer_failed:{e}',
'periods': [],
}
report['modules']['niryaana_shoola_dasha'] = {
'status': 'blocked',
'reason': f'niryaana_shoola_dasha_producer_failed:{e}',
'periods': [],
}
_record_stage_timing(
stage_timings,
'dasha_and_core_varga_stack',
+268
View File
@@ -0,0 +1,268 @@
"""Auditable orchestration for the native tropical Western calculation layers."""
from __future__ import annotations
try: # reached as a script: this directory is on the path, the root is not
import _script_root # noqa: F401
except ModuleNotFoundError: # pragma: no cover - reached as scripts.x instead
try:
from scripts import _script_root # noqa: F401
except ModuleNotFoundError:
pass
import argparse
import json
from datetime import date
from typing import Any, Callable
try:
import swisseph as swe
except ImportError: # pragma: no cover - optional runtime dependency
swe = None
try:
from scripts.classical_astrology import build_classical_astrology_packet, build_classical_usable_loop_audit_packet
from scripts.western_chart_engine import build_tropical_natal_chart
from scripts.western_chart_engine import build_draconic_chart, build_relocation_chart
from scripts.western_timing_engine import (
calculate_annual_profection,
calculate_converse_secondary_progressions,
calculate_converse_solar_arc_directions,
calculate_lunar_return,
calculate_lunar_return_series,
calculate_midpoints,
calculate_parans_status,
calculate_secondary_progressions,
calculate_solar_arc_directions,
calculate_solar_return,
calculate_tertiary_progressed_moon,
calculate_transit_duration_scan,
calculate_transit_to_natal,
calculate_zodiacal_release,
)
except ModuleNotFoundError: # pragma: no cover - direct script execution
from classical_astrology import build_classical_astrology_packet, build_classical_usable_loop_audit_packet
from western_chart_engine import build_tropical_natal_chart
from western_chart_engine import build_draconic_chart, build_relocation_chart
from western_timing_engine import (
calculate_annual_profection,
calculate_converse_secondary_progressions,
calculate_converse_solar_arc_directions,
calculate_lunar_return,
calculate_lunar_return_series,
calculate_midpoints,
calculate_parans_status,
calculate_secondary_progressions,
calculate_solar_arc_directions,
calculate_solar_return,
calculate_tertiary_progressed_moon,
calculate_transit_duration_scan,
calculate_transit_to_natal,
calculate_zodiacal_release,
)
def _executed(inputs: dict[str, Any], calculation: Callable[[], dict[str, Any]]) -> dict[str, Any]:
try:
return {"status": "executed", "input": inputs, "raw_result": calculation()}
except Exception as exc: # pragma: no cover - defensive orchestration boundary
return {
"status": "blocked",
"input": inputs,
"reason": f"{exc.__class__.__name__}: {exc}",
}
def _blocked(reason: str, inputs: dict[str, Any] | None = None) -> dict[str, Any]:
result: dict[str, Any] = {"status": "blocked", "reason": reason}
if inputs:
result["input"] = inputs
return result
def _reference_layer(
reference_date: str | None,
birth: dict[str, Any],
calculation: Callable[[str], dict[str, Any]],
) -> dict[str, Any]:
if not reference_date:
return _blocked("reference_date is required")
return _executed({"reference_date": reference_date}, lambda: calculation(reference_date))
def _overall_status(layers: dict[str, dict[str, Any]]) -> str:
return "complete" if all(layer["status"] == "executed" for layer in layers.values()) else "partial"
def _eclipse_events(reference_date: str) -> dict[str, Any]:
if swe is None:
raise RuntimeError("swisseph is unavailable")
start = date.fromisoformat(reference_date)
jd = swe.julday(start.year, start.month, start.day, 0.0)
events = []
for kind, finder in (("solar", swe.sol_eclipse_when_glob), ("lunar", swe.lun_eclipse_when)):
flags, times = finder(jd)
year, month, day, hour = swe.revjul(times[0])
events.append({"kind": kind, "maximum_utc": f"{year:04d}-{month:02d}-{day:02d}T{int(hour):02d}:{int((hour % 1) * 60):02d}:00Z", "swiss_event_flags": int(flags)})
return {"technique": "eclipse_triggers", "reference_date": reference_date, "events": events, "boundary": "Geometric eclipse triggers only; no predictive interpretation is inferred."}
def build_unified_western_reading(
*,
reference_date: str | None = None,
duration_scan_end_date: str | None = None,
lunar_return_months: int = 1,
relocation_latitude: float | None = None,
relocation_longitude: float | None = None,
relocation_timezone: str | float | int | None = None,
zodiacal_release_lot: str = "spirit",
zodiacal_release_levels: int = 2,
include_classical_usable_loop: bool = False,
classical_event_window_hours: int = 4,
classical_event_step_hours: int = 2,
classical_blind_scoring_manifest: str | None = None,
classical_reviewer_manifest: str | None = None,
**birth: Any,
) -> dict[str, Any]:
"""Run every available native Western layer and preserve raw evidence.
``reference_date`` drives the one-day timing techniques. A duration scan
additionally needs an explicit end date, so a caller cannot unknowingly
request an unbounded scan. Eclipse geometry uses the local Swiss Ephemeris
calculation and does not infer an event interpretation.
"""
layers: dict[str, dict[str, Any]] = {
"natal": _executed(dict(birth), lambda: build_tropical_natal_chart(**birth)),
"classical": _executed(
{"reference_date": reference_date},
lambda: build_classical_astrology_packet(reference_date=reference_date, **birth),
),
"draconic": _executed(dict(birth), lambda: build_draconic_chart(**birth)),
"transits": _reference_layer(
reference_date, birth, lambda target: calculate_transit_to_natal(target_date=target, **birth)
),
"solar_return": _reference_layer(
reference_date,
birth,
lambda target: calculate_solar_return(target_year=date.fromisoformat(target).year, **birth),
),
"secondary_progressions": _reference_layer(
reference_date, birth, lambda target: calculate_secondary_progressions(target_date=target, **birth)
),
"tertiary_moon": _reference_layer(
reference_date, birth, lambda target: calculate_tertiary_progressed_moon(target_date=target, **birth)
),
"solar_arc": _reference_layer(
reference_date, birth, lambda target: calculate_solar_arc_directions(target_date=target, **birth)
),
"midpoints": _reference_layer(
reference_date, birth, lambda target: calculate_midpoints(target_date=target, **birth)
),
"lunar_return": _reference_layer(
reference_date, birth, lambda target: calculate_lunar_return_series(
target_date=target, months=lunar_return_months, **birth
)
),
"parans": _reference_layer(
reference_date, birth, lambda target: calculate_parans_status(target_date=target, **birth)
),
"profection": _reference_layer(
reference_date, birth, lambda target: calculate_annual_profection(target_date=target, **birth)
),
"zodiacal_release": _reference_layer(
reference_date,
birth,
lambda target: calculate_zodiacal_release(
target_date=target,
lot=zodiacal_release_lot,
levels=int(zodiacal_release_levels),
**birth,
),
),
}
if relocation_latitude is None or relocation_longitude is None or relocation_timezone is None:
layers["relocation"] = _blocked(
"relocation_latitude, relocation_longitude, and relocation_timezone are required"
)
else:
layers["relocation"] = _executed(
{
"relocation_latitude": relocation_latitude,
"relocation_longitude": relocation_longitude,
"relocation_timezone": relocation_timezone,
},
lambda: build_relocation_chart(
relocation_latitude=float(relocation_latitude),
relocation_longitude=float(relocation_longitude),
relocation_timezone=relocation_timezone,
**birth,
),
)
if reference_date:
layers["converse"] = _executed(
{"reference_date": reference_date},
lambda: {
"secondary_progressions": calculate_converse_secondary_progressions(
target_date=reference_date, **birth
),
"solar_arc_directions": calculate_converse_solar_arc_directions(
target_date=reference_date, **birth
),
},
)
else:
layers["converse"] = _blocked("reference_date is required")
if not reference_date:
layers["transit_duration"] = _blocked("reference_date is required")
elif not duration_scan_end_date:
layers["transit_duration"] = _blocked("duration_scan_end_date is required")
else:
layers["transit_duration"] = _executed(
{"start_date": reference_date, "end_date": duration_scan_end_date},
lambda: calculate_transit_duration_scan(
start_date=reference_date, end_date=duration_scan_end_date, **birth
),
)
layers["eclipses"] = _reference_layer(reference_date, birth, _eclipse_events)
if include_classical_usable_loop:
layers["classical_usable_loop"] = _executed(
{
"reference_date": reference_date,
"event_window_hours": classical_event_window_hours,
"event_step_hours": classical_event_step_hours,
"blind_scoring_manifest": classical_blind_scoring_manifest,
"reviewer_manifest": classical_reviewer_manifest,
},
lambda: build_classical_usable_loop_audit_packet(
reference_date=reference_date,
event_window_hours=int(classical_event_window_hours),
event_step_hours=int(classical_event_step_hours),
blind_scoring_manifest_path=classical_blind_scoring_manifest,
reviewer_manifest_path=classical_reviewer_manifest,
**birth,
),
)
return {
"system": "western_astrology",
"status": _overall_status(layers),
"reference_date": reference_date,
"layers": layers,
"boundary": "Raw native calculations only; this entrypoint makes no predictive interpretation.",
}
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Run the unified native Western calculation layers.")
parser.add_argument("--input-json", required=True, help="Structured birth and timing input as JSON.")
args = parser.parse_args(argv)
payload = json.loads(args.input_json)
if not isinstance(payload, dict):
raise ValueError("--input-json must decode to an object")
print(json.dumps(build_unified_western_reading(**payload), ensure_ascii=False, default=str))
return 0
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())
+109
View File
@@ -153,6 +153,21 @@ def _birth_zone(value: str | float | int):
return fixed_timezone(timedelta(hours=offset)), f"UTC{offset:+g}"
def _birth_times(
*,
year: int,
month: int,
day: int,
hour: int,
minute: int,
second: int,
timezone: str | float | int,
) -> tuple[datetime, datetime, str]:
zone, timezone_label = _birth_zone(timezone)
local = datetime(year, month, day, hour, minute, second, tzinfo=zone)
utc = local.astimezone(ZoneInfo("UTC"))
return local, utc, timezone_label
def build_tropical_natal_chart(
*,
year: int,
@@ -217,6 +232,100 @@ def build_tropical_natal_chart(
}
def build_draconic_chart(**birth: Any) -> dict[str, Any]:
"""Derive a draconic chart by rotating tropical longitudes to the true node Aries point."""
natal_chart = build_tropical_natal_chart(**birth)
natal = natal_chart["natal"]
node_longitude = natal["planets"]["true_node"]["longitude"]
def _rotate_point(point: dict[str, Any]) -> dict[str, Any]:
rotated = _point(point["longitude"] - node_longitude)
if "house" in point:
rotated["house"] = point["house"]
if "speed_longitude" in point:
rotated["speed_longitude"] = point["speed_longitude"]
rotated["retrograde"] = point.get("retrograde", False)
return rotated
rotated_planets = {name: _rotate_point(point) for name, point in natal["planets"].items()}
rotated_angles = {name: _rotate_point(point) for name, point in natal["angles"].items()}
rotated_houses = [
{"house": row["house"], "cusp": _rotate_point(row["cusp"])}
for row in natal["houses"]
]
aspect_points = {**rotated_planets, "ascendant": rotated_angles["ascendant"], "mc": rotated_angles["mc"]}
return {
"technique": "draconic_chart",
"status": "used",
"source_engine": natal_chart["source_engine"],
"base_zodiac": natal_chart["zodiac"],
"derived_zodiac": "draconic",
"node_basis": "true_node",
"node_longitude": node_longitude,
"draconic": {
"ascendant": rotated_angles["ascendant"],
"mc": rotated_angles["mc"],
"angles": rotated_angles,
"planets": rotated_planets,
"houses": rotated_houses,
"aspects": _aspects(aspect_points),
"distribution": _distribution(rotated_planets),
"house_ruler_chains": _ruler_chains(
[row["cusp"]["longitude"] for row in rotated_houses],
rotated_planets,
),
},
"boundary": "Longitude rotation to the true-node Aries point only; interpretation and timing remain separate audited layers.",
}
def build_relocation_chart(
*,
relocation_latitude: float,
relocation_longitude: float,
relocation_timezone: str | float | int,
house_system: str = "P",
**birth: Any,
) -> dict[str, Any]:
"""Calculate a relocated chart for the same UTC birth instant at a new location."""
local, utc, _ = _birth_times(
year=int(birth["year"]),
month=int(birth["month"]),
day=int(birth["day"]),
hour=int(birth["hour"]),
minute=int(birth["minute"]),
second=int(birth.get("second", 0)),
timezone=birth["timezone"],
)
relocation_zone, relocation_label = _birth_zone(relocation_timezone)
relocation_local = utc.astimezone(relocation_zone)
relocated_chart = build_tropical_natal_chart(
year=relocation_local.year,
month=relocation_local.month,
day=relocation_local.day,
hour=relocation_local.hour,
minute=relocation_local.minute,
second=relocation_local.second,
latitude=float(relocation_latitude),
longitude=float(relocation_longitude),
timezone=relocation_timezone,
house_system=house_system,
)
return {
"technique": "relocation_chart",
"status": "used",
"source_engine": relocated_chart["source_engine"],
"same_birth_utc": utc.isoformat(),
"source_local_time": local.isoformat(),
"relocation_local_time": relocation_local.isoformat(),
"relocation_timezone": relocation_label,
"relocation_coordinates": {
"latitude": float(relocation_latitude),
"longitude": float(relocation_longitude),
},
"relocated_chart": relocated_chart,
"boundary": "Same UTC birth instant with relocated coordinates and timezone; interpretation and relocation outcome judgment remain separate audited layers.",
}
def _request_field(body: dict[str, Any], *keys: str) -> Any:
for key in keys:
if key in body and body[key] not in (None, ""):
+19
View File
@@ -16,6 +16,25 @@ except ImportError:
DEFAULT_MANIFEST = Path("references/real_case_calibration/replay_manifest.json")
def _build_calibration_summary(primary_theme: str = "general") -> dict:
"""Import shim for classical_astrology.
Upstream's scoring body is not in this tree. The caller only reads these
keys and treats a missing gate as blocked, so full-reading can import the
classical packet without inventing a specificity score.
"""
return {
"status": "blocked",
"reason": "specificity_gate_not_ported",
"primary_theme": primary_theme,
"false_positive_audit": {"frozen_negative_holdout": {"status": "blocked"}},
"positive_replay": {},
"positive_holdout": {},
"domain_replay_scoring": {},
"review_pack": {},
}
def build_case(case_id: str, manifest_path: Path = DEFAULT_MANIFEST) -> dict:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
case = next((row for row in manifest["cases"] if row["case_id"] == case_id), None)