astro-mcp/src/astro_mcp/tools.py

"""
MCP tool definitions for astro-mcp.

All tools are async and use the ephemeris client to get astronomical data,
then the astrology module to transform it into astrological structures.
"""

from __future__ import annotations

import logging
from typing import Any

from .server import mcp
from . import astrology
from .ephemeris_client import call_sky_state, extract_bodies

logger = logging.getLogger("astro-mcp.tools")


DEFAULT_ORBS = astrology.DEFAULT_ORBS


# ── Tool: get_planetary_positions ────────────────────────────────────

@mcp.tool()
async def get_planetary_positions(
    datetime: str | None = None,
    lat: float | None = None,
    lon: float | None = None,
    elevation: float = 0.0,
    geocentric: bool = True,
    bodies: list[str] | None = None,
) -> dict[str, Any]:
    """Get planetary positions enhanced with zodiac signs, degrees, and retrograde flags.

    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.
        bodies: Optional list of body names to filter (e.g., ["sun", "moon"]).

    Returns:
        Object with 'input' (echoed params), 'timestamp', 'julian_day',
        and 'bodies' array. Each body includes ecliptic_lon, ecliptic_lat,
        sign, degree_within_sign, retrograde flag, speed_lon, and distance.
    """
    resolved_lat = lat if lat is not None else 0.0
    resolved_lon = lon if lon is not None else 0.0

    sky = await call_sky_state(
        datetime=datetime,
        lat=resolved_lat,
        lon=resolved_lon,
        elevation=elevation,
        geocentric=geocentric,
    )

    if "error" in sky:
        return {"input": {"datetime": datetime, "lat": resolved_lat, "lon": resolved_lon}, "error": sky["error"]}

    raw_bodies = extract_bodies(sky)

    enhanced_bodies = []
    for body in raw_bodies:
        name = body.get("body", "unknown")
        if bodies and name not in bodies:
            continue

        ecl_lon = body.get("ecliptic_lon", 0.0)
        ecl_lat = body.get("ecliptic_lat", 0.0)
        speed_lon = body.get("speed_lon")
        distance_au = body.get("distance_au", 0.0)

        zodiac = astrology.ecliptic_to_zodiac(ecl_lon)
        retrograde = astrology.is_retrograde(speed_lon)

        enhanced_bodies.append({
            "body": name,
            "ecliptic_lon": ecl_lon,
            "ecliptic_lat": ecl_lat,
            "distance_au": distance_au,
            "speed_lon": speed_lon,
            "sign": zodiac["sign"],
            "sign_abbreviation": zodiac["abbreviation"],
            "degree_within_sign": zodiac["degree"],
            "retrograde": retrograde,
        })

    return {
        "input": {
            "datetime": datetime,
            "lat": resolved_lat,
            "lon": resolved_lon,
            "elevation": elevation,
            "geocentric": geocentric,
            "bodies_filter": bodies,
        },
        "timestamp_utc": sky.get("timestamp_utc"),
        "julian_day": sky.get("julian_day"),
        "bodies": enhanced_bodies,
    }


# ── Tool: calculate_natal_chart ──────────────────────────────────────

@mcp.tool()
async def calculate_natal_chart(
    birth_datetime: str,
    latitude: float,
    longitude: float,
    elevation: float = 0.0,
    house_system: str = "placidus",
    orb_limits: dict[str, float] | None = None,
) -> dict[str, Any]:
    """Calculate a complete natal chart from birth data.

    Args:
        birth_datetime: ISO 8601 birth datetime (UTC).
        latitude: Birth latitude in decimal degrees.
        longitude: Birth longitude in decimal degrees.
        elevation: Birth elevation in meters.
        house_system: House system to use (default: Placidus).
        orb_limits: Optional dict of {aspect_name: max_orb_degrees}.

    Returns:
        Complete natal chart structure with planets, houses, aspects, and angles.
    """
    sky = await call_sky_state(
        datetime=birth_datetime,
        lat=latitude,
        lon=longitude,
        elevation=elevation,
        geocentric=True,
    )

    if "error" in sky:
        return {"input": {"birth_datetime": birth_datetime, "latitude": latitude, "longitude": longitude}, "error": sky["error"]}

    raw_bodies = extract_bodies(sky)
    sidereal = sky.get("sidereal_time", {})
    lst_hours = sidereal.get("local_sidereal_time", 0.0)

    # Calculate houses
    houses = astrology.calculate_houses(lst_hours, latitude, house_system)

    # Build planet list with house placement
    planets = []
    for body in raw_bodies:
        ecl_lon = body.get("ecliptic_lon", 0.0)
        ecl_lat = body.get("ecliptic_lat", 0.0)
        speed_lon = body.get("speed_lon")
        zodiac = astrology.ecliptic_to_zodiac(ecl_lon)
        house = astrology.get_house_placement(ecl_lon, houses)

        planets.append({
            "body": body["body"],
            "sign": zodiac["sign"],
            "sign_abbreviation": zodiac["abbreviation"],
            "degree_within_sign": zodiac["degree"],
            "absolute_lon": zodiac["absolute_lon"],
            "ecliptic_lat": ecl_lat,
            "distance_au": body.get("distance_au", 0.0),
            "house": house,
            "retrograde": astrology.is_retrograde(speed_lon),
        })

    # Calculate aspects
    aspect_bodies = [{"name": p["body"], "lon": p["absolute_lon"]} for p in planets]
    aspects = astrology.compute_aspects(aspect_bodies, orb_limits)

    # Format aspects
    formatted_aspects = []
    for asp in aspects:
        formatted_aspects.append({
            "body1": asp["body1"],
            "body2": asp["body2"],
            "aspect": asp["aspect"],
            "orb": asp["orb"],
            "applying": asp["applying"],
            "exactness": asp["exactness"],
        })

    # Calculate angles
    angles = astrology.calculate_angles(lst_hours, latitude)

    return {
        "input": {
            "birth_datetime": birth_datetime,
            "latitude": latitude,
            "longitude": longitude,
            "elevation": elevation,
            "house_system": house_system,
            "orb_limits": orb_limits,
        },
        "chart_type": "natal",
        "planets": planets,
        "houses": houses,
        "aspects": formatted_aspects,
        "angles": angles,
    }


# ── Tool: calculate_transit_chart ────────────────────────────────────

@mcp.tool()
async def calculate_transit_chart(
    birth_datetime: str,
    transit_datetime: str,
    latitude: float,
    longitude: float,
    elevation: float = 0.0,
    house_system: str = "placidus",
    orb_limits: dict[str, float] | None = None,
) -> dict[str, Any]:
    """Calculate a transit chart: transiting planets vs natal positions.

    Args:
        birth_datetime: ISO 8601 birth datetime (UTC).
        transit_datetime: ISO 8601 transit datetime (UTC).
        latitude: Birth latitude in decimal degrees.
        longitude: Birth longitude in decimal degrees.
        elevation: Birth elevation in meters.
        house_system: House system for natal houses (default: Placidus).
        orb_limits: Optional orb configuration.

    Returns:
        Transit chart with transiting planets, aspects to natal, and house placements.
    """
    # Get natal sky state
    natal_sky = await call_sky_state(
        datetime=birth_datetime,
        lat=latitude,
        lon=longitude,
        elevation=elevation,
        geocentric=True,
    )

    # Get transit sky state at birth location
    transit_sky = await call_sky_state(
        datetime=transit_datetime,
        lat=latitude,
        lon=longitude,
        elevation=elevation,
        geocentric=True,
    )

    if "error" in natal_sky:
        return {"error": f"natal: {natal_sky['error']}"}
    if "error" in transit_sky:
        return {"error": f"transit: {transit_sky['error']}"}

    natal_bodies = extract_bodies(natal_sky)
    transit_bodies = extract_bodies(transit_sky)

    # Natal houses from natal LST
    sidereal = natal_sky.get("sidereal_time", {})
    lst_hours = sidereal.get("local_sidereal_time", 0.0)
    houses = astrology.calculate_houses(lst_hours, latitude, house_system)

    # Build natal planets
    natal_planets = []
    for body in natal_bodies:
        ecl_lon = body.get("ecliptic_lon", 0.0)
        zodiac = astrology.ecliptic_to_zodiac(ecl_lon)
        house = astrology.get_house_placement(ecl_lon, houses)
        natal_planets.append({
            "body": body["body"],
            "sign": zodiac["sign"],
            "degree_within_sign": zodiac["degree"],
            "absolute_lon": zodiac["absolute_lon"],
            "house": house,
            "retrograde": astrology.is_retrograde(body.get("speed_lon")),
        })

    # Build transit planets
    transit_planets = []
    for body in transit_bodies:
        ecl_lon = body.get("ecliptic_lon", 0.0)
        zodiac = astrology.ecliptic_to_zodiac(ecl_lon)
        transit_house = astrology.get_house_placement(ecl_lon, houses)
        transit_planets.append({
            "body": body["body"],
            "sign": zodiac["sign"],
            "degree_within_sign": zodiac["degree"],
            "absolute_lon": zodiac["absolute_lon"],
            "natal_house": transit_house,
            "retrograde": astrology.is_retrograde(body.get("speed_lon")),
        })

    # Transit-to-natal aspects
    transit_aspects = []
    for t_body in transit_planets:
        for n_body in natal_planets:
            pair = [
                {"name": f"transit_{t_body['body']}", "lon": t_body["absolute_lon"], "speed_lon": None},
                {"name": f"natal_{n_body['body']}", "lon": n_body["absolute_lon"], "speed_lon": None},
            ]
            pair_aspects = astrology.compute_aspects(pair, orb_limits)
            for asp in pair_aspects:
                transit_aspects.append({
                    "transiting": t_body["body"],
                    "natal": n_body["body"],
                    "aspect": asp["aspect"],
                    "orb": asp["orb"],
                    "exactness": asp["exactness"],
                })

    transit_aspects.sort(key=lambda a: a["orb"])

    return {
        "input": {
            "birth_datetime": birth_datetime,
            "transit_datetime": transit_datetime,
            "latitude": latitude,
            "longitude": longitude,
            "house_system": house_system,
        },
        "chart_type": "transit",
        "natal_planets": natal_planets,
        "transiting_planets": transit_planets,
        "aspects": transit_aspects,
        "houses": houses,
    }


# ── Tool: calculate_synastry_chart ───────────────────────────────────

@mcp.tool()
async def calculate_synastry_chart(
    person1_datetime: str,
    person1_latitude: float,
    person1_longitude: float,
    person2_datetime: str,
    person2_latitude: float,
    person2_longitude: float,
    elevation: float = 0.0,
    house_system: str = "placidus",
    orb_limits: dict[str, float] | None = None,
) -> dict[str, Any]:
    """Calculate a synastry (relationship) chart for two people.

    Args:
        person1_datetime, person1_latitude, person1_longitude: Person 1 birth data.
        person2_datetime, person2_latitude, person2_longitude: Person 2 birth data.
        elevation: Birth elevation in meters.
        house_system: House system (default: Placidus).
        orb_limits: Optional orb configuration.

    Returns:
        Synastry chart with interaspects, house overlays, composite, and Davison charts.
    """
    sky1 = await call_sky_state(datetime=person1_datetime, lat=person1_latitude, lon=person1_longitude, elevation=elevation)
    sky2 = await call_sky_state(datetime=person2_datetime, lat=person2_latitude, lon=person2_longitude, elevation=elevation)

    if "error" in sky1:
        return {"error": f"person1: {sky1['error']}"}
    if "error" in sky2:
        return {"error": f"person2: {sky2['error']}"}

    bodies1 = extract_bodies(sky1)
    bodies2 = extract_bodies(sky2)

    sidereal1 = sky1.get("sidereal_time", {})
    lst1 = sidereal1.get("local_sidereal_time", 0.0)
    houses1 = astrology.calculate_houses(lst1, person1_latitude, house_system)

    sidereal2 = sky2.get("sidereal_time", {})
    lst2 = sidereal2.get("local_sidereal_time", 0.0)
    houses2 = astrology.calculate_houses(lst2, person2_latitude, house_system)

    def build_planet_list(bodies):
        result = []
        for b in bodies:
            ecl_lon = b.get("ecliptic_lon", 0.0)
            z = astrology.ecliptic_to_zodiac(ecl_lon)
            result.append({
                "body": b["body"],
                "sign": z["sign"],
                "degree_within_sign": z["degree"],
                "absolute_lon": z["absolute_lon"],
                "retrograde": astrology.is_retrograde(b.get("speed_lon")),
            })
        return result

    chart1_planets = build_planet_list(bodies1)
    chart2_planets = build_planet_list(bodies2)

    # Interaspects
    interaspects = []
    for p1 in chart1_planets:
        for p2 in chart2_planets:
            pair = [
                {"name": f"p1_{p1['body']}", "lon": p1["absolute_lon"]},
                {"name": f"p2_{p2['body']}", "lon": p2["absolute_lon"]},
            ]
            for asp in astrology.compute_aspects(pair, orb_limits):
                interaspects.append({
                    "person1_planet": p1["body"],
                    "person2_planet": p2["body"],
                    "aspect": asp["aspect"],
                    "orb": asp["orb"],
                    "exactness": asp["exactness"],
                })

    interaspects.sort(key=lambda a: a["orb"])

    # House overlays: person2's planets in person1's houses
    p2_in_p1_houses = []
    for p2 in chart2_planets:
        house = astrology.get_house_placement(p2["absolute_lon"], houses1)
        p2_in_p1_houses.append({
            "planet": p2["body"],
            "house": house,
        })

    p1_in_p2_houses = []
    for p1 in chart1_planets:
        house = astrology.get_house_placement(p1["absolute_lon"], houses2)
        p1_in_p2_houses.append({
            "planet": p1["body"],
            "house": house,
        })

    # Composite chart (midpoint method)
    composite_bodies = astrology.compute_composite_chart(
        [{"name": p["body"], "lon": p["absolute_lon"]} for p in chart1_planets],
        [{"name": p["body"], "lon": p["absolute_lon"]} for p in chart2_planets],
    )
    composite_planets = []
    for cb in composite_bodies:
        z = astrology.ecliptic_to_zodiac(cb["lon"])
        composite_planets.append({
            "body": cb["name"],
            "sign": z["sign"],
            "degree_within_sign": z["degree"],
            "absolute_lon": z["absolute_lon"],
        })

    # Davison chart (date midpoint)
    davison = astrology.compute_davison_chart(person1_datetime, person2_datetime, person1_datetime, person2_datetime)
    davison_mid_lat = (person1_latitude + person2_latitude) / 2
    davison_mid_lon = (person1_longitude + person2_longitude) / 2

    return {
        "input": {
            "person1": {"datetime": person1_datetime, "latitude": person1_latitude, "longitude": person1_longitude},
            "person2": {"datetime": person2_datetime, "latitude": person2_latitude, "longitude": person2_longitude},
            "house_system": house_system,
        },
        "chart_type": "synastry",
        "chart1_natal": {"planets": chart1_planets, "houses": houses1},
        "chart2_natal": {"planets": chart2_planets, "houses": houses2},
        "interaspects": interaspects,
        "house_overlays": {
            "person2_in_person1_houses": p2_in_p1_houses,
            "person1_in_person2_houses": p1_in_p2_houses,
        },
        "composite_chart": {"planets": composite_planets},
        "davison_chart": {
            "date_midpoint_jd": davison["date_midpoint_jd"],
            "latitude_midpoint": davison_mid_lat,
            "longitude_midpoint": davison_mid_lon,
        },
    }


# ── Tool: get_transit_preview ────────────────────────────────────────

@mcp.tool()
async def get_transit_preview(
    person_id: str,
    start_date: str,
    end_date: str,
    event_types: list[str] | None = None,
    orb_limits: dict[str, float] | None = None,
) -> dict[str, Any]:
    """Preview significant transit events for a person over a time range.

    Args:
        person_id: ID of a person in the persons database.
        start_date: ISO date string for the start of the range.
        end_date: ISO date string for the end of the range.
        event_types: Optional filter for event types
                     (exact_aspect, ingress, retrograde_station, lunar_phase).
        orb_limits: Optional orb configuration for aspect detection.

    Returns:
        List of transit events with timestamps, descriptions, and orbs.
    """
    return {
        "input": {
            "person_id": person_id,
            "start_date": start_date,
            "end_date": end_date,
            "event_types": event_types,
        },
        "events": [],
        "_note": "Transit preview not yet implemented -- requires person database (Phase 4)",
    }


# ── Tool: person_manage ─────────────────────────────────────────────

@mcp.tool()
async def person_manage(
    action: str,
    person_id: str | None = None,
    name: str | None = None,
    nickname: str | None = None,
    birth_datetime: str | None = None,
    latitude: float | None = None,
    longitude: float | None = None,
    elevation: float | None = None,
) -> dict[str, Any]:
    """Manage persons in the birth data database.

    Args:
        action: One of: add, get, list, update, delete.
        person_id: Required for get, update, delete.
        name: Person's full name (required for add).
        nickname: Optional short name for quick lookup.
        birth_datetime: ISO 8601 UTC datetime (required for add).
        latitude: Birth latitude (required for add).
        longitude: Birth longitude (required for add).
        elevation: Birth elevation in meters.

    Returns:
        Operation result with person data or error.
    """
    from . import storage

    action = action.lower().strip()

    if action == "add":
        if not name or not birth_datetime or latitude is None or longitude is None:
            return {"error": "add requires: name, birth_datetime, latitude, longitude"}
        person = await storage.add_person(
            name=name,
            birth_datetime=birth_datetime,
            latitude=latitude,
            longitude=longitude,
            elevation=elevation if elevation is not None else 0.0,
            nickname=nickname,
        )
        return {"action": "add", "person": person}

    elif action == "get":
        if not person_id and not nickname:
            return {"error": "get requires: person_id or nickname"}
        person = await storage.get_person(person_id=person_id, nickname=nickname)
        if not person:
            return {"action": "get", "error": "not_found"}
        return {"action": "get", "person": person}

    elif action == "list":
        persons = await storage.list_persons()
        return {"action": "list", "persons": persons, "count": len(persons)}

    elif action == "update":
        if not person_id:
            return {"error": "update requires: person_id"}
        person = await storage.update_person(
            person_id=person_id,
            name=name,
            nickname=nickname,
            birth_datetime=birth_datetime,
            latitude=latitude,
            longitude=longitude,
            elevation=elevation,
        )
        if not person:
            return {"action": "update", "error": "not_found"}
        return {"action": "update", "person": person}

    elif action == "delete":
        if not person_id:
            return {"error": "delete requires: person_id"}
        deleted = await storage.delete_person(person_id)
        if not deleted:
            return {"action": "delete", "error": "not_found"}
        return {"action": "delete", "deleted": True, "person_id": person_id}

    else:
        return {"error": f"unknown action: {action}. Use: add, get, list, update, delete"}


# ── Tool: list_house_systems ─────────────────────────────────────────

@mcp.tool()
def list_house_systems() -> dict[str, Any]:
    """List supported house systems.

    Returns:
        Object with 'systems' array, each containing name and description.
    """
    return {
        "systems": [
            {"id": "placidus", "name": "Placidus", "description": "Most common system; houses based on time divisions of the diurnal arc. Default."},
            {"id": "equal", "name": "Equal House", "description": "Each house is exactly 30 degrees, starting from the ASC."},
            {"id": "whole_sign", "name": "Whole Sign", "description": "Each house corresponds to one full sign. The ASC sign is house 1."},
        ]
    }