|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +"""Find exact mundane (transiting-to-transiting) aspects over a date range. |
| 3 | +
|
| 4 | +A printed astrological calendar's *aspectarian* lists the instant each aspect |
| 5 | +between two moving bodies perfects. This factory produces exactly that: every |
| 6 | +exact planet-to-planet aspect within a range, with its Julian Day / ISO UTC |
| 7 | +timestamp and both bodies' zodiacal state at the event. |
| 8 | +
|
| 9 | +Aspect *times* are zodiac-independent (an ayanamsha shifts both longitudes |
| 10 | +equally, leaving their separation unchanged); the reported longitudes and signs |
| 11 | +follow the requested zodiac. |
| 12 | +
|
| 13 | +Swiss Ephemeris / libephemeris functions used (via :mod:`.utils`): |
| 14 | + - ephe.calc_ut(jd, planet_id, iflag) |
| 15 | +""" |
| 16 | + |
| 17 | +from __future__ import annotations |
| 18 | + |
| 19 | +from datetime import datetime, timezone |
| 20 | +from typing import List, Optional, Tuple |
| 21 | + |
| 22 | +from kerykeion._predictive_utils import jd_to_iso_utc as _jd_to_iso |
| 23 | +from kerykeion.ephemeris_backend import ephe, ephemeris_session |
| 24 | +from kerykeion.mundane_aspects.utils import MundaneAspectEvent, _lon_speed, scan_mundane_aspects |
| 25 | +from kerykeion.schemas.kerykeion_exception import KerykeionException |
| 26 | +from kerykeion.schemas.kr_literals import SIGN_CODES, Sign, SiderealMode, ZodiacType |
| 27 | +from kerykeion.schemas.kr_models import SubscriptableBaseModel |
| 28 | +from kerykeion.settings.chart_defaults import DEFAULT_CHART_ASPECTS_SETTINGS |
| 29 | +from kerykeion.settings.config_constants import POINT_NUMBER_MAP |
| 30 | +from kerykeion.utilities import datetime_to_julian |
| 31 | +from pydantic import Field |
| 32 | + |
| 33 | +# Canonical body order (fastest mean geocentric motion first, the traditional |
| 34 | +# aspectarian convention): each emitted event's ``point_a`` is the pair member |
| 35 | +# that appears earlier here. Also the request vocabulary. |
| 36 | +_CANONICAL_ORDER: Tuple[str, ...] = ( |
| 37 | + "Moon", |
| 38 | + "Mercury", |
| 39 | + "Venus", |
| 40 | + "Sun", |
| 41 | + "Mars", |
| 42 | + "Jupiter", |
| 43 | + "Saturn", |
| 44 | + "Chiron", |
| 45 | + "Uranus", |
| 46 | + "Neptune", |
| 47 | + "Pluto", |
| 48 | + "Mean_North_Lunar_Node", |
| 49 | + "True_North_Lunar_Node", |
| 50 | +) |
| 51 | +_POINT_IDS: dict[str, int] = {name: POINT_NUMBER_MAP[name] for name in _CANONICAL_ORDER} |
| 52 | + |
| 53 | +# Default scan set: the classical ten minus the Moon. The Moon perfects every |
| 54 | +# aspect family to every body roughly once per 27.3-day lap (~75 events/month |
| 55 | +# on its own) — noise for most consumers, so it is opt-in by listing it in |
| 56 | +# ``points``, the same convention as ``SignIngressFactory``. |
| 57 | +_DEFAULT_POINTS: Tuple[str, ...] = ( |
| 58 | + "Sun", "Mercury", "Venus", "Mars", "Jupiter", "Saturn", "Uranus", "Neptune", "Pluto", |
| 59 | +) |
| 60 | + |
| 61 | +# Longitude-aspect vocabulary (name → exact degrees) from the shared chart |
| 62 | +# defaults. Declination aspects (parallel/contra-parallel) are not longitude |
| 63 | +# events and are therefore not in this map — requesting one raises ValueError. |
| 64 | +_ASPECT_DEGREES: dict[str, float] = { |
| 65 | + setting["name"]: float(setting["degree"]) for setting in DEFAULT_CHART_ASPECTS_SETTINGS |
| 66 | +} |
| 67 | + |
| 68 | +# The five Ptolemaic majors, the default aspect set. |
| 69 | +_DEFAULT_ASPECTS: Tuple[str, ...] = ("conjunction", "sextile", "square", "trine", "opposition") |
| 70 | + |
| 71 | + |
| 72 | +def _to_utc_naive(dt: datetime) -> datetime: |
| 73 | + """Normalize an offset-aware datetime to naive UTC (see lunation factory).""" |
| 74 | + if dt.tzinfo is not None: |
| 75 | + dt = dt.astimezone(timezone.utc).replace(tzinfo=None) |
| 76 | + return dt |
| 77 | + |
| 78 | + |
| 79 | +def _validate_zodiac(zodiac_type: ZodiacType, sidereal_mode: Optional[SiderealMode]) -> None: |
| 80 | + """Validate the zodiac configuration before opening an ephemeris session. |
| 81 | +
|
| 82 | + Same contract as ``VoidOfCourseMoonFactory``'s validator: pure validation, |
| 83 | + no global ephemeris state touched. |
| 84 | + """ |
| 85 | + if zodiac_type not in ("Tropical", "Sidereal"): |
| 86 | + raise KerykeionException(f"Unknown zodiac_type: {zodiac_type!r} (expected 'Tropical' or 'Sidereal').") |
| 87 | + |
| 88 | + if zodiac_type == "Sidereal": |
| 89 | + if sidereal_mode is None: |
| 90 | + raise KerykeionException("sidereal_mode is required when zodiac_type='Sidereal'.") |
| 91 | + if sidereal_mode == "USER": |
| 92 | + raise KerykeionException( |
| 93 | + "sidereal_mode='USER' requires custom ayanamsha parameters, which MundaneAspectFactory does not accept." |
| 94 | + ) |
| 95 | + if not hasattr(ephe, f"SIDM_{sidereal_mode}"): |
| 96 | + raise KerykeionException(f"Unknown sidereal_mode: {sidereal_mode!r}.") |
| 97 | + |
| 98 | + |
| 99 | +# ============================================================================= |
| 100 | +# MODELS |
| 101 | +# ============================================================================= |
| 102 | + |
| 103 | + |
| 104 | +class MundaneAspectModel(SubscriptableBaseModel): |
| 105 | + """A single exact mundane aspect between two transiting bodies.""" |
| 106 | + |
| 107 | + point_a: str = Field(description="First body (canonical fast-to-slow order)") |
| 108 | + point_b: str = Field(description="Second body") |
| 109 | + aspect: str = Field(description="Aspect name, e.g. 'square'") |
| 110 | + aspect_degrees: float = Field(description="The aspect's exact angle in degrees") |
| 111 | + julian_day: float = Field(description="Julian Day (UT) of the exact perfection") |
| 112 | + iso_utc: str = Field(description="ISO 8601 UTC datetime of the exact perfection") |
| 113 | + point_a_longitude: float = Field(description="Ecliptic longitude of point_a at the event (requested zodiac)") |
| 114 | + point_b_longitude: float = Field(description="Ecliptic longitude of point_b at the event (requested zodiac)") |
| 115 | + point_a_sign: Sign = Field(description="Sign of point_a at the event") |
| 116 | + point_b_sign: Sign = Field(description="Sign of point_b at the event") |
| 117 | + point_a_retrograde: bool = Field(description="True if point_a is retrograde at the event") |
| 118 | + point_b_retrograde: bool = Field(description="True if point_b is retrograde at the event") |
| 119 | + |
| 120 | + |
| 121 | +class MundaneAspectsCollectionModel(SubscriptableBaseModel): |
| 122 | + """Ordered list of mundane aspects within a Julian Day range.""" |
| 123 | + |
| 124 | + start_jd: float |
| 125 | + end_jd: float |
| 126 | + aspects: List[MundaneAspectModel] |
| 127 | + |
| 128 | + |
| 129 | +# ============================================================================= |
| 130 | +# FACTORY |
| 131 | +# ============================================================================= |
| 132 | + |
| 133 | + |
| 134 | +class MundaneAspectFactory: |
| 135 | + """Find exact mundane (planet-to-planet) aspects within a date range. |
| 136 | +
|
| 137 | + Example: |
| 138 | + >>> from kerykeion import MundaneAspectFactory |
| 139 | + >>> result = MundaneAspectFactory.from_iso_range("2020-12-01", "2020-12-31") |
| 140 | + >>> [(a.point_a, a.aspect, a.point_b) for a in result.aspects if a.point_a == "Jupiter"] |
| 141 | + [('Jupiter', 'conjunction', 'Saturn')] |
| 142 | + """ |
| 143 | + |
| 144 | + @staticmethod |
| 145 | + def from_iso_range( |
| 146 | + start_date: str, |
| 147 | + end_date: str, |
| 148 | + points: Optional[List[str]] = None, |
| 149 | + aspects: Optional[List[str]] = None, |
| 150 | + zodiac_type: ZodiacType = "Tropical", |
| 151 | + sidereal_mode: Optional[SiderealMode] = None, |
| 152 | + ) -> MundaneAspectsCollectionModel: |
| 153 | + """Find mundane aspects between two ISO date(time) strings (treated as UTC). |
| 154 | +
|
| 155 | + Args: |
| 156 | + start_date: ISO date or datetime, e.g. ``"2026-01-01"``. |
| 157 | + end_date: ISO date or datetime; a date-only value means "through the |
| 158 | + end of that UTC day". |
| 159 | + points: Optional subset of body names. Defaults to Sun..Pluto |
| 160 | + (Moon excluded unless explicitly requested). |
| 161 | + aspects: Optional aspect names. Defaults to the five Ptolemaic |
| 162 | + majors (conjunction, sextile, square, trine, opposition). |
| 163 | + zodiac_type: ``"Tropical"`` (default) or ``"Sidereal"``. Event times |
| 164 | + are zodiac-independent; longitudes/signs follow the request. |
| 165 | + sidereal_mode: Ayanamsha when ``zodiac_type='Sidereal'``. |
| 166 | + """ |
| 167 | + try: |
| 168 | + start_dt = _to_utc_naive(datetime.fromisoformat(start_date)) |
| 169 | + end_dt = _to_utc_naive(datetime.fromisoformat(end_date)) |
| 170 | + except (ValueError, TypeError) as exc: |
| 171 | + # Same contract as SignIngressFactory.from_iso_range. |
| 172 | + raise KerykeionException( |
| 173 | + f"Invalid ISO date/datetime for mundane aspect range " |
| 174 | + f"(start_date={start_date!r}, end_date={end_date!r}): {exc}" |
| 175 | + ) from exc |
| 176 | + if "T" not in end_date and "t" not in end_date and " " not in end_date: |
| 177 | + end_dt = end_dt.replace(hour=23, minute=59, second=59, microsecond=999999) |
| 178 | + return MundaneAspectFactory.from_julian_day( |
| 179 | + datetime_to_julian(start_dt), |
| 180 | + datetime_to_julian(end_dt), |
| 181 | + points, |
| 182 | + aspects, |
| 183 | + zodiac_type, |
| 184 | + sidereal_mode, |
| 185 | + ) |
| 186 | + |
| 187 | + @staticmethod |
| 188 | + def from_julian_day( |
| 189 | + start_jd: float, |
| 190 | + end_jd: float, |
| 191 | + points: Optional[List[str]] = None, |
| 192 | + aspects: Optional[List[str]] = None, |
| 193 | + zodiac_type: ZodiacType = "Tropical", |
| 194 | + sidereal_mode: Optional[SiderealMode] = None, |
| 195 | + ) -> MundaneAspectsCollectionModel: |
| 196 | + """Find all mundane aspects in ``[start_jd, end_jd]``, ordered chronologically. |
| 197 | +
|
| 198 | + Raises: |
| 199 | + KerykeionException: For an invalid zodiac configuration, or if the |
| 200 | + ephemeris backend fails mid-scan (most often a date outside the |
| 201 | + available ephemeris range). |
| 202 | + ValueError: If a body/aspect name is unknown or the range is too |
| 203 | + large to scan. |
| 204 | + """ |
| 205 | + # None = default set; an explicit empty list = scan nothing. |
| 206 | + if points is not None: |
| 207 | + invalid = sorted(set(points) - set(_POINT_IDS)) |
| 208 | + if invalid: |
| 209 | + raise ValueError( |
| 210 | + f"Unknown points: {', '.join(invalid)}. " |
| 211 | + f"Valid: {', '.join(_CANONICAL_ORDER)}" |
| 212 | + ) |
| 213 | + requested = set(points) |
| 214 | + else: |
| 215 | + requested = set(_DEFAULT_POINTS) |
| 216 | + # Canonical (fast-to-slow) order fixes point_a/point_b per pair and |
| 217 | + # makes results independent of the caller's list order. |
| 218 | + bodies = [(name, _POINT_IDS[name]) for name in _CANONICAL_ORDER if name in requested] |
| 219 | + |
| 220 | + if aspects is not None: |
| 221 | + invalid = sorted(set(aspects) - set(_ASPECT_DEGREES)) |
| 222 | + if invalid: |
| 223 | + raise ValueError( |
| 224 | + f"Unknown aspects: {', '.join(invalid)}. " |
| 225 | + f"Valid: {', '.join(_ASPECT_DEGREES)}" |
| 226 | + ) |
| 227 | + aspect_pairs = [(name, _ASPECT_DEGREES[name]) for name in dict.fromkeys(aspects)] |
| 228 | + else: |
| 229 | + aspect_pairs = [(name, _ASPECT_DEGREES[name]) for name in _DEFAULT_ASPECTS] |
| 230 | + |
| 231 | + _validate_zodiac(zodiac_type, sidereal_mode) |
| 232 | + |
| 233 | + models: List[MundaneAspectModel] = [] |
| 234 | + if end_jd > start_jd and len(bodies) >= 2 and aspect_pairs: |
| 235 | + # The session holds the ephemeris lock across the whole scan and |
| 236 | + # resets global state on exit; its iflag already includes FLG_SPEED |
| 237 | + # (plus FLG_SIDEREAL for sidereal zodiacs). |
| 238 | + with ephemeris_session(zodiac_type=zodiac_type, sidereal_mode=sidereal_mode) as iflag: |
| 239 | + events = scan_mundane_aspects(start_jd, end_jd, bodies, aspect_pairs, iflag) |
| 240 | + point_ids = dict(bodies) |
| 241 | + models = [MundaneAspectFactory._build(event, point_ids, iflag) for event in events] |
| 242 | + |
| 243 | + return MundaneAspectsCollectionModel( |
| 244 | + start_jd=start_jd, |
| 245 | + end_jd=end_jd, |
| 246 | + aspects=models, |
| 247 | + ) |
| 248 | + |
| 249 | + @staticmethod |
| 250 | + def _build( |
| 251 | + event: MundaneAspectEvent, |
| 252 | + point_ids: dict[str, int], |
| 253 | + iflag: int, |
| 254 | + ) -> MundaneAspectModel: |
| 255 | + """Build the public model for an event: one position lookup per body.""" |
| 256 | + lon_a, speed_a = _lon_speed(event.julian_day, point_ids[event.point_a], iflag) |
| 257 | + lon_b, speed_b = _lon_speed(event.julian_day, point_ids[event.point_b], iflag) |
| 258 | + return MundaneAspectModel( |
| 259 | + point_a=event.point_a, |
| 260 | + point_b=event.point_b, |
| 261 | + aspect=event.aspect, |
| 262 | + aspect_degrees=event.degrees, |
| 263 | + julian_day=event.julian_day, |
| 264 | + iso_utc=_jd_to_iso(event.julian_day), |
| 265 | + point_a_longitude=round(lon_a, 6), |
| 266 | + point_b_longitude=round(lon_b, 6), |
| 267 | + point_a_sign=SIGN_CODES[int(lon_a // 30) % 12], |
| 268 | + point_b_sign=SIGN_CODES[int(lon_b // 30) % 12], |
| 269 | + point_a_retrograde=speed_a < 0.0, |
| 270 | + point_b_retrograde=speed_b < 0.0, |
| 271 | + ) |
0 commit comments