Spaces:
Configuration error
Configuration error
| 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") |