pylot/tests/test_Metadata/test_Metadata.py

338 lines
19 KiB
Python

import os
import unittest
from obspy import UTCDateTime
from obspy.io.xseed import Parser
from obspy.io.xseed.utils import SEEDParserException
from pylot.core.util.dataprocessing import Metadata
from tests.utils import HidePrints
class TestMetadata(unittest.TestCase):
def setUp(self):
self.station_id = 'BW.WETR..HH'
self.time = UTCDateTime('2012-08-01')
metadata_folder = os.path.join('test_data', 'dless_multiple_files', 'metadata1')
self.m = Metadata(metadata_folder)
def test_get_coordinates_sucess(self):
expected = {'Z': {u'elevation': 607.0, u'longitude': 12.87571, u'local_depth': 0.0, u'azimuth': 0.0,
u'latitude': 49.14502, u'dip': -90.0},
'E': {u'azimuth': 90.0, u'dip': 0.0, u'elevation': 607.0, u'latitude': 49.14502,
u'local_depth': 0.0, u'longitude': 12.87571},
'N': {u'azimuth': 0.0, u'dip': 0.0, u'elevation': 607.0, u'latitude': 49.14502, u'local_depth': 0.0,
u'longitude': 12.87571}
}
result = {}
for channel in ('Z', 'N', 'E'):
with HidePrints():
coords = self.m.get_coordinates(self.station_id + channel, time=self.time)
result[channel] = coords
self.assertDictEqual(result[channel], expected[channel])
def test_get_coordinates_sucess_no_time(self):
expected = {'Z': {u'elevation': 607.0, u'longitude': 12.87571, u'local_depth': 0.0, u'azimuth': 0.0,
u'latitude': 49.14502, u'dip': -90.0},
'E': {u'azimuth': 90.0, u'dip': 0.0, u'elevation': 607.0, u'latitude': 49.14502,
u'local_depth': 0.0, u'longitude': 12.87571},
'N': {u'azimuth': 0.0, u'dip': 0.0, u'elevation': 607.0, u'latitude': 49.14502, u'local_depth': 0.0,
u'longitude': 12.87571}
}
result = {}
for channel in ('Z', 'N', 'E'):
with HidePrints():
coords = self.m.get_coordinates(self.station_id + channel)
result[channel] = coords
self.assertDictEqual(result[channel], expected[channel])
class TestMetadataAdding(unittest.TestCase):
"""Tests if adding files and directories to a metadata object works."""
def setUp(self):
self.station_id = 'BW.WETR..HH'
self.metadata_folders = (os.path.join('test_data', 'dless_multiple_files', 'metadata1'),
os.path.join('test_data', 'dless_multiple_files', 'metadata2'))
self.m = Metadata()
def test_add_inventory_folder(self):
"""Test if add_inventory adds the folder to the list of inventories"""
self.m.add_inventory(self.metadata_folders[0])
# adding an inventory folder should append it to the list of inventories
self.assertDictEqual({}, self.m.inventory_files)
self.assertDictEqual({}, self.m.seed_ids)
self.assertEqual([self.metadata_folders[0]], self.m.inventories)
def test_add_inventory_file(self):
"""Test if add_inventory_file adds the folder containing the file to the list of inventories and
if the files is added to inventory_files"""
fpath = os.path.join(self.metadata_folders[0], 'DATALESS.BW.WETR..HHZ')
self.m.add_inventory_file(fpath)
# adding an inventory file should append its folder to the list of inventories and the file to the
self.assertEqual([os.path.join(self.metadata_folders[0], 'DATALESS.BW.WETR..HHZ')],
self.m.inventory_files.keys()) # does the filename exist in inventory files?
self.assertEqual(['data', 'invtype'], self.m.inventory_files[os.path.join(self.metadata_folders[0],
'DATALESS.BW.WETR..HHZ')].keys()) # is the required information attacht to the filename?
self.assertDictEqual({}, self.m.seed_ids)
self.assertEqual([self.metadata_folders[0]], self.m.inventories)
def test_add_inventory_invalid_path(self):
"""Test if adding an inventory that is not an existing directory fails with an exception"""
with self.assertRaises(Exception):
self.m.add_inventory('InvalidDirName')
self.assertEqual([], self.m.inventories) # inventory list should still be empty
def test_add_inventory_file_invalid_path(self):
"""Test if adding a inventory file with an invalid path fails with an exception"""
with self.assertRaises(Exception):
self.m.add_inventory_file('/invalid/file/name')
self.assertEqual([], self.m.inventories) # inventory list should still be empty
class TestMetadataRemoval(unittest.TestCase):
"""Tests if removing files and directories to a metadata object works."""
def setUp(self):
self.station_id = 'BW.WETR..HH'
self.metadata_folders = (os.path.join('test_data', 'dless_multiple_files', 'metadata1'),
os.path.join('test_data', 'dless_multiple_files', 'metadata2'))
self.m = Metadata()
def test_remove_all_inventories(self):
"""Test if function remove_inventory cleans the Metadata object """
# add multiple inventories
for folder in self.metadata_folders:
self.m.add_inventory(folder)
self.m.remove_all_inventories()
self.isEmpty(self.m)
def test_remove_inventory(self):
"""Test if remove_inventory removes single inventories"""
# add multiple inventories
for folder in self.metadata_folders:
self.m.add_inventory(folder)
self.m.remove_inventory(self.metadata_folders[0])
self.assertNotIn(self.metadata_folders[0], self.m.inventories)
self.m.remove_inventory(self.metadata_folders[1])
self.assertNotIn(self.metadata_folders[1], self.m.inventories)
self.isEmpty(self.m)
def test_remove_inventory_not_in_inventory_list(self):
"""Test if remove_inventory does not modify the metadata instance if the given inventory to remove does not
exist in the instance."""
# add multiple inventories
self.m.add_inventory(self.metadata_folders[0])
with HidePrints():
self.m.remove_inventory('metadata_not_existing')
self.assertIn(self.metadata_folders[0], self.m.inventories)
def isEmpty(self, metadata):
"""Asserts if the given metadata object is empty"""
self.assertDictEqual({}, metadata.inventory_files)
self.assertDictEqual({}, metadata.seed_ids)
self.assertEqual([], metadata.inventories)
class TestMetadata_read_single_file(unittest.TestCase):
def setUp(self):
self.station_id = 'BW.WETR..HHZ'
self.metadata_folders = (os.path.join('test_data', 'dless_multiple_files', 'metadata1'),
os.path.join('test_data', 'dless_multiple_files', 'metadata2'))
self.metadata_paths = []
self.m = Metadata()
def test_read_single_file(self):
"""Test if reading a single file works"""
fname = os.path.join(self.metadata_folders[0], 'DATALESS.' + self.station_id)
with HidePrints():
res = self.m.read_single_file(fname)
# method should return true if file is successfully read
self.assertTrue(res)
# list of inventories (folders) should be empty
self.assertEqual([], self.m.inventories)
# list of inventory files should contain the added file
self.assertIn(fname, self.m.inventory_files.keys())
self.assertEqual({}, self.m.seed_ids)
def test_read_single_file_invalid_path(self):
"""Test if reading from a non existing file fails. The filename should not be
added to the metadata object"""
fname = os.path.join("this", "path", "doesnt", "exist")
with HidePrints():
res = self.m.read_single_file(fname)
# method should return None if file reading fails
self.assertIsNone(res)
# list of inventories (folders) should be empty
self.assertEqual([], self.m.inventories)
# list of inventory files should not contain the added file
self.assertNotIn(fname, self.m.inventory_files.keys())
self.assertEqual({}, self.m.seed_ids)
def test_read_single_file_multiple_times(self):
"""Test if reading a file twice doesnt add it twice to the metadata object"""
fname = os.path.join(self.metadata_folders[0], 'DATALESS.' + self.station_id)
with HidePrints():
res1 = self.m.read_single_file(fname)
res2 = self.m.read_single_file(fname)
self.assertTrue(res1)
self.assertIsNone(res2)
self.assertItemsEqual([fname], self.m.inventory_files.keys())
class TestMetadataMultipleTime(unittest.TestCase):
"""Test if stations with multiple metadata entries in a single file are handled correctly.
The user must specify the time where he wants to get metadata.
The station ROTT changed has metadata available at multiple times
LE.ROTT..HNE | 200.00 Hz | Titan 4g-EDR-209, Very Low gain, 200 sps | 2015-01-08 - 2015-03-19 | Lat: 49.1, Lng: 8.1
LE.ROTT..HNE | 200.00 Hz | Titan 4g-EDR-209, Very Low gain, 200 sps | 2015-03-19 - | Lat: 49.1, Lng: 8.1
LE.ROTT..HNN | 200.00 Hz | Titan 4g-EDR-209, Very Low gain, 200 sps | 2015-01-08 - 2015-03-19 | Lat: 49.1, Lng: 8.1
LE.ROTT..HNN | 200.00 Hz | Titan 4g-EDR-209, Very Low gain, 200 sps | 2015-03-19 - | Lat: 49.1, Lng: 8.1
LE.ROTT..HNZ | 200.00 Hz | Titan 4g-EDR-209, Very Low gain, 200 sps | 2015-01-08 - 2015-03-19 | Lat: 49.1, Lng: 8.1
LE.ROTT..HNZ | 200.00 Hz | Titan 4g-EDR-209, Very Low gain, 200 sps | 2015-03-19 - | Lat: 49.1, Lng: 8.1
"""
def setUp(self):
self.seed_id = 'LE.ROTT..HN'
path = os.path.dirname(__file__) # gets path to currently running script
metadata = os.path.join('test_data', 'dless_multiple_times',
'MAGS2_LE_ROTT.dless') # specific subfolder of test data
metadata_path = os.path.join(path, metadata)
self.m = Metadata(metadata_path)
self.p = Parser(metadata_path)
def test_get_metadata_works_without_datetime(self):
"""Test if get_metadata works if multiple metadata entries are available but no time is
specified."""
for channel in ('Z', 'N', 'E'):
with HidePrints():
md = self.m.get_metadata(self.seed_id + channel)
self.assertDictEqual(md['data'].get_inventory(), self.p.get_inventory())
def test_get_metadata_works_with_first_datetime(self):
"""Test if get_metadata works if multiple metadata entries are available and the older time is specified."""
t = UTCDateTime('2015-02-08')
for channel in ('Z', 'N', 'E'):
with HidePrints():
md = self.m.get_metadata(self.seed_id + channel, t)
self.assertDictEqual(md['data'].get_inventory(), self.p.get_inventory())
def test_get_metadata_fails_when_time_before_starttime(self):
"""Tests if get_metadata returns None when given a data that is before the start date
of the metadata"""
with HidePrints():
md = self.m.get_metadata(self.seed_id, UTCDateTime('1960-07-20'))
self.assertIs(md, None)
def test_get_metadata_invalid_seed_id(self):
"""Tes if get metadata returns none when asked for a seed id that does not exist"""
with HidePrints():
res = self.m.get_metadata("this.doesnt..exist")
self.assertIsNone(res)
class TestMetadataMultipleEntries(unittest.TestCase):
"""
The station KB.TMO07 has changed instruments multiple times.
Networks:
KB (KB network)
Stations:
KB.TMO07 (Karlsruhe GPI)
Channels:
KB.TMO07.00.BHE | 50.00 Hz | Streckeisen KABBA-STS-2 | 2004-12-06 - 2005-04-18 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHE | 50.00 Hz | Streckeisen KABBA-STS-2 | 2005-04-18 - 2006-07-18 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHE | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2006-10-10 - 2006-11-14 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHE | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2006-11-24 - 2007-01-12 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHE | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-01-18 - 2007-03-15 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHE | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-10-25 - 2007-11-21 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHE | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-11-21 - 2008-01-17 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHN | 50.00 Hz | Streckeisen KABBA-STS-2 | 2004-12-06 - 2005-04-18 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHN | 50.00 Hz | Streckeisen KABBA-STS-2 | 2005-04-18 - 2006-07-18 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHN | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2006-10-10 - 2006-11-14 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHN | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2006-11-24 - 2007-01-12 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHN | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-01-18 - 2007-03-15 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHN | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-10-25 - 2007-11-21 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHN | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-11-21 - 2008-01-17 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHZ | 50.00 Hz | Streckeisen KABBA-STS-2 | 2004-12-06 - 2005-04-18 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHZ | 50.00 Hz | Streckeisen KABBA-STS-2 | 2005-04-18 - 2006-07-18 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHZ | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2006-10-10 - 2006-11-14 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHZ | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2006-11-24 - 2007-01-12 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHZ | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-01-18 - 2007-03-15 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHZ | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-10-25 - 2007-11-21 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.BHZ | 50.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-11-21 - 2008-01-17 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 100.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-01-12 - 2007-01-18 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 100.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-10-10 - 2007-10-25 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 100.00 Hz | Streckeisen KABBA-STS-2 | 2008-07-11 - 2008-12-05 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 100.00 Hz | Streckeisen KABBA-STS-2 | 2009-05-12 - 2010-02-15 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 100.00 Hz | Streckeisen KABBA-STS-2 | 2010-02-15 - 2010-04-07 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 100.00 Hz | Lennartz KABBA-LE-3D/1 | 2010-04-07 - 2010-08-03 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 200.00 Hz | Streckeisen KABBA-STS-2 | 2010-08-05 - 2010-12-20 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 100.00 Hz | Streckeisen KABBA-STS-2 | 2010-12-20 - 2010-12-22 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 200.00 Hz | Streckeisen KABBA-STS-2 | 2010-12-22 - 2011-04-02 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 200.00 Hz | Streckeisen KABBA-STS-2 | 2011-04-15 - 2012-05-07 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHE | 200.00 Hz | Streckeisen KABBA-STS-2 | 2012-05-07 - | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 100.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-01-12 - 2007-01-18 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 100.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-10-10 - 2007-10-25 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 100.00 Hz | Streckeisen KABBA-STS-2 | 2008-07-11 - 2008-12-05 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 100.00 Hz | Streckeisen KABBA-STS-2 | 2009-05-12 - 2010-02-15 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 100.00 Hz | Streckeisen KABBA-STS-2 | 2010-02-15 - 2010-04-07 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 100.00 Hz | Lennartz KABBA-LE-3D/1 | 2010-04-07 - 2010-08-03 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 200.00 Hz | Streckeisen KABBA-STS-2 | 2010-08-05 - 2010-12-20 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 100.00 Hz | Streckeisen KABBA-STS-2 | 2010-12-20 - 2010-12-22 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 200.00 Hz | Streckeisen KABBA-STS-2 | 2010-12-22 - 2011-04-02 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 200.00 Hz | Streckeisen KABBA-STS-2 | 2011-04-15 - 2012-05-07 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHN | 200.00 Hz | Streckeisen KABBA-STS-2 | 2012-05-07 - | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 100.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-01-12 - 2007-01-18 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 100.00 Hz | Lennartz KABBA-LE-3D/5 | 2007-10-10 - 2007-10-25 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 100.00 Hz | Streckeisen KABBA-STS-2 | 2008-07-11 - 2008-12-05 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 100.00 Hz | Streckeisen KABBA-STS-2 | 2009-05-12 - 2010-02-15 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 100.00 Hz | Streckeisen KABBA-STS-2 | 2010-02-15 - 2010-04-07 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 100.00 Hz | Lennartz KABBA-LE-3D/1 | 2010-04-07 - 2010-08-03 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 200.00 Hz | Streckeisen KABBA-STS-2 | 2010-08-05 - 2010-12-20 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 100.00 Hz | Streckeisen KABBA-STS-2 | 2010-12-20 - 2010-12-22 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 200.00 Hz | Streckeisen KABBA-STS-2 | 2010-12-22 - 2011-04-02 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 200.00 Hz | Streckeisen KABBA-STS-2 | 2011-04-15 - 2012-05-07 | Lat: 49.0, Lng: 8.4
KB.TMO07.00.HHZ | 200.00 Hz | Streckeisen KABBA-STS-2 | 2012-05-07 - | Lat: 49.0, Lng: 8.4
"""
def setUp(self):
self.seed_id = 'KB.TMO07.00.HHZ'
path = os.path.dirname(__file__) # gets path to currently running script
metadata = os.path.join('test_data', 'dless_multiple_instruments',
'MAGS2_KB_TMO07.dless') # specific subfolder of test data
metadata_path = os.path.join(path, metadata)
self.m = Metadata(metadata_path)
self.p = Parser(metadata_path)
def test_get_paz_current_time(self):
"""Test if getting the paz from the metadata object with the current time works"""
t = UTCDateTime()
with HidePrints():
pazm = self.m.get_paz(self.seed_id, t)
pazp = self.p.get_paz(self.seed_id, t)
self.assertEqual(pazm, pazp)
def test_get_paz_past(self):
"""Test if getting paz from metadata object with a time in the past works"""
t = UTCDateTime('2007-01-13')
with HidePrints():
pazm = self.m.get_paz(self.seed_id, t)
pazp = self.p.get_paz(self.seed_id, t)
self.assertEqual(pazm, pazp)
def test_get_paz_time_not_exisiting(self):
"""Test if getting paz from metadata at a time where there is no metadata
available fails correctly"""
with self.assertRaises(SEEDParserException):
with HidePrints():
self.m.get_paz(self.seed_id, UTCDateTime('1990-1-1'))
def test_get_paz_seed_id_not_existing(self):
"""Test if getting paz from a non existing seed id returns None as expected."""
with HidePrints():
res = self.m.get_paz('This.doesnt..exist', UTCDateTime)
self.assertIsNone(res)