diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index aa606941..4e3e88e7 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -45,6 +45,12 @@ try: from scripts.western_oracle_adapter import build_packet_from_oracle_payload except ModuleNotFoundError: # pragma: no cover - script execution path from western_oracle_adapter import build_packet_from_oracle_payload +try: + from scripts.western_chart_engine import build_tropical_western_evidence_packet + from scripts.western_timing_engine import build_timing_techniques +except ModuleNotFoundError: # pragma: no cover - script execution path + from western_chart_engine import build_tropical_western_evidence_packet + from western_timing_engine import build_timing_techniques load_local_env(REPO_ROOT) _LOCAL_MODULE_CACHE = {} @@ -53,27 +59,84 @@ _HIGH_RIGOR_JOB_SCOPE = 'high_rigor_workflow' _UNIFIED_CONSULTATION_ORCHESTRATOR = UnifiedConsultationOrchestrator() -def _western_evidence_packet_from_body(body: dict, route_packet: dict) -> dict | None: +def _western_evidence_packet_from_body( + body: dict, + route_packet: dict, + *, + birth_payload: dict | None = None, +) -> dict | None: explicit_packet = body.get('western_evidence_packet') if isinstance(explicit_packet, dict): return explicit_packet oracle_payload = body.get('western_oracle_payload') or body.get('western_astrology_oracle') - if not isinstance(oracle_payload, dict): + if isinstance(oracle_payload, dict): + try: + return build_packet_from_oracle_payload(oracle_payload, route_packet=route_packet) + except Exception as exc: # pragma: no cover - defensive contract boundary + return { + 'system': 'western_astrology', + 'status': 'blocked', + 'route': dict(route_packet), + 'signals': [], + 'missing_sections': ['western_oracle_payload'], + 'adapter_error': exc.__class__.__name__, + 'boundary': 'Western oracle payload was supplied but could not be normalized.', + } + automatic = body.get('western_mode', body.get('western_auto_compute', 'auto')) + if automatic in {False, 'off', 'external_only'} or body.get('entry_mode') == 'prashna' or not isinstance(birth_payload, dict): return None try: - return build_packet_from_oracle_payload(oracle_payload, route_packet=route_packet) - except Exception as exc: # pragma: no cover - defensive contract boundary + packet = build_tropical_western_evidence_packet( + route_packet=route_packet, + 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=int(birth_payload.get('second', 0)), + latitude=float(birth_payload['lat']), longitude=float(birth_payload['lon']), + timezone=body.get('western_timezone') or birth_payload['tz'], + house_system=str(body.get('western_house_system', 'P')), + ) + timing_request = body.get('western_timing') + if isinstance(timing_request, dict): + birth = { + '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': int(birth_payload.get('second', 0)), + 'latitude': float(birth_payload['lat']), 'longitude': float(birth_payload['lon']), + 'timezone': body.get('western_timezone') or birth_payload['tz'], + 'house_system': str(body.get('western_house_system', 'P')), + } + timing = build_timing_techniques( + **birth, + transit_date=timing_request.get('transit_date'), + solar_return_year=timing_request.get('solar_return_year'), + secondary_progression_date=timing_request.get('secondary_progression_date'), + solar_arc_date=timing_request.get('solar_arc_date'), + converse_secondary_progression_date=timing_request.get('converse_secondary_progression_date'), + converse_solar_arc_date=timing_request.get('converse_solar_arc_date'), + midpoint_date=timing_request.get('midpoint_date'), + lunar_return_start_date=timing_request.get('lunar_return_start_date'), + duration_scan_start_date=timing_request.get('duration_scan_start_date'), + duration_scan_end_date=timing_request.get('duration_scan_end_date'), + parans_date=timing_request.get('parans_date'), + ) + if timing: + packet['timing_techniques'] = timing + packet['sections']['timing_techniques'] = {'status': 'used', 'source_path': 'western.native_timing'} + packet['missing_sections'] = [item for item in packet['missing_sections'] if item != 'timing_techniques'] + packet['boundary'] = ( + 'Native calculations include only explicitly requested transit, solar-return, secondary-progression, ' + 'solar-arc, midpoint, lunar-return and daily duration-scan layers; parans remain blocked until a ' + 'dedicated latitude-aware event solver is implemented. Outputs do not infer outcomes or interpretation.' + ) + return packet + except Exception as exc: # pragma: no cover - defensive boundary return { 'system': 'western_astrology', 'status': 'blocked', 'route': dict(route_packet), 'signals': [], - 'missing_sections': ['western_oracle_payload'], + 'missing_sections': ['native_tropical_calculation'], 'adapter_error': exc.__class__.__name__, - 'boundary': 'Western oracle payload was supplied but could not be normalized.', + 'boundary': 'Native Western natal calculation could not be materialized.', } - - def _consultation_reference_date(body: dict) -> datetime: raw = ( body.get('reference_date') diff --git a/scripts/skill_release_package.py b/scripts/skill_release_package.py index e69065fd..e1bc14bf 100644 --- a/scripts/skill_release_package.py +++ b/scripts/skill_release_package.py @@ -110,6 +110,8 @@ REQUIRED_CONTRACTS = [ "references/oracle/western_oracle_adapter_contract.md", "scripts/user_invocation_acceptance_check.py", "scripts/diagnose_external_engine_adapters.py", + "scripts/western_chart_engine.py", + "scripts/western_timing_engine.py", ] diff --git a/scripts/western_chart_engine.py b/scripts/western_chart_engine.py new file mode 100644 index 00000000..11bbce20 --- /dev/null +++ b/scripts/western_chart_engine.py @@ -0,0 +1,258 @@ +#!/usr/bin/env python3 +"""Native, auditable tropical Western natal-chart calculation. + +This module deliberately uses the project's existing Swiss Ephemeris binding +instead of bundling an AGPL Western astrology library. It is a calculation +layer only: transits, progressions, solar arcs, returns, and interpretation +remain separate evidence layers. +""" + +from __future__ import annotations + +import argparse +import json +from datetime import datetime, timedelta, timezone as fixed_timezone +from typing import Any +from zoneinfo import ZoneInfo + +import swisseph as swe + +try: + from western_evidence_packet import build_western_evidence_packet +except ImportError: # pragma: no cover - package import path + from scripts.western_evidence_packet import build_western_evidence_packet + + +_PLANETS = { + "sun": swe.SUN, + "moon": swe.MOON, + "mercury": swe.MERCURY, + "venus": swe.VENUS, + "mars": swe.MARS, + "jupiter": swe.JUPITER, + "saturn": swe.SATURN, + "uranus": swe.URANUS, + "neptune": swe.NEPTUNE, + "pluto": swe.PLUTO, + "true_node": swe.TRUE_NODE, +} +_SIGNS = ( + "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", + "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces", +) +_ELEMENTS = { + "Aries": "fire", "Leo": "fire", "Sagittarius": "fire", + "Taurus": "earth", "Virgo": "earth", "Capricorn": "earth", + "Gemini": "air", "Libra": "air", "Aquarius": "air", + "Cancer": "water", "Scorpio": "water", "Pisces": "water", +} +_MODES = { + "Aries": "cardinal", "Cancer": "cardinal", "Libra": "cardinal", "Capricorn": "cardinal", + "Taurus": "fixed", "Leo": "fixed", "Scorpio": "fixed", "Aquarius": "fixed", + "Gemini": "mutable", "Virgo": "mutable", "Sagittarius": "mutable", "Pisces": "mutable", +} +_RULERS = { + "Aries": "mars", "Taurus": "venus", "Gemini": "mercury", "Cancer": "moon", + "Leo": "sun", "Virgo": "mercury", "Libra": "venus", "Scorpio": "mars", + "Sagittarius": "jupiter", "Capricorn": "saturn", "Aquarius": "saturn", "Pisces": "jupiter", +} +_ASPECTS = {"conjunction": 0.0, "sextile": 60.0, "square": 90.0, "trine": 120.0, "opposition": 180.0} +_ORB = {"sun": 8.0, "moon": 8.0, "ascendant": 5.0, "mc": 5.0} + + +def _longitude(value: float) -> float: + return float(value) % 360.0 + + +def _point(longitude: float, *, house: int | None = None, speed: float | None = None) -> dict[str, Any]: + longitude = _longitude(longitude) + point = { + "longitude": round(longitude, 6), + "sign": _SIGNS[int(longitude // 30)], + "degree_in_sign": round(longitude % 30, 6), + } + if house is not None: + point["house"] = house + if speed is not None: + point["speed_longitude"] = round(float(speed), 8) + point["retrograde"] = bool(speed < 0) + return point + + +def _house_for_longitude(longitude: float, cusps: list[float]) -> int: + """Return Placidus house by testing each cusp-to-next-cusp circular arc.""" + longitude = _longitude(longitude) + for index, cusp in enumerate(cusps): + start = _longitude(cusp) + end = _longitude(cusps[(index + 1) % 12]) + span = (end - start) % 360.0 + if (longitude - start) % 360.0 < span: + return index + 1 + raise RuntimeError("Unable to assign longitude to a house") # pragma: no cover + + +def _orb_for(left: str, right: str) -> float: + return min(_ORB.get(left, 6.0), _ORB.get(right, 6.0)) + + +def _aspects(points: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: + names = list(points) + found: list[dict[str, Any]] = [] + for index, left in enumerate(names): + for right in names[index + 1:]: + separation = abs(points[left]["longitude"] - points[right]["longitude"]) + separation = min(separation, 360.0 - separation) + allowed_orb = _orb_for(left, right) + for aspect, exact in _ASPECTS.items(): + orb = abs(separation - exact) + if orb <= allowed_orb: + found.append({ + "left": left, + "right": right, + "aspect": aspect, + "exact_degrees": exact, + "separation": round(separation, 6), + "orb": round(orb, 6), + "allowed_orb": allowed_orb, + }) + return sorted(found, key=lambda row: (row["orb"], row["left"], row["right"])) + + +def _distribution(planets: dict[str, dict[str, Any]]) -> dict[str, dict[str, int]]: + elements = {name: 0 for name in ("fire", "earth", "air", "water")} + modes = {name: 0 for name in ("cardinal", "fixed", "mutable")} + for planet in planets.values(): + elements[_ELEMENTS[planet["sign"]]] += 1 + modes[_MODES[planet["sign"]]] += 1 + return {"elements": elements, "modes": modes} + + +def _ruler_chains(cusps: list[float], planets: dict[str, dict[str, Any]]) -> dict[str, list[str]]: + chains: dict[str, list[str]] = {} + for house, cusp in enumerate(cusps, start=1): + sign = _SIGNS[int(_longitude(cusp) // 30)] + chain: list[str] = [] + current = _RULERS[sign] + for _ in range(12): + if current in chain: + break + chain.append(current) + current = _RULERS[planets[current]["sign"]] + chains[str(house)] = chain + return chains + + +def _birth_zone(value: str | float | int): + if isinstance(value, str): + return ZoneInfo(value), value + offset = float(value) + return fixed_timezone(timedelta(hours=offset)), f"UTC{offset:+g}" + + +def build_tropical_natal_chart( + *, + year: int, + month: int, + day: int, + hour: int, + minute: int, + latitude: float, + longitude: float, + timezone: str | float | int, + second: int = 0, + house_system: str = "P", +) -> dict[str, Any]: + """Calculate a tropical natal chart from local birth data using Swiss Ephemeris.""" + if len(house_system) != 1: + raise ValueError("house_system must be a single Swiss Ephemeris house-system letter") + zone, timezone_label = _birth_zone(timezone) + local = datetime(year, month, day, hour, minute, second, tzinfo=zone) + utc = local.astimezone(ZoneInfo("UTC")) + jd_ut = swe.julday(utc.year, utc.month, utc.day, utc.hour + utc.minute / 60 + utc.second / 3600) + flags = swe.FLG_SWIEPH | swe.FLG_SPEED + cusps_raw, ascmc = swe.houses_ex(jd_ut, float(latitude), float(longitude), house_system.encode("ascii"), 0) + cusps = [_longitude(cusp) for cusp in cusps_raw] + planets: dict[str, dict[str, Any]] = {} + for name, planet_id in _PLANETS.items(): + values, _ = swe.calc_ut(jd_ut, planet_id, flags) + position = _point(values[0], house=_house_for_longitude(values[0], cusps), speed=values[3]) + planets[name] = position + angles = { + "ascendant": _point(ascmc[0]), + "mc": _point(ascmc[1]), + "descendant": _point(ascmc[0] + 180.0), + "ic": _point(ascmc[1] + 180.0), + } + aspect_points = {**planets, "ascendant": angles["ascendant"], "mc": angles["mc"]} + natal = { + "ascendant": angles["ascendant"], + "mc": angles["mc"], + "angles": angles, + "planets": planets, + "houses": [{"house": index + 1, "cusp": _point(cusp)} for index, cusp in enumerate(cusps)], + "aspects": _aspects(aspect_points), + "distribution": _distribution(planets), + "house_ruler_chains": _ruler_chains(cusps, planets), + } + return { + "source_engine": "pyswisseph_tropical", + "engine_version": getattr(swe, "version", "unknown"), + "zodiac": "tropical", + "house_system": house_system.upper(), + "calculation_contract": { + "birth_timezone": timezone_label, + "local_birth_time": local.isoformat(), + "utc_birth_time": utc.isoformat(), + "julian_day_ut": round(jd_ut, 8), + "latitude": float(latitude), + "longitude": float(longitude), + "ephemeris": "Swiss Ephemeris via pyswisseph", + }, + "natal": natal, + "boundary": "Natal tropical calculation only; it does not calculate transits, progressions, solar arcs, returns, or interpretation.", + } + + +def build_tropical_western_evidence_packet(*, route_packet: dict[str, Any], **birth: Any) -> dict[str, Any]: + """Wrap direct natal calculation in the existing cross-system packet contract.""" + chart = build_tropical_natal_chart(**birth) + packet = build_western_evidence_packet( + route_packet=route_packet, + natal=chart["natal"], + timing_techniques={}, + signals=[], + ) + packet.update({ + "source_engine": chart["source_engine"], + "calculation": { + "status": "used", + "source_engine": chart["source_engine"], + "zodiac": chart["zodiac"], + "house_system": chart["house_system"], + "contract": chart["calculation_contract"], + }, + "native_chart": chart, + "boundary": chart["boundary"], + }) + return packet + + +def main() -> int: + parser = argparse.ArgumentParser(description="Calculate an auditable tropical Western natal chart.") + for name, kind in (("year", int), ("month", int), ("day", int), ("hour", int), ("minute", int)): + parser.add_argument(f"--{name}", required=True, type=kind) + parser.add_argument("--lat", required=True, type=float) + parser.add_argument("--lon", required=True, type=float) + parser.add_argument("--timezone", required=True) + parser.add_argument("--second", type=int, default=0) + parser.add_argument("--house-system", default="P") + args = parser.parse_args() + print(json.dumps(build_tropical_natal_chart( + year=args.year, month=args.month, day=args.day, hour=args.hour, minute=args.minute, second=args.second, + latitude=args.lat, longitude=args.lon, timezone=args.timezone, house_system=args.house_system, + ), ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/scripts/western_timing_engine.py b/scripts/western_timing_engine.py new file mode 100644 index 00000000..82dc0e48 --- /dev/null +++ b/scripts/western_timing_engine.py @@ -0,0 +1,449 @@ +#!/usr/bin/env python3 +"""Auditable tropical transit and solar-return evidence calculations.""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from typing import Any +from zoneinfo import ZoneInfo + +import swisseph as swe + +try: + from western_chart_engine import _ASPECTS, _PLANETS, _birth_zone, _longitude, _orb_for, _point, build_tropical_natal_chart +except ImportError: # pragma: no cover - package import path + from scripts.western_chart_engine import _ASPECTS, _PLANETS, _birth_zone, _longitude, _orb_for, _point, build_tropical_natal_chart + + +def _target_jd(target_date: str, timezone: str | float | int) -> tuple[float, datetime]: + zone, _ = _birth_zone(timezone) + local = datetime.fromisoformat(target_date).replace(tzinfo=zone) + utc = local.astimezone(ZoneInfo("UTC")) + jd = swe.julday(utc.year, utc.month, utc.day, utc.hour + utc.minute / 60 + utc.second / 3600) + return jd, local + + +def _cross_aspects(transits: dict[str, dict[str, Any]], natal: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: + matches: list[dict[str, Any]] = [] + for transit_name, transit in transits.items(): + for natal_name, point in natal.items(): + separation = abs(transit["longitude"] - point["longitude"]) + separation = min(separation, 360.0 - separation) + allowed_orb = _orb_for(transit_name, natal_name) + for aspect, exact in _ASPECTS.items(): + orb = abs(separation - exact) + if orb <= allowed_orb: + matches.append({ + "transit_planet": transit_name, + "natal_point": natal_name, + "aspect": aspect, + "exact_degrees": exact, + "separation": round(separation, 6), + "orb": round(orb, 6), + "allowed_orb": allowed_orb, + }) + return sorted(matches, key=lambda row: (row["orb"], row["transit_planet"], row["natal_point"])) + + +def calculate_transit_to_natal(*, target_date: str, **birth: Any) -> dict[str, Any]: + """Calculate major tropical transits to natal planets and ASC/MC on a local date.""" + natal_chart = build_tropical_natal_chart(**birth) + jd, local = _target_jd(target_date, birth["timezone"]) + flags = swe.FLG_SWIEPH | swe.FLG_SPEED + planets: dict[str, dict[str, Any]] = {} + for name, planet_id in _PLANETS.items(): + values, _ = swe.calc_ut(jd, planet_id, flags) + planets[name] = _point(values[0], speed=values[3]) + natal_points = { + **natal_chart["natal"]["planets"], + "ascendant": natal_chart["natal"]["angles"]["ascendant"], + "mc": natal_chart["natal"]["angles"]["mc"], + } + return { + "technique": "transits", + "status": "used", + "target_date": target_date, + "target_local_time": local.isoformat(), + "zodiac": "tropical", + "transit_planets": planets, + "aspects": _cross_aspects(planets, natal_points), + "orb_policy": "major aspects 0/60/90/120/180; min(per-point configured orb)", + "boundary": "A dated transit snapshot only; no duration, outcome, or interpretation is inferred.", + } + + +def _jd_to_local(jd_ut: float, timezone: str | float | int) -> datetime: + zone, _ = _birth_zone(timezone) + year, month, day, hour_float = swe.revjul(jd_ut, swe.GREG_CAL) + utc = datetime(year, month, day, tzinfo=ZoneInfo("UTC")) + timedelta(hours=hour_float) + return utc.astimezone(zone) + + +def calculate_solar_return(*, target_year: int, **birth: Any) -> dict[str, Any]: + """Find the exact tropical solar return and calculate its local return chart.""" + natal_chart = build_tropical_natal_chart(**birth) + natal_sun = natal_chart["natal"]["planets"]["sun"]["longitude"] + start_jd = swe.julday(int(target_year), 1, 1, 0.0) + return_jd = swe.solcross_ut(natal_sun, start_jd, swe.FLG_SWIEPH) + return_local = _jd_to_local(return_jd, birth["timezone"]) + return_birth = { + **birth, + "year": return_local.year, + "month": return_local.month, + "day": return_local.day, + "hour": return_local.hour, + "minute": return_local.minute, + "second": return_local.second, + } + return_chart = build_tropical_natal_chart(**return_birth) + returned_sun = return_chart["natal"]["planets"]["sun"]["longitude"] + delta = abs(_longitude(returned_sun - natal_sun)) + delta = min(delta, 360.0 - delta) + return { + "technique": "solar_return", + "status": "used", + "target_year": int(target_year), + "return_julian_day_ut": round(return_jd, 8), + "return_local_time": return_local.isoformat(), + "natal_sun_longitude": natal_sun, + "return_sun_longitude": returned_sun, + "sun_longitude_delta": round(delta, 8), + "return_chart": return_chart, + "boundary": "Exact solar return time and chart only; annual topics require separate audited interpretation.", + } + + +def _birth_jd(**birth: Any) -> float: + zone, _ = _birth_zone(birth["timezone"]) + local = datetime( + int(birth["year"]), int(birth["month"]), int(birth["day"]), + int(birth["hour"]), int(birth["minute"]), int(birth.get("second", 0)), tzinfo=zone, + ) + utc = local.astimezone(ZoneInfo("UTC")) + return swe.julday(utc.year, utc.month, utc.day, utc.hour + utc.minute / 60 + utc.second / 3600) + + +def _progressed_planets(progressed_jd: float) -> dict[str, dict[str, Any]]: + flags = swe.FLG_SWIEPH | swe.FLG_SPEED + planets: dict[str, dict[str, Any]] = {} + for name, planet_id in _PLANETS.items(): + values, _ = swe.calc_ut(progressed_jd, planet_id, flags) + planets[name] = _point(values[0], speed=values[3]) + return planets + + +def calculate_secondary_progressions(*, target_date: str, **birth: Any) -> dict[str, Any]: + """Calculate progressed planets using one ephemeris day per tropical year.""" + natal_chart = build_tropical_natal_chart(**birth) + target_jd, local = _target_jd(target_date, birth["timezone"]) + birth_jd = _birth_jd(**birth) + elapsed_years = (target_jd - birth_jd) / 365.242189 + progressed_jd = birth_jd + elapsed_years + planets = _progressed_planets(progressed_jd) + natal_points = { + **natal_chart["natal"]["planets"], + "ascendant": natal_chart["natal"]["angles"]["ascendant"], + "mc": natal_chart["natal"]["angles"]["mc"], + } + return { + "technique": "secondary_progressions", + "status": "partial", + "method": "one_ephemeris_day_per_tropical_year", + "target_date": target_date, + "target_local_time": local.isoformat(), + "elapsed_tropical_years": round(elapsed_years, 8), + "progressed_julian_day_ut": round(progressed_jd, 8), + "natal_sun_longitude": natal_chart["natal"]["planets"]["sun"]["longitude"], + "progressed_planets": planets, + "aspects": _cross_aspects(planets, natal_points), + "boundary": "Progressed planets only. Progressed angles, lunar phases, stations, duration, and interpretation remain separate audited layers.", + } + + +def calculate_solar_arc_directions(*, target_date: str, **birth: Any) -> dict[str, Any]: + """Direct natal points by the true arc of the secondary progressed Sun.""" + natal_chart = build_tropical_natal_chart(**birth) + progressions = calculate_secondary_progressions(target_date=target_date, **birth) + natal_sun = natal_chart["natal"]["planets"]["sun"]["longitude"] + progressed_sun = progressions["progressed_planets"]["sun"]["longitude"] + arc = _longitude(progressed_sun - natal_sun) + natal_points = { + **natal_chart["natal"]["planets"], + "ascendant": natal_chart["natal"]["angles"]["ascendant"], + "mc": natal_chart["natal"]["angles"]["mc"], + } + directed = {name: _point(point["longitude"] + arc) for name, point in natal_points.items()} + return { + "technique": "solar_arc_directions", + "status": "partial", + "method": "secondary_progressed_sun_arc", + "target_date": target_date, + "natal_sun_longitude": natal_sun, + "progressed_sun_longitude": progressed_sun, + "solar_arc_degrees": round(arc, 8), + "directed_points": directed, + "aspects": _cross_aspects(directed, natal_points), + "boundary": "True secondary-progressed-Sun arc applied to natal planets/ASC/MC. Directional converse, latitude, parans, midpoint, duration, and event interpretation are not inferred.", + } + + +def calculate_converse_secondary_progressions(*, target_date: str, **birth: Any) -> dict[str, Any]: + """Calculate converse progressed planets using one ephemeris day per tropical year backward.""" + natal_chart = build_tropical_natal_chart(**birth) + target_jd, local = _target_jd(target_date, birth["timezone"]) + birth_jd = _birth_jd(**birth) + elapsed_years = (target_jd - birth_jd) / 365.242189 + progressed_jd = birth_jd - elapsed_years + planets = _progressed_planets(progressed_jd) + natal_points = { + **natal_chart["natal"]["planets"], + "ascendant": natal_chart["natal"]["angles"]["ascendant"], + "mc": natal_chart["natal"]["angles"]["mc"], + } + return { + "technique": "converse_secondary_progressions", + "status": "partial", + "method": "one_ephemeris_day_per_tropical_year_backward", + "target_date": target_date, + "target_local_time": local.isoformat(), + "elapsed_tropical_years": round(elapsed_years, 8), + "progressed_julian_day_ut": round(progressed_jd, 8), + "progressed_planets": planets, + "aspects": _cross_aspects(planets, natal_points), + "progressed_angles": { + "status": "blocked", + "reason": "Progressed angle method is not selected; quotidian/solar-arc/Naibod variants are not interchangeable.", + }, + "boundary": "Converse progressed planets only; progressed angles and interpretation remain blocked until a method is selected.", + } + + +def calculate_converse_solar_arc_directions(*, target_date: str, **birth: Any) -> dict[str, Any]: + """Direct natal points backward by the converse secondary-progressed Sun arc.""" + natal_chart = build_tropical_natal_chart(**birth) + progressions = calculate_converse_secondary_progressions(target_date=target_date, **birth) + natal_sun = natal_chart["natal"]["planets"]["sun"]["longitude"] + progressed_sun = progressions["progressed_planets"]["sun"]["longitude"] + arc = _longitude(natal_sun - progressed_sun) + natal_points = { + **natal_chart["natal"]["planets"], + "ascendant": natal_chart["natal"]["angles"]["ascendant"], + "mc": natal_chart["natal"]["angles"]["mc"], + } + directed = {name: _point(point["longitude"] - arc) for name, point in natal_points.items()} + return { + "technique": "converse_solar_arc_directions", + "status": "partial", + "method": "converse_secondary_progressed_sun_arc", + "target_date": target_date, + "natal_sun_longitude": natal_sun, + "converse_progressed_sun_longitude": progressed_sun, + "converse_solar_arc_degrees": round(arc, 8), + "directed_points": directed, + "aspects": _cross_aspects(directed, natal_points), + "boundary": "Backward solar arc applied to natal planets/ASC/MC. Interpretation and parans remain separate audited layers.", + } + + +def _midpoint_longitude(first: float, second: float) -> float: + diff = _longitude(second - first) + if diff > 180.0: + diff -= 360.0 + return _longitude(first + diff / 2.0) + + +def calculate_midpoints(*, target_date: str | None = None, orb: float = 1.5, **birth: Any) -> dict[str, Any]: + """Calculate natal midpoint tree and optional transit conjunction/opposition hits.""" + natal_chart = build_tropical_natal_chart(**birth) + natal_points = { + **natal_chart["natal"]["planets"], + "ascendant": natal_chart["natal"]["angles"]["ascendant"], + "mc": natal_chart["natal"]["angles"]["mc"], + } + names = [name for name in [*_PLANETS.keys(), "ascendant", "mc"] if name in natal_points] + midpoints: dict[str, dict[str, Any]] = {} + for index, first_name in enumerate(names): + for second_name in names[index + 1:]: + key = f"{first_name}/{second_name}" + lon = _midpoint_longitude(natal_points[first_name]["longitude"], natal_points[second_name]["longitude"]) + midpoints[key] = _point(lon) + result: dict[str, Any] = { + "technique": "midpoints", + "status": "used", + "method": "shortest_arc_direct_midpoints", + "orb_degrees": float(orb), + "natal_midpoints": midpoints, + "boundary": "Midpoint geometry only; hits are conjunction/opposition contacts, not interpretations.", + } + if target_date: + transit = calculate_transit_to_natal(target_date=target_date, **birth) + hits: list[dict[str, Any]] = [] + for transit_name, transit_point in transit["transit_planets"].items(): + for midpoint_name, midpoint in midpoints.items(): + separation = abs(transit_point["longitude"] - midpoint["longitude"]) + separation = min(separation, 360.0 - separation) + for aspect, exact in {"conjunction": 0.0, "opposition": 180.0}.items(): + hit_orb = abs(separation - exact) + if hit_orb <= orb: + hits.append({ + "transit_planet": transit_name, + "midpoint": midpoint_name, + "aspect": aspect, + "orb": round(hit_orb, 6), + "separation": round(separation, 6), + }) + result["target_date"] = target_date + result["transit_midpoint_hits"] = sorted(hits, key=lambda row: (row["orb"], row["transit_planet"], row["midpoint"])) + return result + + +def calculate_lunar_return(*, start_date: str, **birth: Any) -> dict[str, Any]: + """Find the next exact tropical lunar return after a local start date.""" + natal_chart = build_tropical_natal_chart(**birth) + natal_moon = natal_chart["natal"]["planets"]["moon"]["longitude"] + start_jd, _ = _target_jd(start_date, birth["timezone"]) + return_jd = swe.mooncross_ut(natal_moon, start_jd, swe.FLG_SWIEPH) + return_local = _jd_to_local(return_jd, birth["timezone"]) + return_birth = { + **birth, + "year": return_local.year, + "month": return_local.month, + "day": return_local.day, + "hour": return_local.hour, + "minute": return_local.minute, + "second": return_local.second, + } + return_chart = build_tropical_natal_chart(**return_birth) + returned_moon = return_chart["natal"]["planets"]["moon"]["longitude"] + delta = abs(_longitude(returned_moon - natal_moon)) + delta = min(delta, 360.0 - delta) + return { + "technique": "lunar_return", + "status": "used", + "method": "Swiss Ephemeris mooncross_ut tropical longitude", + "start_date": start_date, + "return_julian_day_ut": round(return_jd, 8), + "return_local_time": return_local.isoformat(), + "natal_moon_longitude": natal_moon, + "return_moon_longitude": returned_moon, + "moon_longitude_delta": round(delta, 8), + "return_chart": return_chart, + "boundary": "Exact lunar return time and chart only; monthly topics require separate audited interpretation.", + } + + +def calculate_transit_duration_scan(*, start_date: str, end_date: str, max_days: int = 370, **birth: Any) -> dict[str, Any]: + """Scan daily transit-to-natal aspect activity and group consecutive windows.""" + start = datetime.fromisoformat(start_date) + end = datetime.fromisoformat(end_date) + if end < start: + raise ValueError("end_date must be on or after start_date") + days = (end.date() - start.date()).days + 1 + if days > max_days: + raise ValueError(f"duration scan range exceeds max_days={max_days}") + daily_hits: list[dict[str, Any]] = [] + active: dict[tuple[str, str, str], dict[str, Any]] = {} + windows: list[dict[str, Any]] = [] + for offset in range(days): + current = (start + timedelta(days=offset)).date().isoformat() + transit = calculate_transit_to_natal(target_date=current, **birth) + keys = set() + for aspect in transit["aspects"]: + key = (aspect["transit_planet"], aspect["natal_point"], aspect["aspect"]) + keys.add(key) + if key not in active: + active[key] = {"start_date": current, "min_orb": aspect["orb"]} + else: + active[key]["min_orb"] = min(active[key]["min_orb"], aspect["orb"]) + for key in list(active): + if key not in keys: + row = active.pop(key) + windows.append({ + "transit_planet": key[0], + "natal_point": key[1], + "aspect": key[2], + "start_date": row["start_date"], + "end_date": (start + timedelta(days=offset - 1)).date().isoformat(), + "min_orb": round(row["min_orb"], 6), + }) + daily_hits.append({"date": current, "hit_count": len(transit["aspects"]), "aspects": transit["aspects"]}) + final_date = end.date().isoformat() + for key, row in active.items(): + windows.append({ + "transit_planet": key[0], + "natal_point": key[1], + "aspect": key[2], + "start_date": row["start_date"], + "end_date": final_date, + "min_orb": round(row["min_orb"], 6), + }) + return { + "technique": "transit_duration_scan", + "status": "used", + "method": "daily local-midnight transit snapshots grouped into consecutive aspect windows", + "start_date": start_date, + "end_date": end_date, + "days_scanned": days, + "daily_hits": daily_hits, + "windows": sorted(windows, key=lambda row: (row["start_date"], row["min_orb"], row["transit_planet"])), + "boundary": "Daily scan only; exact ingress/egress times require sub-daily root finding.", + } + + +def calculate_parans_status(*, target_date: str | None = None, **birth: Any) -> dict[str, Any]: + return { + "technique": "parans", + "status": "blocked", + "target_date": target_date, + "reason": "Parans need a dedicated rising/setting/culminating engine and latitude-aware event solver; not yet implemented in this repository.", + } + + +def build_timing_techniques( + *, + transit_date: str | None = None, + solar_return_year: int | None = None, + secondary_progression_date: str | None = None, + solar_arc_date: str | None = None, + converse_secondary_progression_date: str | None = None, + converse_solar_arc_date: str | None = None, + midpoint_date: str | None = None, + lunar_return_start_date: str | None = None, + duration_scan_start_date: str | None = None, + duration_scan_end_date: str | None = None, + parans_date: str | None = None, + **birth: Any, +) -> dict[str, Any]: + """Materialize only the requested, independently auditable timing layers.""" + techniques: dict[str, Any] = {} + if transit_date: + techniques["transits"] = calculate_transit_to_natal(target_date=transit_date, **birth) + if solar_return_year is not None: + techniques["solar_return"] = calculate_solar_return(target_year=int(solar_return_year), **birth) + if secondary_progression_date: + techniques["secondary_progressions"] = calculate_secondary_progressions( + target_date=secondary_progression_date, **birth + ) + if solar_arc_date: + techniques["solar_arc_directions"] = calculate_solar_arc_directions(target_date=solar_arc_date, **birth) + if converse_secondary_progression_date: + techniques["converse_secondary_progressions"] = calculate_converse_secondary_progressions( + target_date=converse_secondary_progression_date, **birth + ) + if converse_solar_arc_date: + techniques["converse_solar_arc_directions"] = calculate_converse_solar_arc_directions( + target_date=converse_solar_arc_date, **birth + ) + if midpoint_date: + techniques["midpoints"] = calculate_midpoints(target_date=midpoint_date, **birth) + if lunar_return_start_date: + techniques["lunar_return"] = calculate_lunar_return(start_date=lunar_return_start_date, **birth) + if duration_scan_start_date and duration_scan_end_date: + techniques["transit_duration_scan"] = calculate_transit_duration_scan( + start_date=duration_scan_start_date, + end_date=duration_scan_end_date, + **birth, + ) + if parans_date: + techniques["parans"] = calculate_parans_status(target_date=parans_date, **birth) + return techniques diff --git a/tests/test_western_chart_engine.py b/tests/test_western_chart_engine.py new file mode 100644 index 00000000..53bd88e1 --- /dev/null +++ b/tests/test_western_chart_engine.py @@ -0,0 +1,107 @@ +"""Regression tests for the native tropical Western chart calculator.""" + +from __future__ import annotations + +from scripts.western_chart_engine import ( + build_tropical_natal_chart, + build_tropical_western_evidence_packet, +) +from scripts.jyotish_api_server import _western_evidence_packet_from_body +from scripts.skill_release_package import _edition_files + + +_BIRTH = { + "year": 1993, + "month": 4, + "day": 17, + "hour": 14, + "minute": 49, + "latitude": 36.683333, + "longitude": 114.35, + "timezone": "Asia/Shanghai", +} + + +def test_native_engine_calculates_auditable_tropical_natal_chart() -> None: + chart = build_tropical_natal_chart(**_BIRTH) + + assert chart["source_engine"] == "pyswisseph_tropical" + assert chart["zodiac"] == "tropical" + assert chart["house_system"] == "P" + assert chart["natal"]["planets"]["sun"]["sign"] == "Aries" + assert 26 < chart["natal"]["planets"]["sun"]["longitude"] < 28 + assert set(chart["natal"]["angles"]) == {"ascendant", "mc", "descendant", "ic"} + assert len(chart["natal"]["houses"]) == 12 + assert all(1 <= planet["house"] <= 12 for planet in chart["natal"]["planets"].values()) + assert chart["natal"]["aspects"] + assert all(aspect["orb"] <= aspect["allowed_orb"] for aspect in chart["natal"]["aspects"]) + + +def test_native_engine_marks_timing_and_interpretation_boundaries() -> None: + packet = build_tropical_western_evidence_packet(**_BIRTH, route_packet={"primary_theme": "career"}) + + assert packet["status"] == "partial" + assert packet["calculation"]["status"] == "used" + assert packet["calculation"]["source_engine"] == "pyswisseph_tropical" + assert "timing_techniques" in packet["missing_sections"] + assert "signals" in packet["missing_sections"] + assert "does not calculate transits" in packet["boundary"] + + +def test_workflow_auto_materializes_native_western_natal_without_external_json() -> None: + packet = _western_evidence_packet_from_body( + {"entry_mode": "direct_chart", "western_mode": "auto"}, + {"primary_theme": "career"}, + birth_payload={ + "year": 1993, "month": 4, "day": 17, "hour": 14, "minute": 49, + "second": 0, "lat": 36.683333, "lon": 114.35, "tz": 8, + }, + ) + + assert packet is not None + assert packet["source_engine"] == "pyswisseph_tropical" + assert packet["status"] == "partial" + + +def test_workflow_does_not_auto_attach_natal_western_data_to_prashna() -> None: + packet = _western_evidence_packet_from_body( + {"entry_mode": "prashna", "western_mode": "auto"}, + {"primary_theme": "career"}, + birth_payload={ + "year": 1993, "month": 4, "day": 17, "hour": 14, "minute": 49, + "second": 0, "lat": 36.683333, "lon": 114.35, "tz": 8, + }, + ) + + assert packet is None + + +def test_workflow_adds_only_explicit_western_timing_layers() -> None: + packet = _western_evidence_packet_from_body( + { + "entry_mode": "direct_chart", + "western_timing": { + "transit_date": "2026-07-09", + "solar_return_year": 2026, + "secondary_progression_date": "2026-07-09", + "solar_arc_date": "2026-07-09", + }, + }, + {"primary_theme": "career"}, + birth_payload={ + "year": 1993, "month": 4, "day": 17, "hour": 14, "minute": 49, + "second": 0, "lat": 36.683333, "lon": 114.35, "tz": 8, + }, + ) + + assert set(packet["timing_techniques"]) == { + "transits", "solar_return", "secondary_progressions", "solar_arc_directions", + } + assert packet["sections"]["timing_techniques"]["status"] == "used" + + +def test_release_editions_include_native_western_calculator() -> None: + assert "scripts/western_chart_engine.py" in _edition_files("basic_git") + assert "scripts/western_chart_engine.py" in _edition_files("premium_cloud_drive") + assert "scripts/western_timing_engine.py" in _edition_files("basic_git") + assert "scripts/western_timing_engine.py" in _edition_files("premium_cloud_drive") diff --git a/tests/test_western_timing_engine.py b/tests/test_western_timing_engine.py new file mode 100644 index 00000000..7510c6eb --- /dev/null +++ b/tests/test_western_timing_engine.py @@ -0,0 +1,124 @@ +"""Regression tests for native Western timing calculations.""" + +from __future__ import annotations + +from scripts.western_timing_engine import ( + build_timing_techniques, + calculate_converse_secondary_progressions, + calculate_converse_solar_arc_directions, + calculate_lunar_return, + calculate_midpoints, + calculate_parans_status, + calculate_secondary_progressions, + calculate_solar_arc_directions, + calculate_solar_return, + calculate_transit_duration_scan, + calculate_transit_to_natal, +) + + +_BIRTH = { + "year": 1993, "month": 4, "day": 17, "hour": 14, "minute": 49, + "latitude": 36.683333, "longitude": 114.35, "timezone": "Asia/Shanghai", +} + + +def test_transit_to_natal_emits_orb_auditable_aspects() -> None: + transit = calculate_transit_to_natal(**_BIRTH, target_date="2026-07-09") + + assert transit["technique"] == "transits" + assert transit["target_date"] == "2026-07-09" + assert transit["aspects"] + assert all(row["orb"] <= row["allowed_orb"] for row in transit["aspects"]) + + +def test_solar_return_calculates_return_moment_and_chart() -> None: + solar_return = calculate_solar_return(**_BIRTH, target_year=2026) + + assert solar_return["technique"] == "solar_return" + assert solar_return["target_year"] == 2026 + assert solar_return["return_chart"]["natal"]["planets"]["sun"]["sign"] == "Aries" + assert solar_return["sun_longitude_delta"] < 0.001 + + +def test_timing_builder_only_contains_requested_techniques() -> None: + timing = build_timing_techniques(**_BIRTH, transit_date="2026-07-09", solar_return_year=2026) + + assert set(timing) == {"transits", "solar_return"} + + +def test_secondary_progressions_use_declared_day_for_year_contract() -> None: + progressions = calculate_secondary_progressions(**_BIRTH, target_date="2026-07-09") + + assert progressions["technique"] == "secondary_progressions" + assert progressions["method"] == "one_ephemeris_day_per_tropical_year" + assert progressions["progressed_planets"]["sun"]["longitude"] != progressions["natal_sun_longitude"] + assert progressions["aspects"] + + +def test_solar_arc_uses_secondary_progressed_sun_arc() -> None: + directions = calculate_solar_arc_directions(**_BIRTH, target_date="2026-07-09") + + assert directions["technique"] == "solar_arc_directions" + assert directions["method"] == "secondary_progressed_sun_arc" + assert 0 < directions["solar_arc_degrees"] < 40 + assert directions["directed_points"]["sun"]["longitude"] != directions["natal_sun_longitude"] + + +def test_converse_progressions_and_solar_arc_are_auditable() -> None: + progressions = calculate_converse_secondary_progressions(**_BIRTH, target_date="2026-07-09") + assert progressions["technique"] == "converse_secondary_progressions" + assert progressions["progressed_angles"]["status"] == "blocked" + directions = calculate_converse_solar_arc_directions(**_BIRTH, target_date="2026-07-09") + assert directions["technique"] == "converse_solar_arc_directions" + assert 0 < directions["converse_solar_arc_degrees"] < 40 + assert directions["directed_points"]["sun"]["longitude"] != directions["natal_sun_longitude"] + + +def test_midpoints_emit_geometry_and_optional_transit_hits() -> None: + midpoints = calculate_midpoints(**_BIRTH, target_date="2026-07-09") + assert midpoints["technique"] == "midpoints" + assert "sun/moon" in midpoints["natal_midpoints"] + assert isinstance(midpoints["transit_midpoint_hits"], list) + + +def test_lunar_return_calculates_next_exact_return_chart() -> None: + lunar_return = calculate_lunar_return(**_BIRTH, start_date="2026-07-01") + assert lunar_return["technique"] == "lunar_return" + assert lunar_return["moon_longitude_delta"] < 0.01 + assert lunar_return["return_chart"]["natal"]["planets"]["moon"]["sign"] + + +def test_transit_duration_scan_groups_daily_windows() -> None: + scan = calculate_transit_duration_scan(**_BIRTH, start_date="2026-07-01", end_date="2026-07-03") + assert scan["technique"] == "transit_duration_scan" + assert scan["days_scanned"] == 3 + assert len(scan["daily_hits"]) == 3 + assert isinstance(scan["windows"], list) + + +def test_parans_are_explicitly_blocked_until_solver_exists() -> None: + parans = calculate_parans_status(**_BIRTH, target_date="2026-07-09") + assert parans["technique"] == "parans" + assert parans["status"] == "blocked" + + +def test_timing_builder_can_emit_advanced_layers() -> None: + timing = build_timing_techniques( + **_BIRTH, + converse_secondary_progression_date="2026-07-09", + converse_solar_arc_date="2026-07-09", + midpoint_date="2026-07-09", + lunar_return_start_date="2026-07-01", + duration_scan_start_date="2026-07-01", + duration_scan_end_date="2026-07-02", + parans_date="2026-07-09", + ) + assert { + "converse_secondary_progressions", + "converse_solar_arc_directions", + "midpoints", + "lunar_return", + "transit_duration_scan", + "parans", + } <= set(timing)