diff --git a/scripts/ephemeris_adapter_contract.py b/scripts/ephemeris_adapter_contract.py index a56b48ae..48845cd4 100644 --- a/scripts/ephemeris_adapter_contract.py +++ b/scripts/ephemeris_adapter_contract.py @@ -14,6 +14,7 @@ import sys from dataclasses import asdict, dataclass from pathlib import Path from typing import Any +import importlib.util SCRIPT_DIR = Path(__file__).resolve().parent @@ -168,6 +169,21 @@ def run_swisseph_python(case: dict[str, Any]) -> dict[str, Any]: def run_candidate_backend(case: dict[str, Any], backend: str) -> dict[str, Any]: if backend == "swisseph_python": return run_swisseph_python(case) + if backend == "vedastro_service_adapter_candidate": + spec = importlib.util.spec_from_file_location( + "vedastro_service_adapter", + SCRIPT_DIR / "vedastro_service_adapter.py", + ) + if spec is None or spec.loader is None: + return { + "backend": backend, + "available": False, + "status": "service_adapter_missing", + "reason": "vedastro_service_adapter.py is missing", + } + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module.run_case(case["id"]) return { "backend": backend, "available": False, diff --git a/scripts/vedastro_service_adapter.py b/scripts/vedastro_service_adapter.py new file mode 100644 index 00000000..d1ad66c8 --- /dev/null +++ b/scripts/vedastro_service_adapter.py @@ -0,0 +1,154 @@ +#!/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 +from pathlib import Path +from typing import Any + + +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", + }, +} + + +def schema() -> dict[str, Any]: + return { + "adapter": "vedastro_service_adapter", + "backend": "vedastro_service_adapter_candidate", + "transport": "http_json_service_boundary", + "default_timeout_seconds": 8, + "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", + ], + } + + +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", + }, + } + + +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] + return { + "backend": "vedastro_service_adapter_candidate", + "available": False, + "status": "service_execution_not_implemented", + "reason": "Endpoint is configured, but the real HTTP execution/normalization layer is intentionally not implemented yet.", + "request_preview": { + **case, + "body_list": ["Sun", "Moon", "Ascendant", "Rahu", "Ketu"], + }, + "source_metadata": { + "transport": "http_json_service_boundary", + "endpoint": endpoint, + "provenance_mode": "external_service_candidate", + }, + } + + +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()) diff --git a/tests/test_vedastro_service_adapter_executor.py b/tests/test_vedastro_service_adapter_executor.py new file mode 100644 index 00000000..61e900cd --- /dev/null +++ b/tests/test_vedastro_service_adapter_executor.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_vedastro_service_adapter_executor_schema_is_declared() -> None: + completed = subprocess.run( + [sys.executable, "scripts/vedastro_service_adapter.py", "--print-schema"], + cwd=ROOT, + text=True, + capture_output=True, + timeout=120, + check=False, + ) + + assert completed.returncode == 0, completed.stderr or completed.stdout + report = json.loads(completed.stdout) + assert report["adapter"] == "vedastro_service_adapter" + assert report["transport"] == "http_json_service_boundary" + assert report["default_timeout_seconds"] >= 3 + assert "endpoint" in report["required_env"] + assert "ayanamsa_policy" in report["request_contract"] + assert "bodies" in report["response_contract"] + assert "source_metadata" in report["response_contract"] + + +def test_vedastro_service_adapter_returns_controlled_unconfigured_status() -> None: + completed = subprocess.run( + [sys.executable, "scripts/vedastro_service_adapter.py", "--case", "beijing_first_use_demo"], + cwd=ROOT, + text=True, + capture_output=True, + timeout=120, + check=False, + ) + + assert completed.returncode == 0, completed.stderr or completed.stdout + report = json.loads(completed.stdout) + assert report["backend"] == "vedastro_service_adapter_candidate" + assert report["available"] is False + assert report["status"] == "service_endpoint_not_configured" + assert "VEDASTRO" in report["reason"] + + +def test_ephemeris_adapter_contract_can_call_vedastro_candidate_stub() -> None: + completed = subprocess.run( + [sys.executable, "scripts/ephemeris_adapter_contract.py", "vedastro_service_adapter_candidate"], + cwd=ROOT, + text=True, + capture_output=True, + timeout=120, + check=False, + ) + + assert completed.returncode == 0, completed.stderr or completed.stdout + report = json.loads(completed.stdout) + assert report["candidate_backend"] == "vedastro_service_adapter_candidate" + first = report["rows"][0]["candidate_backend"] + assert first["backend"] == "vedastro_service_adapter_candidate" + assert first["status"] == "service_endpoint_not_configured"