116 lines
4.4 KiB
Python
116 lines
4.4 KiB
Python
"""CelesTrak TLE Collector
|
|
|
|
Collects satellite TLE (Two-Line Element) data from CelesTrak.org.
|
|
Free, no authentication required.
|
|
"""
|
|
|
|
import json
|
|
from typing import Dict, Any, List
|
|
import httpx
|
|
|
|
from app.core.satellite_tle import build_tle_lines_from_elements
|
|
from app.services.collectors.base import BaseCollector
|
|
|
|
|
|
class CelesTrakTLECollector(BaseCollector):
|
|
name = "celestrak_tle"
|
|
priority = "P2"
|
|
module = "L3"
|
|
frequency_hours = 24
|
|
data_type = "satellite_tle"
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
return "https://celestrak.org/NORAD/elements/gp.php"
|
|
|
|
async def fetch(self) -> List[Dict[str, Any]]:
|
|
satellite_groups = [
|
|
"starlink",
|
|
"gps-ops",
|
|
"galileo",
|
|
"glonass",
|
|
"beidou",
|
|
"leo",
|
|
"geo",
|
|
"iridium-next",
|
|
]
|
|
|
|
all_satellites = []
|
|
|
|
async with httpx.AsyncClient(timeout=120.0) as client:
|
|
for group in satellite_groups:
|
|
try:
|
|
url = f"https://celestrak.org/NORAD/elements/gp.php?GROUP={group}&FORMAT=json"
|
|
response = await client.get(url)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if isinstance(data, list):
|
|
all_satellites.extend(data)
|
|
print(f"CelesTrak: Fetched {len(data)} satellites from group '{group}'")
|
|
except Exception as e:
|
|
print(f"CelesTrak: Error fetching group '{group}': {e}")
|
|
|
|
if not all_satellites:
|
|
return self._get_sample_data()
|
|
|
|
print(f"CelesTrak: Total satellites fetched: {len(all_satellites)}")
|
|
|
|
# Return raw data - base.run() will call transform()
|
|
return all_satellites
|
|
|
|
def transform(self, raw_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
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("RA_OF_ASC_NODE"),
|
|
eccentricity=item.get("ECCENTRICITY"),
|
|
arg_of_perigee=item.get("ARG_OF_PERICENTER"),
|
|
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("OBJECT_ID"),
|
|
"epoch": item.get("EPOCH"),
|
|
"mean_motion": item.get("MEAN_MOTION"),
|
|
"eccentricity": item.get("ECCENTRICITY"),
|
|
"inclination": item.get("INCLINATION"),
|
|
"raan": item.get("RA_OF_ASC_NODE"),
|
|
"arg_of_perigee": item.get("ARG_OF_PERICENTER"),
|
|
"mean_anomaly": item.get("MEAN_ANOMALY"),
|
|
"classification_type": item.get("CLASSIFICATION_TYPE"),
|
|
"bstar": item.get("BSTAR"),
|
|
"mean_motion_dot": item.get("MEAN_MOTION_DOT"),
|
|
"mean_motion_ddot": item.get("MEAN_MOTION_DDOT"),
|
|
"ephemeris_type": item.get("EPHEMERIS_TYPE"),
|
|
# Prefer the original TLE lines when the source provides them.
|
|
# If they are missing, store a normalized TLE pair built once on the backend.
|
|
"tle_line1": item.get("TLE_LINE1") or tle_line1,
|
|
"tle_line2": item.get("TLE_LINE2") or tle_line2,
|
|
},
|
|
}
|
|
)
|
|
return transformed
|
|
|
|
def _get_sample_data(self) -> List[Dict[str, Any]]:
|
|
return [
|
|
{
|
|
"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.0,
|
|
},
|
|
]
|