Source code for astroquery.dace.core

# Licensed under a 3-clause BSD style license - see LICENSE.rst

from collections import defaultdict
from json import JSONDecodeError
from astropy.table import Table
from ..query import BaseQuery
from ..utils import async_to_sync
from . import conf

__all__ = ['Dace', 'DaceClass']


class ParseException(Exception):
    """Raised when parsing Dace data fails"""



[docs]
@async_to_sync
class DaceClass(BaseQuery):
    """
    DACE class to access DACE (Data Analysis Center for Exoplanets) data based in Geneva Observatory
    """

    __DACE_URL = conf.server
    __DACE_TIMEOUT = conf.timeout
    __OBSERVATION_ENDPOINT = 'ObsAPI/observation/'
    __RADIAL_VELOCITIES_ENDPOINT = __OBSERVATION_ENDPOINT + 'radialVelocities/'
    __HEADERS = {'Accept': 'application/json'}


[docs]
    def query_radial_velocities_async(self, object_name):
        """
        Get radial velocities data for an object_name. For example : HD40307

        Parameters
        ----------
        object_name : str
            The target you want radial velocities data

        Returns
        -------
        response : a ``requests.Response`` from DACE
        """
        return self._request("GET", ''.join([self.__DACE_URL, self.__RADIAL_VELOCITIES_ENDPOINT, object_name]),
                             timeout=self.__DACE_TIMEOUT, headers=self.__HEADERS)


    def _parse_result(self, response, *, verbose=False):
        try:
            json_data = response.json()
            dace_dict = self.transform_data_as_dict(json_data)
            return Table(dace_dict)
        except JSONDecodeError as error:
            raise ParseException("Failed to parse DACE result. Request response : {}".format(response.text)) from error


[docs]
    @staticmethod
    def transform_data_as_dict(json_data):
        """
        Internally DACE data are provided using protobuff. The format is a list of parameters. Here we parse
        these data to give to the user something more readable and ignore the internal stuff
        """
        data = defaultdict(list)
        parameters = json_data.get('parameters')
        for parameter in parameters:
            variable_name = parameter.get('variableName')
            double_values = parameter.get('doubleValues')
            int_values = parameter.get('intValues')
            string_values = parameter.get('stringValues')
            bool_values = parameter.get('boolValues')
            # Only one type of values can be present. So we look for the next occurence not None
            values = next(values_list for values_list in [double_values, int_values, string_values, bool_values] if
                          values_list is not None)
            data[variable_name].extend(values)

            error_values = parameter.get('minErrorValues')  # min or max is symmetric
            if error_values is not None:
                data[variable_name + '_err'].extend(error_values)
        return data




Dace = DaceClass()
Navigation

Source code for astroquery.dace.core
