source: SH_SHM/trunk/util/quakeml/parse.py @ 318

#! /usr/bin/python
# -*- coding: UTF8 -*-

import sys
try:
    from lxml import etree
except ImportError:
    from xml.etree import ElementTree

from decimal import Decimal

class Event(object):
    """
    Seismic event.
    """

    def __str__(self):
        return str(self.__dict__)

class QuakeML(object):
    """
    Read access for QuakeML file-like objects.

    >>> q = QuakeML(open("example.xml"), True)
    >>> q.rawdata[40:92]
    'quakeml xmlns="http://quakeml.org/xmlns/quakeml/1.0"'
    >>> len(q.events)
    9
    >>> q.events[0].origin
    '2010-07-07T04:03:57.7+00:00'
    >>> q.events[3].description
    'TAJIKISTAN'
    >>> q.events[1].agencyOrigin
    'smi:smi-registry/organization/NNC'
    >>> q.events[2].agencyMagnitude
    'smi:smi-registry/organization/GFZ'
    """

    namespace = {"qml": "http://quakeml.org/xmlns/quakeml/1.0"}

    def __init__(self, stream, debug=False):
        """
        Read stream content (raw).
        """

        # There are two debug levels:
        # "1": normal debugging
        # "2": additionally write XML input data to "debug-dump.xml"
        self.debugLevel = debug

        try:
            self.rawdata = stream.read()
            self.debug("Data input length %u bytes." % len(self.rawdata))
        except:
            quit("input not readable.")

        if self.debugLevel > 1:
            try:
                open("debug-dump.xml", "w").write(self.rawdata)
                self.debug("Raw XML data written to 'debug-dump.xml'")
            except Exception, e:
                self.debug("Raw XML data dump requested, BUT an error "
                           "occurred. %s" % e)

        self.events = {}
        self._initXML()
        self._parseEvents()

    def debug(self, text, nontext=None, spaces=0):
        """
        Simple debug function.
        """

        if not self.debugLevel:
            return

        print >> sys.stderr, text

        if not nontext:
            return

        if type(nontext) == dict:
            for i in nontext:
                print >> sys.stderr, " "*spaces + i, nontext[i]

    def _initXML(self):
        """
        Inititialize XML processor(s).
        
        Depending on available modules the method "xpathns" is set to the
        corresponding helper function:

        __lxml_xpathns - uses lxml.etree xpath method
        __xml_findallns - uses xml.etree.ElementTree findall method

        Both methods add the quakeml namespace to the path automatically.
        """

        if "etree" in globals().keys():
            # use lxml
            try:
                parser = etree.XMLParser(remove_blank_text=True)
                self.xml = etree.fromstring(self.rawdata, parser)
                self.debug("LXML parser initiated.")
                self.xpathns = self.__lxml_xpathns
            except Exception, e:
                quit("error while parsing XML: " + e)
        else:
            # use built-in xml.etree
            try:
                self.xml = ElementTree.XML(self.rawdata)
                self.debug("Standard XML parser initiated.")
                self.xpathns = self.__xml_findallns
            except Exception, e:
                quit("error while parsing XML: " + e)

    def __lxml_xpathns(self, xml, path, select=0):
        """
        Helper function to handle global quakeml namespace for lxml.
        """
        # add namespace prefix
        path = "/qml:".join(path.split("/"))
        
        if select == None:
            return xml.xpath(path, namespaces=self.namespace)
        else:
            return xml.xpath(path, namespaces=self.namespace)[select]

    def __xml_findallns(self, xml, path, select=0):
        """
        Helper function for namespace handling (standard xml module).
        """
        x = "/{%s}" % self.namespace.values()[0]
        path = x.join(path.split("/"))

        if select == None:
            return xml.findall(path)
        else:
            return xml.findall(path)[select]
    
    def _parseEvents(self):
        """
        Loop through events inside XML data schema.
        """

        for event in self.xpathns(self.xml, "./eventParameters/event", select=None):
            # mostly there is more than one origin
            self.debug("Processing event %s" % event.attrib["publicID"])
            origins = self.xpathns(event, "./origin", select=None)
            for o in origins:
                publicID = o.attrib["publicID"]
                self.debug(" - origin %s" % publicID)

                self.events[publicID] = e = Event()
                # there's only a global description of source region
                e.description = self.xpathns(event, "./description/text").text
                # save publicID of associated event
                e.eventID = event.attrib["publicID"]
                e.publicID = publicID

                # detailed information of single origins
                e.origin = self.xpathns(o, "./time/value").text
                e.longitude = Decimal(self.xpathns(o, "./longitude/value").text)
                e.latitude = Decimal(self.xpathns(o, "./latitude/value").text)
                e.depth = self.xpathns(o, "./depth/value").text
                e.mode = self.xpathns(o, "./evaluationMode").text
                e.agencyOrigin = self.xpathns(o, "./creationInfo/authorURI").text

            # add associated magnitude to origin
            magnitudes = self.xpathns(event, "./magnitude", select=None)
            for m in magnitudes:
                originID = self.xpathns(m, "./originID").text
                e = self.events[originID]
                e.magnitude = Decimal(self.xpathns(m, "./mag/value").text)
                e.magnitudeType = self.xpathns(m, "./type").text
                e.agencyMagnitude = self.xpathns(m, "./creationInfo/authorURI").text

        # sort events by origin (results in a list)
        self.events = self.events.values()
        self.events.sort(key=lambda x: x.origin)

        if not self.debugLevel:
            return

        for e in self.events:
            self.debug("Origin details:", vars(e), 2)

if __name__ == "__main__":
    import doctest
    errors, _ = doctest.testmod(exclude_empty=True)