Source code for sunpy.net.hek.hek

"""
Facilities to interface with the Heliophysics Events Knowledgebase.
"""
import json
import codecs
import urllib
import inspect
from itertools import chain

import astropy.table
from astropy.table import Row

import sunpy.net._attrs as core_attrs
from sunpy import log
from sunpy.net import attr
from sunpy.net.base_client import BaseClient, QueryResponseTable
from sunpy.net.hek import attrs
from sunpy.time import parse_time
from sunpy.util import dict_keys_same, unique
from sunpy.util.xml import xml_to_dict

__all__ = ['HEKClient', 'HEKTable', 'HEKRow']

DEFAULT_URL = 'https://www.lmsal.com/hek/her?'


def _freeze(obj):
    """ Create hashable representation of result dict. """
    if isinstance(obj, dict):
        return tuple((k, _freeze(v)) for k, v in obj.items())
    if isinstance(obj, list):
        return tuple(_freeze(elem) for elem in obj)
    return obj


[docs] class HEKClient(BaseClient): """ Provides access to the Heliophysics Event Knowledgebase (HEK). The HEK stores solar feature and event data generated by algorithms and human observers. """ # FIXME: Expose fields in .attrs with the right types # that is, not all StringParamWrapper! default = { 'cosec': '2', # Return .json 'cmd': 'search', 'type': 'column', 'event_type': '**', } # Default to full disk. attrs.walker.apply(attrs.SpatialRegion(), {}, default) def __init__(self, url=DEFAULT_URL): self.url = url def _download(self, data): """ Download all data, even if paginated. """ page = 1 results = [] new_data = data.copy() # Override the default name of the operatorX, where X is a number. for key in data.keys(): if "operator" in key: new_data[f"op{key.split('operator')[-1]}"] = new_data.pop(key) while True: new_data['page'] = page url = self.url + urllib.parse.urlencode(new_data) log.debug(f'Opening {url}') fd = urllib.request.urlopen(url) try: result = codecs.decode(fd.read(), encoding='utf-8', errors='replace') result = json.loads(result) except Exception as e: raise OSError("Failed to load return from the HEKClient.") from e finally: fd.close() results.extend(result['result']) if not result['overmax']: if len(results) > 0: table = astropy.table.Table(dict_keys_same(results)) table = self._parse_times(table) return table else: return astropy.table.Table() page += 1 @staticmethod def _parse_times(table): # All time columns from https://www.lmsal.com/hek/VOEvent_Spec.html time_keys = ['event_endtime', 'event_starttime', 'event_peaktime'] for tkey in time_keys: if tkey in table.colnames and not any(time == "" for time in table[tkey]): table[tkey] = parse_time(table[tkey]) table[tkey].format = 'iso' return table
[docs] def search(self, *args, **kwargs): """ Retrieves information about HEK records matching the criteria given in the query expression. If multiple arguments are passed, they are connected with AND. The result of a query is a list of unique HEK Response objects that fulfill the criteria. Examples ------- >>> from sunpy.net import attrs as a, Fido >>> timerange = a.Time('2011/08/09 07:23:56', '2011/08/09 12:40:29') >>> res = Fido.search(timerange, a.hek.FL, a.hek.FRM.Name == "SWPC") # doctest: +REMOTE_DATA >>> res # doctest: +SKIP <sunpy.net.fido_factory.UnifiedResponse object at ...> Results from 1 Provider: <BLANKLINE> 2 Results from the HEKClient: SOL_standard active ... skel_startc2 sum_overlap_scores ------------------------------ ------ ... ------------ ------------------ SOL2011-08-09T07:19:00L227C090 true ... None 0 SOL2011-08-09T07:48:00L296C073 true ... None 0 <BLANKLINE> <BLANKLINE> """ query = attr.and_(*args) data = attrs.walker.create(query, {}) ndata = [] for elem in data: new = self.default.copy() new.update(elem) ndata.append(new) if len(ndata) == 1: return HEKTable(self._download(ndata[0]), client=self) else: return HEKTable(self._merge(self._download(data) for data in ndata), client=self)
def _merge(self, responses): """ Merge responses, removing duplicates. """ return list(unique(chain.from_iterable(responses), _freeze))
[docs] def fetch(self, *args, **kwargs): """ This is a no operation function as this client does not download data. """ return NotImplemented
@classmethod def _attrs_module(cls): return 'hek', 'sunpy.net.hek.attrs' @classmethod def _can_handle_query(cls, *query): required = {core_attrs.Time} optional = {i[1] for i in inspect.getmembers(attrs, inspect.isclass)} - required qr = tuple(x for x in query if not isinstance(x, attrs.EventType)) return cls.check_attr_types_in_query(qr, required, optional)
[docs] class HEKRow(Row): """ Handles the response from the HEK. Each HEKRow object is a subclass of `~astropy.table.Row`. The column-row key-value pairs correspond to the HEK feature/event properties and their values, for that record from the HEK. Each HEKRow object also has extra properties that relate HEK concepts to VSO concepts. """ @property def vso_time(self): return core_attrs.Time( self['event_starttime'], self['event_endtime'] ) @property def vso_instrument(self): if self['obs_instrument'] == 'HEK': raise ValueError("No instrument contained.") return core_attrs.Instrument(self['obs_instrument']) @property def vso_all(self): return attr.and_(self.vso_time, self.vso_instrument)
[docs] def get_voevent(self, as_dict=True, base_url="http://www.lmsal.com/hek/her?"): """Retrieves the VOEvent object associated with a given event and returns it as either a Python dictionary or an XML string.""" # Build URL params = { "cmd": "export-voevent", "cosec": 1, "ivorn": self['kb_archivid'] } url = base_url + urllib.parse.urlencode(params) # Query and read response response = urllib.request.urlopen(url).read() # Return a string or dict if as_dict: return xml_to_dict(response) else: return response
[docs] def get(self, key, default=None): try: return self[key] except KeyError: return default
[docs] class HEKTable(QueryResponseTable): """ A container for data returned from HEK searches. """ Row = HEKRow