"""Annual data package v1 (Gitea #152 ruling 2026-11-01, phase 1 of #153 / #154). One export for the annual report: year charts, annual periods, natal long periods and slow-planet transits, with separate time semantics. It only calls existing producers; no astrology rule is defined here. Producers (one per block): - year chart, Muntha, year lord, Sahams, Tajika yogas, Tripataki, annual strength, Mudda: ``annual_tajika_pack.build_annual_tajika_pack`` on the report of ``cmd_solar_return`` (``solar_return.solar_return_full_report``); - Patyayini: ``annual_native.patyayini.build_native_patyayini`` (the native annual entry named by the ``tajika-annual`` skill; never the equal-period function in ``extended_dashas.py``); - Tajika monthly charts: ``annual_native.monthly_charts.build_monthly_charts`` (the Sun advancing 30 degrees at a time); - Tithi Pravesha (independent annual view, own start rule): ``annual_native.tithi_pravesh.build_tithi_pravesh``; - Vimshottari MD/AD/PD: ``jyotish_engine.cmd_dasha`` (birth moment); - Narayana: ``narayana_dasha.narayana_dasha_full_report`` (product legacy profile, labelled by ``narayana_legacy_label``) and ``narayana_rath`` (Sanjay Rath profile, display only); - slow transits: positions from ``jyotish_engine._calc_sidereal_planets_for_jd`` (the chart engine's sidereal frame, honours node mode and ayanamsa), sampled once a day and refined by bisection. Time semantics: every boundary is an instant with UTC, local time, offset and Julian day. Local time is the return location's zone (the birth place in v1). Each period system keeps its own year length; rows keep their native bounds and carry ``in_report_range`` for the part inside the report range. Nothing is stretched to fit a year chart. Every block has ``status`` computed | blocked | not_applicable with a reason. A blocked block never fails the response. """ from __future__ import annotations import json import math from collections.abc import Callable, Iterable from datetime import UTC, date, datetime, timedelta, timezone from types import SimpleNamespace from typing import Any from zoneinfo import ZoneInfo SCHEMA = "jyotish.annual_package.v1" ANNUAL_PACKAGE_VERSION = 1 RANGE_KINDS = ("birthday_year", "calendar_year", "next_12_months") RETURN_LOCATION_MODES = ("birth_place",) SLOW_BODIES = ("Jupiter", "Saturn", "Rahu", "Ketu") LOOP_BODIES = ("Jupiter", "Saturn") NATAL_POINTS = ("Lagna", "Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu") SIGNS = ( "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces", ) SIGN_LORDS = { "Aries": "Mars", "Taurus": "Venus", "Gemini": "Mercury", "Cancer": "Moon", "Leo": "Sun", "Virgo": "Mercury", "Libra": "Venus", "Scorpio": "Mars", "Sagittarius": "Jupiter", "Capricorn": "Saturn", "Aquarius": "Saturn", "Pisces": "Jupiter", } PATYAYINI_YEAR_DAYS = 365.25 VIMSHOTTARI_YEAR_DAYS = 365.25 NARAYANA_AGE_YEAR_DAYS = 365.25 TRANSIT_MARGIN_DAYS = 220 BISECT_STEPS = 20 # one day / 2**20 ~ 0.08 s LOOKBACK_MONTHS = 3 _J2000 = datetime(2000, 1, 1, 12, tzinfo=UTC) # Keys that hold reading sentences, not data. Interpretation is the report # phase; the package keeps numbers, names, rules and bases only. _TEXT_KEYS = frozenset({"interpretation"}) class AnnualPackageInputError(ValueError): """Bad request input; the API maps it to HTTP 400.""" # ── time helpers ──────────────────────────────────────────────────────────── def _zone(timezone_id: str | None, tz_hours: float): if timezone_id: return ZoneInfo(timezone_id) return timezone(timedelta(hours=float(tz_hours))) def _zone_id(zone) -> str | None: return getattr(zone, "key", None) def _jd_from_utc(moment: datetime) -> float: return 2451545.0 + (moment.astimezone(UTC) - _J2000).total_seconds() / 86400.0 def _utc_from_jd(jd: float) -> datetime: return _J2000 + timedelta(days=float(jd) - 2451545.0) def _fmt_offset(delta: timedelta | None) -> str: seconds = int((delta or timedelta()).total_seconds()) sign = "+" if seconds >= 0 else "-" seconds = abs(seconds) return f"{sign}{seconds // 3600:02d}:{(seconds % 3600) // 60:02d}" def instant(moment: datetime, zone) -> dict[str, Any]: """An exact boundary: UTC, local time in ``zone``, its offset, Julian day.""" utc = moment.astimezone(UTC) rounded = (utc + timedelta(microseconds=500_000)).replace(microsecond=0) local = rounded.astimezone(zone) return { "utc": rounded.strftime("%Y-%m-%dT%H:%M:%SZ"), "local": local.isoformat(), "utc_offset": _fmt_offset(local.utcoffset()), "timezone_id": _zone_id(zone), "jd_ut": round(_jd_from_utc(utc), 8), } def _parse_naive(value: Any) -> datetime | None: if isinstance(value, datetime): return value.replace(tzinfo=None) if not isinstance(value, str) or not value.strip(): return None try: return datetime.fromisoformat(value.strip().replace("T", " ")).replace(tzinfo=None) except ValueError: return None def _elapsed_mapper(anchor_naive: datetime, anchor_utc: datetime) -> Callable[[Any], datetime | None]: """Map a producer's naive local boundary to UTC by elapsed time. Mudda, Patyayini and Vimshottari add day counts to a naive local start. The day count is the rule; read as elapsed time from the anchor instant it stays right across a daylight-saving change. """ def convert(value: Any) -> datetime | None: parsed = _parse_naive(value) if parsed is None: return None return anchor_utc + (parsed - anchor_naive) return convert def _clip(start: datetime, end: datetime, range_start: datetime, range_end: datetime, zone) -> dict | None: lo, hi = max(start, range_start), min(end, range_end) if lo >= hi: return None return {"start": instant(lo, zone), "end": instant(hi, zone)} def _add_months(day: date, months: int) -> date: index = day.month - 1 + months year, month = day.year + index // 12, index % 12 + 1 for candidate in (day.day, 30, 29, 28): try: return date(year, month, min(day.day, candidate)) except ValueError: continue raise ValueError("unreachable month arithmetic") def _local_midnight(day: date, zone) -> datetime: return datetime(day.year, day.month, day.day, tzinfo=zone).astimezone(UTC) # ── block helpers ─────────────────────────────────────────────────────────── def block(status: str, producer: str, *, reason: str | None = None, **fields: Any) -> dict[str, Any]: if status not in ("computed", "blocked", "not_applicable"): raise ValueError(f"unknown block status {status}") out: dict[str, Any] = {"status": status, "producer": producer} if reason is not None or status != "computed": out["reason"] = reason or "unspecified" out.update(fields) return out def _strip_text(value: Any) -> Any: if isinstance(value, dict): return {key: _strip_text(item) for key, item in value.items() if key not in _TEXT_KEYS} if isinstance(value, list): return [_strip_text(item) for item in value] return value def _json_default(value: Any) -> Any: if isinstance(value, datetime): return value.isoformat() if isinstance(value, date): return value.isoformat() return str(value) def _plain(value: Any) -> Any: return json.loads(json.dumps(_strip_text(value), default=_json_default, ensure_ascii=False)) def _pack_block(field: Any, producer: str) -> dict[str, Any]: """Normalize an ``annual_tajika_pack`` field into a package block.""" if not isinstance(field, dict): return block("blocked", producer, reason="pack_field_missing") status = field.get("status") if status == "conflict": return block( "blocked", producer, reason=f"producer_conflict:{field.get('reason')}", producer_status="conflict", conflict_values=_plain([item.get("value") for item in field.get("values") or [] if isinstance(item, dict)]), ) if status == "blocked": detail = field.get("blocked_detail") if isinstance(field.get("blocked_detail"), dict) else None extra = {"blocked_detail": _plain(detail)} if detail else {} return block("blocked", producer, reason=str(field.get("reason") or "producer_blocked"), producer_status="blocked", **extra) if status == "not_applicable": return block("not_applicable", producer, reason=str(field.get("reason") or "not_applicable")) data = field.get("data") if isinstance(data, dict) and data.get("error"): return block("blocked", producer, reason=f"producer_error:{data['error']}") if isinstance(data, dict) and data.get("status") == "blocked": return block("blocked", producer, reason=str(data.get("reason") or "producer_blocked"), blocked_detail=_plain(data)) if data is None: return block("blocked", producer, reason="pack_field_without_data") return block("computed", producer, producer_status=status, data=_plain(data)) def _guard(producer: str, func: Callable[[], dict[str, Any]]) -> dict[str, Any]: try: return func() except Exception as exc: # noqa: BLE001 - a producer failure is a blocked block, not a 500 return block("blocked", producer, reason=f"{exc.__class__.__name__}: {exc}") def _sign_idx(longitude: float) -> int: return int(float(longitude) % 360.0 // 30) % 12 def _house(sign_idx: int | None, base_idx: int | None) -> int | None: if sign_idx is None or base_idx is None: return None return (int(sign_idx) - int(base_idx)) % 12 + 1 def _wrap(delta: float) -> float: return ((float(delta) + 180.0) % 360.0) - 180.0 # ── modules (scripts/ is on sys.path in the API and tests) ────────────────── def _engine(): try: import jyotish_engine except ImportError: # pragma: no cover from scripts import jyotish_engine return jyotish_engine def _module(name: str): import importlib try: return importlib.import_module(name) except ImportError: # pragma: no cover return importlib.import_module(f"scripts.{name}") # ── request ───────────────────────────────────────────────────────────────── def normalize_request( *, birth: dict[str, Any], ayanamsa: str, node_mode: str, report_range: Any, observation_date: Any, return_location: Any, ) -> dict[str, Any]: """Validate the package-specific fields. Birth fields arrive validated.""" obs = _parse_day(observation_date, "observation_date") if observation_date else date.today() raw_range = report_range if report_range is not None else {"kind": "birthday_year"} if not isinstance(raw_range, dict): raise AnnualPackageInputError("report_range must be an object") kind = raw_range.get("kind") or "birthday_year" if kind not in RANGE_KINDS: raise AnnualPackageInputError(f"report_range.kind must be one of {', '.join(RANGE_KINDS)}") year = raw_range.get("year") if year is not None: if isinstance(year, bool) or not isinstance(year, (int, str)) or not str(year).strip().lstrip("-").isdigit(): raise AnnualPackageInputError("report_range.year must be an integer") year = int(year) if not int(birth["year"]) <= year <= 2400: raise AnnualPackageInputError("report_range.year must be between the birth year and 2400") start_date = raw_range.get("start_date") if kind == "next_12_months": start_date = _parse_day(start_date, "report_range.start_date") if start_date else obs elif start_date is not None: raise AnnualPackageInputError("report_range.start_date is only used with next_12_months") raw_location = return_location if return_location is not None else {"mode": "birth_place"} if not isinstance(raw_location, dict): raise AnnualPackageInputError("return_location must be an object") mode = raw_location.get("mode") or "birth_place" if mode not in RETURN_LOCATION_MODES: raise AnnualPackageInputError("return_location.mode must be birth_place in annual_package v1") if node_mode not in ("mean", "true"): raise AnnualPackageInputError("node_mode must be mean or true") return { "birth": dict(birth), "ayanamsa": ayanamsa, "node_mode": node_mode, "report_range": {"kind": kind, "year": year, "start_date": start_date}, "observation_date": obs, "return_location": {"mode": mode}, } def _parse_day(value: Any, key: str) -> date: if isinstance(value, date): return value if not isinstance(value, str): raise AnnualPackageInputError(f"{key} must be YYYY-MM-DD") try: return datetime.strptime(value.strip()[:10], "%Y-%m-%d").date() except ValueError as exc: raise AnnualPackageInputError(f"{key} must be YYYY-MM-DD") from exc # ── package ───────────────────────────────────────────────────────────────── class _Context: def __init__(self, request: dict[str, Any]): birth = request["birth"] self.request = request self.birth = birth self.ayanamsa = request["ayanamsa"] self.node_mode = request["node_mode"] self.tz_hours = float(birth["tz"]) self.timezone_id = birth.get("timezone_id") self.zone = _zone(self.timezone_id, self.tz_hours) self.birth_naive = datetime( int(birth["year"]), int(birth["month"]), int(birth["day"]), int(birth["hour"]), int(birth["minute"]), int(birth.get("second") or 0), ) self.birth_utc = (self.birth_naive - timedelta(hours=self.tz_hours)).replace(tzinfo=UTC) self.observation_date: date = request["observation_date"] self._packs: dict[int, dict[str, Any]] = {} # The pack payload, as `_annual_payload_from_args` builds it for reports. def pack_payload(self, year: int) -> dict[str, Any]: b = self.birth payload = { "birth": { "date": f"{int(b['year']):04d}-{int(b['month']):02d}-{int(b['day']):02d}", "time": f"{int(b['hour']):02d}:{int(b['minute']):02d}:{int(b.get('second') or 0):02d}", "utc_offset": _fmt_offset(timedelta(hours=self.tz_hours)), "latitude": float(b["lat"]), "longitude": float(b["lon"]), "coordinate_precision": "coordinates", }, "settings": { "ayanamsa": self.ayanamsa, "node_mode": self.node_mode, "house_system": "whole_sign", "position_mode": "legacy", "dasha_year_days": 365.25, "solar_return_location_mode": "birth_place", "annual_year_policy": "solar_return_exact", }, "target_year": int(year), "age": int(year) - int(b["year"]), } if self.timezone_id: payload["birth"]["timezone_id"] = self.timezone_id return payload def year_pack(self, year: int) -> dict[str, Any]: """Solar-return report and annual pack for one target year (cached).""" if year in self._packs: return self._packs[year] entry: dict[str, Any] = {"year": year} try: pack_mod = _module("annual_tajika_pack") cmd_solar_return = _module("cmd_solar_return").cmd_solar_return payload = self.pack_payload(year) # Same args the pack builds for itself; the report is passed in so # the full annual chart (speeds, nodes) stays readable here. report = cmd_solar_return(pack_mod._args_from_payload(payload)) if not isinstance(report, dict): report = {"error": "solar_return_non_dict"} entry["payload"] = payload entry["report"] = report entry["pack"] = pack_mod.build_annual_tajika_pack(payload, solar_return_report=report) sr = report.get("solar_return") if isinstance(report.get("solar_return"), dict) else {} jd = sr.get("jd_ut") if report.get("error") or not isinstance(jd, (int, float)) or isinstance(jd, bool): entry["error"] = str(report.get("error") or "solar_return_jd_missing") else: entry["return_utc"] = _utc_from_jd(float(jd)) entry["return_jd"] = float(jd) except Exception as exc: # noqa: BLE001 - recorded as a blocked year chart entry["error"] = f"{exc.__class__.__name__}: {exc}" self._packs[year] = entry return entry def return_utc(self, year: int) -> datetime | None: return self.year_pack(year).get("return_utc") def engine_args(self, today: date | None = None) -> SimpleNamespace: b = self.birth return SimpleNamespace( year=int(b["year"]), month=int(b["month"]), day=int(b["day"]), hour=int(b["hour"]), minute=int(b["minute"]), second=int(b.get("second") or 0), lat=float(b["lat"]), lon=float(b["lon"]), tz=self.tz_hours, node_mode=self.node_mode, ayanamsa=self.ayanamsa, moon_lon=None, nakshatra=None, pada=None, birthdate=None, today=(today or self.observation_date).strftime("%Y-%m-%d"), ) def build_annual_package(request: dict[str, Any]) -> dict[str, Any]: """Build the annual package from a ``normalize_request`` result.""" ctx = _Context(request) natal = _natal_chart(ctx) report_range, years = _report_range(ctx) range_start = report_range.pop("_start", None) range_end = report_range.pop("_end", None) year_charts = [_year_chart(ctx, year, range_start, range_end) for year in years] if range_start is not None and range_end is not None: natal_periods = { "vimshottari": _guard("jyotish_engine.cmd_dasha", lambda: _vimshottari(ctx, range_start, range_end)), "narayana": _guard("narayana_dasha / narayana_rath", lambda: _narayana(ctx, natal, range_start, range_end)), } slow = _guard("jyotish_engine._calc_sidereal_planets_for_jd", lambda: _slow_transits(ctx, natal, range_start, range_end)) tithi = [ _guard("annual_native.tithi_pravesh.build_tithi_pravesh", lambda y=year: _tithi_pravesha(ctx, y, range_start, range_end)) for year in years ] else: reason = "report_range_blocked" natal_periods = { "vimshottari": block("blocked", "jyotish_engine.cmd_dasha", reason=reason), "narayana": block("blocked", "narayana_dasha / narayana_rath", reason=reason), } slow = block("blocked", "jyotish_engine._calc_sidereal_planets_for_jd", reason=reason) tithi = [] package = { "schema": SCHEMA, "annual_package_version": ANNUAL_PACKAGE_VERSION, "parameters": _parameters(ctx, natal), "report_range": report_range, "year_charts": year_charts, "natal_periods": natal_periods, "slow_transits": slow, "independent_annual_views": { "tithi_pravesha": block( "computed" if tithi else "blocked", "annual_native.tithi_pravesh.build_tithi_pravesh", reason=None if tithi else "report_range_blocked", start_rule="Sun and Moon return to the natal Sun-Moon phase near the birthday (own start rule, not the solar return)", location_rule="birth place, birth UTC offset (the producer's own rule; not the Tajika return-location option)", years=tithi, ), }, "computation_boundaries": _boundaries(ctx), "not_included": NOT_INCLUDED, } package = _plain(package) package["block_index"] = list(_walk_blocks(package, "")) package["blocked_count"] = sum(1 for row in package["block_index"] if row["status"] == "blocked") return package def _parameters(ctx: _Context, natal: dict[str, Any]) -> dict[str, Any]: b = ctx.birth return { "birth": { "local_datetime": ctx.birth_naive.isoformat(), "second": int(b.get("second") or 0), "utc_offset_hours": ctx.tz_hours, "timezone_id": ctx.timezone_id, # The chart engine reads the birth clock with the fixed offset `tz`, # so the birth instant is shown in that offset (an IANA zone may # have had daylight saving then, e.g. Asia/Shanghai 1986-1991). "instant": instant(ctx.birth_utc, timezone(timedelta(hours=ctx.tz_hours))), "latitude": float(b["lat"]), "longitude": float(b["lon"]), }, "ayanamsa": ctx.ayanamsa, "node_mode": ctx.node_mode, "house_system": "whole_sign", "return_location": { "mode": ctx.request["return_location"]["mode"], "options_v1": list(RETURN_LOCATION_MODES), "latitude": float(b["lat"]), "longitude": float(b["lon"]), "timezone_id": ctx.timezone_id, "utc_offset_hours": ctx.tz_hours, }, "observation_date": ctx.observation_date.isoformat(), "natal_chart": natal, } def _natal_chart(ctx: _Context) -> dict[str, Any]: def run() -> dict[str, Any]: chart, asc_idx, _jd, ayanamsa_value = _engine()._compute_chart_from_args(ctx.engine_args()) if not isinstance(chart, dict): return block("blocked", "jyotish_engine._compute_chart_from_args", reason="natal_chart_unavailable") planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {} asc = chart.get("ascendant") if isinstance(chart.get("ascendant"), dict) else {} points = {"Lagna": float(asc.get("degree_raw", asc.get("lon")))} for name in NATAL_POINTS[1:]: row = planets.get(name) if isinstance(row, dict) and row.get("degree_raw") is not None: points[name] = float(row["degree_raw"]) return block( "computed", "jyotish_engine._compute_chart_from_args", lagna_sign_idx=int(asc_idx), lagna_sign=SIGNS[int(asc_idx)], moon_sign_idx=_sign_idx(points["Moon"]) if "Moon" in points else None, ayanamsa_degrees=ayanamsa_value, longitudes=points, # Rounded `degree` is what cmd_narayana_dasha and the full reading # hand to Narayana; kept so that block uses the same input. narayana_input_longitudes={ name: row["degree"] for name, row in planets.items() if isinstance(row, dict) and isinstance(row.get("degree"), (int, float)) }, ) return _guard("jyotish_engine._compute_chart_from_args", run) # ── report range ──────────────────────────────────────────────────────────── def _report_range(ctx: _Context) -> tuple[dict[str, Any], list[int]]: spec = ctx.request["report_range"] kind, zone = spec["kind"], ctx.zone obs = ctx.observation_date requested: dict[str, Any] = {"kind": kind} if kind == "birthday_year": year = spec["year"] if spec["year"] is not None else _return_year_covering(ctx, _local_midnight(obs, zone)) requested["year"] = year start, end = ctx.return_utc(year) if year is not None else None, ctx.return_utc(year + 1) if year is not None else None basis = "solar return of the year (inclusive) to the next solar return (exclusive)" elif kind == "calendar_year": year = spec["year"] if spec["year"] is not None else obs.year requested["year"] = year start, end = _local_midnight(date(year, 1, 1), zone), _local_midnight(date(year + 1, 1, 1), zone) basis = "local midnight 1 January to local midnight 1 January of the next year" else: first = spec["start_date"] requested["start_date"] = first.isoformat() start, end = _local_midnight(first, zone), _local_midnight(_add_months(first, 12), zone) basis = "local midnight of start_date to local midnight twelve calendar months later" out: dict[str, Any] = {"kind": kind, "requested": requested, "basis": basis, "end_exclusive": True, "timezone": {"timezone_id": _zone_id(zone), "basis": "return_location_zone"}} if start is None or end is None: out.update(block("blocked", "solar_return.find_solar_return_ut", reason="solar_return_instant_unavailable")) return out, [requested["year"]] if requested.get("year") is not None else [] years = [] first_year = start.astimezone(zone).year - 1 for year in range(first_year, end.astimezone(zone).year + 1): begin, finish = ctx.return_utc(year), ctx.return_utc(year + 1) if begin is None or finish is None: continue if begin < end and finish > start: years.append(year) covered = _coverage_gap(ctx, years, start, end) birthdays = [ctx.return_utc(year) for year in years if start < ctx.return_utc(year) < end] lookback = _local_midnight(_add_months(obs, -LOOKBACK_MONTHS), zone) observed = _local_midnight(obs, zone) out.update(block( "computed" if covered is None else "blocked", "annual_package._report_range", reason=covered, start=instant(start, zone), end=instant(end, zone), duration_days=round((end - start).total_seconds() / 86400.0, 6), year_chart_years=years, crosses_birthday=bool(birthdays), birthday_boundaries=[instant(moment, zone) for moment in birthdays], observation={ "date": obs.isoformat(), "instant": instant(observed, zone), "position": "before_range" if observed < start else "after_range" if observed >= end else "inside_range", "lookback_months": LOOKBACK_MONTHS, "lookback_start": instant(lookback, zone), "rule": "lookback = [observation - 3 months, observation); prospective = [observation, range end)", }, )) out["_start"], out["_end"] = start, end return out, years def _return_year_covering(ctx: _Context, moment: datetime) -> int | None: local_year = moment.astimezone(ctx.zone).year for year in (local_year, local_year - 1): begin, finish = ctx.return_utc(year), ctx.return_utc(year + 1) if begin is not None and finish is not None and begin <= moment < finish: return year return None def _coverage_gap(ctx: _Context, years: list[int], start: datetime, end: datetime) -> str | None: if not years: return "no_year_chart_covers_range" cursor = start for year in years: begin = ctx.return_utc(year) if begin > cursor: return f"year_chart_gap_before_{year}" cursor = max(cursor, ctx.return_utc(year + 1)) return None if cursor >= end else "year_chart_gap_at_range_end" # ── year chart ────────────────────────────────────────────────────────────── def _year_chart(ctx: _Context, year: int, range_start: datetime | None, range_end: datetime | None) -> dict[str, Any]: entry = ctx.year_pack(year) nxt = ctx.year_pack(year + 1) base = {"target_year": year, "completed_years": year - int(ctx.birth["year"])} if entry.get("error") or nxt.get("error"): reason = entry.get("error") or f"next_solar_return_unavailable:{nxt.get('error')}" return {**base, **block("blocked", "solar_return.solar_return_full_report", reason=str(reason))} start, end = entry["return_utc"], nxt["return_utc"] report, pack = entry["report"], entry["pack"] location = report.get("annual_location") if isinstance(report.get("annual_location"), dict) else {} zone = _zone(location.get("timezone_id"), float(location.get("utc_offset") or ctx.tz_hours)) sr = report["solar_return"] validity = { "start": instant(start, zone), "end": instant(end, zone), "end_exclusive": True, "end_source": "next_solar_return", "duration_days": round((end - start).total_seconds() / 86400.0, 6), } overlap = _clip(start, end, range_start, range_end, zone) if range_start and range_end else None houses = report.get("annual_houses") if isinstance(report.get("annual_houses"), dict) else {} local_start = _parse_naive(sr.get("dt_local")) to_utc = _elapsed_mapper(local_start, start) if local_start else None blocks = { "chart": _guard("solar_return.calc_solar_return_chart", lambda: _chart_block(report, houses, sr)), "muntha": _guard("solar_return._muntha_from_natal_ascendant", lambda: _muntha_block(pack, houses)), "year_lord": _pack_block(pack.get("year_lord"), "tajika_year_lord.select_annual_year_lord (Panchadhikari)"), "annual_strength": _strength_block(pack), "sahams": _pack_block(pack.get("sahams"), "tajika.calc_all_sahams"), "tajika_yogas": _pack_block(pack.get("tajika_named_yogas"), "tajika_named_yoga_authority.build_tajika_named_yoga_authority"), "tripataki": _tripataki_block(pack), "mudda_dasha": _guard("tajika_mudda.annual_mudda_dasha", lambda: _mudda_block(pack, to_utc, zone, range_start, range_end)), "patyayini": _guard("annual_native.patyayini.build_native_patyayini", lambda: _patyayini_block(houses, local_start, to_utc, zone, range_start, range_end)), "monthly_charts": _guard("annual_native.monthly_charts.build_monthly_charts", lambda: _monthly_block(ctx, year, sr, houses, location, end, zone, range_start, range_end)), } return { **base, **block("computed", "annual_tajika_pack.build_annual_tajika_pack"), "validity": validity, "in_report_range": overlap, "return_location": { "mode": "birth_place", "latitude": location.get("latitude"), "longitude": location.get("longitude"), "timezone_id": location.get("timezone_id"), "utc_offset_at_return": _fmt_offset(timedelta(hours=float(location.get("utc_offset") or 0.0))), }, "solar_return": { "instant": instant(start, zone), "target_sun_longitude": sr.get("sun_lon"), "target_sun_frame": sr.get("target_sun_frame"), "solver": sr.get("method"), "residual_degrees": sr.get("error_deg"), }, "blocks": blocks, } def _chart_block(report: dict, houses: dict, sr: dict) -> dict[str, Any]: chart = report.get("sr_chart") if isinstance(report.get("sr_chart"), dict) else None if not chart or houses.get("status") == "blocked": return block("blocked", "solar_return.calc_solar_return_chart", reason=str(houses.get("reason") or "annual_chart_missing")) asc_lon = float(houses["ascendant_longitude"]) asc_idx = _sign_idx(asc_lon) planets = {} for name, row in (chart.get("planets") or {}).items(): if not isinstance(row, dict) or row.get("degree_raw") is None: continue lon = float(row["degree_raw"]) % 360.0 planets[name] = { "longitude": lon, "sign": SIGNS[_sign_idx(lon)], "degree_in_sign": lon % 30.0, "house": _house(_sign_idx(lon), asc_idx), "speed": row.get("speed"), "retrograde": row.get("retrograde"), "nakshatra": row.get("nakshatra"), } lord = SIGN_LORDS[SIGNS[asc_idx]] lord_row = planets.get(lord) or {} return block( "computed", "solar_return.calc_solar_return_chart", ascendant={"longitude": asc_lon, "sign": SIGNS[asc_idx], "degree_in_sign": asc_lon % 30.0}, ascendant_lord={"planet": lord, "sign": lord_row.get("sign"), "house": lord_row.get("house"), "longitude": lord_row.get("longitude")}, planets=planets, houses=_plain(houses.get("houses")), house_system="whole_sign", node_mode_of_chart=report.get("node_mode", "mean"), ) def _muntha_block(pack: dict, houses: dict) -> dict[str, Any]: field = _pack_block(pack.get("muntha"), "solar_return._muntha_from_natal_ascendant") if field["status"] != "computed": return field data = field["data"] asc_idx = _sign_idx(houses["ascendant_longitude"]) if houses.get("ascendant_longitude") is not None else None lord = data.get("muntha_lord") lord_lon = (houses.get("planet_longitudes") or {}).get(lord) lord_idx = _sign_idx(lord_lon) if isinstance(lord_lon, (int, float)) else None field["data"] = { **data, "muntha_house_from_annual_ascendant": _house(data.get("muntha_sign_idx"), asc_idx), "muntha_lord_placement": { "planet": lord, "sign": SIGNS[lord_idx] if lord_idx is not None else None, "house_from_annual_ascendant": _house(lord_idx, asc_idx), "longitude": lord_lon, }, } return field def _strength_block(pack: dict) -> dict[str, Any]: field = _pack_block(pack.get("tajika_strength"), "tajika.calc_tajika_strength_layers") field["scope"] = "annual chart Panchavargiya and Harsha Bala; not natal Shadbala" return field def _tripataki_block(pack: dict) -> dict[str, Any]: raw = pack.get("tripataki_chakra") producer = "annual_native.tripataki_chakra.build_tripataki_chakra" if not isinstance(raw, dict): return block("blocked", producer, reason="tripataki_missing") if raw.get("status") == "blocked": return block("blocked", producer, reason=str(raw.get("reason") or "producer_blocked")) return block("computed", producer, producer_status=raw.get("status"), data=_plain(raw)) def _period_rows(rows: Iterable[dict], to_utc, zone, range_start, range_end, *, keys: tuple[str, ...]) -> list[dict]: out = [] for row in rows: if not isinstance(row, dict): continue start, end = to_utc(row.get("start")), to_utc(row.get("end")) if start is None or end is None: continue item = {key: row.get(key) for key in keys if key in row} item["start"] = instant(start, zone) item["end"] = instant(end, zone) item["in_report_range"] = _clip(start, end, range_start, range_end, zone) if range_start and range_end else None out.append(item) return out def _mudda_block(pack: dict, to_utc, zone, range_start, range_end) -> dict[str, Any]: producer = "tajika_mudda.annual_mudda_dasha" field = _pack_block(pack.get("mudda_dasha"), producer) if field["status"] != "computed": return field if to_utc is None: return block("blocked", producer, reason="solar_return_local_time_missing") data = field["data"] start, end = to_utc(data.get("annual_start")), to_utc(data.get("annual_end")) return block( "computed", producer, producer_status=data.get("status"), year_length={ "basis": data.get("year_length_basis"), "days": data.get("annual_duration_days"), "start": instant(start, zone), "end": instant(end, zone), "note": "Mudda keeps its own year; it is not stretched to the next solar return", }, starting_lord=data.get("starting_lord"), starting_lord_basis=data.get("starting_lord_basis"), rule=data.get("rule"), completed_years=data.get("completed_years"), natal_moon_longitude=data.get("natal_moon_longitude"), natal_moon_nakshatra_number=data.get("natal_moon_nakshatra_number"), balance_profile=data.get("balance_profile"), balance_fraction_remaining=data.get("balance_fraction_remaining"), independent_of_year_lord=data.get("independent_of_year_lord"), source=data.get("source"), periods=_period_rows(data.get("periods") or [], to_utc, zone, range_start, range_end, keys=("order", "lord", "duration_days", "full_duration_days")), antardasha_periods=_period_rows(data.get("antardasha_periods") or [], to_utc, zone, range_start, range_end, keys=("main_order", "sub_order", "main_lord", "sub_lord", "duration_days")), ) def _patyayini_block(houses: dict, local_start, to_utc, zone, range_start, range_end) -> dict[str, Any]: producer = "annual_native.patyayini.build_native_patyayini" if houses.get("ascendant_longitude") is None or not isinstance(houses.get("planet_longitudes"), dict): return block("blocked", producer, reason="annual_chart_longitudes_missing") if local_start is None or to_utc is None: return block("blocked", producer, reason="solar_return_local_time_missing") build = _module("annual_native.patyayini").build_native_patyayini raw = build( ascendant_longitude=float(houses["ascendant_longitude"]), planet_longitudes=houses["planet_longitudes"], start=local_start, year_days=PATYAYINI_YEAR_DAYS, ) end = to_utc(local_start + timedelta(days=PATYAYINI_YEAR_DAYS)) return block( "computed", producer, producer_status=raw.get("status"), model=raw.get("model"), year_length={ "basis": "fixed_365_25_days", "days": PATYAYINI_YEAR_DAYS, "start": instant(to_utc(local_start), zone), "end": instant(end, zone), "note": "the native entry's fixed 365.25-day year; clipped for display, never stretched", }, inputs=[ {"point": row.get("lord"), "longitude": row.get("longitude"), "sign": row.get("sign"), "degree_in_sign": row.get("degree_in_sign")} for row in raw.get("periods") or [] ], periods=_period_rows(raw.get("periods") or [], to_utc, zone, range_start, range_end, keys=("order", "lord")), antardasha_periods=_period_rows(raw.get("antardasha_periods") or [], to_utc, zone, range_start, range_end, keys=("main", "sub", "main_order", "sub_order")), ) def _monthly_block(ctx, year, sr, houses, location, next_return, zone, range_start, range_end) -> dict[str, Any]: producer = "annual_native.monthly_charts.build_monthly_charts" annual_sun = (houses.get("planet_longitudes") or {}).get("Sun") if not isinstance(annual_sun, (int, float)): return block("blocked", producer, reason="annual_sun_longitude_missing") raw = _module("annual_native.monthly_charts").build_monthly_charts( annual_jd=float(sr["jd_ut"]), birth_sun_longitude=float(annual_sun), latitude=float(location.get("latitude", ctx.birth["lat"])), longitude=float(location.get("longitude", ctx.birth["lon"])), tz_hours=float(location.get("utc_offset") or ctx.tz_hours), ayanamsa=ctx.ayanamsa, node_mode=ctx.node_mode, target_year=year, timezone_id=location.get("timezone_id"), ) rows = [row for row in raw.get("rows") or [] if isinstance(row, dict)] starts = [(_utc_from_jd(row["jd_ut"]) if isinstance(row.get("jd_ut"), (int, float)) else None) for row in rows] out_rows = [] for index, row in enumerate(rows): start = starts[index] if row.get("status") != "partial_verified" or start is None: out_rows.append({"month_index": row.get("month_index"), **block("blocked", producer, reason=str(row.get("reason") or "monthly_root_failed"))}) continue end = starts[index + 1] if index + 1 < len(starts) and starts[index + 1] is not None else ( next_return if index == len(rows) - 1 else None) out_rows.append({ "month_index": row.get("month_index"), "status": "computed", "start": instant(start, zone), "end": instant(end, zone) if end else None, "end_source": "next_monthly_chart" if index < len(rows) - 1 else "next_solar_return", "in_report_range": _clip(start, end, range_start, range_end, zone) if end and range_start and range_end else None, "target_sun_longitude": row.get("target_sun_longitude"), "sun_longitude": row.get("sun_longitude"), "ascendant": row.get("ascendant"), "planet_longitudes": row.get("planet_longitudes"), }) ready = sum(1 for row in out_rows if row.get("status") == "computed") return block( "computed" if ready == 12 else "blocked", producer, reason=None if ready == 12 else f"expected 12 monthly charts, got {ready}", producer_status=raw.get("status"), step="solar_30_degrees", step_note="each chart starts when the Sun has advanced another 30 degrees from the annual Sun; not a calendar month, not a lunar return", rows=out_rows, ) # ── independent annual view ───────────────────────────────────────────────── def _tithi_pravesha(ctx: _Context, year: int, range_start, range_end) -> dict[str, Any]: producer = "annual_native.tithi_pravesh.build_tithi_pravesh" raw = _module("annual_native.tithi_pravesh").build_tithi_pravesh(ctx.pack_payload(year)) zone = timezone(timedelta(hours=ctx.tz_hours)) rows = [] for row in raw.get("rows") or []: if not isinstance(row, dict) or not isinstance(row.get("jd_ut"), (int, float)): continue moment = _utc_from_jd(row["jd_ut"]) rows.append({ "instant": instant(moment, zone), "in_report_range": bool(range_start <= moment < range_end), "tithi": row.get("tithi"), "phase_degrees": row.get("phase_degrees"), "phase_delta_degrees": row.get("phase_delta_degrees"), "planet_longitudes": row.get("planet_longitudes"), }) if raw.get("status") == "blocked" or not rows: return {"target_year": year, **block("blocked", producer, reason=str(raw.get("reason") or "no_candidate"))} return { "target_year": year, **block("computed", producer, producer_status=raw.get("status")), "natal_phase_degrees": raw.get("natal_phase_degrees"), "candidate_count": len(rows), "candidates": rows, "candidate_note": "every candidate the producer found is listed; none is chosen here", } # ── natal long periods ────────────────────────────────────────────────────── def _vimshottari(ctx: _Context, range_start: datetime, range_end: datetime) -> dict[str, Any]: producer = "jyotish_engine.cmd_dasha" result = _engine().cmd_dasha(ctx.engine_args()) timeline = result.get("timeline") if isinstance(result, dict) else None if not isinstance(timeline, list) or not timeline: return block("blocked", producer, reason=str((result or {}).get("error") or "timeline_missing")) to_utc = _elapsed_mapper(ctx.birth_naive, ctx.birth_utc) zone = ctx.zone def overlapping(rows, level): out = [] for row in rows or []: start = to_utc(row.get("start_datetime")) end = to_utc(row.get("end_datetime")) if start is None or end is None or not (start < range_end and end > range_start): continue out.append((row, {"level": level, "lord": row.get("lord"), "start": instant(start, zone), "end": instant(end, zone), "in_report_range": _clip(start, end, range_start, range_end, zone)})) return out rows = [] for md, md_row in overlapping(timeline, "MD"): rows.append(md_row) for ad, ad_row in overlapping(md.get("antardasha_timeline"), "AD"): ad_row["mahadasha_lord"] = md.get("lord") rows.append(ad_row) for _pd, pd_row in overlapping(ad.get("pratyantar_dasha_timeline"), "PD"): pd_row["mahadasha_lord"] = md.get("lord") pd_row["antardasha_lord"] = ad.get("lord") rows.append(pd_row) return block( "computed", producer, moon_nakshatra=result.get("moon_nakshatra"), start_basis="birth_moment", year_length={"days": VIMSHOTTARI_YEAR_DAYS, "basis": "cmd_dasha 365.25-day year"}, boundary_time_basis="birth local clock (fixed birth offset) mapped to UTC by elapsed time", rows=rows, ) def _age_mapper(ctx: _Context) -> Callable[[Any], datetime | None]: def convert(age: Any) -> datetime | None: if isinstance(age, bool) or not isinstance(age, (int, float)): return None return ctx.birth_utc + timedelta(days=float(age) * NARAYANA_AGE_YEAR_DAYS) return convert def _age_rows(rows, to_utc, zone, range_start, range_end, *, level: str, extra: tuple[str, ...] = ()) -> list[dict]: out = [] for row in rows or []: if not isinstance(row, dict): continue start, end = to_utc(row.get("start_age")), to_utc(row.get("end_age")) if start is None or end is None or not (start < range_end and end > range_start): continue item = {"level": level, "sign": row.get("sign"), "lord": row.get("lord"), "start_age": row.get("start_age"), "end_age": row.get("end_age"), "start": instant(start, zone), "end": instant(end, zone), "in_report_range": _clip(start, end, range_start, range_end, zone)} for key in extra: if key in row: item[key] = row[key] out.append(item) return out def _narayana(ctx: _Context, natal: dict, range_start: datetime, range_end: datetime) -> dict[str, Any]: producer = "narayana_dasha / narayana_rath" if natal.get("status") != "computed": return block("blocked", producer, reason="natal_chart_blocked") lagna = int(natal["lagna_sign_idx"]) lons = dict(natal["narayana_input_longitudes"]) to_utc = _age_mapper(ctx) zone = ctx.zone age_at_obs = (_local_midnight(ctx.observation_date, zone) - ctx.birth_utc).total_seconds() / 86400.0 / NARAYANA_AGE_YEAR_DAYS common = { "age_axis": {"year_days": NARAYANA_AGE_YEAR_DAYS, "basis": "date = birth instant + age x 365.25 days (the engine's current_age convention)"}, } def legacy() -> dict[str, Any]: nd = _module("narayana_dasha") label = _module("narayana_legacy_label") report = nd.narayana_dasha_full_report(lagna_sign_idx=lagna, planet_lons=lons, current_age=max(age_at_obs, 0.0), birth_year=int(ctx.birth["year"])) labelled = label.label_legacy_narayana(report) dense = [row for row in report.get("dense_boundaries") or [] if isinstance(row, dict)] last_end = max((float(row["end_age"]) for row in dense if isinstance(row.get("end_age"), (int, float))), default=None) rows = (_age_rows([r for r in dense if r.get("level") == "MD"], to_utc, zone, range_start, range_end, level="MD") + _age_rows([r for r in dense if r.get("level") == "AD"], to_utc, zone, range_start, range_end, level="AD", extra=("parent_md",))) if last_end is not None and to_utc(last_end) < range_end: return block("blocked", "narayana_dasha.narayana_dasha_full_report", reason="report_range_beyond_single_cycle") return block("computed", "narayana_dasha.narayana_dasha_full_report", profile=labelled.get("profile", label.LEGACY_PROFILE), profile_note=label.LEGACY_NOTE_EN, used_by_product_timeline=True, age_precision="MD ages exact sums of whole years; AD ages rounded by the producer to 4 decimals (about 53 minutes)", rows=rows) def rath() -> dict[str, Any]: nr = _module("narayana_rath") result = nr.calc_narayana_rath(lagna % 12, lons, cycle_count=2) if not result.get("mahadasha"): return block("blocked", "narayana_rath.calc_narayana_rath", reason=str(result.get("reason") or result.get("status") or "seed_undecided"), producer_status=result.get("status"), alternative_seed_signs=sorted((result.get("alternatives") or {}).keys())) md_rows = _age_rows(result["mahadasha"], to_utc, zone, range_start, range_end, level="MD") ad_rows: list[dict] = [] ad_blocked: list[dict] = [] for md in result["mahadasha"]: start, end = to_utc(md.get("start_age")), to_utc(md.get("end_age")) if start is None or end is None or not (start < range_end and end > range_start): continue ad = nr.antardasha(md, lons) if ad.get("status") != "selected": ad_blocked.append({"parent_md": md.get("sign"), "reason": ad.get("reason")}) continue for row in _age_rows(ad.get("rows"), to_utc, zone, range_start, range_end, level="AD"): row["parent_md"] = md.get("sign") ad_rows.append(row) return block("computed", "narayana_rath.calc_narayana_rath", producer_status=result.get("status"), seed_sign=result.get("seed_sign"), seed_basis=result.get("seed_basis"), direction=result.get("direction"), used_by_product_timeline=False, profile_note="Sanjay Rath profile (astrologer's ruling); shown alongside, product timeline not switched", antardasha_blocked=ad_blocked, rows=md_rows + ad_rows) return block("computed", producer, **common, legacy=_guard("narayana_dasha.narayana_dasha_full_report", legacy), rath=_guard("narayana_rath.calc_narayana_rath", rath)) # ── slow transits ─────────────────────────────────────────────────────────── class _Positions: """Daily-sampled sidereal positions in the chart engine's frame.""" def __init__(self, ctx: _Context): self.engine = _engine() self.ctx = ctx self.cache: dict[float, dict[str, tuple[float, bool]]] = {} def at(self, jd: float) -> dict[str, tuple[float, bool]]: key = round(jd, 9) if key not in self.cache: data, _ = self.engine._calc_sidereal_planets_for_jd( jd, node_mode=self.ctx.node_mode, include_ketu=True, ayanamsa_name=self.ctx.ayanamsa) self.cache[key] = { name: (float(row["degree_raw"]) % 360.0, bool(row.get("retrograde"))) for name, row in (data or {}).items() if isinstance(row, dict) and row.get("degree_raw") is not None } return self.cache[key] def lon(self, jd: float, body: str) -> float: return self.at(jd)[body][0] def retro(self, jd: float, body: str) -> bool: return self.at(jd)[body][1] def _bisect(pred: Callable[[float], bool], lo: float, hi: float) -> float: """``pred(lo)`` holds, ``pred(hi)`` does not; return the first instant where it stops.""" for _ in range(BISECT_STEPS): mid = (lo + hi) / 2.0 if pred(mid): lo = mid else: hi = mid return hi def _slow_transits(ctx: _Context, natal: dict, range_start: datetime, range_end: datetime) -> dict[str, Any]: producer = "jyotish_engine._calc_sidereal_planets_for_jd" if natal.get("status") != "computed": return block("blocked", producer, reason="natal_chart_blocked") zone = ctx.zone pos = _Positions(ctx) jd_start, jd_end = _jd_from_utc(range_start), _jd_from_utc(range_end) scan_start = math.floor(jd_start - TRANSIT_MARGIN_DAYS) + 0.5 days = int(math.ceil(jd_end + TRANSIT_MARGIN_DAYS - scan_start)) samples = [scan_start + offset for offset in range(days + 1)] lagna_idx = natal["lagna_sign_idx"] moon_idx = natal.get("moon_sign_idx") def in_range(jd: float) -> bool: return jd_start <= jd < jd_end def at_instant(jd: float) -> dict[str, Any]: return instant(_utc_from_jd(jd), zone) events: list[dict[str, Any]] = [] stations: dict[str, list[dict[str, Any]]] = {body: [] for body in SLOW_BODIES} for body in SLOW_BODIES: for lo, hi in zip(samples, samples[1:], strict=False): s0, s1 = _sign_idx(pos.lon(lo, body)), _sign_idx(pos.lon(hi, body)) if s0 != s1: jd = _bisect(lambda t, b=body, s=s0: _sign_idx(pos.lon(t, b)) == s, lo, hi) if in_range(jd): to_idx = _sign_idx(pos.lon(jd, body)) events.append({ "kind": "ingress", "body": body, "instant": at_instant(jd), "from_sign": SIGNS[s0], "to_sign": SIGNS[to_idx], "to_house_from_lagna": _house(to_idx, lagna_idx), "to_house_from_moon": _house(to_idx, moon_idx), "retrograde": pos.retro(jd, body), "axis_group": "rahu_ketu" if body in ("Rahu", "Ketu") else body.lower(), }) r0, r1 = pos.retro(lo, body), pos.retro(hi, body) if r0 != r1 and body != "Ketu": # Ketu is Rahu + 180 deg; its stations are Rahu's jd = _bisect(lambda t, b=body, r=r0: pos.retro(t, b) == r, lo, hi) station = { "kind": "station", "body": body, "instant": at_instant(jd), "_jd": jd, "motion": "direct_to_retrograde" if r1 else "retrograde_to_direct", "longitude": pos.lon(jd, body), "sign": SIGNS[_sign_idx(pos.lon(jd, body))], } stations[body].append(station) if in_range(jd): events.append({k: v for k, v in station.items() if k != "_jd"}) events.sort(key=lambda row: (row["instant"]["jd_ut"], row["body"], row["kind"])) def crossings(body: str, target: float, lo_jd: float, hi_jd: float) -> list[float]: found = [] window = [jd for jd in samples if lo_jd - 1 <= jd <= hi_jd + 1] for lo, hi in zip(window, window[1:], strict=False): f0, f1 = _wrap(pos.lon(lo, body) - target), _wrap(pos.lon(hi, body) - target) if abs(f0) < 30 and abs(f1) < 30 and ((f0 < 0 <= f1) or (f1 < 0 <= f0)): negative = f0 < 0 jd = _bisect(lambda t, neg=negative: (_wrap(pos.lon(t, body) - target) < 0) == neg, lo, hi) if lo_jd <= jd < hi_jd: found.append(jd) return found natal_points = natal["longitudes"] contact_rows = [] for body in SLOW_BODIES: for point, target in natal_points.items(): hits = crossings(body, target, jd_start, jd_end) if not hits: continue contact_rows.append({ "body": body, "natal_point": point, "natal_longitude": target, "contact": "exact_conjunction_by_longitude", "count": len(hits), "repeated": len(hits) > 1, "crossings": [{"instant": at_instant(jd), "retrograde": pos.retro(jd, body)} for jd in hits], "axis_group": "rahu_ketu" if body in ("Rahu", "Ketu") else body.lower(), }) loops = [] first_sample, last_sample = samples[0], samples[-1] for body in LOOP_BODIES: rows = stations[body] for index, station in enumerate(rows): if station["motion"] != "direct_to_retrograde" or index + 1 >= len(rows): continue direct = rows[index + 1] if direct["motion"] != "retrograde_to_direct": continue lon_r, lon_d = station["longitude"], direct["longitude"] entry = crossings(body, lon_d, first_sample, station["_jd"]) exit_ = crossings(body, lon_r, direct["_jd"], last_sample) loop_start = entry[-1] if entry else station["_jd"] loop_end = exit_[0] if exit_ else direct["_jd"] if not (loop_start < jd_end and loop_end > jd_start): continue arc = (lon_r - lon_d) % 360.0 inside = [ {"natal_point": point, "natal_longitude": target, "crossings": [{"instant": at_instant(jd), "retrograde": pos.retro(jd, body), "in_report_range": in_range(jd)} for jd in crossings(body, target, loop_start - 1, loop_end + 1)]} for point, target in natal_points.items() if (target - lon_d) % 360.0 <= arc ] loops.append({ "body": body, "station_retrograde": {"instant": station["instant"], "longitude": lon_r, "sign": station["sign"]}, "station_direct": {"instant": direct["instant"], "longitude": lon_d, "sign": direct["sign"]}, "retraced_arc": {"from_longitude": lon_d, "to_longitude": lon_r, "degrees": arc}, "pre_shadow_entry": at_instant(entry[-1]) if entry else None, "post_shadow_exit": at_instant(exit_[0]) if exit_ else None, "shadow_note": None if entry and exit_ else "shadow boundary outside the scan window", "natal_points_in_arc": inside, }) start_positions = {} for body in SLOW_BODIES: lon = pos.lon(jd_start, body) idx = _sign_idx(lon) start_positions[body] = {"longitude": lon, "sign": SIGNS[idx], "degree_in_sign": lon % 30.0, "retrograde": pos.retro(jd_start, body), "house_from_lagna": _house(idx, lagna_idx), "house_from_moon": _house(idx, moon_idx)} return block( "computed", producer, bodies=list(SLOW_BODIES), node_mode=ctx.node_mode, ayanamsa=ctx.ayanamsa, method="daily sample at 12:00 UT, then bisection to under one second; events closer than one day apart may merge", reference_points={"lagna_sign": SIGNS[lagna_idx], "moon_sign": SIGNS[moon_idx] if moon_idx is not None else None, "house_basis": "whole sign from natal Lagna and from natal Moon"}, positions_at_range_start=start_positions, events=events, natal_point_contacts=contact_rows, retrograde_loops=loops, retrograde_loop_scope="Jupiter and Saturn; Rahu/Ketu are excluded (their mean motion is retrograde)", ) # ── static boundaries ─────────────────────────────────────────────────────── def _boundaries(ctx: _Context) -> dict[str, Any]: return { "instant_format": "every boundary has utc, local (ISO with offset), utc_offset, timezone_id and jd_ut", "local_zone": "return location zone (birth place in v1); IANA zone when timezone_id is given, else the fixed birth offset", "year_chart_validity": "this solar return (inclusive) to the next solar return (exclusive)", "clipping": "period rows keep their native bounds; in_report_range is the part inside the report range; nothing is stretched", "period_year_lengths": [ {"system": "mudda_dasha", "days": 365.25, "anchor": "solar return instant"}, {"system": "patyayini", "days": PATYAYINI_YEAR_DAYS, "anchor": "solar return instant"}, {"system": "vimshottari", "days": VIMSHOTTARI_YEAR_DAYS, "anchor": "birth moment"}, {"system": "narayana", "days": NARAYANA_AGE_YEAR_DAYS, "anchor": "birth moment, age axis"}, ], "naive_local_to_utc": "producers that add day counts to a naive local start are mapped to UTC by elapsed time from the anchor instant", "monthly_charts": "Sun advancing 30 degrees per chart from the annual Sun; month 12 ends at the next solar return", "slow_transit_frame": "jyotish_engine._calc_sidereal_planets_for_jd (same frame as the natal chart), node mode and ayanamsa as requested", "candidate_windows": "not built here; the report phase builds them from these instants", } NOT_INCLUDED = [ {"item": "varsha_narayana", "reason": "no native producer in this repository (upstream yinduzhanxing has one; not synced)"}, {"item": "yoga_and_nakshatra_pravesha", "reason": "research-only upstream; the ruling keeps them out of the report"}, {"item": "external_engine_replay", "reason": "PyJHora annual replay runs only in the full-data report child process; not executed here"}, {"item": "residence_return_location", "reason": "v1 offers the birth place only; return_location keeps the shape for a later comparison"}, {"item": "transit_ashtakavarga_vedha_sade_sati", "reason": "report phase; BAV/SAV come from /api/ashtakavarga, Vedha and Sade Sati phase are not executed here"}, {"item": "candidate_windows", "reason": "report phase (#153)"}, {"item": "real_case_calibration", "reason": "blocked per the 2026-11-01 ruling; no accuracy claim"}, ] def _walk_blocks(value: Any, path: str): if isinstance(value, dict): if value.get("status") in ("computed", "blocked", "not_applicable") and "producer" in value: yield {"path": path or "$", "status": value["status"], "producer": value["producer"], "reason": value.get("reason")} for key, item in value.items(): if key in ("block_index",): continue yield from _walk_blocks(item, f"{path}.{key}" if path else key) elif isinstance(value, list): for index, item in enumerate(value): yield from _walk_blocks(item, f"{path}[{index}]")