# -*- coding: utf-8 -*-
'''WebGeoCalc API module.'''
import os
import requests
from .errors import APIError, APIReponseError, KernelSetNotFound, TooManyKernelSets
from .types import ColumnResult, KernelSetDetails, get_type
from .vars import ESA_URL, JPL_URL
[docs]class Api:
'''WebGeoCalc API object.
Parameters
----------
url : str, optional
API root URL.
Use ``WGC_URL`` global environment variable if present.
If not, fallback on :py:obj:`JPL_URL`:
``https://wgc2.jpl.nasa.gov:8443/webgeocalc/api``
'''
def __init__(self, url=''):
self.url = str(url) if url != '' else os.environ.get('WGC_URL', JPL_URL)
self._kernel_sets = None
self._meta = None
def __str__(self):
return self.url
def __repr__(self):
return f'<{self.__class__.__name__}> {self}'
def __getitem__(self, key):
if key in self.metadata:
return self.metadata[key]
raise KeyError(key)
[docs] def get(self, url):
'''Generic GET request on the API.
Parameters
----------
url: str
Request url with GET method.
Returns
-------
list or tuple:
Parsed json API response. See: :py:func:`read`.
Raises
------
requests.reponse.HTMLError
If HTML error is thrown by the API (HTML code not equal 200)
Example
-------
>>> API.get('/kernel-sets')
[<KernelSetDetails> Solar System Kernels (id: 1), ...]
'''
response = requests.get(self.url + url)
if response.ok:
return self.read(response.json())
return response.raise_for_status()
[docs] def post(self, url, payload):
'''Generic POST request on the API.
Parameters
----------
url: str
Request url with POST method.
payload: dict
Calculation payload data.
Returns
-------
list or tuple:
Parsed json API response. See: :py:func:`read`.
Raises
------
requests.reponse.HTMLError
If HTML error is thrown by the API (HTML code not equal 200)
Example
-------
>>> API.post('/calculation/new', payload=calculation_payload) # doctest: +SKIP
('0788aba2-d4e5-4028-9ef1-4867ad5385e0', 'COMPLETE')
'''
response = requests.post(self.url + url, json=payload)
if response.ok:
return self.read(response.json())
return response.raise_for_status()
[docs] @staticmethod
def read(json):
'''Read content from API JSON reponse.
Parameters
----------
json: dict
API JSON response
Returns
-------
list or tuple
Parsed content of the JSON API response.
If the response contents ``resultType`` and ``items`` keys, then
return type will be:
[`Result objects`: :obj:`webgeocalc.types.ResultType`]
If response contents ``result``, then return type will be:
(`Calculation id`: str, `Phase`: str)
If the response contents ``columns`` and ``rows``, then
return type will be:
(`Columns`: [:obj:`webgeocalc.types.ColumnResult`],
`Results rows`: [[int, float or str], ...])
Else an error is thrown.
Raises
------
APIError
If the API status is not ``OK``.
APIReponseError
If the format of the API response is unexpected.
'''
if 'status' not in json:
return json
if json['status'] != 'OK':
raise APIError(json['error']['shortDescription'])
keys = json.keys()
if 'resultType' in keys and 'items' in json:
dtype = get_type(json['resultType'])
return [dtype(item) for item in json['items']]
if 'result' in keys:
phase = json['result']['phase']
if phase == 'QUEUED':
phase += f' | POSITION: {json["result"]["position"]}'
return json['calculationId'], phase
if 'columns' in keys and 'rows' in keys:
cols = [ColumnResult(col) for col in json['columns']]
return cols, json['rows']
raise APIReponseError(json)
[docs] def kernel_sets(self):
'''Get list of all kernel sets available on the webgeocalc server.
``GET: /kernel-sets``
Returns
-------
[:obj:`webgeocalc.types.KernelSetDetails`]
List of kernel sets.
'''
if self._kernel_sets is None:
self._kernel_sets = self.get('/kernel-sets')
return self._kernel_sets
[docs] def kernel_set(self, kernel_set):
'''Get kernel set by ``caption`` name or kernel set ``id``.
Parameters
----------
kernel_set: str or int
Kernel set ``name`` or ``id``.
Returns
--------
:obj:`webgeocalc.types.KernelSetDetails`
Kernel set details object.
'''
kernel_sets = list(filter(lambda x: kernel_set in x, self.kernel_sets()))
if len(kernel_sets) == 1:
return kernel_sets[0]
if len(kernel_sets) > 1:
raise TooManyKernelSets(kernel_set, kernel_sets)
raise KernelSetNotFound(kernel_set)
[docs] def kernel_set_id(self, kernel_set):
'''Extract kernel set ``id`` based on ``id``, ``name`` or `object`.
Parameters
----------
kernel_set: str, int or :obj:`webgeocalc.types.KernelSetDetails`
Kernel sets ``name``, ``id`` or `object`.
Returns
-------
int
Kernel set ``id``.
'''
if isinstance(kernel_set, int):
return kernel_set
if isinstance(kernel_set, str):
return int(self.kernel_set(kernel_set))
if isinstance(kernel_set, KernelSetDetails):
return int(kernel_set)
raise TypeError(f"'kernel_set' must be a 'int', a 'str' of a 'KernelSetDetails' object:\n' + \
'>>> Type({kernel_set}) = {type(kernel_set)}")
[docs] def bodies(self, kernel_set):
'''Get list of bodies available in a kernel set.
``GET: /kernel-set/{kernelSetId}/bodies``
Parameters
----------
kernel_set: str, int or :obj:`webgeocalc.types.KernelSetDetails`
Kernel sets ``name``, ``id`` or `object`.
Returns
-------
[:obj:`webgeocalc.types.BodyData`]
List of bodies in the requested kernel set.
'''
kernel_set_id = self.kernel_set_id(kernel_set)
return self.get(f'/kernel-set/{kernel_set_id}/bodies')
[docs] def frames(self, kernel_set):
'''Get list of frames available in a kernel set.
``GET: /kernel-set/{kernelSetId}/frames``
Parameters
----------
kernel_set: str, int or :obj:`webgeocalc.types.KernelSetDetails`
Kernel sets ``name``, ``id`` or `object`.
Returns
-------
[:obj:`webgeocalc.types.FrameData`]
List of frames in the requested kernel set.
'''
kernel_set_id = self.kernel_set_id(kernel_set)
return self.get(f'/kernel-set/{kernel_set_id}/frames')
[docs] def instruments(self, kernel_set):
'''Get list of instruments available in a kernel set.
``GET: /kernel-set/{kernelSetId}/instruments``
Parameters
----------
kernel_set: str, int or :obj:`webgeocalc.types.KernelSetDetails`
Kernel sets ``name``, ``id`` or `object`.
Returns
-------
[:obj:`webgeocalc.types.InstrumentData`]
List of instruments in the requested kernel set.
'''
kernel_set_id = self.kernel_set_id(kernel_set)
return self.get(f'/kernel-set/{kernel_set_id}/instruments')
[docs] def new_calculation(self, payload):
'''Starts a new calculation.
``POST: /calculation/new``
Parameters
----------
payload: dict
Calculation payload.
Returns
-------
(str, str)
Tuple of the calculation phase:
``(calculation-id, phase)``
See: :py:func:`read`.
Example
-------
>>> API.calculation_new(calculation_payload) # doctest: +SKIP
('0788aba2-d4e5-4028-9ef1-4867ad5385e0', 'LOADING_KERNELS')
'''
return self.post('/calculation/new', payload)
[docs] def phase_calculation(self, calculation_id):
'''Gets the phase of a calculation.
``GET: /calculation/{id}``
Parameters
----------
calculation_id: str
Calculation id.
Returns
-------
(str, str)
Tuple of the calculation phase:
``(calculation-id, phase)``
See: :py:func:`read`.
Example
-------
>>> API.phase_calculation('0788aba2-d4e5-4028-9ef1-4867ad5385e0') # noqa: E501 # doctest: +SKIP
('0788aba2-d4e5-4028-9ef1-4867ad5385e0', 'COMPLETE')
'''
return self.get(f'/calculation/{calculation_id}')
[docs] def cancel_calculation(self, calculation_id):
'''Cancels a previously requested calculation, should this not be completed, or its results..
``GET: /calculation/{id}/cancel``
Calculations that have been already ``DISPATCHED``, previously ``CANCELLED`` or
that have ``EXPIRED`` cannot be cancelled, and requesting it will produce an `error`.
Parameters
----------
calculation_id: str
Calculation id.
Returns
-------
(str, str)
Tuple of the calculation phase:
``(calculation-id, phase)``
See: :py:func:`read`.
Example
-------
>>> API.phase_calculation('0788aba2-d4e5-4028-9ef1-4867ad5385e0') # noqa: E501 # doctest: +SKIP
('0788aba2-d4e5-4028-9ef1-4867ad5385e0', 'CANCELLED')
'''
return self.get(f'/calculation/{calculation_id}/cancel')
[docs] def results_calculation(self, calculation_id):
'''Gets the results of a complete calculation.
``GET: /calculation/{id}/results``
Parameters
----------
calculation_id: str
Calculation id.
Returns
-------
([:obj:`webgeocalc.types.ColumnResult`], [[int, float or str], ...])
Tuple of calculation results:
``(columns, rows)``
See: :py:func:`read`.
Example
-------
>>> API.results_calculation('0788aba2-d4e5-4028-9ef1-4867ad5385e0') # noqa: E501 # doctest: +SKIP
([<ColumnResult> UTC calendar date, ...], [['2000-01-01 00:00:00.000000 UTC', ...], [...]])
'''
return self.get(f'/calculation/{calculation_id}/results')
@property
def metadata(self):
"""API metadata."""
if self._meta is None:
self._meta = self.get('/')
return self._meta
# Export default API object
API = Api()
JPL_API = Api(JPL_URL)
ESA_API = Api(ESA_URL)