import astropy.units as u
from sunpy.coordinates.utils import get_rectangle_coordinates
from sunpy.net._attrs import Time, Wavelength
from sunpy.net.attr import AttrAnd, AttrComparison, AttrOr, AttrWalker, DataAttr, SimpleAttr
__all__ = ['Series', 'Protocol', 'Notify', 'Segment', 'PrimeKey', 'Cutout', "Keyword"]
# Define a custom __dir__ to restrict tab-completion to __all__
def __dir__():
return __all__
[docs]
class Series(SimpleAttr):
"""
The JSOC Series to Download.
This is the list of `Series <http://jsoc.stanford.edu/JsocSeries_DataProducts_map.html>`__.
"""
[docs]
class PrimeKey(DataAttr):
"""
Prime Keys
Parameters
----------
label : str
value : str
"""
def __init__(self, label, value):
super().__init__()
self.label = label
self.value = value
def __repr__(self):
return f"{object.__repr__(self)}" + "\n" + f"{self.label, self.value}"
[docs]
def collides(self, other):
return False
class KeywordComparison(AttrComparison):
"""
Allows comparison filtering of the JSOC Keywords with the ability to specify the comparison operator.
Parameters
----------
name : str
operator : str
value : Numeric
"""
[docs]
class Keyword(SimpleAttr):
"""
Allows comparison filtering of the JSOC Keywords.
Parameters
----------
value : str
"""
def __lt__(self, other):
return KeywordComparison(self.value, '<', other)
def __le__(self, other):
return KeywordComparison(self.value, '<=', other)
def __gt__(self, other):
return KeywordComparison(self.value, '>', other)
def __ge__(self, other):
return KeywordComparison(self.value, '>=', other)
def __eq__(self, other):
return KeywordComparison(self.value, '=', other)
def __ne__(self, other):
return KeywordComparison(self.value, '!=', other)
[docs]
def collides(self, other):
return isinstance(other, Keyword)
[docs]
class Segment(SimpleAttr):
"""
Segments choose which files to download when there are more than
one present for each record e.g. 'image'.
"""
[docs]
def collides(self, other):
return False
[docs]
class Protocol(SimpleAttr):
"""
The type of download to request one of
("FITS", "JPEG", "MPG", "MP4", or "as-is").
Only FITS is supported, the others will require extra keywords.
"""
[docs]
class Notify(SimpleAttr):
"""
An email address to get a notification to when JSOC has staged your request.
"""
def __init__(self, value):
super().__init__(value)
if value is None:
raise ValueError("Notify attribute must contain an email address")
if value.find('@') == -1:
raise ValueError("Notify attribute must contain an '@' symbol "
"to be a valid email address")
self.value = value
[docs]
class Cutout(DataAttr):
"""
Select a cutout region.
The JSOC allows for users to request cutouts. This process is performed server
side so as to allow users to download only the portions of the full-disk images
they are interested in. For a detailed explanation of the routine
used to perform these cutouts on the JSOC server, see
http://jsoc.stanford.edu/doxygen_html/group__im__patch.html.
Parameters
----------
bottom_left : `~astropy.coordinates.SkyCoord`
Coordinate for the bottom left corner of the cutout.
top_right : `~astropy.coordinates.SkyCoord`, optional
Coordinate for the top right corner of the cutout. If this is
not specified, both ``width`` and ``height`` must both be specified.
width : `~astropy.units.Quantity`, optional
Width of the cutout. If this parameter, along with ``height``, is
not specified, ``top_right`` must be specified.
height : `~astropy.units.Quantity`, optional
Height of the cutout. If this parameter, along with ``width``, is
not specified, ``top_right`` must be specified.
tracking : `bool`, optional
If True, the field of view follows the rotation of the Sun
register : `bool`, optional
If True, use sub-pixel registration when cropping to the target location.
nan_off_limb : `bool`, optional
If True, all off-limb pixels are set to NaN
See Also
--------
sunpy.coordinates.utils.get_rectangle_coordinates
"""
@u.quantity_input
def __init__(self, bottom_left, top_right=None, width: u.arcsec = None,
height: u.arcsec = None, tracking=False, register=False,
nan_off_limb=False):
super().__init__()
bl, tr = get_rectangle_coordinates(bottom_left, top_right=top_right, width=width, height=height)
self.value = {
't_ref': bl.obstime.isot,
# JSOC input is disable tracking so take the negative
't': int(not tracking),
'r': int(register),
'c': int(nan_off_limb),
'locunits': 'arcsec',
'boxunits': 'arcsec',
'x': ((bl.Tx + tr.Tx) / 2).to('arcsec').value,
'y': ((bl.Ty + tr.Ty) / 2).to('arcsec').value,
'width': (tr.Tx - bl.Tx).to('arcsec').value,
'height': (tr.Ty - bl.Ty).to('arcsec').value,
}
[docs]
def collides(self, other):
return isinstance(other, self.__class__)
walker = AttrWalker()
@walker.add_creator(AttrOr)
def _create1(wlk, query):
qblocks = []
for iattr in query.attrs:
qblocks.extend(wlk.create(iattr))
return qblocks
@walker.add_creator(AttrAnd, DataAttr)
def _create(wlk, query):
map_ = {}
wlk.apply(query, map_)
return [map_]
@walker.add_applier(AttrAnd)
def _apply(wlk, query, imap):
for iattr in query.attrs:
wlk.apply(iattr, imap)
@walker.add_applier(SimpleAttr)
def _apply1(wlk, query, imap):
imap[query.__class__.__name__.lower()] = query.value
@walker.add_applier(PrimeKey)
def _apply1(wlk, query, imap):
key = 'primekey'
if key in imap:
imap[key][query.label] = query.value
else:
imap[key] = {query.label: query.value}
@walker.add_applier(Keyword)
def _apply1(wlk, query, imap):
raise ValueError(f"Keyword '{query.value}' needs to have a comparison to a value.")
@walker.add_applier(KeywordComparison)
def _apply1(wlk, query, imap):
key = 'keyword'
if key in imap:
imap[key][query.name] = {"operator": query.operator, "value": query.value}
else:
imap[key] = {f"{query.name}": {"operator": query.operator, "value": query.value}}
@walker.add_applier(Segment)
def _apply1(wlk, query, imap):
key = 'segment'
if key in imap:
imap[key].append(query.value)
else:
imap[key] = [query.value]
@walker.add_applier(Cutout)
def _apply1(wlk, query, imap):
imap[query.__class__.__name__.lower()] = query.value
@walker.add_applier(Time)
def _apply1(wlk, query, imap):
imap['start_time'] = query.start
imap['end_time'] = query.end
@walker.add_applier(Wavelength)
def _apply1(wlk, query, imap):
if query.min != query.max:
raise ValueError(
"For JSOC queries Wavelength.min must equal Wavelength.max")
imap[query.__class__.__name__.lower()] = query.min
