Source code for

Access the Helio Event Catalogue
from __future__ import print_function, absolute_import

from import WellBehavedHttpTransport
from import parser
from sunpy.time import parse_time
from suds.client import Client as C
import suds
from import parse_single_table

from sunpy.extern import six
from sunpy.extern.six.moves import range, input

__author__ = 'Michael Malocha'

__all__ = ['HECClient']

def suds_unwrapper(wrapped_data):
    Removes suds wrapping from returned xml data

    When grabbing data via votable_interceptor.last_payload from the
    suds.client.Client module, it returns the xml data in an un-helpful
    "<s:Envelope>" that needs to be removed. This function politely cleans
    it up.

    wrapped_data : `str`
        Contains the wrapped xml results from a WSDL query

    unwrapped : `str`
        The xml results with the wrapper removed
    if six.PY3 and not isinstance(wrapped_data, str):
        wrapped_data = wrapped_data.decode("utf-8")
    HEADER = '<?xml version="1.0" encoding="UTF-8"?>\n'
    CATCH_1 = '<VOTABLE'
    CATCH_2 = '</VOTABLE>\n'
    # Now going to find the locations of the CATCHes in the wrapped_data
    pos_1 = wrapped_data.find(CATCH_1)
    pos_2 = wrapped_data.find(CATCH_2)
    unwrapped = HEADER + wrapped_data[pos_1:pos_2] + CATCH_2
    return unwrapped

def votable_handler(xml_table):
    Returns a VOtable object from a VOtable style xml string

    In order to get a VOtable object, it has to be parsed from an xml file or
    file-like object. This function creates a file-like object via the
    StringIO module, writes the xml data to it, then passes the file-like
    object to parse_single_table() from the module
    and thereby creates a VOtable object.

    xml_table : str
        Contains the VOtable style xml data

    votable : ``
        A properly formatted VOtable object

    >>> from import hec
    >>> temp = hec.suds_unwrapper(xml_string)  # doctest: +SKIP
    >>> type(temp)  # doctest: +SKIP
    >>> temp = hec.votable_handler(temp)  # doctest: +SKIP
    >>> type(temp)  # doctest: +SKIP
    fake_file = six.BytesIO()
    votable = parse_single_table(fake_file)
    return votable

class VotableInterceptor(suds.plugin.MessagePlugin):
    Adapted example from
    def __init__(self, *args, **kwargs):
        self.last_payload = None

    def received(self, context):
        # received xml as a string
        self.last_payload = six.u(suds_unwrapper(context.reply))
        # clean up reply to prevent parsing
        context.reply = ""
        return context

[docs]class HECClient(object): """ A client class used to interface with and query HELIO webservices. """ def __init__(self, link=None): """ The constructor; establishes the webservice link for the client Initializes the client with a weblink Parameters ---------- link : str Contains URL to valid WSDL endpoint Examples -------- >>> from import hec >>> hc = hec.HECClient() # doctest: +REMOTE_DATA """ if link is None: # The default wsdl file link = parser.wsdl_retriever() self.votable_interceptor = VotableInterceptor() self.hec_client = C(link, plugins=[self.votable_interceptor], transport=WellBehavedHttpTransport())
[docs] def time_query(self, start_time, end_time, table=None, max_records=None): """ The simple interface to query the wsdl service. Used to utilize the service's TimeQuery() method, this is a simple interface between the sunpy module library and the web-service's API. Parameters ---------- start_time : str The datetime where the query window opens end_time : str The datetime where the query window closes table : str The table to query from. If the table is unknown, the user will be prompted to pick from a list of tables. max_records: int The maximum number of desired records. Returns ------- results: `` Table containing the results from the query Examples -------- >>> from import hec >>> hc = hec.HECClient() # doctest: +REMOTE_DATA >>> start = '2005/01/03' >>> end = '2005/12/03' >>> temp = hc.time_query(start, end, max_records=10) # doctest: +REMOTE_DATA """ while table is None: table = self.make_table_list() start_time = parse_time(start_time) end_time = parse_time(end_time) self.hec_client.service.TimeQuery(STARTTIME=start_time.isoformat(), ENDTIME=end_time.isoformat(), FROM=table, MAXRECORDS=max_records) results = votable_handler(self.votable_interceptor.last_payload) return results
[docs] def get_table_names(self): """ Returns a list of the available tables to query. Returns the names of all the tables that can be queried via the webservice. Returns ------- tables.array: `` A VOtable table of available tables names Examples -------- >>> from import hec >>> hc = hec.HECClient() # doctest: +REMOTE_DATA >>> print(hc.get_table_names()) # doctest: +REMOTE_DATA [('timed_see_flare',) ('hi_event',) ('yohkoh_flare_list',) ('wind_mfi_bs_crossing_time',) ('seeds_soho',) ('seeds_stb',) ... ('rhessi_hxr_flare',) ('cactus_soho_flow',) ('cactus_soho_cme',) ('stereob_het_sep',)] """ self.hec_client.service.getTableNames() tables = votable_handler(self.votable_interceptor.last_payload) return tables.array
[docs] def make_table_list(self): """ Creates a list of table names and prompts the user for a choice This takes the table of table names from get_table_names(), creates a list of the names, sorts them, then presents the tables in a convenient menu for the user to choose from. It returns a string containing the name of the table that the user picked. Returns ------- temp: `str` contains the name of the table that the user picked. Examples -------- >>> from import hec >>> hc = hec.HECClient() # doctest: +REMOTE_DATA >>> hc.make_table_list() # doctest: +REMOTE_DATA """ table_list = [] tables = self.get_table_names() for i in tables: table = i[0] if len(table) > 0: table_list.append(table) table_list.sort() for index, table in enumerate(table_list): print(('{number:3d}) {table}'.format(number=index + 1, table=table))) while True: stdinput = input("\nPlease enter a table number between 1 and " "{elem:d} " "('e' to exit): ".format(elem=len(table_list))) if stdinput.lower() == "e" or stdinput.lower() == "exit": temp = None break temp = [int(s) for s in stdinput.split() if s.isdigit()] temp = temp[0] - 1 if temp in range(0, len(table_list)): temp = table_list[temp] break else: print("Choice outside of bounds") return temp