# -*- coding: utf-8 -*- """Tests for `lemi_metadata` module.""" import openpyxl import pickle import unittest from obspy import UTCDateTime from pathlib import Path from lemi2seed.lemi_data import LemiData from lemi2seed.lemi_metadata import LemiMetadata from lemi2seed.logging import setup_logger from lemi2seed.metadata_category import Elec, Mag, Aux OUTPUT_MSEED = Path(__file__).resolve().parent.joinpath('MSEED') OUTPUT_LOG = Path(__file__).resolve().parent.joinpath('LOG') OUTPUT_XML = Path(__file__).resolve().parent.joinpath('STATIONXML') TEST_DIR = Path(__file__).resolve().parent.joinpath('test_data') SCR_DIR = "lemi2seed.lemi_metadata" # Set up logging logger = setup_logger(SCR_DIR) class TestLemiMetadata(unittest.TestCase): """Test suite for LemiMetadata class.""" def setUp(self): """Set up test fixtures""" lemi_data = LemiData(TEST_DIR.joinpath("EM", "TEST5"), OUTPUT_MSEED, OUTPUT_LOG) lemi_data.prep_data() self.data = lemi_data self.path2md = TEST_DIR.joinpath("METADATA") self.path2md_fail_1 = TEST_DIR.joinpath("METADATA_NO_FIELD_SHEET") self.path2md_fail_2 = TEST_DIR.joinpath("METADATA_CORRUPTED") self.file = self.path2md.joinpath('LEMI_Install_Sheet.xlsx') self.md = LemiMetadata(self.path2md, OUTPUT_XML, self.data) file_ = self.path2md.joinpath('metadata_fields.pkl') with open(file_, 'rb') as fin: self.md_fields_no_reformat = pickle.load(fin) file_ = self.path2md.joinpath('metadata_fields_reformatted.pkl') with open(file_, 'rb') as fin: self.md_fields = pickle.load(fin) def test_scan_path2md_field_sheet_provided(self): """Test basic functionality of scan_path2md.""" md = LemiMetadata(self.path2md, OUTPUT_XML, self.data) self.assertEqual(*md.filenames, self.file) def test_scan_path2md_no_field_sheet_provided(self): """Test basic functionality of scan_path2md.""" with self.assertLogs(logger, level='ERROR') as cmd: LemiMetadata(self.path2md_fail_1, OUTPUT_XML, self.data) msg = ("No field sheet found under the following path - {}. All " "metadata will have to be provided using the GUI!" .format(self.path2md_fail_1)) self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])]) def test_get_md_field_pass(self): """Test basic functionality of get_md_field.""" workbook = openpyxl.load_workbook(self.file, data_only=True) sheet = workbook.active self.assertEqual(LemiMetadata.get_md_field(sheet, 'B3'), 'EM') workbook.close() def test_get_md_field_fail(self): """Test basic functionality of get_md_field.""" workbook = openpyxl.load_workbook(self.file, data_only=True) sheet = workbook.active with self.assertLogs(logger, level='ERROR'): self.assertIsNone(LemiMetadata.get_md_field(sheet, 12)) workbook.close() def test_identify_field_sheet_pass(self): """Test basic functionality of identify_field_sheet.""" workbook = openpyxl.load_workbook(self.file, data_only=True) sheet = workbook.active sheet_type = self.md.identify_field_sheet(sheet, 'LEMI_Install_Sheet.xlsx') self.assertEqual(sheet_type, 'install_sheet') workbook.close() def test_identify_field_sheet_fail(self): """Test basic functionality of identify_field_sheet.""" md = LemiMetadata(self.path2md_fail_2, OUTPUT_XML, self.data) file_ = self.path2md_fail_2.joinpath('LEMI_Install_Sheet_2.xlsx') workbook = openpyxl.load_workbook(file_, data_only=True) sheet = workbook.active with self.assertLogs(logger, level='WARNING') as cmd: md.identify_field_sheet(sheet, file_) msg = ("The following file {} does not have the proper header. The " "provided spread sheet templates were not used or their " "layout was modified. Skipping file!".format(file_)) self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])]) workbook.close() def test_reformat_run(self): """Test basic functionality of reformat_run.""" run_fields = self.md_fields_no_reformat['Run'] file_ = self.path2md.joinpath('run_fields_reformatted.pkl') with open(file_, 'rb') as fin: reformatted = pickle.load(fin) self.assertDictEqual(self.md.reformat_run(run_fields), reformatted) def test_reformat_elec(self): """Test basic functionality of reformat_elec.""" elec_fields = self.md_fields_no_reformat['Elec'] file_ = self.path2md.joinpath('electric_fields_reformatted.pkl') with open(file_, 'rb') as fin: reformatted = pickle.load(fin) self.assertDictEqual(self.md.reformat_elec(elec_fields), reformatted) def test_reformat_mag(self): """Test basic functionality of reformat_mag.""" mag_fields = self.md_fields_no_reformat['Mag'] file_ = self.path2md.joinpath('magnetic_fields_reformatted.pkl') with open(file_, 'rb') as fin: reformatted = pickle.load(fin) self.assertDictEqual(self.md.reformat_mag(mag_fields), reformatted) def test_reformat_md_dict(self): """Test basic functionality of reformat_md_dict.""" self.assertDictEqual(self.md.reformat_md_dict(self.md_fields_no_reformat), self.md_fields) def test_from_field_sheets_failed_to_open(self): """Test basic functionality of from_field_sheets""" self.md.filenames = [self.path2md_fail_2.joinpath('LEMI_Install_Sheet_1.xlsx')] with self.assertLogs(logger, level='ERROR') as cmd: self.md.from_field_sheets msg = ("Failed to open field sheet: {} - Skipping file - Exception: " "File is not a zip file".format(self.md.filenames[0])) self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])]) def test_from_field_sheets_duplicate_sheet_type(self): """Test basic functionality of from_field_sheets""" self.md.filenames = [self.file] * 2 with self.assertLogs(logger, level='WARNING') as cmd: self.md.from_field_sheets msg = ("Already parsed the install sheet - You may have more than one " "install sheet - Skipping {}!".format(self.file)) self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])]) def test_from_field_sheets_expected_workflow(self): """Test basic functionality of from_field_sheets""" self.assertDictEqual(self.md.from_field_sheets, self.md_fields) def test_populate_net_md_props(self): """Test basic functionality of populate_net_md_props.""" self.md.populate_net_md_props(self.md_fields['Net']) self.assertSetEqual(self.md.net.md_missing, {'geo_name', 'start', 'project', 'name'}) self.assertSetEqual(self.md.net.md_invalid, {'start'}) for key, val in self.md_fields['Net'].items(): if key not in self.md.net.md_invalid: self.assertEqual(getattr(self.md.net, key), val) self.assertIsNone(self.md.net.start) def test_populate_sta_md_props(self): """Test basic functionality of populate_sta_md_props.""" self.md.populate_sta_md_props(self.md_fields['Sta']) self.assertSetEqual(self.md.sta.md_missing, {'end', 'elev', 'lon', 'start', 'geo_name'}) self.assertSetEqual(self.md.sta.md_invalid, {'elev', 'lon', 'start'}) for key, val in self.md_fields['Sta'].items(): if key not in self.md.sta.md_invalid: self.assertEqual(getattr(self.md.sta, key), val) self.assertEqual(self.md.sta.run_list, 'a, b, c, d') self.assertIsNone(self.md.sta.elev) self.assertIsNone(self.md.sta.start) def test_init_run_md_props(self): """Test basic functionality of init_run_md_props.""" time_period_starts = [UTCDateTime(2020, 9, 30, 21, 5), UTCDateTime(2020, 9, 30, 21, 12), UTCDateTime(2020, 9, 30, 21, 14), UTCDateTime(2020, 10, 1, 0, 0)] time_period_ends = [UTCDateTime(2020, 9, 30, 21, 11, 1), UTCDateTime(2020, 9, 30, 21, 13, 45), UTCDateTime(2020, 9, 30, 21, 27, 59), UTCDateTime(2020, 10, 1, 0, 5, 59)] self.md.init_run_md_props() for ind, run in enumerate(self.md.run): self.assertEqual(run.start, time_period_starts[ind]) self.assertEqual(run.end, time_period_ends[ind]) self.assertEqual(run.resource_id, 'mt.run.id:' + self.md.run_list[ind]) def test_populate_run_md_props(self): """Test basic functionality of populate_run_md_props.""" self.md.populate_run_md_props(self.md_fields['Run']) self.assertEqual(self.md.run[0].comps_rec, 'E1, E2, E3, E4, Hx, Hy, Hz') self.assertEqual(self.md.run[0].datalogger_sn, 110) self.assertSetEqual(self.md.run[0].md_missing, set()) self.assertSetEqual(self.md.run[0].md_invalid, set()) for run in self.md.run[1:]: self.assertIsNone(run.comps_rec) self.assertIsNone(run.datalogger_sn) self.assertSetEqual(run.md_missing, {'datalogger_sn', 'comps_rec'}) self.assertSetEqual(run.md_invalid, {'datalogger_sn', 'comps_rec'}) def test_init_cha_md_props_elec(self): """Test basic functionality of init_cha_md_props.""" cha = self.md.init_cha_md_props('elec', 'a') self.assertIsInstance(cha, Elec) self.assertEqual(cha.elev, round(self.md.data_stats['elev'], 3)) self.assertEqual(cha.lat, round(self.md.data_stats['lat'], 3)) self.assertEqual(cha.lon, round(self.md.data_stats['lon'], 3)) self.assertEqual(cha.run_id, 'a') def test_init_cha_md_props_mag(self): """Test basic functionality of init_cha_md_props.""" cha = self.md.init_cha_md_props('mag', 'a') self.assertIsInstance(cha, Mag) self.assertEqual(cha.elev, round(self.md.data_stats['elev'], 3)) self.assertEqual(cha.lat, round(self.md.data_stats['lat'], 3)) self.assertEqual(cha.lon, round(self.md.data_stats['lon'], 3)) self.assertEqual(cha.run_id, 'a') def test_init_cha_md_props_aux(self): """Test basic functionality of init_cha_md_props.""" cha = self.md.init_cha_md_props('aux', 'a') self.assertIsInstance(cha, Aux) self.assertEqual(cha.elev, round(self.md.data_stats['elev'], 3)) self.assertEqual(cha.lat, round(self.md.data_stats['lat'], 3)) self.assertEqual(cha.lon, round(self.md.data_stats['lon'], 3)) self.assertEqual(cha.run_id, 'a') def test_populate_elec_md_props_one_run_set(self): """Test basic functionality of populate_elec_md_props.""" self.md.populate_run_md_props(self.md_fields['Run']) self.md.populate_elec_md_props(self.md_fields['Elec']) file_ = self.path2md.joinpath('electric_metadata_properties_1.pkl') with open(file_, 'rb') as fin: expected = pickle.load(fin) self.assertListEqual(self.md.elec, expected) def test_populate_elec_md_props_two_runs_set(self): """Test basic functionality of populate_elec_md_props.""" self.md.populate_run_md_props(self.md_fields['Run']) self.md.run[1].comps_rec = 'E1, E2, Hx, Hy, Hz' specs = 'Borin STELTH 4 - Silver-Silver Chloride' self.md_fields['Elec']['Run_b_Elec_Pair_1']['cha_num'] = 'E1' self.md_fields['Elec']['Run_b_Elec_Pair_2']['cha_num'] = 'E2' self.md_fields['Elec']['Run_b_Elec_Pair_3']['cha_num'] = 'E3' self.md_fields['Elec']['Run_b_Elec_Pair_4']['cha_num'] = 'E4' self.md_fields['Elec']['Run_b_Elec_Pair_1']['inst_specs'] = specs self.md_fields['Elec']['Run_b_Elec_Pair_2']['inst_specs'] = specs self.md.populate_elec_md_props(self.md_fields['Elec']) file_ = self.path2md.joinpath('electric_metadata_properties_2.pkl') with open(file_, 'rb') as fin: expected = pickle.load(fin) self.assertListEqual(self.md.elec, expected) def test_populate_mag_md_props_one_run_set(self): """Test basic functionality of populate_mag_md_props.""" self.md.populate_run_md_props(self.md_fields['Run']) self.md.populate_mag_md_props(self.md_fields['Mag']) file_ = self.path2md.joinpath('magnetic_metadata_properties_1.pkl') with open(file_, 'rb') as fin: expected = pickle.load(fin) self.assertListEqual(self.md.mag, expected) def test_populate_mag_md_props_two_runs_set(self): """Test basic functionality of populate_mag_md_props.""" self.md.populate_run_md_props(self.md_fields['Run']) self.md.run[1].comps_rec = 'E1, E2, Hx, Hy, Hz' specs = 'LEMI-039' self.md_fields['Mag']['Run_b_Mag']['inst_specs'] = specs self.md.populate_mag_md_props(self.md_fields['Mag']) file_ = self.path2md.joinpath('magnetic_metadata_properties_2.pkl') with open(file_, 'rb') as fin: expected = pickle.load(fin) self.assertListEqual(self.md.mag, expected) def test_populate_aux_md_props_one_run_set(self): """Test basic functionality of populate_aux_md_props.""" self.md.populate_run_md_props(self.md_fields['Run']) self.md.populate_aux_md_props() file_ = self.path2md.joinpath('auxiliary_metadata_properties.pkl') with open(file_, 'rb') as fin: expected = pickle.load(fin) self.assertListEqual(self.md.aux, expected) def test_filter_cha_cha_not_populated_for_given_run(self): """Test basic functionality of filter_cha.""" self.assertListEqual(self.md.filter_cha('aux', 'a'), []) def test_filter_cha_cha_populated_for_given_run(self): """Test basic functionality of filter_cha.""" cha_names = ['LEH', 'LKH', 'LKF', 'GNS', 'GST', 'LCE'] self.md.populate_md_props(self.md_fields) cha_filtered = self.md.filter_cha('aux', 'a') self.assertEqual(len(cha_filtered), 6) for ind, cha in enumerate(cha_filtered): self.assertEqual(cha.cha_name, cha_names[ind]) def test_get_cha_inds_cha_not_populated_for_given_run(self): """Test basic functionality of get_cha_inds.""" self.assertListEqual(self.md.get_cha_inds('aux', 'a'), []) def test_get_cha_inds_cha_populated_for_given_run(self): """Test basic functionality of get_cha_inds.""" self.md.populate_md_props(self.md_fields) cha_inds = self.md.get_cha_inds('aux', 'a') self.assertListEqual(cha_inds, [0, 1, 2, 3, 4, 5]) def test_get_comps_rec_no_elec(self): """Test basic functionality of get_comps_rec.""" type_ = 'E' cha_type = 'electric field' run_id = 'b' self.md.populate_run_md_props(self.md_fields['Run']) with self.assertLogs(logger, level='WARNING') as cmd: self.md.get_comps_rec(type_, run_id) msg = (f"No {cha_type} data recorded for run '{run_id}'. If you " f"did record {cha_type} data for that run, please update " "your list of recorded components at the station level " "accordingly!") self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])]) def test_get_comps_rec_no_elec_but_mag(self): """Test basic functionality of get_comps_rec.""" type_ = 'E' cha_type = 'electric field' run_id = 'b' self.md.populate_run_md_props(self.md_fields['Run']) self.md.run[1].comps_rec = 'Hx, Hy, Hz' with self.assertLogs(logger, level='WARNING') as cmd: self.md.get_comps_rec(type_, run_id) msg = (f"No {cha_type} data recorded for run '{run_id}'. If you " f"did record {cha_type} data for that run, please update " "your list of recorded components at the station level " "accordingly!") self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])]) def test_get_comps_rec_elec(self): """Test basic functionality of get_comps_rec.""" type_ = 'E' run_id = 'a' self.md.populate_run_md_props(self.md_fields['Run']) comps = self.md.get_comps_rec(type_, run_id) self.assertEqual(comps, ['E1', 'E2', 'E3', 'E4']) def test_get_comps_rec_missing_mag(self): """Test basic functionality of get_comps_rec.""" type_ = 'H' cha_type = 'magnetic field' run_id = 'b' self.md.populate_run_md_props(self.md_fields['Run']) self.md.run[1].comps_rec = 'Hx, Hy' with self.assertLogs(logger, level='WARNING') as cmd: self.md.get_comps_rec(type_, run_id) msg = ("LEMI-039 fluxgate is a 3-component magnetometer. If you did " f"record {cha_type} data for run {run_id}, data from all " "magnetic components (Hx, Hy and Hz) were recorded. Please " "update your list of recorded components at the station level.") self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])]) def test_get_comps_rec_mag(self): """Test basic functionality of get_comps_rec.""" type_ = 'H' run_id = 'a' self.md.populate_run_md_props(self.md_fields['Run']) comps = self.md.get_comps_rec(type_, run_id) self.assertEqual(comps, ['Hx', 'Hy', 'Hz']) def test_match_num_e_pairs(self): """Test basic functionality of match_num_e_pairs.""" comps = ['E1', 'E2', 'E3', 'E4'] run_id = 'a' self.md.populate_md_props(self.md_fields) with self.assertLogs(logger, level='ERROR') as cmd: self.md.match_num_e_pairs(comps, run_id, 2) msg = (f"Invalid number of electrode pairs (run '{run_id}')! The " "number of electrode pairs does not match the number of " "electric channels in your list of recorded components at the " "station/run level.") self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])]) for ind, elec in enumerate(self.md.elec): self.assertSetEqual(elec.md_invalid, {'run_a_num_e_pairs'}) self.md.match_num_e_pairs(comps, run_id, 4) for elec in self.md.elec: self.assertSetEqual(elec.md_invalid, set()) def test_match_comps_rec(self): """Test basic functionality of match_comps_rec.""" comps = ['E1', 'E2'] run_id = 'a' self.md.populate_md_props(self.md_fields) with self.assertLogs(logger, level='ERROR') as cmd: self.md.match_comps_rec(self.md.elec, comps, run_id) msg = (f"Invalid electric channel numbers (run '{run_id}')! The " "channel numbers don't match the electric channels in your " "list of recorded components at the station/run level.") self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])]) for e_cha in self.md.elec: self.assertSetEqual(e_cha.md_invalid, {'cha_num'}) def test_update_cha_num(self): """Test basic functionality of update_cha_num.""" cha = self.md.init_cha_md_props('elec', 'a') self.assertIsNone(cha.cha_num) updated_cha = LemiMetadata.update_cha_num(cha, 'E1') self.assertEqual(updated_cha.cha_num, 'E1') def test_update_comp_cha_name(self): """Test basic functionality of update_comp_cha_name.""" cha = self.md.init_cha_md_props('mag', 'a') self.assertIsNone(cha.cha_name) self.assertIsNone(cha.comp) updated_cha = LemiMetadata.update_comp_cha_name(cha, 'Hx') self.assertEqual(updated_cha.cha_name, 'LFN') self.assertEqual(updated_cha.comp, 'Hx') def test_get_e_infos(self): """Test basic functionality of get_e_infos.""" comps = ['Ex', 'Ey', 'Ex', 'Ey', 'Ex', 'Ey'] self.md.populate_run_md_props(self.md_fields['Run']) self.md.run[1].comps_rec = 'E1, E2, Hx, Hy, Hz' cha = self.md.init_cha_md_props('elec', 'a') self.md.elec.extend([self.md.update_cha_num(cha, f'E{i+1}') for i in range(4)]) cha = self.md.init_cha_md_props('elec', 'b') self.md.elec.extend([self.md.update_cha_num(cha, f'E{i+1}') for i in range(2)]) for ind, elec in enumerate(self.md.elec): elec.comp = comps[ind] e_infos = self.md.get_e_infos() self.assertDictEqual(e_infos, {'a': {'E1': 'Ex', 'E2': 'Ey', 'E3': 'Ex', 'E4': 'Ey'}, 'b': {'E1': 'Ex', 'E2': 'Ey'}, 'c': {}, 'd': {}}) def test_update_location(self): """Test basic functionality of update_location.""" comps = ['Ex', 'Ey', 'Ex', 'Ey'] location_codes = ['00', '00', '01', '01'] self.md.populate_run_md_props(self.md_fields['Run']) cha = self.md.init_cha_md_props('elec', 'a') self.md.elec.extend([self.md.update_cha_num(cha, f'E{i+1}') for i in range(4)]) for ind, elec in enumerate(self.md.elec): elec.comp = comps[ind] self.assertEqual(elec.loc_code, '') self.md.update_loc() for ind, elec in enumerate(self.md.elec): self.assertEqual(elec.loc_code, location_codes[ind]) def test_flag_md_missing_no_skip(self): """Test basic functionality of flag_md_missing.""" LemiMetadata.flag_md_missing(self.md.net) self.assertSetEqual(self.md.net.md_missing, {'geo_name', 'start', 'archive_net', 'project', 'end', 'name'}) def test_flag_md_missing_skip(self): """Test basic functionality of flag_md_missing.""" LemiMetadata.flag_md_missing(self.md.net, ['geo_name']) self.assertSetEqual(self.md.net.md_missing, {'start', 'archive_net', 'project', 'end', 'name'}) def test_dc2dict(self): """Test basic functionality of dc2dict.""" cha = self.md.init_cha_md_props('elec', 'a') self.assertDictEqual(self.md.dc2dict(cha), {'elev': 2201.725, 'lat': 34.048, 'lon': -107.128, 'inst_manufacturer': None, 'inst_model': None, 'inst_specs': None, 'inst_type': None, 'meas_azimuth': None, 'meas_tilt': None, 'sample_rate': 1.0, 'cha_num': None, 'contact_resistance_end': None, 'contact_resistance_start': None, 'dc_end': None, 'dc_start': None, 'dipole_len': None, 'neg_elec_dir': None, 'neg_elec_sn': None, 'pos_elec_dir': None, 'pos_elec_sn': None}) def test_for_gui_md_not_populated(self): """Test basic functionality of for_gui.""" file_ = self.path2md.joinpath('gui_metadata_not_populated.pkl') with open(file_, 'rb') as fin: expected = pickle.load(fin) self.assertDictEqual(self.md.for_gui, expected) def test_for_gui_md_populated(self): """Test basic functionality of for_gui.""" self.md.populate_md_props(self.md_fields) file_ = self.path2md.joinpath('gui_metadata_populated.pkl') with open(file_, 'rb') as fin: expected = pickle.load(fin) self.assertDictEqual(self.md.for_gui, expected) def test_save_md(self): """Test basic functionality of save_md.""" file_ = self.path2md.joinpath('lemi_metadata.pkl') self.md.save_md(file_) self.assertTrue(file_.is_file()) def test_load_md(self): """Test basic functionality of load_md.""" file_ = self.path2md.joinpath('lemi_metadata.pkl') self.md.save_md(file_) md = LemiMetadata.load_md(file_) self.assertEqual(md.net, self.md.net) self.assertEqual(md.sta, self.md.sta) self.assertListEqual(md.run, self.md.run) self.assertListEqual(md.elec, self.md.elec) self.assertListEqual(md.mag, self.md.mag) self.assertListEqual(md.aux, self.md.aux) def test_update_azimuth_tilt(self): """Test basic functionality of update_azimuth_tilt.""" self.md.populate_md_props(self.md_fields) self.md.update_azimuth_tilt() for cha in self.md.filter_cha('mag', 'a'): if cha.comp == 'Hx': self.assertEqual(cha.meas_tilt, 0.0) self.assertEqual(cha.meas_azimuth, 20) if cha.comp == 'Hy': self.assertEqual(cha.meas_tilt, 0.0) self.assertEqual(cha.meas_azimuth, 110.0) if cha.comp == 'Hz': self.assertEqual(cha.meas_tilt, -1.0) self.assertEqual(cha.meas_azimuth, 0.0) def tearDown(self): """Tear down test fixtures""" file_ = self.path2md.joinpath('lemi_metadata.pkl') if file_.is_file(): file_.unlink()