Add native Western astrology engines

This commit is contained in:
732642856
2026-07-16 20:20:10 +08:00
parent 4d2802c338
commit 87a07699de
6 changed files with 1011 additions and 8 deletions
+71 -8
View File
@@ -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')
+2
View File
@@ -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",
]
+258
View File
@@ -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())
+449
View File
@@ -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
+107
View File
@@ -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")
+124
View File
@@ -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)