# -*- 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_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_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_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 tearDown(self):
"""Tear down test fixtures"""
file_ = self.path2md.joinpath('lemi_metadata.pkl')
if file_.is_file():
file_.unlink()