Spaces:
Configuration error
Configuration error
File size: 6,905 Bytes
3dc2617 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 |
import json
import logging
import requests
from typing import Dict, Any, List, Optional
from datetime import datetime, timedelta
from .config import logger
class SpaceTrackApi:
"""
Cliente para la API de Space-Track que maneja solicitudes
y respuestas en formato JSON para obtener información sobre objetos espaciales,
TLEs, y datos de conjunciones.
"""
def __init__(self):
self.logger = logging.getLogger("Orbix.SpaceTrackApi")
from .config import SPACE_TRACK_USER, SPACE_TRACK_PASS
self.BASE_URL = "https://www.space-track.org"
self.AUTH_URL = f"{self.BASE_URL}/ajaxauth/login"
self.username = SPACE_TRACK_USER
self.password = SPACE_TRACK_PASS
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json"
})
def _login(self) -> bool:
"""
Inicia sesión en la API de Space-Track.
Returns:
True si el login fue exitoso, False en caso contrario
"""
try:
payload = {
"identity": self.username,
"password": self.password
}
response = self.session.post(self.AUTH_URL, data=payload)
response.raise_for_status()
return True
except requests.RequestException as e:
self.logger.error(f"Error al iniciar sesión en Space-Track: {str(e)}")
return False
def _process_response(self, data: Any) -> Dict[str, Any]:
"""
Procesa la respuesta de la API y la convierte en un formato estándar.
Args:
data: Datos de respuesta de la API
Returns:
Datos procesados en formato estándar
"""
if isinstance(data, list):
return {"data": data}
return {"data": [data]} if data else {"data": []}
def get_satellite_catalog(self, limit: Optional[int] = None) -> Dict[str, Any]:
"""
Obtiene el catálogo de satélites de Space-Track.
Args:
limit: Número máximo de satélites a obtener
Returns:
Información sobre satélites
"""
if not self._login():
return {"error": "Error de autenticación"}
try:
url = f"{self.BASE_URL}/basicspacedata/query/class/satcat"
params = {"format": "json"}
if limit is not None:
params["limit"] = limit
response = self.session.get(url, params=params)
response.raise_for_status()
return self._process_response(response.json())
except requests.RequestException as e:
self.logger.error(f"Error al obtener catálogo de satélites: {str(e)}")
return {"error": str(e)}
def get_latest_tle(self, norad_id: int) -> Dict[str, Any]:
"""
Obtiene el TLE más reciente para un satélite específico.
Args:
norad_id: ID NORAD del satélite
Returns:
TLE más reciente del satélite
"""
if not self._login():
return {"error": "Error de autenticación"}
try:
url = f"{self.BASE_URL}/basicspacedata/query/class/tle_latest/NORAD_CAT_ID/{norad_id}/orderby/EPOCH desc/limit/1"
params = {"format": "json"}
response = self.session.get(url, params=params)
response.raise_for_status()
return self._process_response(response.json())
except requests.RequestException as e:
self.logger.error(f"Error al obtener TLE para el satélite {norad_id}: {str(e)}")
return {"error": str(e)}
def get_conjunction_data(self, days_from_now: int = 7) -> Dict[str, Any]:
"""
Obtiene datos de conjunciones (posibles colisiones) para los próximos días.
Args:
days_from_now: Número de días hacia adelante para obtener datos
Returns:
Datos de conjunciones
"""
if not self._login():
return {"error": "Error de autenticación"}
try:
now = datetime.utcnow()
future = now + timedelta(days=days_from_now)
start_date = now.strftime("%Y-%m-%d")
end_date = future.strftime("%Y-%m-%d")
url = f"{self.BASE_URL}/basicspacedata/query/class/cdm_public/CREATION_DATE/>/{start_date}/CREATION_DATE/</{end_date}"
params = {"format": "json"}
response = self.session.get(url, params=params)
response.raise_for_status()
return self._process_response(response.json())
except requests.RequestException as e:
self.logger.error(f"Error al obtener datos de conjunciones: {str(e)}")
return {"error": str(e)}
def get_decay_data(self, days_from_now: int = 30) -> Dict[str, Any]:
"""
Obtiene datos de decaimiento orbital para los próximos días.
Args:
days_from_now: Número de días hacia adelante para obtener datos
Returns:
Datos de decaimiento orbital
"""
if not self._login():
return {"error": "Error de autenticación"}
try:
now = datetime.utcnow()
future = now + timedelta(days=days_from_now)
start_date = now.strftime("%Y-%m-%d")
end_date = future.strftime("%Y-%m-%d")
url = f"{self.BASE_URL}/basicspacedata/query/class/decay/DECAY_DATE/>/{start_date}/DECAY_DATE/</{end_date}"
params = {"format": "json"}
response = self.session.get(url, params=params)
response.raise_for_status()
return self._process_response(response.json())
except requests.RequestException as e:
self.logger.error(f"Error al obtener datos de decaimiento orbital: {str(e)}")
return {"error": str(e)}
def authenticate(self):
"""
Autentica con la API de Space-Track usando credenciales.
"""
try:
# Intentar obtener credenciales de Streamlit secrets
import streamlit as st
username = st.secrets["space_track"]["username"]
password = st.secrets["space_track"]["password"]
except (KeyError, ImportError):
# Fallback a variables de entorno
username = os.getenv("SPACE_TRACK_USERNAME")
password = os.getenv("SPACE_TRACK_PASSWORD")
if not username or not password:
raise ValueError("Credenciales de Space-Track no encontradas") |