# -*- coding: utf-8 -*-
"""ssc package for Space Physics WebServices Client."""
__author__ = """Alexis Jeandet"""
__email__ = 'alexis.jeandet@member.fsf.org'
__version__ = '0.1.0'
import logging
import time
from datetime import datetime, timedelta
from typing import Optional, Dict
import xml.etree.ElementTree as ET
import numpy as np
from speasy.core import EnsureUTCDateTime
from speasy.core.cache import Cacheable, CacheCall, CACHE_ALLOWED_KWARGS
from speasy.core.dataprovider import DataProvider, ParameterRangeCheck, GET_DATA_ALLOWED_KWARGS, register_provider
from speasy.core.datetime_range import DateTimeRange
from speasy.core.inventory.indexes import ParameterIndex, SpeasyIndex
from speasy.core.proxy import Proxyfiable, GetProduct, PROXY_ALLOWED_KWARGS
from speasy.core.requests_scheduling import SplitLargeRequests
from speasy.products.variable import SpeasyVariable, VariableTimeAxis, DataContainer
from ...core import AllowedKwargs, http
[docs]
def drop_namespace(xml):
for elem in xml.iter():
tag_elements = elem.tag.split("}")
# Removing name spaces and attributes
elem.tag = tag_elements[1]
elem.attrib.clear()
return xml
[docs]
def parse_inventory(inventory: str):
xml = drop_namespace(ET.fromstring(inventory))
return [
{
item.tag: item.text
for item in obs
}
for obs in xml.findall('Observatory')
]
log = logging.getLogger(__name__)
def _is_valid(xml):
# SSCWeb occasionally returns a valid XML document that doesn't match the
# expected <Result><StatusCode>.../<Result> shape (e.g. some other fault
# format); .find() returns None rather than raising for a missing element,
# so every level here must be checked before descending further.
result = xml.find('Result')
if result is None:
return False
status_code = result.find('StatusCode')
status_subcode = result.find('StatusSubCode')
return (status_code is not None and status_code.text.lower() == 'success'
and status_subcode is not None and status_subcode.text.lower() == 'success')
[docs]
def parse_trajectory(trajectory: str) -> Optional[SpeasyVariable]:
xml = drop_namespace(ET.fromstring(trajectory))
if not _is_valid(xml):
return None
data = xml.find('Result').find('Data')
if data is None:
return None
coordinates = data.find('Coordinates')
if coordinates is None:
return None
coordinate_system = coordinates.find('CoordinateSystem')
if coordinate_system is None:
return None
values = np.array(
[list(map(lambda v: float(v.text), coordinates.findall(axis))) for axis in ('X', 'Y', 'Z')]).transpose()
time_axis = np.array([np.datetime64(v.text[:-1], 'ns') for v in data.findall('Time')])
return SpeasyVariable(
axes=[VariableTimeAxis(values=time_axis)],
values=DataContainer(values,
name='Position',
meta={'CoordinateSystem': coordinate_system.text.upper(),
'UNITS': 'km'}),
columns=['X', 'Y', 'Z']
)
# SSCWeb sometimes answers 200 OK with a response that doesn't parse into a
# variable (see _is_valid/parse_trajectory) even though the transport layer
# already retries on HTTP-level failures (speasy.core.http's Retry). This is
# a second, small retry budget for that content-level anomaly.
SSC_MALFORMED_RESPONSE_RETRY_COUNT = 2
SSC_MALFORMED_RESPONSE_RETRY_DELAY_SECONDS = 1
def _make_cache_entry_name(prefix: str, product: str, start_time: str, **kwargs):
return f"{prefix}/{product}/{kwargs.get('coordinate_system', 'gse')}/{start_time}"
[docs]
def get_parameter_args(start_time: datetime, stop_time: datetime, product: str, **kwargs):
return {'path': f"sscweb/{product}", 'start_time': f'{start_time.isoformat()}',
'stop_time': f'{stop_time.isoformat()}', 'coordinate_system': kwargs.get('coordinate_system', 'gse')}
[docs]
def make_index(meta: Dict):
name = meta.pop('Name')
meta['start_date'] = meta.pop('StartTime')
meta['stop_date'] = meta.pop('EndTime')
node = ParameterIndex(name=name, provider="ssc", uid=meta['Id'], meta=meta)
return node
[docs]
@register_provider
class SscWebservice(DataProvider):
NAME = 'ssc'
ALIASES = ('sscweb',)
BASE_URL = "https://sscweb.gsfc.nasa.gov"
def __init__(self):
self.__url = f"{self.BASE_URL}/WS/sscr/2"
DataProvider.__init__(self)
[docs]
def build_inventory(self, root: SpeasyIndex):
inv = list(map(make_index, self.get_observatories()))
root.Trajectories = SpeasyIndex(name='Trajectories', provider='ssc', uid='Trajectories',
meta={item.Id: item for item in inv})
return root
[docs]
@CacheCall(cache_retention=7 * 24 * 60 * 60, is_pure=True)
def get_observatories(self):
res = http.get(f"{self.__url}/observatories", headers={"Accept": "application/xml"})
if not res.ok:
return None
return parse_inventory(res.text)
[docs]
def version(self, product):
return 4
[docs]
def parameter_range(self, parameter_id: str or ParameterIndex) -> Optional[DateTimeRange]:
"""Get product time range.
Parameters
----------
parameter_id: str or ParameterIndex
parameter id
Returns
-------
Optional[DateTimeRange]
Data time range
Examples
--------
>>> import speasy as spz
>>> spz.ssc.parameter_range("solarorbiter")
<DateTimeRange: 2020-02-10T04:56:30+00:00 -> ...>
"""
return self._parameter_range(parameter_id)
[docs]
def get_data(self, product: str, start_time: datetime, stop_time: datetime, coordinate_system: str = 'gse',
debug=False, **kwargs) -> Optional[SpeasyVariable]:
var = self._get_orbit(product=product, start_time=start_time, stop_time=stop_time,
coordinate_system=coordinate_system, debug=debug, **kwargs)
return var
@AllowedKwargs(
PROXY_ALLOWED_KWARGS + CACHE_ALLOWED_KWARGS + GET_DATA_ALLOWED_KWARGS + ['coordinate_system',
'debug'])
@EnsureUTCDateTime()
@ParameterRangeCheck()
@Cacheable(prefix="ssc_orbits", fragment_hours=lambda x: 24, version=version, entry_name=_make_cache_entry_name)
@SplitLargeRequests(threshold=lambda x: timedelta(days=60))
@Proxyfiable(GetProduct, get_parameter_args)
def _get_orbit(self, product: str, start_time: datetime, stop_time: datetime, coordinate_system: str = 'gse',
debug=False, extra_http_headers: Dict or None = None) -> Optional[SpeasyVariable]:
request_stop_time = stop_time
if request_stop_time - start_time < timedelta(days=1):
# pad to a full day to match this method's own cache fragment size
# (fragment_hours=24 above), then slice back to what was actually
# asked for below
request_stop_time += timedelta(days=1)
url = f"{self.__url}/locations/{product}/{start_time.strftime('%Y%m%dT%H%M%SZ')},{request_stop_time.strftime('%Y%m%dT%H%M%SZ')}/{coordinate_system.lower()}/"
headers = {"Accept": "application/xml"}
if extra_http_headers is not None:
headers.update(extra_http_headers)
for attempt in range(1 + SSC_MALFORMED_RESPONSE_RETRY_COUNT):
log.debug(f"Requesting {url} (attempt {attempt + 1})")
res = http.get(url, headers=headers)
if res.ok:
maybe_var = parse_trajectory(res.text)
if maybe_var is not None:
return maybe_var[start_time:stop_time]
if attempt < SSC_MALFORMED_RESPONSE_RETRY_COUNT:
log.warning(f"Unexpected SSCWeb response for {url}, retrying")
time.sleep(SSC_MALFORMED_RESPONSE_RETRY_DELAY_SECONDS)
else:
log.warning(f"Giving up on {url} after {attempt + 1} attempts")
return None