asteroidnet/data_access/skybot_client.py

"""
AsteroidNET SkyBoT Client (data_access.skybot_client)

Queries the IMCCE SkyBoT service to find all known solar system objects
in a given sky field at a given epoch. Used both for:
  1. Removing known asteroids from candidate lists (catalog_matcher)
  2. Labeling archival FITS frames for ML training (training.dataset_builder)

SkyBoT covers ephemerides for all known SSOs from 1889 to 2060.

Reference: https://ssp.imcce.fr/webservices/skybot/
"""
from __future__ import annotations

import logging
from typing import Optional

import requests
from astropy.coordinates import SkyCoord
from astropy.table import Table
from astropy.time import Time
import astropy.units as u

logger = logging.getLogger(__name__)

_SKYBOT_URL = "https://ssp.imcce.fr/webservices/skybot/api/conesearch.php"


def query_skybot(
    center: SkyCoord,
    radius: u.Quantity,
    epoch: Time,
    observer: str = "500",
    config: Optional[dict] = None,
) -> Table:
    """
    Cone-search the SkyBoT service for known solar system objects.

    Parameters
    ----------
    center : SkyCoord
        Field center.
    radius : Quantity
        Search radius (e.g. 30*u.arcmin).
    epoch : Time
        UTC observation epoch (use mid-exposure time).
    observer : str
        MPC observatory code. '500' = geocenter.
        Use 'F51' for Pan-STARRS, '695' for Palomar.
    config : dict, optional
        Pipeline configuration dict.

    Returns
    -------
    Table
        Astropy Table with columns: Number, Name, RA(h), DE(deg),
        Type, Mv, posunc, ErrRA, ErrDE, d, dRA, dDE, Rgeo, Rhel
        Returns empty Table if no objects found or service unavailable.
    """
    url = (config or {}).get("data_access", {}).get("skybot", {}).get("url", _SKYBOT_URL)
    observer = (config or {}).get("data_access", {}).get("skybot", {}).get(
        "default_observer", observer)

    # SkyBoT expects RA in degrees, Dec in degrees, epoch as JD (UTC)
    ra_deg  = center.icrs.ra.deg
    dec_deg = center.icrs.dec.deg
    radius_deg = radius.to(u.deg).value
    jd_utc = epoch.utc.jd

    params = {
        "EPOCH": f"{jd_utc:.6f}",
        "-ra": f"{ra_deg:.6f}",
        "-dec": f"{dec_deg:.6f}",
        "-bd": f"{radius_deg:.4f}",
        "-loc": observer,
        "-mime": "votable",
        "-filter": "0",          # 0 = all object types
        "-refsys": "EQJ2000",
    }

    logger.info(
        "SkyBoT query: RA=%.4f Dec=%.4f r=%.2f' epoch=%s obs=%s",
        ra_deg, dec_deg, radius.to(u.arcmin).value, epoch.utc.isot, observer
    )

    try:
        resp = requests.get(url, params=params, timeout=30)
        resp.raise_for_status()

        if "No solar system object" in resp.text or len(resp.text) < 100:
            logger.debug("SkyBoT: no known SSOs in field")
            return Table()

        from astropy.io.votable import parse_single_table
        import io
        votable = parse_single_table(io.BytesIO(resp.content))
        table = votable.to_table()
        logger.info("SkyBoT: found %d known SSOs in field", len(table))
        return table

    except requests.exceptions.RequestException as exc:
        logger.warning("SkyBoT query failed (network): %s — proceeding without known-object removal", exc)
        return Table()
    except Exception as exc:
        logger.warning("SkyBoT parse error: %s — proceeding without known-object removal", exc)
        return Table()


def skybot_table_to_skycoord(table: Table) -> Optional[SkyCoord]:
    """Convert SkyBoT result table to SkyCoord array for cross-matching."""
    if len(table) == 0:
        return None
    ra_col  = "RA(h)" if "RA(h)" in table.colnames else "RA"
    dec_col = "DE(deg)" if "DE(deg)" in table.colnames else "Dec"
    return SkyCoord(
        ra=table[ra_col],
        dec=table[dec_col],
        unit=(u.hourangle if "h" in ra_col else u.deg, u.deg),
        frame="icrs",
    )