ephemeris-mcp/src/ephemeris_mcp/server.py

"""
Ephemeris MCP server — FastAPI + FastMCP entry point.

Provides Swiss Ephemeris computation as a service: planetary positions,
house cusps, lunar state, sidereal time, solar events, and more.
Downstream consumers (e.g. astro-mcp, satrack-mcp) call these tools
instead of bundling the Swiss Ephemeris themselves.

All MCP tools are defined here. The ephemeris engine and cache are in
ephemeris.py and storage.py respectively.
"""

from __future__ import annotations

import logging
from datetime import datetime, timezone

from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings

from . import config
from .ephemeris import (
    BODIES,
    close as ephemeris_close,
    get_constellation_at_ecliptic as _compute_constellation_at_ecliptic,
    get_lunar_state as _compute_lunar_state,
    get_planetary_positions as _compute_planetary_positions,
    get_sidereal_time as _compute_sidereal_time,
    get_solar_events as _compute_solar_events,
    init as ephemeris_init,
)
from .storage import cache_key, get_cache

import swisseph as swe

logger = logging.getLogger("ephemeris-mcp")

# ── FastMCP + FastAPI ──────────────────────────────────────────────

cache = get_cache()
mcp = FastMCP(
    "ephemeris-mcp",
    transport_security=TransportSecuritySettings(
        enable_dns_rebinding_protection=False,
    ),
)


# ── Helpers ─────────────────────────────────────────────────────────

def _now_jd() -> float:
    from swisseph import julday

    now = datetime.now(timezone.utc)
    return julday(now.year, now.month, now.day, now.hour + now.minute / 60 + now.second / 3600)


def _parse_datetime(dt_str: str | None) -> float:
    """Parse ISO datetime string to Julian Day. Defaults to now."""
    if dt_str is None:
        return _now_jd()
    from swisseph import julday

    dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return julday(dt.year, dt.month, dt.day, dt.hour + dt.minute / 60 + dt.second / 3600)


def _parse_date(date_str: str) -> float:
    """Parse ISO date string to Julian Day at noon."""
    from swisseph import julday

    dt = datetime.fromisoformat(date_str)
    return julday(dt.year, dt.month, dt.day, 12.0)


def _sign_name(lon: float) -> str:
    signs = ["Aries","Taurus","Gemini","Cancer","Leo","Virgo","Libra","Scorpio","Sagittarius","Capricorn","Aquarius","Pisces"]
    return signs[int(lon / 30) % 12]

def _sign_abbr(lon: float) -> str:
    return _sign_name(lon)[:3]

def _utc_date_from_datetime(dt_str: str | None) -> str:
    """Convert an ISO datetime string to a UTC date string."""
    if dt_str is None:
        return __import__("datetime").datetime.now(timezone.utc).date().isoformat()

    dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc).date().isoformat()


def _default_location(lat: float | None = None, lon: float | None = None) -> tuple[float, float]:
    resolved_lat = getattr(config, "DEFAULT_LAT", None) if lat is None else lat
    resolved_lon = getattr(config, "DEFAULT_LON", None) if lon is None else lon
    return (resolved_lat if resolved_lat is not None else 0.0, resolved_lon if resolved_lon is not None else 0.0)


def _tool_names() -> list[str]:
    return [
        "get_planetary_positions",
        "get_solar_events",
        "get_lunar_state",
        "get_moon_phase",
        "get_sidereal_time",
        "get_constellation_at_ecliptic",
        "list_available_bodies",
        "get_sky_state",
    ]


# ── Group A — Celestial Mechanics ──────────────────────────────────

@mcp.tool()
def get_planetary_positions(
    datetime: str | None = None,
    lat: float | None = None,
    lon: float | None = None,
    elevation: float = 0.0,
    geocentric: bool = True,
) -> dict:
    """
    Get positions of all major solar system bodies.

    Args:
        datetime: ISO 8601 datetime (UTC). Defaults to now.
        lat: Observer latitude in decimal degrees.
        lon: Observer longitude in decimal degrees.
        elevation: Observer elevation in meters.
        geocentric: If True, return geocentric positions instead of topocentric.

    Returns:
        Object with 'input' (echoed params), 'timestamp', and 'bodies' array.
    """
    lat, lon = _default_location(lat, lon)
    jd = _parse_datetime(datetime)

    ck = cache_key(
        "get_planetary_positions",
        cache_version=3,
        datetime=datetime or "now",
        elevation=elevation,
        geocentric=geocentric,
        lat=lat,
        lon=lon,
    )
    cached = cache.get(ck)
    if cached:
        logger.info(f"cache hit: {ck}")
        return {
            "input": {
                "datetime": datetime,
                "lat": lat,
                "lon": lon,
                "elevation": elevation,
                "geocentric": geocentric,
            },
            **cached,
        }

    positions = _compute_planetary_positions(jd, lat, lon, elevation, geocentric)

    result = {
        "input": {
            "datetime": datetime,
            "lat": lat,
            "lon": lon,
            "elevation": elevation,
            "geocentric": geocentric,
        },
        "timestamp_utc": datetime or __import__("datetime").datetime.now(timezone.utc).isoformat(),
        "julian_day": round(jd, 6),
        "bodies": positions,
    }
    cache.set(
        ck,
        {
            "timestamp_utc": result["timestamp_utc"],
            "julian_day": result["julian_day"],
            "bodies": positions,
        },
        config.CACHE_PLANETARY,
    )
    return result


@mcp.tool()
def get_solar_events(
    date: str,
    lat: float | None = None,
    lon: float | None = None,
) -> dict:
    """
    Get solar events (sunrise, sunset, solar noon, twilight) for a date and location.

    Args:
        date: ISO date string (YYYY-MM-DD).
        lat: Observer latitude in decimal degrees.
        lon: Observer longitude in decimal degrees.

    Returns:
        Object with all event times as ISO datetime strings and day length.
    """
    lat, lon = _default_location(lat, lon)
    jd = _parse_date(date)

    ck = cache_key("get_solar_events", date=date, lat=lat, lon=lon)
    cached = cache.get(ck)
    if cached:
        return {"input": {"date": date, "lat": lat, "lon": lon}, **cached}

    events = _compute_solar_events(jd, lat, lon)

    result = {
        "input": {"date": date, "lat": lat, "lon": lon},
        "date": date,
        "julian_day_base": round(jd, 6),
        "events_jd": events,
    }
    cache.set(
        ck,
        {"date": date, "events_jd": events, "julian_day_base": round(jd, 6)},
        config.CACHE_SOLAR,
    )
    return result


@mcp.tool()
def get_lunar_state(
    datetime: str | None = None,
    lat: float | None = None,
    lon: float | None = None,
) -> dict:
    """
    Get current lunar phase and position.

    Args:
        datetime: ISO 8601 datetime (UTC). Defaults to now.
        lat: Observer latitude in decimal degrees.
        lon: Observer longitude in decimal degrees.

    Returns:
        Object with phase name, illumination fraction, age, and position.
    """
    lat, lon = _default_location(lat, lon)
    jd = _parse_datetime(datetime)

    ck = cache_key("get_lunar_state", cache_version=3, datetime=datetime or "now", lat=lat, lon=lon)
    cached = cache.get(ck)
    cached_next_major_phase = cached.get("lunar_state", {}).get("next_major_phase") if cached else None
    if cached_next_major_phase and cached_next_major_phase.get("phase_name") and cached_next_major_phase.get("at_utc") and cached_next_major_phase.get("in_text"):
        return {"input": {"datetime": datetime, "lat": lat, "lon": lon}, **cached}
    if cached:
        cache.delete(ck)

    state = _compute_lunar_state(jd, lat, lon)

    result = {
        "input": {"datetime": datetime, "lat": lat, "lon": lon},
        "timestamp_utc": datetime or __import__("datetime").datetime.now(timezone.utc).isoformat(),
        "julian_day": round(jd, 6),
        "lunar_state": state,
    }
    cache.set(
        ck,
        {"timestamp_utc": result["timestamp_utc"], "julian_day": result["julian_day"], "lunar_state": state},
        config.CACHE_LUNAR,
    )
    return result


@mcp.tool()
def get_sidereal_time(
    datetime: str | None = None,
    lat: float | None = None,
    lon: float | None = None,
) -> dict:
    """
    Get sidereal time and obliquity of the ecliptic.

    Args:
        datetime: ISO 8601 datetime (UTC). Defaults to now.
        lat: Observer latitude.
        lon: Observer longitude.

    Returns:
        Object with Greenwich and local sidereal time, and obliquity.
    """
    lat, lon = _default_location(lat, lon)
    jd = _parse_datetime(datetime)

    ck = cache_key("get_sidereal_time", datetime=datetime or "now", lat=lat, lon=lon)
    cached = cache.get(ck)
    if cached:
        return {"input": {"datetime": datetime, "lat": lat, "lon": lon}, **cached}

    result_st = _compute_sidereal_time(jd, lat, lon)

    result = {
        "input": {"datetime": datetime, "lat": lat, "lon": lon},
        "timestamp_utc": datetime or __import__("datetime").datetime.now(timezone.utc).isoformat(),
        "julian_day": round(jd, 6),
        **result_st,
    }
    cache.set(
        ck,
        {"timestamp_utc": result["timestamp_utc"], "julian_day": result["julian_day"], **result_st},
        config.CACHE_SIDEREAL,
    )
    return result


@mcp.tool()
def get_constellation_at_ecliptic(ecliptic_lon: float) -> dict:
    """
    Look up IAU constellation at a given ecliptic longitude.

    Args:
        ecliptic_lon: Ecliptic longitude in degrees (0-360).

    Returns:
        Object with constellation abbreviation, full name, and position within.
    """
    result = _compute_constellation_at_ecliptic(ecliptic_lon)
    return {"input": {"ecliptic_lon": ecliptic_lon}, **result}


@mcp.tool()
def list_available_bodies(category: str | None = None) -> dict:
    """
    List all computable celestial bodies.

    This is a lightweight discovery tool for the v0.2 core slice.
    """
    bodies = []
    for name in BODIES:
        bodies.append(
            {
                "name": name,
                "full_name": name.replace("_", " ").title(),
                "available": True,
            }
        )

    if category:
        category_lower = category.lower()
        if category_lower == "planets":
            bodies = [b for b in bodies if b["name"] in {"sun", "mercury", "venus", "mars", "jupiter", "saturn", "uranus", "neptune", "pluto"}]
        elif category_lower == "lunar":
            bodies = [b for b in bodies if b["name"] == "moon"]
        elif category_lower == "major":
            bodies = [b for b in bodies if b["name"] in {"sun", "moon", "mercury", "venus", "mars", "jupiter", "saturn"}]

    return {"bodies": bodies}


@mcp.tool()
def get_moon_phase(datetime: str | None = None, lat: float | None = None, lon: float | None = None) -> dict:
    """
    Convenience lunar alias for mcporter users.

    Phase naming is driven by lunar age, while illumination is reported as
    a separate descriptive value.
    """
    lat, lon = _default_location(lat, lon)
    state = get_lunar_state(datetime=datetime, lat=lat, lon=lon)
    return {
        "input": {"datetime": datetime, "lat": lat, "lon": lon},
        "phase_name": state["lunar_state"]["phase_name"],
        "illumination_fraction": state["lunar_state"]["illumination_fraction"],
        "age_days": state["lunar_state"]["age_days"],
        "next_major_phase": state["lunar_state"]["next_major_phase"],
        "ecliptic_lon": state["lunar_state"]["ecliptic_lon"],
        "ecliptic_lat": state["lunar_state"]["ecliptic_lat"],
        "distance_km": state["lunar_state"]["distance_km"],
    }


@mcp.tool()
def get_sky_state(
    datetime: str | None = None,
    lat: float | None = None,
    lon: float | None = None,
    elevation: float = 0.0,
    geocentric: bool = True,
    house_system: str | None = None,
) -> dict:
    """
    Return a consolidated raw sky-state snapshot for downstream chart engines.

    Combines planetary positions, lunar state, sidereal time, and solar events
    into a single response. Optionally includes house cusps and angles computed
    server-side via the Swiss Ephemeris when a house system is specified.

    This is the primary tool for downstream consumers (e.g. astro-mcp) that
    need a full sky-state from a single call.

    Args:
        datetime: ISO 8601 datetime (UTC). Defaults to now.
        lat: Observer latitude in decimal degrees.
        lon: Observer longitude in decimal degrees.
        elevation: Observer elevation in meters.
        geocentric: If True, return geocentric positions instead of topocentric.
        house_system: Optional house system code (single uppercase letter).
            When provided, includes 'houses' key with 12 cusps and angles
            (ASC, MC, DSC, IC). All 22 Swiss Ephemeris house systems:
            P=Placidus, K=Koch, E=Equal (0° Aries), W=Whole Sign,
            A=Alcabitius, C=Campanus, M=Morinus, R=Porphyry,
            T=Polich/Page, U=Krusinski-Pisa, V=Vehlow Equal,
            X=Meridian, Y=Horizontal, H=Azimuthal, O=Equal (MC),
            F=Carter poli-eq, D=Equal (15° Aries), G=Gauquelin sectors,
            I=Sunshine, J=Sunshine alt, L=Pullen SD, N=Equal/1,
            Q=Pullen SR, S=Sripati, Z=APC houses.
    """
    lat, lon = _default_location(lat, lon)
    date = _utc_date_from_datetime(datetime)
    jd = _parse_datetime(datetime)

    ck = cache_key(
        "get_sky_state",
        cache_version=2,
        datetime=datetime or "now",
        date=date,
        elevation=elevation,
        geocentric=geocentric,
        lat=lat,
        lon=lon,
        house_system=house_system,
    )
    cached = cache.get(ck)
    if cached:
        return {
            "input": {
                "datetime": datetime,
                "date": date,
                "lat": lat,
                "lon": lon,
                "elevation": elevation,
                "geocentric": geocentric,
                "house_system": house_system,
            },
            **cached,
        }

    planetary_positions = get_planetary_positions(
        datetime=datetime,
        lat=lat,
        lon=lon,
        elevation=elevation,
        geocentric=geocentric,
    )
    lunar_state = get_lunar_state(datetime=datetime, lat=lat, lon=lon)
    sidereal_time = get_sidereal_time(datetime=datetime, lat=lat, lon=lon)
    solar_events = get_solar_events(date=date, lat=lat, lon=lon)

    result = {
        "input": {
            "datetime": datetime,
            "date": date,
            "lat": lat,
            "lon": lon,
            "elevation": elevation,
            "geocentric": geocentric,
            "house_system": house_system,
        },
        "timestamp_utc": datetime or __import__("datetime").datetime.now(timezone.utc).isoformat(),
        "julian_day": round(jd, 6),
        "planetary_positions": planetary_positions,
        "solar_events": solar_events,
        "lunar_state": lunar_state,
        "sidereal_time": sidereal_time,
    }

    # Compute house cusps when requested
    if house_system is not None:
        hs_code = house_system.upper().encode()[:1]
        logger.info(f"house_system={house_system!r} hs_code={hs_code!r} type={type(hs_code)}")
        try:
            cusps, ascmc = swe.houses(jd, lat, lon, hs_code)
            houses = []
            for i in range(12):
                houses.append({
                    "house": i + 1,
                    "absolute_lon": cusps[i],
                    "sign": _sign_name(cusps[i]),
                    "abbreviation": _sign_abbr(cusps[i]),
                    "degree": cusps[i] % 30.0,
                })
            result["houses"] = {
                "system": house_system.upper(),
                "cusps": houses,
                "ascendant": ascmc[0],
                "midheaven": ascmc[1],
                "descendant": ascmc[0] + 180.0 if ascmc[0] < 180.0 else ascmc[0] - 180.0,
                "imum_coeli": ascmc[1] + 180.0 if ascmc[1] < 180.0 else ascmc[1] - 180.0,
                "armc": ascmc[2] if len(ascmc) > 2 else None,
                "equatorial_ascendant": ascmc[3] if len(ascmc) > 3 else None,
                "vertex": ascmc[4] if len(ascmc) > 4 else None,
                "co_ascendant": ascmc[5] if len(ascmc) > 5 else None,
                "polar_ascendant": ascmc[6] if len(ascmc) > 6 else None,
            }
        except Exception as exc:
            logger.warning(f"house calculation failed for system {house_system!r}: {exc}")
            result["houses"] = {"system": house_system.upper(), "error": str(exc)}

    cache.set(
        ck,
        {
            "timestamp_utc": result["timestamp_utc"],
            "julian_day": result["julian_day"],
            "planetary_positions": planetary_positions,
            "solar_events": solar_events,
            "lunar_state": lunar_state,
            "sidereal_time": sidereal_time,
            "houses": result.get("houses"),
        },
        config.CACHE_SKY,
    )
    return result


def create_app() -> FastAPI:
    """
    Build the FastAPI app for the v0.2 core slice.
    """
    config.LOG_DIR.mkdir(parents=True, exist_ok=True)
    logging.basicConfig(
        filename=str(config.LOG_DIR / "server.log"),
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )

    ephemeris_init()
    app = FastAPI(title="ephemeris-mcp")
    app.mount("/mcp", mcp.sse_app())

    @app.get("/health")
    def health() -> dict:
        return {"ok": True, "server": "ephemeris-mcp", "port": config.PORT}

    @app.get("/")
    def root() -> dict:
        return {
            "server": "ephemeris-mcp",
            "status": "ready",
            "tools": _tool_names(),
            "mcp": {
                "sse": "/mcp/sse",
                "messages": "/mcp/messages",
            },
        }

    return app