# -*- coding: utf-8 -*-
"""CDA_Webservice package for Space Physics WebServices Client."""
__author__ = """Alexis Jeandet"""
__email__ = 'alexis.jeandet@member.fsf.org'
__version__ = '0.1.0'
import logging
import re
from datetime import datetime, timedelta
from typing import Dict, Optional, Tuple
from packaging.version import Version
from speasy.core import AllowedKwargs, EnsureUTCDateTime, make_utc_datetime
from speasy.core import http, url_utils
from email.utils import format_datetime
from speasy.config import cdaweb as cda_cfg
from speasy.core.codecs import get_codec
from speasy.core.cache import CACHE_ALLOWED_KWARGS, CacheCall, UnversionedProviderCache
from speasy.core.dataprovider import (GET_DATA_ALLOWED_KWARGS, DataProvider,
ParameterRangeCheck, register_provider)
from speasy.core.datetime_range import DateTimeRange
from speasy.core.inventory.indexes import (DatasetIndex, ParameterIndex,
SpeasyIndex)
from speasy.core.proxy import PROXY_ALLOWED_KWARGS, GetProduct, Proxyfiable
from speasy.core.requests_scheduling import SplitLargeRequests
from speasy.core.direct_archive_downloader import first_file, get_product as direct_archive_get_product
from speasy.products.variable import SpeasyVariable
from ._direct_archive import to_direct_archive_params
log = logging.getLogger(__name__)
_burst_regex = re.compile("(.*MMS.*FPI.*BRST.*|.*MMS.*SCM.*BRST.*)")
def _is_burst_product(product: ParameterIndex or str) -> bool:
if isinstance(product, ParameterIndex):
product = product.spz_uid()
return bool(_burst_regex.match(str(product)))
def _is_virtual(dataset: DatasetIndex, product: ParameterIndex) -> bool:
"""True when this product can only ever be reached through an API call, never a direct
file archive. Two independent reasons: CDAWeb marks the parameter itself as computed
(VIRTUAL='TRUE', never backed by any file), or the dataset's own CDAWeb catalog entry has
no filenaming pattern at all -- some datasets are pure redirects to another NASA service
(e.g. the "_LinkTo_...sscweb" ephemeris entries, no <access> node whatsoever) or have an
<access> node that never got a filenaming attribute (e.g. some "SPDF Helioweb"-sourced
trajectory datasets). Both leave the DatasetIndex without a `.filenaming` attribute.
"""
return product.__dict__.get('VIRTUAL', 'FALSE').upper() == 'TRUE' or not hasattr(dataset, 'filenaming')
@CacheCall(cache_retention=timedelta(days=7), is_pure=True)
def _codec_can_read(file_url: str, codec: str, variable: str) -> bool:
"""Does the codec really get that variable out of that file?
Listing the variables is not enough: the ICON_L2-7_IVM-A files list 125 of them and then raise
while pyistp walks the axes of half. Only the load answers the question for sure.
"""
try:
return get_codec(codec).load_variable(variable, file=file_url) is not None
except IOError:
raise # transient, and the caller must not remember it as a verdict on the dataset
except Exception as e:
log.warning(f"The {codec} codec failed on {file_url}: {type(e).__name__}: {e}")
return False
def _codec_can_read_archive(archive_params: Dict, variable: str, start_time: datetime,
stop_time: datetime) -> bool:
"""Can this product be read from the first file the archive serves for that range?
Asks a real file rather than looking at what get_data() returned: an empty result only means
the interval holds no data, while a file the codec can't get the variable out of is a property
of the dataset. The retention window gives a dataset that later becomes readable another try.
"""
codec = archive_params['codec']
if get_codec(codec) is None:
log.warning(f"No {codec} codec available, switching to web service")
return False
sample_file = first_file(start_time=start_time, stop_time=stop_time, **archive_params)
if sample_file is None:
return False
try:
return _codec_can_read(sample_file, codec, variable)
except IOError as e:
log.warning(f"Failed to probe {sample_file} with the {codec} codec, switching to web service: {e}")
return False
def _archive_params_for(dataset: DatasetIndex, product_index: ParameterIndex, variable: str,
start_time: datetime, stop_time: datetime) -> Optional[Dict]:
"""Direct archive parameters when this product can really be served from files, None otherwise."""
if _is_virtual(dataset, product_index):
return None
params = to_direct_archive_params(file_naming=dataset.filenaming, subdivided_by=dataset.subdividedby,
url=dataset.url)
if params is None:
return None
if 'codec' not in params:
return {**params, 'master_cdf_url': dataset.mastercdf}
return params if _codec_can_read_archive(params, variable, start_time, stop_time) else None
def _large_request_max_duration(product):
if _is_burst_product(product):
return timedelta(hours=2)
else:
return timedelta(days=7)
def _cache_fragment_size(product):
if _is_burst_product(product):
return 2
else:
return 12
[docs]
class CdaWebException(Exception):
def __init__(self, text):
super(CdaWebException, self).__init__(text)
[docs]
def get_parameter_args_ws(start_time: datetime, stop_time: datetime, product: str, **_):
return {
'path': f"cdaweb/{product}",
'start_time': f'{start_time.isoformat()}',
'stop_time': f'{stop_time.isoformat()}',
'method': 'API',
}
[docs]
@register_provider
class CdaWebservice(DataProvider):
NAME = 'cda'
ALIASES = ('cdaweb',)
BASE_URL = "https://cdaweb.gsfc.nasa.gov"
def __init__(self):
self.__url = f"{self.BASE_URL}/WS/cdasr/1"
DataProvider.__init__(self, min_proxy_version=Version("0.13.0"))
self._cdf_codec = get_codec('application/x-cdf')
[docs]
def build_inventory(self, root: SpeasyIndex):
from ._inventory_builder import build_inventory
root = build_inventory(root=root)
return root
[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.cda.parameter_range("AC_H0_MFI/BGSEc")
<DateTimeRange: 1997-09-02T00:00:12+00:00 -> ...>
"""
return self._parameter_range(parameter_id)
[docs]
def dataset_range(self, dataset_id: str or DatasetIndex) -> Optional[DateTimeRange]:
"""Get product time range.
Parameters
----------
dataset_id: str or DatasetIndex
parameter id
Returns
-------
Optional[DateTimeRange]
Data time range
Examples
--------
>>> import speasy as spz
>>> spz.cda.dataset_range("AC_H0_MFI")
<DateTimeRange: 1997-09-02T00:00:12+00:00 -> ...>
"""
return self._dataset_range(dataset_id)
def _to_dataset_and_variable(self, index_or_str: ParameterIndex or str) -> Tuple[str, str]:
if isinstance(index_or_str, ParameterIndex):
index_or_str = index_or_str.spz_uid()
if type(index_or_str) is str:
if '/' in index_or_str:
parts = index_or_str.split('/')
if len(parts) == 2:
return parts[0], parts[1]
for pos in range(1, len(parts)):
ds = '/'.join(parts[:pos])
var = '/'.join(parts[pos:])
if (ds in self.flat_inventory.datasets) and (index_or_str in self.flat_inventory.parameters):
return ds, var
raise ValueError(
f"Given string is ambiguous, it contains several '/', tried all combinations but failed to find a matching dataset/variable pair in inventory: {index_or_str}")
raise ValueError(f"Given string does not look like a CDA dataset/variable pair: {index_or_str}")
raise TypeError(f"Wrong type for {index_or_str}, expecting a string or a SpeasyIndex, got {type(index_or_str)}")
def _dl_variable(self,
dataset: str, variable: str,
start_time: datetime, stop_time: datetime, if_newer_than: datetime or None = None,
extra_http_headers: Dict or None = None) -> Optional[SpeasyVariable]:
start_time, stop_time = start_time.strftime('%Y%m%dT%H%M%SZ'), stop_time.strftime('%Y%m%dT%H%M%SZ')
fmt = "cdf"
url = f"{self.__url}/dataviews/sp_phys/datasets/{url_utils.quote(dataset, safe='')}/data/{start_time},{stop_time}/{url_utils.quote(variable, safe='')}?format={fmt}"
headers = {"Accept": "application/json"}
if if_newer_than is not None:
# If-Modified-Since must be a valid HTTP-date (RFC 7231), datetime.ctime()
# is not one and may make servers answer 400
headers["If-Modified-Since"] = format_datetime(make_utc_datetime(if_newer_than), usegmt=True)
if extra_http_headers is not None:
headers.update(extra_http_headers)
resp = http.get(url, headers=headers)
log.debug(resp.url)
if resp.status_code == 200 and 'FileDescription' in resp.json():
return self._cdf_codec.load_variable(file=resp.json()['FileDescription'][0]['Name'], variable=variable)
elif not resp.ok:
if resp.status_code == 404 and "No data available" in resp.json().get('Message', [""])[0]:
log.warning(f"Got 404 'No data available' from CDAWeb with {url}")
return None
raise CdaWebException(f'Failed to get data with request: {url}, got {resp.status_code} HTTP response')
else:
return None
@UnversionedProviderCache(prefix="cda", fragment_hours=_cache_fragment_size, cache_retention=timedelta(days=7))
@SplitLargeRequests(threshold=_large_request_max_duration)
@Proxyfiable(GetProduct, get_parameter_args_ws)
def _get_data_with_ws(self, product, start_time: datetime, stop_time: datetime,
if_newer_than: datetime or None = None,
extra_http_headers: Dict or None = None) -> Optional[SpeasyVariable]:
dataset, variable = self._to_dataset_and_variable(product)
return self._dl_variable(start_time=start_time, stop_time=stop_time, dataset=dataset,
variable=variable, if_newer_than=if_newer_than, extra_http_headers=extra_http_headers)
def _get_data_with_direct_archive(self, product, start_time: datetime, stop_time: datetime, mode_is_best: bool,
if_newer_than: datetime or None = None,
extra_http_headers: Dict or None = None) -> Optional[SpeasyVariable]:
dataset, variable = self._to_dataset_and_variable(product)
dataset = self.flat_inventory.datasets[dataset]
if type(product) is str:
product_index = self.flat_inventory.parameters[product]
else:
product_index = product
archive_params = _archive_params_for(dataset, product_index, variable, start_time, stop_time)
log.debug(f"Trying to get {product} with direct_archive method, archive_params={archive_params}")
if archive_params is not None:
return direct_archive_get_product(variable=variable, start_time=start_time, stop_time=stop_time,
**archive_params)
if not mode_is_best:
log.warning(f"Can't get {product} without web service, switching to web service")
return self._get_data_with_ws(product=product, start_time=start_time, stop_time=stop_time,
if_newer_than=if_newer_than, extra_http_headers=extra_http_headers)
[docs]
@AllowedKwargs(
PROXY_ALLOWED_KWARGS + CACHE_ALLOWED_KWARGS + GET_DATA_ALLOWED_KWARGS + ['if_newer_than', 'method'])
@EnsureUTCDateTime()
@ParameterRangeCheck()
def get_data(self, product, start_time: datetime, stop_time: datetime, if_newer_than: datetime or None = None,
extra_http_headers: Dict or None = None, method: Optional[str] = None, **kwargs) -> Optional[
SpeasyVariable]:
method = method or cda_cfg.preferred_access_method.get()
if method.upper() in ('FILE', 'BEST'):
return self._get_data_with_direct_archive(product=product, start_time=start_time, stop_time=stop_time,
if_newer_than=if_newer_than,
extra_http_headers=extra_http_headers,
mode_is_best=method.upper() == 'BEST')
else:
return self._get_data_with_ws(product=product, start_time=start_time, stop_time=stop_time,
if_newer_than=if_newer_than, extra_http_headers=extra_http_headers, **kwargs)
[docs]
def get_variable(self, dataset: str, variable: str, start_time: datetime or str, stop_time: datetime or str,
**kwargs) -> \
Optional[SpeasyVariable]:
return self.get_data(f"{dataset}/{variable}", start_time, stop_time, **kwargs)