astro-mcp/tests/test_step2_local_ephemeris.py

"""
Step 2: Call the LOCAL ephemeris-mcp server (run.sh) via MCP SSE.
Compare returned houses/angles against ground truth from Step 1.

Requires: ephemeris-mcp running locally via run.sh on port 7015.
"""
from __future__ import annotations

import asyncio
import json
import os
import sys
from datetime import timedelta

from mcp import ClientSession
from mcp.client.sse import sse_client

# ── Ground truth from Step 1 ──────────────────────────────────────────

REFERENCE = {
    "trump": {
        "label": "Donald Trump",
        "datetime_utc": "1946-06-14T14:54:00",
        "lat": 40.6413,
        "lon": -73.7781,
        "asc": 149.972371,
        "mc": 54.387979,
    },
    "einstein": {
        "label": "Albert Einstein",
        "datetime_utc": "1879-03-14T10:37:00",
        "lat": 48.39841,
        "lon": 9.99155,
        "asc": 98.923507,
        "mc": 339.337618,
    },
    "chaka": {
        "label": "Chaka Khan",
        "datetime_utc": "1953-03-24T03:05:00",
        "lat": 41.87811,
        "lon": -87.62980,
        "asc": 218.923115,
        "mc": 137.470885,
    },
}

TOLERANCE = 0.01  # degrees

EPHEMERIS_URL = os.environ.get(
    "EPHEMERIS_MCP_URL", "http://127.0.0.1:7015/mcp/sse"
)


def _payload_from_result(result):
    payload = getattr(result, "structuredContent", None)
    if isinstance(payload, dict):
        return payload
    content_items = getattr(result, "content", []) or []
    for item in content_items:
        text = getattr(item, "text", None)
        if isinstance(text, str) and text.strip():
            try:
                decoded = json.loads(text)
            except Exception:
                continue
            if isinstance(decoded, dict):
                return decoded
    return {}


async def call_ephemeris(datetime_utc, lat, lon, house_system="placidus"):
    url = EPHEMERIS_URL
    if not url.endswith("/mcp/sse"):
        url = url.rstrip("/") + "/mcp/sse"
    async with sse_client(url, timeout=15.0, sse_read_timeout=15.0) as streams:
        async with ClientSession(
            *streams, read_timeout_seconds=timedelta(seconds=15)
        ) as session:
            await session.initialize()
            result = await session.call_tool(
                "get_sky_state",
                {
                    "datetime": datetime_utc,
                    "lat": lat,
                    "lon": lon,
                    "elevation": 0.0,
                    "geocentric": True,
                    "house_system": house_system,
                },
            )
            return _payload_from_result(result)


def _fmt(lon):
    signs = ["Ari", "Tau", "Gem", "Can", "Leo", "Vir",
             "Lib", "Sco", "Sag", "Cap", "Aqu", "Pis"]
    idx = int(lon // 30) % 12
    deg = lon % 30.0
    return f"{signs[idx]} {deg:05.2f}° (abs {lon:.6f}°)"


async def main():
    print("=" * 70)
    print("STEP 2: Local ephemeris-mcp server (run.sh) via MCP")
    print(f"  URL: {EPHEMERIS_URL}")
    print("=" * 70)

    all_ok = True
    for key, ref in REFERENCE.items():
        print(f"\n--- {ref['label']} ---")
        print(f"  Input: UTC={ref['datetime_utc']}  lat={ref['lat']}  lon={ref['lon']}")

        try:
            sky = await call_ephemeris(ref["datetime_utc"], ref["lat"], ref["lon"])
        except Exception as e:
            print(f"  ERROR: Could not reach server: {e}")
            all_ok = False
            continue

        houses_data = sky.get("houses", {})
        if not houses_data or "error" in houses_data:
            print(f"  ERROR: Server returned error: {houses_data}")
            all_ok = False
            continue

        server_asc = houses_data.get("ascendant")
        server_mc = houses_data.get("midheaven")
        cusps = houses_data.get("cusps", [])
        server_cusp1 = cusps[0]["absolute_lon"] if cusps else None

        print(f"  Server ASC:  {_fmt(server_asc)}" if server_asc else "  Server ASC:  MISSING")
        print(f"  Server MC:   {_fmt(server_mc)}" if server_mc else "  Server MC:   MISSING")
        print(f"  Server H1:   {_fmt(server_cusp1)}" if server_cusp1 else "  Server H1:   MISSING")
        print(f"  Expected ASC: {_fmt(ref['asc'])}")
        print(f"  Expected MC:  {_fmt(ref['mc'])}")

        asc_diff = abs(server_asc - ref["asc"]) if server_asc else float("inf")
        mc_diff = abs(server_mc - ref["mc"]) if server_mc else float("inf")

        asc_ok = asc_diff < TOLERANCE
        mc_ok = mc_diff < TOLERANCE

        print(f"  ASC diff: {asc_diff:.6f}°  {'OK' if asc_ok else 'MISMATCH'}")
        print(f"  MC  diff: {mc_diff:.6f}°  {'OK' if mc_ok else 'MISMATCH'}")

        if not asc_ok or not mc_ok:
            all_ok = False

    print("\n" + "=" * 70)
    if all_ok:
        print("RESULT: LOCAL EPHEMERIS SERVER IS CORRECT")
    else:
        print("RESULT: LOCAL EPHEMERIS SERVER HAS MISMATCHES")
    print("=" * 70)


if __name__ == "__main__":
    asyncio.run(main())