""" NASA SBDB (Small-Body Database) Close-Approach Data API Service Fetches close approach events for asteroids and comets API Docs: https://ssd-api.jpl.nasa.gov/doc/cad.html """ import logging import httpx from typing import List, Dict, Optional, Any from datetime import datetime, timedelta from app.config import settings logger = logging.getLogger(__name__) class NasaSbdbService: """NASA Small-Body Database Close-Approach Data API client""" def __init__(self): self.base_url = "https://ssd-api.jpl.nasa.gov/cad.api" self.timeout = settings.nasa_api_timeout or 30 async def get_close_approaches( self, date_min: Optional[str] = None, date_max: Optional[str] = None, dist_max: Optional[str] = "0.2", # Max distance in AU (Earth-Moon distance ~0.0026 AU) body: Optional[str] = None, sort: str = "date", limit: Optional[int] = None, fullname: bool = False ) -> List[Dict[str, Any]]: """ Query NASA SBDB Close-Approach Data API Args: date_min: Minimum approach date (YYYY-MM-DD or 'now') date_max: Maximum approach date (YYYY-MM-DD) dist_max: Maximum approach distance in AU (default 0.2 AU) body: Filter by specific body (e.g., 'Earth') sort: Sort by 'date', 'dist', 'dist-min', etc. limit: Maximum number of results fullname: Return full designation names Returns: List of close approach events """ params = { "dist-max": dist_max, "sort": sort, "fullname": "true" if fullname else "false" } if date_min: params["date-min"] = date_min if date_max: params["date-max"] = date_max if body: params["body"] = body if limit: params["limit"] = str(limit) logger.info(f"Querying NASA SBDB for close approaches: {params}") # Use proxy if configured proxies = settings.proxy_dict if proxies: logger.info(f"Using proxy for NASA SBDB API") try: async with httpx.AsyncClient(timeout=self.timeout, proxies=proxies) as client: response = await client.get(self.base_url, params=params) response.raise_for_status() data = response.json() if "data" not in data: logger.warning("No data field in NASA SBDB response") return [] # Parse response fields = data.get("fields", []) rows = data.get("data", []) events = [] for row in rows: event = dict(zip(fields, row)) events.append(event) logger.info(f"Retrieved {len(events)} close approach events from NASA SBDB") return events except httpx.HTTPStatusError as e: logger.error(f"NASA SBDB API HTTP error: {e.response.status_code} - {e.response.text}") return [] except httpx.TimeoutException: logger.error(f"NASA SBDB API timeout after {self.timeout}s") return [] except Exception as e: logger.error(f"Error querying NASA SBDB: {e}") return [] def parse_event_to_celestial_event(self, sbdb_event: Dict[str, Any], approach_body: str = "Earth") -> Dict[str, Any]: """ Parse NASA SBDB event data to CelestialEvent format Args: sbdb_event: Event data from NASA SBDB API approach_body: Name of the body being approached (e.g., "Earth", "Mars") SBDB fields typically include: - des: Object designation - orbit_id: Orbit ID - jd: Julian Date of close approach - cd: Calendar date (YYYY-MMM-DD HH:MM) - dist: Nominal approach distance (AU) - dist_min: Minimum approach distance (AU) - dist_max: Maximum approach distance (AU) - v_rel: Relative velocity (km/s) - v_inf: Velocity at infinity (km/s) - t_sigma_f: Time uncertainty (formatted string) - h: Absolute magnitude - fullname: Full object name (if requested) """ try: # Extract fields designation = sbdb_event.get("des", "Unknown") fullname = sbdb_event.get("fullname", designation) cd = sbdb_event.get("cd", "") # Calendar date string dist = sbdb_event.get("dist", "") # Nominal distance in AU dist_min = sbdb_event.get("dist_min", "") v_rel = sbdb_event.get("v_rel", "") # Note: NASA API doesn't return the approach body, so we use the parameter body = approach_body # Parse date (format: YYYY-MMM-DD HH:MM) event_time = datetime.strptime(cd, "%Y-%b-%d %H:%M") # Create title title = f"{fullname} Close Approach to {body}" # Create description desc_parts = [ f"Asteroid/Comet {fullname} will make a close approach to {body}.", f"Nominal distance: {dist} AU", ] if dist_min: desc_parts.append(f"Minimum distance: {dist_min} AU") if v_rel: desc_parts.append(f"Relative velocity: {v_rel} km/s") description = " ".join(desc_parts) # Store all technical details in JSONB details = { "designation": designation, "orbit_id": sbdb_event.get("orbit_id"), "julian_date": sbdb_event.get("jd"), "nominal_dist_au": dist, "dist_min_au": dist_min, "dist_max_au": sbdb_event.get("dist_max"), "relative_velocity_km_s": v_rel, "v_inf": sbdb_event.get("v_inf"), "time_sigma": sbdb_event.get("t_sigma_f"), "absolute_magnitude": sbdb_event.get("h"), "approach_body": body } return { "body_id": designation, # Will need to map to celestial_bodies.id "title": title, "event_type": "approach", "event_time": event_time, "description": description, "details": details, "source": "nasa_sbdb" } except Exception as e: logger.error(f"Error parsing SBDB event: {e}") return None # Singleton instance nasa_sbdb_service = NasaSbdbService()