root/Jyotisha
2
0
Fork 0
Code Issues Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

feat: use domain calculation contract in commercial API

Browse Source
This commit is contained in:
732642856
2026-07-16 14:14:25 +08:00
parent 274fdfe32a
commit 61655fe060
4 changed files with 345 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
references/cross_project_contract/sync_policy.v1.json
+1
View File
@@ -29,6 +29,7 @@
"references/cross_project_contract/sync_ledger.json",
"references/cross_project_contract/sync_policy.v1.json",
"scripts/cross_project_contract.py",
"scripts/domain_calculation_service.py",
"scripts/cross_project_sync_status.py",
"tests/test_cross_project_contract.py",
"tests/test_cross_project_sync_status.py"
scripts/domain_calculation_service.py
+260
View File
@@ -0,0 +1,260 @@
#!/usr/bin/env python3
"""Canonical calculation service shared by CLI, REST, and MCP adapters."""
from __future__ import annotations
import hashlib
import json
import math
import threading
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo
import swisseph as swe
from ayanamsa_utils import apply_ayanamsa, normalize_ayanamsa_name
from dasha_analyzer import build_dasha_timeline, lon_to_nakshatra
from jyotish_engine import SIGNS, compute_chart_data
from sade_sati import calc_sade_sati_complete
CONTRACT_VERSION = "1.0.0"
_SWISSEPH_LOCK = threading.RLock()
_PLANET_IDS = {"Saturn": swe.SATURN}
class CalculationError(ValueError):
pass
class TimezoneInferenceError(CalculationError):
pass
def _canonical_hash(payload: dict[str, Any]) -> str:
encoded = json.dumps(
payload,
ensure_ascii=True,
sort_keys=True,
separators=(",", ":"),
default=str,
).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
def _lookup_timezone_name(lat: float, lon: float) -> str | None:
try:
from timezonefinder import TimezoneFinder
except ImportError as exc:
raise TimezoneInferenceError("timezone inference dependency unavailable") from exc
return TimezoneFinder().timezone_at(lng=lon, lat=lat)
def infer_timezone_offset(*, lat: float, lon: float, local_datetime: datetime) -> float:
if not (-90 <= lat <= 90 and -180 <= lon <= 180):
raise TimezoneInferenceError("timezone inference received invalid coordinates")
tz_name = _lookup_timezone_name(lat, lon)
if not tz_name:
raise TimezoneInferenceError("timezone inference returned no IANA zone")
try:
offset = local_datetime.replace(tzinfo=ZoneInfo(tz_name)).utcoffset()
except Exception as exc:
raise TimezoneInferenceError("timezone inference failed for IANA zone") from exc
if offset is None:
raise TimezoneInferenceError("timezone inference returned no UTC offset")
return offset.total_seconds() / 3600.0
def _normalized_request(payload: dict[str, Any]) -> dict[str, Any]:
requested_node = str(payload.get("node_mode", payload.get("nodeMode", "mean"))).lower()
if requested_node not in {"mean", "true"}:
raise CalculationError("node_mode must be mean or true")
ayanamsa = normalize_ayanamsa_name(payload.get("ayanamsa", "lahiri"))
local_dt = datetime(
int(payload["year"]),
int(payload["month"]),
int(payload["day"]),
int(float(payload.get("hour", 0))),
int(float(payload.get("minute", 0))),
int(float(payload.get("second", 0))),
)
lat = float(payload["lat"])
lon = float(payload["lon"])
tz_requested = payload.get("tz")
timezone_source = "explicit_offset"
if tz_requested in {None, ""}:
tz = infer_timezone_offset(lat=lat, lon=lon, local_datetime=local_dt)
timezone_source = "iana_inferred"
else:
tz = float(tz_requested)
if not math.isfinite(tz) or not -14 <= tz <= 14:
raise CalculationError("tz must be a finite offset between -14 and 14")
return {
"year": local_dt.year,
"month": local_dt.month,
"day": local_dt.day,
"hour": int(float(payload.get("hour", 0))),
"minute": int(float(payload.get("minute", 0))),
"second": int(float(payload.get("second", 0))),
"lat": lat,
"lon": lon,
"tz": tz,
"timezone_source": timezone_source,
"ayanamsa": ayanamsa,
"node_mode": requested_node,
}
def _contract(requested: dict[str, Any], effective: dict[str, Any], *, algorithm: str) -> dict[str, Any]:
return {
"contract_version": CONTRACT_VERSION,
"algorithm": algorithm,
"requested": requested,
"effective": effective,
}
def compute_chart(payload: dict[str, Any]) -> dict[str, Any]:
request = _normalized_request(payload)
with _SWISSEPH_LOCK:
chart, _asc_idx, _jd, _ayanamsa = compute_chart_data(
request["year"],
request["month"],
request["day"],
request["hour"],
request["minute"],
request["lat"],
request["lon"],
request["tz"],
node_mode=request["node_mode"],
second=request["second"],
ayanamsa_name=request["ayanamsa"],
)
if not isinstance(chart, dict):
raise CalculationError("canonical chart calculation failed")
for planet in chart.get("planets", {}).values():
if not isinstance(planet, dict) or "error" in planet:
continue
planet.setdefault("lon", planet.get("degree_raw", planet.get("degree")))
if planet.get("sign") in SIGNS:
planet.setdefault("sign_idx", SIGNS.index(planet["sign"]))
birth = chart.get("birth_info", {})
effective = {
"ayanamsa": birth.get("ayanamsa_name", request["ayanamsa"]),
"node_mode": birth.get("node_mode", request["node_mode"]),
"timezone_offset": request["tz"],
"timezone_source": request["timezone_source"],
"ephemeris_source": "swisseph_calc_ut",
"ephemeris_flags_verified": False,
}
requested = {
"ayanamsa": payload.get("ayanamsa", "lahiri"),
"node_mode": payload.get("node_mode", payload.get("nodeMode", "mean")),
"timezone_offset": payload.get("tz"),
}
contract = _contract(requested, effective, algorithm="sidereal_natal_chart")
hash_payload = {
"contract": contract,
"birth": birth,
"ascendant": chart.get("ascendant"),
"planets": chart.get("planets"),
}
chart["calculation_contract"] = contract
chart["result_hash"] = _canonical_hash(hash_payload)
return chart
def compute_vimshottari_timeline(
*, birth_dt: datetime, moon_lon: float, current_date: datetime | None = None
) -> dict[str, Any]:
nak_info, progress, pada = lon_to_nakshatra(float(moon_lon) % 360)
timeline, elapsed, remaining, start_lord = build_dasha_timeline(
birth_dt.strftime("%Y-%m-%d"), nak_info, progress
)
periods = [
{
"lord": period["lord"],
"years": period["years"],
"start": period["start"].strftime("%Y-%m-%d"),
"end": period["end"].strftime("%Y-%m-%d"),
}
for period in timeline
]
contract = _contract(
{"moon_longitude": float(moon_lon) % 360},
{"year_basis_days": 365.25, "nakshatra": nak_info[0], "pada": pada},
algorithm="vimshottari_birth_balance",
)
result = {
"periods": periods,
"birth_balance": {
"lord": start_lord,
"elapsed_years": elapsed,
"remaining_years": remaining,
},
"calculation_contract": contract,
}
result["result_hash"] = _canonical_hash(result)
return result
def compute_transit_longitude(
*, planet: str, reference_date: str, tz: float, ayanamsa: str = "lahiri"
) -> dict[str, Any]:
if planet not in _PLANET_IDS:
raise CalculationError(f"unsupported transit planet: {planet}")
try:
local_dt = datetime.strptime(reference_date[:10], "%Y-%m-%d").replace(hour=12)
except (TypeError, ValueError) as exc:
raise CalculationError("reference_date must be YYYY-MM-DD") from exc
ayanamsa_name = normalize_ayanamsa_name(ayanamsa)
with _SWISSEPH_LOCK:
apply_ayanamsa(ayanamsa_name, swe)
jd = swe.julday(
local_dt.year,
local_dt.month,
local_dt.day,
12.0 - float(tz),
)
ayanamsa_value = swe.get_ayanamsa(jd)
position, flags = swe.calc_ut(jd, _PLANET_IDS[planet])
longitude = (position[0] - ayanamsa_value) % 360
return {
"planet": planet,
"longitude": longitude,
"reference_date": reference_date[:10],
"ayanamsa": ayanamsa_name,
"timezone_offset": float(tz),
"swisseph_return_flags": int(flags),
"data_layer": "true_transit_positions",
}
def compute_sade_sati(
*,
moon_degree: float,
asc_degree: float,
reference_date: str,
tz: float,
ayanamsa: str = "lahiri",
) -> dict[str, Any]:
transit = compute_transit_longitude(
planet="Saturn",
reference_date=reference_date,
tz=tz,
ayanamsa=ayanamsa,
)
result = calc_sade_sati_complete(
float(moon_degree) % 360,
float(asc_degree) % 360,
transit["longitude"],
datetime.strptime(reference_date[:10], "%Y-%m-%d"),
)
result["transit_saturn_lon"] = transit["longitude"]
result["provenance"] = transit
result["calculation_contract"] = _contract(
{"reference_date": reference_date[:10], "ayanamsa": ayanamsa, "tz": tz},
transit,
algorithm="sade_sati_true_saturn_transit",
)
result["result_hash"] = _canonical_hash(result)
return result
scripts/jyotish_api_server.py
+26 -1
View File
@@ -4405,6 +4405,25 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
except Exception as e:
import logging
logging.warning(f"[api_server] yoga expansion detection failed: {e}")
calculation_service = _load_local_module('domain_calculation_service')
canonical_chart = calculation_service.compute_chart({
'year': year,
'month': month,
'day': day,
'hour': hour,
'minute': minute,
'second': second,
'lat': lat,
'lon': lon,
'tz': tz,
'ayanamsa': body.get('ayanamsa', 'lahiri'),
'node_mode': body.get('node_mode', body.get('nodeMode', 'mean')),
})
canonical_dasha = calculation_service.compute_vimshottari_timeline(
birth_dt=birth_dt,
moon_lon=moon_lon,
current_date=datetime.utcnow(),
)
result = {
'success': True, 'version': '6.9.15',
'birth': {
@@ -4420,7 +4439,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
'ayanamsa': round(ayanamsa, 4),
'ayanamsa_name': ayanamsa_name,
'ayanamsa_display': ayanamsa_display,
'node_mode': body.get('node_mode', body.get('nodeMode', 'mean')),
'node_mode': canonical_chart['calculation_contract']['effective']['node_mode'],
},
'ascendant': {
'sign': asc_sign,
@@ -4435,6 +4454,10 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
'remaining_years': round(remaining, 2),
'total_years': total_years,
'start_date': dasha_start.isoformat() if hasattr(dasha_start, 'isoformat') else str(dasha_start),
'periods': canonical_dasha['periods'],
'birth_balance': canonical_dasha['birth_balance'],
'calculation_contract': canonical_dasha['calculation_contract'],
'result_hash': canonical_dasha['result_hash'],
},
'yogas': yogas,
'sade_sati': sade_sati,
@@ -4443,6 +4466,8 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
'special_lagnas': special_lagnas,
'available_dashas': dasha_list,
'dasha_count': len(dasha_list),
'calculation_contract': canonical_chart['calculation_contract'],
'result_hash': canonical_chart['result_hash'],
}
result['modules'] = {
'chart': {
tests/test_commercial_domain_calculation_contract.py
+58
View File
@@ -0,0 +1,58 @@
from __future__ import annotations
import sys
from pathlib import Path
import pytest
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
if str(SCRIPTS) not in sys.path:
sys.path.insert(0, str(SCRIPTS))
import domain_calculation_service as calculation_service # noqa: E402
import jyotish_api_server # noqa: E402
from jyotish_api_server import JyotishAPIHandler # noqa: E402
BIRTH = {
"year": 1990,
"month": 1,
"day": 1,
"hour": 12,
"minute": 0,
"second": 0,
"lat": 28.6139,
"lon": 77.2090,
"tz": 5.5,
"ayanamsa": "lahiri",
}
def test_domain_chart_exposes_effective_params_and_result_hash() -> None:
mean = calculation_service.compute_chart({**BIRTH, "node_mode": "mean"})
true = calculation_service.compute_chart({**BIRTH, "node_mode": "true"})
assert mean["calculation_contract"]["effective"]["ayanamsa"] == "lahiri"
assert true["calculation_contract"]["effective"]["node_mode"] == "true"
assert mean["planets"]["Rahu"]["lon"] != pytest.approx(true["planets"]["Rahu"]["lon"], abs=1e-8)
assert mean["result_hash"] != true["result_hash"]
def test_api_chart_response_uses_domain_contract_hash(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("JYOTISH_API_CHART_CACHE_TTL_SECONDS", "0")
monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "0")
monkeypatch.setattr(
jyotish_api_server,
"_attach_vedastro_main_entry_overview",
lambda result, _birth: result,
)
expected = calculation_service.compute_chart({**BIRTH, "node_mode": "true"})
rest = JyotishAPIHandler.__new__(JyotishAPIHandler)._compute_chart_sync(
{**BIRTH, "node_mode": "true", "transit_date": "2026-07-11"}
)
assert rest["result_hash"] == expected["result_hash"]
assert rest["birth"]["node_mode"] == "true"
assert rest["calculation_contract"]["effective"]["node_mode"] == "true"