1
# Licensed under a 3-clause BSD style license - see LICENSE.rst
2 1
"""
3
Sub-query to obtain test parameters from service providers.
4

5
In case USVO service is unstable, it does the following:
6

7
    #. Try USVO production server.
8
    #. If SR > 0.1, force SR to be 0.1.
9
    #. If fails, use RA=0 DEC=0 SR=0.1.
10

11
"""
12
# STDLIB
13 1
import warnings
14 1
from collections import OrderedDict
15

16
# ASTROPY
17 1
from astropy.table import Table
18 1
from astropy.utils.data import get_readable_fileobj
19 1
from astropy.utils.exceptions import AstropyUserWarning
20

21
# LOCAL
22 1
from astroquery.utils.commons import ASTROPY_LT_4_1
23

24 1
__all__ = ['parse_cs']
25

26

27 1
def parse_cs(ivoid, cap_index=1):
28
    """Return test query pars as dict for given IVO ID and capability index."""
29 0
    if isinstance(ivoid, bytes):  # ASTROPY_LT_4_1
30 0
        ivoid = ivoid.decode('ascii')
31

32
    # Production server.
33 0
    url = ("http://vao.stsci.edu/regtap/tapservice.aspx/sync?lang=adql&"
34
           "query=select%20detail_xpath%2Cdetail_value%20from%20"
35
           "rr.res_detail%20where%20"
36
           "ivoid%3D%27{0}%27%20and%20cap_index={1}%20and%20"
37
           "detail_xpath%20in%20%28%27/capability/testQuery/ra%27%2C"
38
           "%27/capability/testQuery/dec%27%2C%27/capability/testQuery/sr%27"
39
           "%29".format(ivoid, cap_index))
40

41 0
    urls_failed = False
42 0
    default_sr = 0.1
43

44 0
    try:
45 0
        with get_readable_fileobj(url, encoding='binary',
46
                                  show_progress=False) as fd:
47 0
            t_query = Table.read(fd, format='votable')
48
    except Exception as e:  # pragma: no cover
49
        urls_failed = True
50
        urls_errmsg = '{0} raised {1}, using default'.format(
51
            url, str(e))
52

53 0
    if not urls_failed:
54 0
        try:
55 0
            xpath = t_query['detail_xpath']
56 0
            if ASTROPY_LT_4_1:
57 0
                ra = float(
58
                    t_query[xpath == b'/capability/testQuery/ra']['detail_value'])
59 0
                dec = float(
60
                    t_query[xpath == b'/capability/testQuery/dec']['detail_value'])
61 0
                sr = float(
62
                    t_query[xpath == b'/capability/testQuery/sr']['detail_value'])
63
            else:
64 0
                ra = float(
65
                    t_query[xpath == '/capability/testQuery/ra']['detail_value'])
66 0
                dec = float(
67
                    t_query[xpath == '/capability/testQuery/dec']['detail_value'])
68 0
                sr = float(
69
                    t_query[xpath == '/capability/testQuery/sr']['detail_value'])
70

71
            # Handle big SR returning too big a table for some queries, causing
72
            # tests to fail due to timeout.
73 0
            if sr > default_sr:
74 0
                warnings.warn(
75
                    'SR={0} is too large, using SR={1} for {2},{3}'.format(
76
                        sr, default_sr, ivoid, cap_index), AstropyUserWarning)
77 0
                sr = default_sr
78

79 0
            d = OrderedDict({'RA': ra, 'DEC': dec, 'SR': sr})
80

81
        except Exception as e:  # pragma: no cover
82
            urls_failed = True
83
            urls_errmsg = ('Failed to retrieve test query parameters for '
84
                           '{0},{1}, using default: {2}'.format(ivoid, cap_index, str(e)))
85

86
    # If no test query found, use default
87
    if urls_failed:  # pragma: no cover
88
        d = OrderedDict({'RA': 0, 'DEC': 0, 'SR': default_sr})
89
        warnings.warn(urls_errmsg, AstropyUserWarning)
90

91 0
    return d

Read our documentation on viewing source code .

Loading