"""Space-Track TLE Collector. Collects satellite TLE (Two-Line Element) data from Space-Track.org. API documentation: https://www.space-track.org/documentation """ from typing import Any, Dict, List from urllib.parse import urlparse import httpx from app.core.data_sources import get_data_sources_config from app.core.logging import get_logger from app.core.satellite_tle import build_tle_lines_from_elements from app.services.collectors.base import BaseCollector logger = get_logger(__name__, service="collector") class SpaceTrackTLECollector(BaseCollector): name = "spacetrack_tle" priority = "P2" module = "L3" frequency_hours = 24 data_type = "satellite_tle" @property def query_url(self) -> str: config = get_data_sources_config() if self._resolved_url: return self._resolved_url return config.get_yaml_url("spacetrack_tle") @property def site_root(self) -> str: config = get_data_sources_config() configured_root = config.get_yaml_value("spacetrack.base_url") if isinstance(configured_root, str) and configured_root: return configured_root.rstrip("/") parsed = urlparse(self.query_url) return f"{parsed.scheme}://{parsed.netloc}".rstrip("/") @property def login_url(self) -> str: return f"{self.site_root}/ajaxauth/login" @property def probe_url(self) -> str: return f"{self.site_root}/basicspacedata/query/class/gp/NORAD_CAT_ID/25544/format/json" async def fetch(self) -> List[Dict[str, Any]]: from app.core.config import settings username = settings.SPACETRACK_USERNAME password = settings.SPACETRACK_PASSWORD if not username or not password: logger.warning_event( "Space-Track credentials are not configured; using sample data", event="collector.spacetrack.credentials_missing", ) return self._get_sample_data() logger.info_event( "Space-Track TLE fetch started", event="collector.spacetrack.fetch.start", ) try: async with httpx.AsyncClient( timeout=120.0, follow_redirects=True, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "application/json, text/html, */*", "Accept-Language": "en-US,en;q=0.9", "Referer": f"{self.site_root}/", }, ) as client: await client.get(f"{self.site_root}/") login_response = await client.post( self.login_url, data={ "identity": username, "password": password, }, ) logger.info_event( "Space-Track login response received", event="collector.spacetrack.login.response", context={"status_code": login_response.status_code}, ) if login_response.status_code == 403: logger.warning_event( "Space-Track login returned forbidden; trying alternate method", event="collector.spacetrack.login.forbidden", ) async with httpx.AsyncClient( timeout=120.0, follow_redirects=True, ) as alt_client: await alt_client.get(f"{self.site_root}/") alt_login = await alt_client.post( self.login_url, data={ "identity": username, "password": password, }, ) logger.info_event( "Space-Track alternate login response received", event="collector.spacetrack.alt_login.response", context={"status_code": alt_login.status_code}, ) if alt_login.status_code == 200: tle_response = await alt_client.get(self.probe_url) if tle_response.status_code == 200: data = tle_response.json() logger.info_event( "Space-Track alternate query completed", event="collector.spacetrack.alt_query.completed", context={"record_count": len(data)}, ) return data if login_response.status_code != 200: logger.warning_event( "Space-Track login failed; using sample data", event="collector.spacetrack.login.failed", context={"status_code": login_response.status_code}, ) return self._get_sample_data() tle_response = await client.get(self.probe_url) logger.info_event( "Space-Track TLE query response received", event="collector.spacetrack.query.response", context={"status_code": tle_response.status_code}, ) if tle_response.status_code != 200: logger.warning_event( "Space-Track TLE query failed; using sample data", event="collector.spacetrack.query.failed", context={"status_code": tle_response.status_code}, ) return self._get_sample_data() data = tle_response.json() logger.info_event( "Space-Track TLE fetch completed", event="collector.spacetrack.fetch.completed", context={"record_count": len(data)}, ) return data except Exception as e: logger.warning_event( "Space-Track TLE fetch failed; using sample data", event="collector.spacetrack.fetch.failed", context={"error": str(e)}, ) return self._get_sample_data() def transform(self, raw_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Transform TLE data to internal format""" transformed = [] for item in raw_data: tle_line1, tle_line2 = build_tle_lines_from_elements( norad_cat_id=item.get("NORAD_CAT_ID"), epoch=item.get("EPOCH"), inclination=item.get("INCLINATION"), raan=item.get("RAAN"), eccentricity=item.get("ECCENTRICITY"), arg_of_perigee=item.get("ARG_OF_PERIGEE"), mean_anomaly=item.get("MEAN_ANOMALY"), mean_motion=item.get("MEAN_MOTION"), ) transformed.append( { "name": item.get("OBJECT_NAME", "Unknown"), "reference_date": item.get("EPOCH", ""), "metadata": { "norad_cat_id": item.get("NORAD_CAT_ID"), "international_designator": item.get("INTL_DESIGNATOR"), "epoch": item.get("EPOCH"), "mean_motion": item.get("MEAN_MOTION"), "eccentricity": item.get("ECCENTRICITY"), "inclination": item.get("INCLINATION"), "raan": item.get("RAAN"), "arg_of_perigee": item.get("ARG_OF_PERIGEE"), "mean_anomaly": item.get("MEAN_ANOMALY"), "ephemeris_type": item.get("EPHEMERIS_TYPE"), "classification_type": item.get("CLASSIFICATION_TYPE"), "element_set_no": item.get("ELEMENT_SET_NO"), "rev_at_epoch": item.get("REV_AT_EPOCH"), "bstar": item.get("BSTAR"), "mean_motion_dot": item.get("MEAN_MOTION_DOT"), "mean_motion_ddot": item.get("MEAN_MOTION_DDOT"), # Prefer original lines from the source, but keep a backend-built pair as a stable fallback. "tle_line1": item.get("TLE_LINE1") or item.get("TLE1") or tle_line1, "tle_line2": item.get("TLE_LINE2") or item.get("TLE2") or tle_line2, }, } ) return transformed def _get_sample_data(self) -> List[Dict[str, Any]]: """Return sample TLE data for testing""" return [ { "name": "ISS (ZARYA)", "norad_cat_id": 25544, "international_designator": "1998-067A", "epoch": "2026-03-13T00:00:00Z", "mean_motion": 15.49872723, "eccentricity": 0.0006292, "inclination": 51.6400, "raan": 315.0000, "arg_of_perigee": 100.0000, "mean_anomaly": 260.0000, }, { "name": "STARLINK-1000", "norad_cat_id": 44720, "international_designator": "2019-029AZ", "epoch": "2026-03-13T00:00:00Z", "mean_motion": 15.79234567, "eccentricity": 0.0001234, "inclination": 53.0000, "raan": 120.0000, "arg_of_perigee": 90.0000, "mean_anomaly": 270.0000, }, ]