Commit 63439fe8 authored by Derek Homeier's avatar Derek Homeier
Browse files

Implementation of filter attributes and search function for KISClient

parent f6cabb71
"""
Search attributes which are specific to the `sdc.net.KISClient`.
Other attributes provided by `sunpy.net.attrs` are supported by the client.
"""
import collections.abc
import astropy.units as u
from astropy.time import Time
from sunpy.net import _attrs
from sunpy.net import attr as _attr
from sunpy.coordinates.frames import Helioprojective
from sunpy.coordinates.utils import get_rectangle_coordinates
from sunpy.time import parse_time
from sunpy.time.time import _variables_for_parse_time_docstring
__all__ = ['DataProduct', 'ObsName', 'Date', 'Filter', 'PolStates', 'Telescope', 'Target',
'ExposureTime', 'HelioProjLon', 'HelioProjLat', 'AtmosR0', 'Theta', 'Mu',
'SpectralResolution', 'SpatialResolution', 'TemporalResolution',
'NDimensions', 'PolXel', 'SpatialXel1', 'SpatialXel2', 'SpectralXel', 'TimeXel']
# SimpleAttrs
# description.DATAPRODUCT_TYPE
class DataProduct(_attr.SimpleAttr):
"""
Type of data product ('image', 'cube', 'timeseries' etc.) 'DATA_PRODUCT_TYPE'.
Parameters
----------
dataproduct_type : `str`
Description of the type of data product.
"""
type_name = "dataproduct_type"
def __init__(self, dataproduct_type: str):
super().__init__(dataproduct_type)
# OBS_NAME
class ObsName(_attr.SimpleAttr):
"""
Unique observation identifier for a dataset 'OBS_NAME'.
Parameters
----------
obsname : `str`
A unique identifier like 'instrument_YYMMDD_HHMMSS[FFF]_[further_fields]'.
"""
type_name = "obs_name"
def __init__(self, obsname: str):
super().__init__(obsname)
class Date(_attr.SimpleAttr):
"""
Specifies a date within the time range of the observation.
Parameters
----------
time : {parse_time_types}
The start datetime in a format parseable by `~sunpy.time.parse_time` or
an `~astropy.time.Time` object.
"""
type_name = "date_"
def __init__(self, time):
if isinstance(time, Time):
self.time = time
else:
self.time = parse_time(time)
super().__init__(self.time)
def __hash__(self):
if not isinstance(self.time, collections.abc.Hashable):
# The hash is the hash of the time
return hash((self.time.jd1, self.time.jd2, self.time.scale))
else:
return super().__hash__()
def collides(self, other):
# Use exact type checking here, because otherwise it collides with all
# subclasses of itself which can have completely different search
# meanings.
return type(other) is type(self)
def __xor__(self, other):
if not isinstance(other, self.__class__):
raise TypeError
return super().__xor__(other)
def pad(self, timedelta):
return type(self)(self.time - timedelta, self.time + timedelta)
def __repr__(self):
time = self.time.iso
return f'<sunpy.net.attrs.sdc.Date({str(time)})>'
# description.FILTER
class Filter(_attr.SimpleAttr):
"""
Name of the filter used during the observation.
Parameters
----------
filter: `str`
A string containing the filter name.
"""
def __init__(self, filter: str):
super().__init__(filter)
class PolStates(_attr.SimpleAttr):
"""
A list of polarisation states (e.g. 'IQUV') 'POL_STATES'.
Parameters
----------
polstates : int
The string listing the polarisation states.
"""
type_name = "pol_states"
def __init__(self, polstates):
super().__init__(polstates)
def collides(self, other):
return isinstance(other, self.__class__)
# description.TELESCOPE
class Telescope(_attr.SimpleAttr):
"""
Specifies the name of the telescope for the search.
Parameters
----------
name : `str`
Notes
-----
The telescope for each instrument can be found under the description in
`attrs.Instrument`.
"""
def __init__(self, name):
if not isinstance(name, str):
raise ValueError("Telescope names must be strings")
super().__init__(name)
# description.TARGET
class Target(_attr.SimpleAttr):
"""
Solar feature name from the observation.
Parameters
----------
target: `str`
A string describing the targeted solar feature.
"""
def __init__(self, target: str):
super().__init__(target)
# Range Attrs
class AtmosR0(_attr.Range):
"""
The atmospheric coherence length during observation 'ATMOS_R0_[MIN|MAX]'.
Parameters
----------
atmosmin : `u.Quantity`
The minimum value of the coherence length to search between.
atmosmax : `u.Quantity`, optional
The maximum value of the coherence length to search between;
defaults to same value as the minimum.
"""
type_name = "atmos_r0_"
@u.quantity_input
def __init__(self, atmosmin: u.cm, atmosmax: u.cm = None):
if atmosmax is None:
atmosmax = atmosmin
super().__init__(atmosmin, atmosmax)
self.min = atmosmin.to_value(u.cm)
self.max = atmosmax.to_value(u.cm)
def collides(self, other):
return isinstance(other, self.__class__)
class ExposureTime(_attr.Range):
"""
Accumulated exposure time 'EXPTIME'.
Parameters
----------
expmin : `u.Quantity`
The minimum value of the exposure time to search between.
expmax : `u.Quantity`
The maximum value of the exposure time to search between.
"""
type_name = "exptime"
@u.quantity_input
def __init__(self, expmin: u.s, expmax: u.s):
super().__init__(expmin, expmax)
self.min = expmin.to_value(u.s)
self.max = expmax.to_value(u.s)
def collides(self, other):
return isinstance(other, self.__class__)
class HelioProjLon(_attr.Range):
"""
The helioprojective westward angle (longitude) 'HPLN_TAN_[MIN|MAX]'.
Parameters
----------
hpln_tan_min : `u.Quantity`
The minimum value of the helioprojective longitude to search in;
hpln_tan_max : `u.Quantity`, optional
The maximum value of the helioprojective longitude to search in;
defaults to same value as the minimum.
"""
type_name = "hpln_tan_"
def __init__(self, hpln_tan_min: u.arcsec, hpln_tan_max: u.arcsec = None):
if hpln_tan_max is None:
hpln_tan_max = hpln_tan_min
super().__init__(hpln_tan_min, hpln_tan_max)
self.min = hpln_tan_min.to_value(u.arcsec)
self.max = hpln_tan_max.to_value(u.arcsec)
def collides(self, other):
return isinstance(other, self.__class__)
class HelioProjLat(_attr.Range):
"""
The helioprojective northward angle (latitude) 'HPLT_TAN_[MIN|MAX]'.
Parameters
----------
hplt_tan_min : `u.Quantity`
The minimum value of the helioprojective latitude to search in;
hplt_tan_max : `u.Quantity`, optional
The maximum value of the helioprojective latitude to search in;
defaults to same value as the minimum.
"""
type_name = "hplt_tan_"
def __init__(self, hplt_tan_min: u.arcsec, hplt_tan_max: u.arcsec = None):
if hplt_tan_max is None:
hplt_tan_max = hplt_tan_min
super().__init__(hplt_tan_min, hplt_tan_max)
self.min = hplt_tan_min.to_value(u.arcsec)
self.max = hplt_tan_max.to_value(u.arcsec)
def collides(self, other):
return isinstance(other, self.__class__)
class SpectralResolution(_attr.Range):
"""
The spectral resolving power 'EM_RES_POWER', equals to lambda/delta lambda.
Parameters
----------
spectralmin : int
The minimum value of the spectral resolution to search between.
spectralmax : int, optional
The maximum value of the spectral resolution to search between;
defaults to same value as the minimum.
"""
type_name = "em_res_power"
def __init__(self, spectralmin, spectralmax=None):
if spectralmax is None:
spectralmax = spectralmin
super().__init__(spectralmin, spectralmax)
def collides(self, other):
return isinstance(other, self.__class__)
class SpatialResolution(_attr.Range):
"""
The spatial (angular) resolution of an observation 'S_RESOLUTION'.
Parameters
----------
spatialmin : `u.Quantity`
The minimum value of the spatial resolution to search between.
spatialmax : `u.Quantity`
The maximum value of the spatial resolution to search between.
"""
type_name = "s_resolution"
def __init__(self, spatialmin: u.arcsec, spatialmax: u.arcsec):
super().__init__(spatialmin, spatialmax)
self.min = spatialmin.to_value(u.arcsec)
self.max = spatialmax.to_value(u.arcsec)
def collides(self, other):
return isinstance(other, self.__class__)
class TemporalResolution(_attr.Range):
"""
The temporal resolution (minimal interval between two points along the time axis)
for an observation 'T_RESOLUTION'.
Parameters
----------
temporalmin : `u.Quantity`
The minimum value of the time resolution to search between.
temporalmax : `u.Quantity`
The maximum value of the time resolution to search between.
"""
type_name = "t_resolution"
def __init__(self, temporalmin: u.s, temporalmax: u.s):
super().__init__(temporalmin, temporalmax)
self.min = temporalmin.to_value(u.s)
self.max = temporalmax.to_value(u.s)
def collides(self, other):
return isinstance(other, self.__class__)
class Theta(_attr.Range):
"""
The angle between surface normal and the line of sight to the observer
for a given point.
Parameters
----------
thetamin : `u.Quantity`
The minimum value of the surface normal angle to search between.
thetamax : `u.Quantity`
The maximum value of the surface normal angle to search between.
"""
def __init__(self, thetamin: u.deg, thetamax: u.deg):
super().__init__(thetamin, thetamax)
self.min = thetamin.to_value(u.deg)
self.max = thetamax.to_value(u.deg)
def collides(self, other):
return isinstance(other, self.__class__)
class Mu(_attr.Range):
"""
The cosine of the angle between surface normal and the line of sight to
the observer for a given point.
Parameters
----------
mumin : float
The minimum value of the surface normal angle to search between.
mumax : float
The maximum value of the surface normal angle to search between.
"""
def __init__(self, mumin, mumax):
super().__init__(mumin, mumax)
def collides(self, other):
return isinstance(other, self.__class__)
class NDimensions(_attr.Range):
"""
The number of dimensions in the data cube 'N_DIMENSIONS'.
Parameters
----------
ndimmin : int
The minimum value of the number of dimensions to search between.
ndimmax : int, optional
The maximum value of the number of dimensions to search between;
defaults to same value as the minimum.
"""
type_name = "pol_xel"
def __init__(self, ndimmin, ndimmax=None):
if ndimmax is None:
ndimmax = ndimmin
super().__init__(ndimmin, ndimmax)
def collides(self, other):
return isinstance(other, self.__class__)
class PolXel(_attr.Range):
"""
The number of polarisation states present in the dataset 'POL_XEL'.
Parameters
----------
polxelmin : int
The minimum value of the number of polarisation states to search between.
polxelmax : int, optional
The maximum value of the number of polarisation states to search between;
defaults to same value as the minimum.
"""
type_name = "pol_xel"
def __init__(self, polxelmin, polxelmax=None):
if polxelmax is None:
polxelmax = polxelmin
super().__init__(polxelmin, polxelmax)
def collides(self, other):
return isinstance(other, self.__class__)
class SpatialXel1(_attr.Range):
"""
The number of elements along spatial axis 1 'S_XEL1'.
Parameters
----------
spatialxelmin : int
The minimum value of the number of spatial elements to search between.
spatialxelmax : int, optional
The maximum value of the number of spatial elements to search between;
defaults to same value as the minimum.
"""
type_name = "s_xel1"
def __init__(self, spatialxelmin, spatialxelmax=None):
if spatialxelmax is None:
spatialxelmax = spatialxelmin
super().__init__(spatialxelmin, spatialxelmax)
def collides(self, other):
return isinstance(other, self.__class__)
class SpatialXel2(_attr.Range):
"""
The number of elements along spatial axis 2 'S_XEL2'.
Parameters
----------
spatialxelmin : int
The minimum value of the number of spatial elements to search between.
spatialxelmax : int, optional
The maximum value of the number of spatial elements to search between;
defaults to same value as the minimum.
"""
type_name = "s_xel1"
def __init__(self, spatialxelmin, spatialxelmax=None):
if spatialxelmax is None:
spatialxelmax = spatialxelmin
super().__init__(spatialxelmin, spatialxelmax)
def collides(self, other):
return isinstance(other, self.__class__)
class SpectralXel(_attr.Range):
"""
The number of elements along the spectral axis 'EM_XEL'.
Parameters
----------
spectralxelmin : int
The minimum value of the number of spectral elements to search between.
spectralxelmax : int, optional
The maximum value of the number of spectral elements to search between;
defaults to same value as the minimum.
"""
type_name = "em_xel"
def __init__(self, spectralxelmin, spectralxelmax=None):
if spectralxelmax is None:
spectralxelmax = spectralxelmin
super().__init__(spectralxelmin, spectralxelmax)
def collides(self, other):
return isinstance(other, self.__class__)
class TimeXel(_attr.Range):
"""
The number of temporal elements spanning the time axis 'T_XEL'.
Parameters
----------
timexelmin : int
The minimum value of the number of time elements to search between.
timexelmax : int, optional
The maximum value of the number of time elements to search between;
defaults to same value as the minimum.
"""
type_name = "t_xel"
def __init__(self, timexelmin, timexelmax=None):
if timexelmax is None:
timexelmax = timexelmin
super().__init__(timexelmin, timexelmax)
def collides(self, other):
return isinstance(other, self.__class__)
# -*- coding: utf-8 -*-
import warnings
import astropy.units as u
from astropy.time import Time
import sunpy.net.attrs as a
from sunpy.net.attr import AttrAnd, AttrOr, AttrWalker, DataAttr
from sunpy.net.attr import AttrAnd, AttrOr, AttrMeta, AttrWalker, DataAttr
from sunpy.net.base_client import BaseClient, QueryResponseTable
from sunpy.util.exceptions import SunpyUserWarning
# from sunpy.net.attrs import Instrument, Level, Physobs, Provider, Time, Wavelength
import attrs as sattrs
import urllib.parse
import urllib.request
from urllib.error import HTTPError, URLError
walker = AttrWalker()
@walker.add_creator(AttrOr)
def create_or(wlk, tree):
""""""
results = []
def create_from_or(wlk, tree):
"""
For each subtree under the OR, create a new set of query params.
"""
params = []
for sub in tree.attrs:
results.append(wlk.create(sub))
sub_params = wlk.create(sub)
# Strip out one layer of nesting of lists
# This means that create always returns a list of dicts.
if isinstance(sub_params, list) and len(sub_params) == 1:
sub_params = sub_params[0]
params.append(sub_params)
return results
return params
@walker.add_creator(AttrAnd, DataAttr)
def create_and(wlk, tree):
""""""
result = dict()
wlk.apply(tree, result)
return [result]
def create_new_param(wlk, tree):
params = dict()
# Use the apply dispatcher to convert the attrs to their query parameters
wlk.apply(tree, params)
return [params]
@walker.add_applier(AttrAnd)
def iterate_over_and(wlk, tree, params):
for sub in tree.attrs:
wlk.apply(sub, params)
# Map type_name from SunPy Attrs to observation description fields:
# CALIB_LEVEL - Degree of data processing
# BTYPE - IVOA Unified Content Desciptor for data ('phot.flux')
# T_RESOLUTIION - (Minimal) sampling interval along the time axis [s]
#
# The default field/key name is given by Attr.type_name.upper();
# keys ending in '_' have the special significance of setting both
# `KEY_MIN` and `KEY_MAX` description fields.
_obs_fields = {'level': 'CALIB_LEVEL', 'physobs': 'BTYPE', 'sample': 'T_RESOLUTION',
'time': 'DATE', 'wave': 'WAVELENGTH_'}
def _str_val(value, regex=False):
"""
Format `value` for query string.
"""
if isinstance(value, (int, float, complex)):
return f"{value:g}"
elif isinstance(value, Time):
return f"'{value.isot}'"
elif regex:
return f"{{'$regex':'{str.lower(value)}}}'"
else:
return f"'{str.lower(value)}'"
def _update_val(dictionary, key, value, block='description', regex=False):
"""
Update dictionary with field_name:value string in format parseable by BSON filter.
"""
strval = _str_val(value, regex)
if key in ('POL_STATES'):
strval = strval.upper()
# Special case for fields having a _MIN and _MAX form - require MIN <= value <= MAX:
if key.endswith('_'):