# -*- coding: utf-8 -*-
"""Tests for `lemi_metadata` module."""
import logging
import openpyxl
import pickle
import unittest
from obspy import UTCDateTime
from pathlib import Path
from lemi2seed.lemi_data import LemiData, VALID_COMPONENTS
from lemi2seed.lemi_metadata import (BaseSurvey, Survey, BaseStation, Station_,
Run, Electric, Magnetic, Auxiliary,
LemiMetadata, log_file_path)
from lemi2seed.utils import MSG_E_CHA
OUTPUT_MSEED = Path(__file__).resolve().parent.joinpath('MSEED')
OUTPUT_XML = Path(__file__).resolve().parent.joinpath('STATIONXML')
TEST_DIR = Path(__file__).resolve().parent.joinpath('test_data')
SCR_DIR = "lemi2seed.lemi_metadata"
logging.config.fileConfig(log_file_path)
logger = logging.getLogger(SCR_DIR)
class TestBaseSurvey(unittest.TestCase):
"""Test suite for BaseSurvey data class."""
def setUp(self):
"""Set up test fixtures"""
lemi_data = LemiData(TEST_DIR.joinpath("EM", "TEST1"), OUTPUT_MSEED)
lemi_data.prep_data()
self.data_stats = lemi_data.stats
def test_validate_time_period_start_not_utc(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = 2021.167
bs = BaseSurvey()
self.assertFalse(bs.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_start_greater_than_acquisition_start(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = UTCDateTime('2020-10-01T00:00:00.000000Z')
bs = BaseSurvey()
self.assertFalse(bs.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_start_valid(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
bs = BaseSurvey()
self.assertTrue(bs.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_end_not_utc(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = 2021.167
bs = BaseSurvey()
self.assertFalse(bs.validate_time_period_end(metadata_input, data_input))
def test_validate_time_period_end_lower_than_acquisition_end(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
bs = BaseSurvey()
self.assertFalse(bs.validate_time_period_end(metadata_input, data_input))
def test_validate_time_period_end_valid(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = UTCDateTime('2020-10-02T00:00:00.000000Z')
bs = BaseSurvey()
self.assertTrue(bs.validate_time_period_end(metadata_input, data_input))
class TestSurvey(unittest.TestCase):
"""Test suite for Survey data class."""
def test_validate_archive_network_undefined(self):
"""Test basic functionality of validate_archive_network."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_archive_network(None)
msg = "The network code should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_archive_network_erroneous_type(self):
"""Test basic functionality of validate_archive_network."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_archive_network(12)
msg = "The network code should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_archive_network_invalid(self):
"""Test basic functionality of validate_archive_network."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_archive_network('EMX')
msg = "The network code should be two alphanumeric character long."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_archive_network_valid(self):
"""Test basic functionality of validate_archive_network."""
self.assertTrue(Survey.validate_archive_network('EM'))
def test_validate_citation_dataset_doi_undefined(self):
"""Test basic functionality of validate_citation_dataset_doi."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_citation_dataset_doi(None)
msg = "The DOI number(s) should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_citation_dataset_doi_erroneous_type(self):
"""Test basic functionality of validate_citation_dataset_doi."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_citation_dataset_doi(10.7914)
msg = "The DOI number(s) should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_citation_dataset_doi_invalid_doi(self):
"""Test basic functionality of validate_citation_dataset_doi."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_citation_dataset_doi('10.7914/SN/EM')
msg = ("Invalid DOI(s). The DOI number(s) provided by the archive "
"should be strings formatted as follows: 'scheme: path'.")
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_citation_dataset_doi_invalid_dois(self):
"""Test basic functionality of validate_citation_dataset_doi."""
dois = '10.7914/SN/EM, DOI:10.3421/SN/EG'
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_citation_dataset_doi(dois)
msg = ("Invalid DOI(s). The DOI number(s) provided by the archive "
"should be strings formatted as follows: 'scheme: path'.")
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_citation_dataset_doi_valid_doi(self):
"""Test basic functionality of validate_citation_dataset_doi."""
doi = 'DOI:10.7914/SN/EM'
self.assertTrue(Survey.validate_citation_dataset_doi(doi))
def test_validate_citation_dataset_doi_valid_dois(self):
"""Test basic functionality of validate_citation_dataset_doi."""
dois = 'DOI:10.7914/SN/EM, DOI:10.3421/SN/EG'
self.assertTrue(Survey.validate_citation_dataset_doi(dois))
def test_validate_project_lead_email_undefined(self):
"""Test basic functionality of validate_project_lead_email."""
self.assertTrue(Survey.validate_project_lead_email(None))
def test_validate_project_lead_email_erroneous_type(self):
"""Test basic functionality of validate_project_lead_email."""
with self.assertLogs(logger, level='ERROR') as cmd:
Survey.validate_project_lead_email(12)
msg = "The project lead email(s) should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_project_lead_email_invalid_email(self):
"""Test basic functionality of validate_project_lead_email."""
email = 'mpasscal.edu'
self.assertFalse(Survey.validate_project_lead_email(email))
def test_validate_project_lead_email_invalid_emails(self):
"""Test basic functionality of validate_project_lead_email."""
emails = 'mpasscal.edu, d@passcal.edu'
self.assertFalse(Survey.validate_project_lead_email(emails))
def test_validate_project_lead_email_valid_email(self):
"""Test basic functionality of validate_citation_dataset_doi."""
email = 'm@passcal.edu'
self.assertTrue(Survey.validate_project_lead_email(email))
def test_validate_project_lead_email_valid_emails(self):
"""Test basic functionality of validate_citation_dataset_doi."""
emails = 'm@passcal.edu, d@passcal.edu'
self.assertTrue(Survey.validate_project_lead_email(emails))
class TestBaseStation(unittest.TestCase):
"""Test suite for BaseStation data class."""
def setUp(self):
"""Set up test fixtures"""
data_input = 2201.77
min_elev, max_elev = [abs(data_input) * x for x in [0.99, 1.01]]
param = ("Station elevation", -500, 8500, min_elev, max_elev, 'm', '')
self.param = param
def test_validate_geographics_undefined(self):
"""Test basic functionality of validate_geographics."""
with self.assertLogs(logger, level='ERROR') as cmd:
BaseStation.validate_geographics(self.param, None)
msg = "Station elevation should be a float. "
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_geographics_erroneous_type(self):
"""Test basic functionality of validate_geographics."""
with self.assertLogs(logger, level='ERROR') as cmd:
BaseStation.validate_geographics(self.param, 'a')
msg = "Station elevation should be a float. "
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_geographics_not_in_range(self):
"""Test basic functionality of validate_geographics."""
geo, min_range, max_range, min_val, max_val, units, msg = self.param
with self.assertLogs(logger, level='ERROR') as cmd:
BaseStation.validate_geographics(self.param, 9000)
msg = ("Unexpected {0}! The {0} should be between {1}{3} and {2}{3}."
.format(geo, min_range, max_range, units))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_geographics_do_not_match_logger_recorded_metadata(self):
"""Test basic functionality of validate_geographics."""
geo = 'Station elevation'
with self.assertLogs(logger, level='ERROR') as cmd:
BaseStation.validate_geographics(self.param, 2250)
msg = ("Unexpected {0}! Provided {0} should roughly match the {0} "
"recorded by the on-site GPS.".format(geo))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
class TestStation_(unittest.TestCase):
"""Test suite for Station_ data class."""
def test_validate_archive_id_undefined(self):
"""Test basic functionality of validate_archive_id."""
with self.assertLogs(logger, level='ERROR') as cmd:
Station_.validate_archive_id(None)
msg = "The station name should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_archive_id_erroneous_type(self):
"""Test basic functionality of validate_archive_id."""
with self.assertLogs(logger, level='ERROR') as cmd:
Station_.validate_archive_id(12)
msg = "The station name should be a string."
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_archive_id_invalid(self):
"""Test basic functionality of validate_archive_id."""
with self.assertLogs(logger, level='ERROR') as cmd:
Station_.validate_archive_id('KELLYA')
msg = ("The station name should be between three and five "
"alphanumeric character long.")
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_archive_id_valid(self):
"""Test basic functionality of validate_archive_id."""
self.assertTrue(Station_.validate_archive_id('KELLY'))
def test_validate_submitter_email_undefined(self):
"""Test basic functionality of validate_submitter_email."""
self.assertTrue(Station_.validate_submitter_email(None))
class TestRun(unittest.TestCase):
"""Test suite for Run data class."""
def setUp(self):
"""Set up test fixtures"""
self.run = Run(resource_id='mt.run.id:a')
lemi_data = LemiData(TEST_DIR.joinpath("EM", "TEST1"), OUTPUT_MSEED)
lemi_data.prep_data()
self.data_stats = lemi_data.stats
def test_validate_components_recorded_undefined(self):
"""Test basic functionality of validate_components_recorded."""
with self.assertLogs(logger, level='ERROR') as cmd:
self.run.validate_components_recorded(None)
msg = ("The list of components recorded for run '{}' should be a "
"string.".format(self.run.run_id))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_components_recorded_erroneous_type(self):
"""Test basic functionality of validate_components_recorded."""
with self.assertLogs(logger, level='ERROR') as cmd:
self.run.validate_components_recorded(12)
msg = ("The list of components recorded for run '{}' should be a "
"string.".format(self.run.run_id))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_components_recorded_invalid(self):
"""Test basic functionality of validate_components_recorded."""
components = 'E1, E2, Hx, Hy, Hn'
with self.assertLogs(logger, level='ERROR') as cmd:
self.run.validate_components_recorded(components)
msg = ("Some of the listed components for run '{}' are not valid. "
"List of valid components: {}."
.format(self.run.run_id, VALID_COMPONENTS))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_components_recorded_valid(self):
"""Test basic functionality of validate_components_recorded."""
components = 'E1, E2, Hx, Hy, Hz'
self.assertTrue(self.run.validate_components_recorded(components))
def test_validate_time_period_start_lower_than_acquisition_start(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
self.assertFalse(self.run.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_start_valid(self):
"""Test basic functionality of validate_time_period_start."""
data_input = self.data_stats['time_period_start']
metadata_input = UTCDateTime('2020-10-01T00:00:00.000000Z')
self.assertTrue(self.run.validate_time_period_start(metadata_input, data_input))
def test_validate_time_period_end_greater_than_acquisition_end(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = UTCDateTime('2020-10-02T00:00:00.000000Z')
self.assertFalse(self.run.validate_time_period_end(metadata_input, data_input))
def test_validate_time_period_end_valid(self):
"""Test basic functionality of validate_time_period_end."""
data_input = self.data_stats['time_period_end']
metadata_input = UTCDateTime('2020-09-30T00:00:00.000000Z')
self.assertTrue(self.run.validate_time_period_end(metadata_input, data_input))
class TestElectric(unittest.TestCase):
"""Test suite for Electric data class."""
def setUp(self):
"""Set up test fixtures"""
self.elec = Electric(channel_number='E1', run_id='a')
self.metadata_invalid = {'positive_electrode_direction',
'negative_electrode_direction'}
self.param = ("contact resistance (start)", 0, 3000, 'Ω')
def test_set_electric_component_channel_undefined_direction(self):
"""Test basic functionality of set_electric_component_channel."""
self.elec.positive_electrode_direction = None
self.elec.negative_electrode_direction = 'West'
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.set_electric_component_channel()
msg = ("The direction of the positive and negative electrodes should "
"be either: North, South, East or West. Please, provide "
"electrode direction(s) for run '{0}' and channel number "
"'{1}'! {2}".format(self.elec.run_id, self.elec.channel_number, MSG_E_CHA))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
self.assertIsNone(self.elec.component)
self.assertIsNone(self.elec.channel_name)
self.assertSetEqual(self.elec.metadata_invalid, self.metadata_invalid)
def test_set_electric_component_channel_invalid_direction_pair(self):
"""Test basic functionality of set_electric_component_channel."""
self.elec.positive_electrode_direction = 'North'
self.elec.negative_electrode_direction = 'West'
with self.assertLogs(logger, level='WARNING') as cmd:
self.elec.set_electric_component_channel()
msg = ("By convention, one electrode pair is installed in a "
"north-south direction and the other pair in a east-west "
"direction (check run '{0}' and channel number '{1}')!"
.format(self.elec.run_id, self.elec.channel_number))
self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
self.assertIsNone(self.elec.component)
self.assertIsNone(self.elec.channel_name)
self.assertSetEqual(self.elec.metadata_invalid, self.metadata_invalid)
def test_set_electric_component_channel_same_direction(self):
"""Test basic functionality of set_electric_component_channel."""
self.elec.positive_electrode_direction = 'North'
self.elec.negative_electrode_direction = 'North'
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.set_electric_component_channel()
msg = ("The direction of the positive and negative electrodes in a "
"given pair cannot be the same (check run '{0}' and channel "
"number '{1}')!".format(self.elec.run_id, self.elec.channel_number))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
self.assertIsNone(self.elec.component)
self.assertIsNone(self.elec.channel_name)
self.assertSetEqual(self.elec.metadata_invalid, self.metadata_invalid)
def test_set_electric_component_channel_valid(self):
"""Test basic functionality of set_electric_component_channel."""
self.elec.positive_electrode_direction = 'North'
self.elec.negative_electrode_direction = 'South'
self.elec.metadata_invalid = self.metadata_invalid
self.elec.set_electric_component_channel()
self.assertEqual(self.elec.component, 'Ex')
self.assertEqual(self.elec.channel_name, 'LQN')
self.assertSetEqual(self.elec.metadata_invalid, set())
def test_set_electrode_info_default(self):
"""Test basic functionality of set_electrode_info."""
self.elec.instrument_specs = 'Borin STELTH 4 - Silver-Silver Chloride'
self.elec.set_electrode_info()
self.assertEqual(self.elec.instrument_manufacturer, "Borin")
self.assertEqual(self.elec.instrument_model, "STELTH 4")
self.assertEqual(self.elec.instrument_type, "Silver-Silver Chloride")
def test_set_electrode_info_user_defined(self):
"""Test basic functionality of set_electrode_info."""
self.elec.instrument_specs = 'Manufacturer: a - Model: b - Type: c'
self.elec.set_electrode_info()
self.assertEqual(self.elec.instrument_manufacturer, "a")
self.assertEqual(self.elec.instrument_model, "b")
self.assertEqual(self.elec.instrument_type, "c")
def test_validate_e_property_undefined(self):
"""Test basic functionality of validate_e_property."""
e_prop, min_range, max_range, units = self.param
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.validate_e_property(self.param, None)
msg = ("The {0} for run '{1}' and channel number '{2}' should be a "
"float. {3}".format(e_prop, self.elec.run_id,
self.elec.channel_number, MSG_E_CHA))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_e_property_erroneous_type(self):
"""Test basic functionality of validate_e_property."""
e_prop, min_range, max_range, units = self.param
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.validate_e_property(self.param, 'a')
msg = ("The {0} for run '{1}' and channel number '{2}' should be a "
"float. {3}".format(e_prop, self.elec.run_id,
self.elec.channel_number, MSG_E_CHA))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_e_property_inferior_to_min_value(self):
"""Test basic functionality of validate_e_property."""
e_prop, min_range, max_range, units = self.param
with self.assertLogs(logger, level='ERROR') as cmd:
self.elec.validate_e_property(self.param, -10)
msg = ("The {0} for run '{1}' and for channel number '{2}' should be "
"positive.".format(e_prop, self.elec.run_id, self.elec.channel_number))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_validate_e_property_superior_to_max_value(self):
"""Test basic functionality of validate_e_property."""
e_prop, min_range, max_range, units = self.param
with self.assertLogs(logger, level='WARNING') as cmd:
self.elec.validate_e_property(self.param, 5000)
msg = ("The {0} for run '{1}' and for channel number '{2}' should be "
"less than {3}{4}. A {0} > {3}{4} is indicative of poor "
"contact.".format(e_prop, self.elec.run_id,
self.elec.channel_number, max_range, units))
self.assertEqual(cmd.output, [":".join(['WARNING', SCR_DIR, msg])])
def test_validate_e_property_valid(self):
"""Test basic functionality of validate_e_property."""
self.assertTrue(self.elec.validate_e_property(self.param, 500))
def test_validate_dipole_length_empty_data_inputs(self):
"""Test basic functionality of validate_dipole_length."""
user_input = 99
data_input = {}
self.assertTrue(self.elec.validate_dipole_length(user_input, data_input))
def test_validate_dipole_length_mismatch_between_user_and_data_inputs(self):
"""Test basic functionality of validate_dipole_length."""
user_input = 99
data_input = {'E1': 101, 'E2': 102, 'E3': 99, 'E4': 95}
with self.assertLogs(logger, level='ERROR') as cmd:
valid = self.elec.validate_dipole_length(user_input, data_input)
msg = ("The dipole length for run '{0}' and channel number '{1}' does "
"not match the dipole length value that was configured using the "
"LEMI424 logger in the field (99 != 101)! Update value using the "
"GUI.".format(self.elec.run_id, self.elec.channel_number))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
self.assertFalse(valid)
def test_validate_dipole_length_match_between_user_and_data_inputs(self):
"""Test basic functionality of validate_dipole_length."""
user_input = 99
data_input = {'E1': 99, 'E2': 102, 'E3': 99, 'E4': 95}
self.assertTrue(self.elec.validate_dipole_length(user_input, data_input))
class TestMagnetic(unittest.TestCase):
"""Test suite for Magnetic data class."""
def setUp(self):
"""Set up test fixtures"""
self.mag = Magnetic()
def test_set_magnetic_info_default(self):
"""Test basic functionality of set_magnetic_info."""
self.mag.instrument_specs = 'LEMI-039'
self.mag.set_magnetic_info()
self.assertEqual(self.mag.instrument_manufacturer, "LEMI LLC.")
self.assertEqual(self.mag.instrument_model, "LEMI-039")
self.assertEqual(self.mag.instrument_type,
"3-component analog magnetometer")
def test_set_magnetic_info_user_defined(self):
"""Test basic functionality of set_magnetic_info."""
self.mag.instrument_specs = 'Manufacturer: a - Model: b - Type: c'
self.mag.set_magnetic_info()
self.assertEqual(self.mag.instrument_manufacturer, "a")
self.assertEqual(self.mag.instrument_model, "b")
self.assertEqual(self.mag.instrument_type, "c")
class TestAuxiliary(unittest.TestCase):
"""Test suite for Auxiliary data class."""
def setUp(self):
"""Set up test fixtures"""
self.aux = Auxiliary()
def test_set_auxiliary_info(self):
"""Test basic functionality of set_auxiliary_info."""
self.aux.set_auxiliary_info()
self.assertEqual(self.aux.instrument_manufacturer, "LEMI LLC.")
self.assertEqual(self.aux.instrument_model, "LEMI-424")
self.assertEqual(self.aux.instrument_type, "long-period 32-bit")
def test_set_data_type(self):
"""Test basic functionality of set_data_type."""
self.aux.set_data_type()
self.assertEqual(self.aux.data_type, "HEALTH")
def test_set_measurement_azimuth(self):
"""Test basic functionality of set_measurement_azimuth."""
self.aux.set_measurement_azimuth()
self.assertEqual(self.aux.measurement_azimuth, 0.0)
def test_set_measurement_tilt(self):
"""Test basic functionality of set_measurement_tilt."""
self.aux.set_measurement_tilt()
self.assertEqual(self.aux.measurement_tilt, 0.0)
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)
lemi_data.prep_data()
self.data = lemi_data
self.path2metadata = TEST_DIR.joinpath("METADATA")
self.path2metadata_fail_1 = TEST_DIR.joinpath("METADATA_NO_FIELD_SHEET")
self.path2metadata_fail_2 = TEST_DIR.joinpath("METADATA_CORRUPTED")
self.file = self.path2metadata.joinpath('LEMI_Install_Sheet.xlsx')
self.metadata = LemiMetadata(self.path2metadata, OUTPUT_XML, self.data)
file_ = self.path2metadata.joinpath('metadata_fields.pkl')
with open(file_, 'rb') as fin:
self.metadata_fields_no_reformat = pickle.load(fin)
file_ = self.path2metadata.joinpath('metadata_fields_reformatted.pkl')
with open(file_, 'rb') as fin:
self.metadata_fields = pickle.load(fin)
def test_scan_path2metadata_field_sheet_provided(self):
"""Test basic functionality of scan_path2metadata."""
metadata = LemiMetadata(self.path2metadata, OUTPUT_XML, self.data)
self.assertEqual(*metadata.filenames, self.file)
def test_scan_path2metadata_no_field_sheet_provided(self):
"""Test basic functionality of scan_path2metadata."""
with self.assertLogs(logger, level='ERROR') as cmd:
LemiMetadata(self.path2metadata_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.path2metadata_fail_1))
self.assertEqual(cmd.output, [":".join(['ERROR', SCR_DIR, msg])])
def test_get_metadata_field_pass(self):
"""Test basic functionality of get_metadata_field."""
workbook = openpyxl.load_workbook(self.file, data_only=True)
sheet = workbook.active
self.assertEqual(LemiMetadata.get_metadata_field(sheet, 'B3'), 'EM')
workbook.close()
def test_get_metadata_field_fail(self):
"""Test basic functionality of get_metadata_field."""
workbook = openpyxl.load_workbook(self.file, data_only=True)
sheet = workbook.active
with self.assertLogs(logger, level='ERROR'):
self.assertIsNone(LemiMetadata.get_metadata_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.metadata.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."""
metadata = LemiMetadata(self.path2metadata_fail_2, OUTPUT_XML, self.data)
file_ = self.path2metadata_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:
metadata.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_dataclass2dict(self):
"""Test basic functionality of dataclass2dict."""
channel = self.metadata.init_channel_metadata_properties('electric', 'a')
self.assertDictEqual(LemiMetadata.dataclass2dict(channel),
{'elevation': 2201.7248070562296,
'latitude': 34.048390237958735,
'longitude': -107.12844894427468,
'instrument_manufacturer': None,
'instrument_model': None,
'instrument_specs': None,
'instrument_type': None,
'measurement_azimuth': None,
'measurement_tilt': None,
'sample_rate': 1.0,
'channel_number': None,
'contact_resistance_end': None,
'contact_resistance_start': None,
'dc_end': None,
'dc_start': None,
'dipole_length': None,
'negative_electrode_direction': None,
'negative_electrode_serial_number': None,
'positive_electrode_direction': None,
'positive_electrode_serial_number': None})
def test_filter_channel(self):
"""Test basic functionality of filter_channel."""
self.assertListEqual(self.metadata.filter_channel('auxiliary', 'a'), [])
def test_update_channelnumber(self):
"""Test basic functionality of update_channelnumber."""
channel = self.metadata.init_channel_metadata_properties('electric', 'a')
self.assertIsNone(channel.channel_number)
updated_channel = LemiMetadata.update_channelnumber(channel, 'E1')
self.assertEqual(updated_channel.channel_number, 'E1')
def test_for_gui(self):
"""Test basic functionality of for_gui."""
file_ = self.path2metadata.joinpath('gui_metadata_not_populated.pkl')
with open(file_, 'rb') as fin:
expected = pickle.load(fin)
self.assertDictEqual(self.metadata.for_gui, expected)
def test_reformat_run(self):
"""Test basic functionality of reformat_run."""
run_fields = self.metadata_fields_no_reformat['Run']
file_ = self.path2metadata.joinpath('run_fields_reformatted.pkl')
with open(file_, 'rb') as fin:
reformatted = pickle.load(fin)
self.assertDictEqual(self.metadata.reformat_run(run_fields), reformatted)
def test_reformat_electric(self):
"""Test basic functionality of reformat_electric."""
electric_fields = self.metadata_fields_no_reformat['Electric']
file_ = self.path2metadata.joinpath('electric_fields_reformatted.pkl')
with open(file_, 'rb') as fin:
reformatted = pickle.load(fin)
self.assertDictEqual(self.metadata.reformat_electric(electric_fields), reformatted)
def test_reformat_magnetic(self):
"""Test basic functionality of reformat_magnetic."""
magnetic_fields = self.metadata_fields_no_reformat['Magnetic']
file_ = self.path2metadata.joinpath('magnetic_fields_reformatted.pkl')
with open(file_, 'rb') as fin:
reformatted = pickle.load(fin)
self.assertDictEqual(self.metadata.reformat_magnetic(magnetic_fields), reformatted)
def test_reformat_metadata_dict(self):
"""Test basic functionality of reformat_metadata_dict."""
self.assertDictEqual(self.metadata.reformat_metadata_dict(self.metadata_fields_no_reformat),
self.metadata_fields)
def test_from_field_sheets_failed_to_open(self):
"""Test basic functionality of from_field_sheets"""
self.metadata.filenames = [self.path2metadata_fail_2.joinpath('LEMI_Install_Sheet_1.xlsx')]
with self.assertLogs(logger, level='ERROR') as cmd:
self.metadata.from_field_sheets
msg = ("Failed to open field sheet: {} - Skipping file - Exception: "
"File is not a zip file".format(self.metadata.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.metadata.filenames = [self.file] * 2
with self.assertLogs(logger, level='WARNING') as cmd:
self.metadata.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.metadata.from_field_sheets, self.metadata_fields)
def test_init_run_metadata_properties(self):
"""Test basic functionality of init_run_metadata_properties."""
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.metadata.init_run_metadata_properties()
for ind, run in enumerate(self.metadata.run):
self.assertEqual(run.time_period_start, time_period_starts[ind])
self.assertEqual(run.time_period_end, time_period_ends[ind])
self.assertEqual(run.resource_id, 'mt.run.id:' + self.metadata.run_list[ind])
def test_init_channel_metadata_properties_electric(self):
"""Test basic functionality of init_channel_metadata_properties."""
channel = self.metadata.init_channel_metadata_properties('electric', 'a')
self.assertIsInstance(channel, Electric)
self.assertEqual(channel.elevation, self.metadata.data_stats['elevation'])
self.assertEqual(channel.latitude, self.metadata.data_stats['latitude'])
self.assertEqual(channel.longitude, self.metadata.data_stats['longitude'])
self.assertEqual(channel.run_id, 'a')
def test_init_channel_metadata_properties_magnetic(self):
"""Test basic functionality of init_channel_metadata_properties."""
channel = self.metadata.init_channel_metadata_properties('magnetic', 'a')
self.assertIsInstance(channel, Magnetic)
self.assertEqual(channel.elevation, self.metadata.data_stats['elevation'])
self.assertEqual(channel.latitude, self.metadata.data_stats['latitude'])
self.assertEqual(channel.longitude, self.metadata.data_stats['longitude'])
self.assertEqual(channel.run_id, 'a')
def test_init_channel_metadata_properties_auxiliary(self):
"""Test basic functionality of init_channel_metadata_properties."""
channel = self.metadata.init_channel_metadata_properties('auxiliary', 'a')
self.assertIsInstance(channel, Auxiliary)
self.assertEqual(channel.elevation, self.metadata.data_stats['elevation'])
self.assertEqual(channel.latitude, self.metadata.data_stats['latitude'])
self.assertEqual(channel.longitude, self.metadata.data_stats['longitude'])
self.assertEqual(channel.run_id, 'a')