Spaces:
Configuration error
Configuration error
File size: 9,064 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 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 |
import json
import logging
import requests
from typing import Dict, Any, List, Optional
from datetime import datetime
from .config import logger
class SSCApi:
"""
Cliente para la API del Space Science Center (SSC) que maneja solicitudes
y respuestas en formato JSON con las peculiaridades requeridas por la API.
"""
def __init__(self):
self.logger = logging.getLogger("Orbix.SSCApi")
from .config import SSC_API_URL, SSC_API_KEY
self.BASE_URL = SSC_API_URL
self.API_KEY = SSC_API_KEY
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json",
"X-Api-Key": self.API_KEY
})
def _format_time(self, dt: datetime) -> str:
"""
Formatea una fecha y hora en el formato requerido por la API SSC.
"""
return dt.strftime("%Y-%m-%dT%H:%M:%S.000+00:00")
def create_data_request(self,
satellites: List[str],
start_time: datetime,
end_time: datetime,
resolution_factor: int = 1) -> Dict[str, Any]:
"""
Crea una solicitud de datos en el formato JSON requerido por la API SSC.
Args:
satellites: Lista de IDs de satélites
start_time: Tiempo de inicio para los datos
end_time: Tiempo de fin para los datos
resolution_factor: Factor de resolución para los datos (1=máxima resolución)
Returns:
Diccionario con la solicitud formateada para la API SSC
"""
# Crear especificaciones de satélites en el formato requerido
satellite_specs = [
[
"gov.nasa.gsfc.sscweb.schema.SatelliteSpecification",
{
"Id": sat_id,
"ResolutionFactor": resolution_factor
}
] for sat_id in satellites
]
# Construir la solicitud completa con el formato específico requerido
request_data = [
"gov.nasa.gsfc.sscweb.schema.DataRequest",
{
"Description": "Orbital data request from Orbix",
"TimeInterval": [
"gov.nasa.gsfc.sscweb.schema.TimeInterval",
{
"Start": [
"javax.xml.datatype.XMLGregorianCalendar",
self._format_time(start_time)
],
"End": [
"javax.xml.datatype.XMLGregorianCalendar",
self._format_time(end_time)
]
}
],
"Satellites": [
"java.util.ArrayList",
satellite_specs
],
"OutputOptions": [
"gov.nasa.gsfc.sscweb.schema.OutputOptions",
{
"CoordinateSystem": "gse",
"AllLocationFilters": True,
"RegionFilterDistance": 0.0,
"MinMaxPoints": 2
}
]
}
]
return request_data
def get_satellite_data(self,
satellites: List[str],
start_time: datetime,
end_time: datetime,
resolution_factor: int = 1) -> Dict[str, Any]:
"""
Obtiene datos de satélites de la API SSC.
Args:
satellites: Lista de IDs de satélites
start_time: Tiempo de inicio para los datos
end_time: Tiempo de fin para los datos
resolution_factor: Factor de resolución para los datos (1=máxima resolución)
Returns:
Datos de satélites procesados
"""
request_data = self.create_data_request(
satellites, start_time, end_time, resolution_factor
)
try:
response = self.session.post(
f"{self.BASE_URL}/locations",
json=request_data
)
response.raise_for_status()
return self._process_response(response.json())
except requests.RequestException as e:
self.logger.error(f"Error al obtener datos de satélites: {str(e)}")
return {"error": str(e)}
def _process_response(self, response_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Procesa la respuesta de la API SSC y la convierte a un formato más
utilizable para el sistema Orbix.
Args:
response_data: Datos de respuesta de la API SSC
Returns:
Datos procesados en formato compatible con Orbix
"""
try:
# Extraer los datos relevantes de la estructura JSON compleja
# La estructura exacta dependerá de la respuesta de la API
result = {}
# Verificar si hay un error en la respuesta
if "Exception" in str(response_data):
error_msg = str(response_data)
self.logger.error(f"Error en la respuesta de la API SSC: {error_msg}")
return {"error": error_msg}
# Extraer datos de satélites
# Nota: Esta parte puede necesitar ajustes según la estructura exacta de la respuesta
if isinstance(response_data, list) and len(response_data) > 1:
data_container = response_data[1]
if "Result" in data_container and isinstance(data_container["Result"], list):
result_data = data_container["Result"][1]
# Procesar datos para cada satélite
satellites_data = {}
for sat_data in result_data:
if isinstance(sat_data, list) and len(sat_data) > 1:
sat_id = sat_data[1].get("Id")
if sat_id:
# Extraer coordenadas y tiempos
coordinates = sat_data[1].get("Coordinates", [])
times = sat_data[1].get("Time", [])
# Convertir a formato utilizable
trajectory_data = []
for i in range(min(len(coordinates), len(times))):
if i < len(coordinates) and i < len(times):
point = {
"time": times[i],
"x": coordinates[i][0],
"y": coordinates[i][1],
"z": coordinates[i][2]
}
trajectory_data.append(point)
satellites_data[sat_id] = trajectory_data
result["satellites"] = satellites_data
return result
except Exception as e:
self.logger.error(f"Error al procesar la respuesta: {str(e)}")
return {"error": str(e)}
def get_available_satellites(self) -> List[str]:
"""
Obtiene la lista de satélites disponibles en la API SSC.
Returns:
Lista de IDs de satélites disponibles
"""
try:
response = self.session.get(f"{self.BASE_URL}/observatories")
response.raise_for_status()
# Procesar la respuesta para extraer los IDs de satélites
# La estructura exacta dependerá de la respuesta de la API
satellites = []
response_data = response.json()
# Extraer IDs de satélites de la respuesta
# Nota: Esta parte puede necesitar ajustes según la estructura exacta de la respuesta
if isinstance(response_data, list) and len(response_data) > 1:
data_container = response_data[1]
if "Observatory" in data_container and isinstance(data_container["Observatory"], list):
for sat in data_container["Observatory"][1]:
if isinstance(sat, list) and len(sat) > 1:
sat_id = sat[1].get("Id")
if sat_id:
satellites.append(sat_id)
return satellites
except requests.RequestException as e:
self.logger.error(f"Error al obtener satélites disponibles: {str(e)}")
return [] |