Small fixes and cleanup to fdsnws importer

shakyground2/import_fdsnws_eq.py

@@ -17,10 +17,9 @@ Can run standalone:
 
 """
 
-import datetime
 import os
 import sys
-from typing import List, Optional, Union
+from typing import List, Union
 import urllib.request as ul
 from xml.etree import ElementTree as ET
 import yaml
@@ -60,17 +59,15 @@ def fetch_fm(root: ET.Element, ns: dict, preferredfmID: str) -> list:
             _perror("Oops: no <nodalPlanes> object seen")
             break
 
-        d: List[dict] = [{} for i in range(2)]
+        d: List[dict] = [dict(), dict()]
         for k in range(2):
             plane = np.find("ns:nodalPlane" + str(k + 1), ns)
-            #d[k] = dict()
-            if not plane:
+            if plane is None:
                 continue
             for child in plane:
                 found = child.find("ns:value", ns)
-                print(found)
-                #if not found:
-                #    continue
+                if found is None:
+                    continue
                 v = found.text
                 tag = child.tag
                 tag = tag[tag.index("}") + 1 :]
@@ -89,10 +86,10 @@ def fetch_magnitude(
             continue
 
     child = m.find("ns:mag", ns)
-    if not child:
+    if child is None:
         return None
     value = child.find("ns:value", ns)
-    if not value:
+    if value is None:
         return None
     v = value.text
     try:
@@ -154,14 +151,11 @@ def fetch_quakeml_ws(evid: str) -> str:
         )
     )
 
-    # print(url)
     req = ul.Request(url)
     u = ul.urlopen(req)
     buf = u.read().decode("utf8")
 
     print("Got", len(buf), "char(s).")
-    with open("tmp.xml", "w") as fid:
-        fid.write(str(buf))
     return buf
 
 
@@ -188,7 +182,6 @@ def fetch_quakeml(path: str) -> Union[dict, None]:
     ns = QUAKEML_NS
 
     event = root[0][0]
-    # .find("ns:event", ns)
     if not event:
         print("Couldn't get an event!")
         return None
@@ -202,14 +195,14 @@ def fetch_quakeml(path: str) -> Union[dict, None]:
         _perror("Oops, couldn't get event id from " + event.attrib["publicID"])
 
     preferredoriginIDElem = root[0][0].find("ns:preferredOriginID", ns)
-
-    if not preferredoriginIDElem:
+    if preferredoriginIDElem is None:
+        _perror("Oops, couldn't find the preferred origin ID")
         return None
     preferredoriginID = preferredoriginIDElem.text
 
     preferredmagID = None
     preferredMagnitudeIDElem = root[0][0].find("ns:preferredMagnitudeID", ns)
-    if preferredMagnitudeIDElem:
+    if preferredMagnitudeIDElem is not None:
         try:
             preferredmagID = preferredMagnitudeIDElem.text
         except AttributeError:
@@ -217,7 +210,7 @@ def fetch_quakeml(path: str) -> Union[dict, None]:
     preferredfmID = None
     try:
         preferredfmIDElem = root[0][0].find("ns:preferredFocalMechanismID", ns)
-        if preferredfmIDElem:
+        if preferredfmIDElem is not None:
             preferredfmID = preferredfmIDElem.text
     except AttributeError:
         pass
@@ -277,6 +270,8 @@ http://geofon.gfz-potsdam.de/eqinfo/list.php
         sys.exit(1)
 
     ev_info = fetch_quakeml(evid)
+    if ev_info is None:
+        print("Got no dictionary from QuakeML")
     outfile = evid + ".yaml"
     with open(outfile, "w") as fid:
         fid.write(yaml.safe_dump(ev_info, default_flow_style=False))

tests/test_quakexml.py

@@ -44,8 +44,9 @@ class QuakeMLReadTestCase(unittest.TestCase):
         """
         evid = "gfz2021ekhv"
         result = fetch_quakeml(evid)
-        print(result)
         self.assertTrue(isinstance(result, dict))
+        self.assertGreater(len(result), 3)
+        self.assertEqual(result["id"], evid)
 
     def test_read_no_eventid(self):
         """
@@ -54,7 +55,23 @@ class QuakeMLReadTestCase(unittest.TestCase):
         """
         infile = "/no/such/file"
         with self.assertRaises(IOError):
-            result = fetch_quakeml(infile)
+            fetch_quakeml(infile)
+
+    def test_read_good_eventid2(self):
+        """
+        Might fail if the GEOFON fdsnws-event webservice is unavailable.
+        Does the dictionary have the values expected for this event?
+        """
+        evid = "gfz2021htpl"
+        result = fetch_quakeml(evid)
+        self.assertTrue(isinstance(result, dict))
+        self.assertGreater(len(result), 3)
+        self.assertEqual(result["date"], "2021-04-21")
+        self.assertEqual(len(result["focalmechanism"]), 2)
+        self.assertEqual(result["id"], "gfz2021htpl")
+        self.assertLess(abs(float(result["magnitude"]) - 4.465), 0.01)
+        self.assertEqual(len(result["origin"]), 3)
+        self.assertTrue(result["time"].startswith("08:08:23"))
 
     def tearDown(self):
         try: