#!/usr/bin/env python3 """Minimal VedAstro service-boundary adapter skeleton. This module does not replace the local SwissEph path. It only defines the request/response schema and a controlled "not configured" status so the workspace can evolve from research notes to an executable adapter contract. """ from __future__ import annotations import argparse import json import os import socket from pathlib import Path from typing import Any from urllib import request, error ROOT = Path(__file__).resolve().parents[1] PARITY_CASES = { "beijing_first_use_demo": { "year": 1990, "month": 1, "day": 1, "hour": 12, "minute": 0, "lat": 39.9042, "lon": 116.4074, "tz": 8.0, "ayanamsa_policy": "lahiri", "node_policy": "mean", }, "delhi_lagna_boundary": { "year": 1984, "month": 10, "day": 31, "hour": 6, "minute": 30, "lat": 28.6139, "lon": 77.2090, "tz": 5.5, "ayanamsa_policy": "lahiri", "node_policy": "mean", }, "new_york_moon_boundary": { "year": 2001, "month": 9, "day": 11, "hour": 8, "minute": 46, "lat": 40.7128, "lon": -74.0060, "tz": -4.0, "ayanamsa_policy": "lahiri", "node_policy": "mean", }, } DEFAULT_TIMEOUT_SECONDS = 8 TIMEOUT_ENV = "VEDASTRO_TIMEOUT_SECONDS" RETRY_POLICY = { "max_attempts": 2, "backoff_seconds": 1, "retry_on": ["timeout", "429", "502", "503", "504"], } ALLOW_NETWORK_ENV = "VEDASTRO_ENABLE_NETWORK" def _timeout_seconds() -> float: raw = os.environ.get(TIMEOUT_ENV, "").strip() if not raw: return DEFAULT_TIMEOUT_SECONDS try: return float(raw) except ValueError: return DEFAULT_TIMEOUT_SECONDS def schema() -> dict[str, Any]: request_example = { **PARITY_CASES["beijing_first_use_demo"], "body_list": ["Sun", "Moon", "Ascendant", "Rahu", "Ketu"], } return { "adapter": "vedastro_service_adapter", "backend": "vedastro_service_adapter_candidate", "transport": "http_json_service_boundary", "default_timeout_seconds": DEFAULT_TIMEOUT_SECONDS, "retry_policy": RETRY_POLICY, "required_env": { "endpoint": "VEDASTRO_API_ENDPOINT", "api_key_optional": "VEDASTRO_API_KEY", }, "request_contract": [ "year", "month", "day", "hour", "minute", "lat", "lon", "tz", "ayanamsa_policy", "node_policy", "body_list", ], "response_contract": [ "backend", "available", "status", "ayanamsa_value", "node_policy", "body_list", "bodies", "source_metadata", ], "request_example": request_example, "provenance_contract": { "external_service": True, "required_fields": [ "endpoint", "transport", "provenance_mode", "retry_policy", "timeout_seconds", ], }, } def _unconfigured(reason: str) -> dict[str, Any]: return { "backend": "vedastro_service_adapter_candidate", "available": False, "status": "service_endpoint_not_configured", "reason": reason, "source_metadata": { "transport": "http_json_service_boundary", "endpoint_env": "VEDASTRO_API_ENDPOINT", "api_key_env": "VEDASTRO_API_KEY", "provenance_mode": "external_service_candidate", "timeout_seconds": _timeout_seconds(), "retry_policy": RETRY_POLICY, }, } def _request_preview(case: dict[str, Any]) -> dict[str, Any]: return { **case, "body_list": ["Sun", "Moon", "Ascendant", "Rahu", "Ketu"], } def _normalize_success(payload: dict[str, Any], endpoint: str) -> dict[str, Any]: return { "backend": "vedastro_service_adapter_candidate", "available": True, "status": "ok", "ayanamsa_value": payload.get("ayanamsa_value"), "node_policy": payload.get("node_policy"), "body_list": payload.get("body_list"), "bodies": payload.get("bodies"), "source_metadata": { "transport": "http_json_service_boundary", "endpoint": endpoint, "provenance_mode": "external_service_candidate", "timeout_seconds": _timeout_seconds(), "retry_policy": RETRY_POLICY, **(payload.get("source_metadata") or {}), }, } def run_case(case_id: str) -> dict[str, Any]: if case_id not in PARITY_CASES: return { "backend": "vedastro_service_adapter_candidate", "available": False, "status": "unknown_case_id", "reason": f"Unknown parity case: {case_id}", } endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip() if not endpoint: return _unconfigured("VEDASTRO_API_ENDPOINT is not configured; adapter skeleton stops before network access.") case = PARITY_CASES[case_id] request_preview = _request_preview(case) if os.environ.get(ALLOW_NETWORK_ENV, "").strip().lower() not in {"1", "true", "yes"}: return { "backend": "vedastro_service_adapter_candidate", "available": False, "status": "network_execution_disabled", "reason": f"{ALLOW_NETWORK_ENV} is not enabled; adapter stops after building request/provenance metadata.", "request_preview": request_preview, "source_metadata": { "transport": "http_json_service_boundary", "endpoint": endpoint, "provenance_mode": "external_service_candidate", "timeout_seconds": _timeout_seconds(), "retry_policy": RETRY_POLICY, "network_execution_env": ALLOW_NETWORK_ENV, }, } try: req = request.Request( endpoint, data=json.dumps(request_preview).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) with request.urlopen(req, timeout=_timeout_seconds()) as resp: raw = resp.read().decode("utf-8") payload = json.loads(raw) except error.HTTPError as exc: return { "backend": "vedastro_service_adapter_candidate", "available": False, "status": "http_error", "reason": f"VedAstro adapter HTTP error: {exc.code}", "request_preview": request_preview, "source_metadata": { "transport": "http_json_service_boundary", "endpoint": endpoint, "provenance_mode": "external_service_candidate", "timeout_seconds": _timeout_seconds(), "retry_policy": RETRY_POLICY, "network_execution_env": ALLOW_NETWORK_ENV, }, } except error.URLError as exc: return { "backend": "vedastro_service_adapter_candidate", "available": False, "status": "network_error", "reason": f"VedAstro adapter network error: {exc.reason}", "request_preview": request_preview, "source_metadata": { "transport": "http_json_service_boundary", "endpoint": endpoint, "provenance_mode": "external_service_candidate", "timeout_seconds": _timeout_seconds(), "retry_policy": RETRY_POLICY, "network_execution_env": ALLOW_NETWORK_ENV, }, } except (TimeoutError, socket.timeout): return { "backend": "vedastro_service_adapter_candidate", "available": False, "status": "timeout", "reason": "VedAstro adapter timed out", "request_preview": request_preview, "source_metadata": { "transport": "http_json_service_boundary", "endpoint": endpoint, "provenance_mode": "external_service_candidate", "timeout_seconds": _timeout_seconds(), "retry_policy": RETRY_POLICY, "network_execution_env": ALLOW_NETWORK_ENV, }, } except json.JSONDecodeError: return { "backend": "vedastro_service_adapter_candidate", "available": False, "status": "invalid_json", "reason": "VedAstro adapter received non-JSON response", "request_preview": request_preview, "source_metadata": { "transport": "http_json_service_boundary", "endpoint": endpoint, "provenance_mode": "external_service_candidate", "timeout_seconds": _timeout_seconds(), "retry_policy": RETRY_POLICY, "network_execution_env": ALLOW_NETWORK_ENV, }, } return _normalize_success(payload, endpoint) def main() -> int: parser = argparse.ArgumentParser(description="VedAstro service adapter skeleton") parser.add_argument("--print-schema", action="store_true") parser.add_argument("--case", default="beijing_first_use_demo") args = parser.parse_args() result = schema() if args.print_schema else run_case(args.case) print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())