diff --git a/tests/test_model/test_handling_data.py b/tests/test_model/test_handling_data.py index d92a87533465ee5c992dd9afe4f2c1a3b05d5538..62af41beb5678280764d8ae464ae3f2e4a37f4f4 100644 --- a/tests/test_model/test_handling_data.py +++ b/tests/test_model/test_handling_data.py @@ -13,10 +13,13 @@ from sohstationviewer.model.handling_data import ( readSOHTrace, readMPTrace, readWaveformTrace, + readWaveformMSeed, ) from sohstationviewer.model.reftek.from_rt2ms.core import Reftek130 TEST_DATA_DIR = Path(__file__).parent.parent.joinpath('test_data') + + # tempfile.tempdir = './tempdir' @@ -31,16 +34,19 @@ class TestHandlingData(TestCase): cls.rt130_trace = cls.rt130_stream[0] q330_dir = TEST_DATA_DIR.joinpath('Q330-sample/day_vols_AX08') - cls.q330_file = q330_dir.joinpath('AX08.XA..VKI.2021.186') - cls.mseed_stream = read_ms(cls.q330_file) - cls.mseed_trace = cls.mseed_stream[0] + cls.q330_soh_file = q330_dir.joinpath('AX08.XA..VKI.2021.186') + cls.mseed_soh_stream = read_ms(cls.q330_soh_file) + cls.mseed_soh_trace = cls.mseed_soh_stream[0] + cls.q330_waveform_file = q330_dir.joinpath('AX08.XA..LHE.2021.186') + cls.mseed_waveform_stream = read_ms(cls.q330_waveform_file) + cls.mseed_waveform_trace = cls.mseed_waveform_stream[0] # @expectedFailure # def test_read_sohmseed(self): # self.fail() def test_read_soh_trace_processed_trace_have_all_needed_info(self): - processed_trace = readSOHTrace(self.mseed_trace) + processed_trace = readSOHTrace(self.mseed_soh_trace) with self.subTest('test_processed_trace_have_all_needed_info'): expected_key_list = [ 'chanID', @@ -55,7 +61,7 @@ class TestHandlingData(TestCase): ) def test_read_soh_trace_times_calculated_correctly(self): - processed_trace = readSOHTrace(self.mseed_trace) + processed_trace = readSOHTrace(self.mseed_soh_trace) if isclose(processed_trace['startTmEpoch'], 0, abs_tol=0.0001): self.assertAlmostEqual(processed_trace['times'][0], 0) else: @@ -175,3 +181,97 @@ class TestHandlingData(TestCase): # @skip # def test_find_tpstm(self): # self.fail() + + +class TestReadWaveformMSeed(TestCase): + @classmethod + def setUpClass(cls) -> None: + q330_dir = TEST_DATA_DIR.joinpath('Q330-sample/day_vols_AX08') + cls.q330_waveform_file = q330_dir.joinpath('AX08.XA..LHE.2021.186') + cls.mseed_waveform_stream = read_ms(cls.q330_waveform_file) + cls.mseed_waveform_trace = cls.mseed_waveform_stream[0] + + def setUp(self) -> None: + self.station_id = self.mseed_waveform_trace.stats['station'] + self.channel_id = self.mseed_waveform_trace.stats['channel'] + # This list is only ever written to, so we can keep it empty + self.traces_info = [] + + self.temp_dir = tempfile.TemporaryDirectory() + + patcher = patch( + 'sohstationviewer.model.handling_data.readWaveformTrace') + self.addCleanup(patcher.stop) + self.mock_readWaveformTrace = patcher.start() + self.mock_readWaveformTrace.return_value = { + 'startTmEpoch': 0, + 'endTmEpoch': 0, + } + + def tearDown(self) -> None: + self.temp_dir.cleanup() + + def test_all_traces_are_processed(self): + data_time = [12512512, 12215125121] + readWaveformMSeed( + self.q330_waveform_file, self.q330_waveform_file.name, + self.station_id, self.channel_id, self.traces_info, data_time, + self.temp_dir.name) + self.assertEqual(len(self.traces_info), len(self.mseed_waveform_stream)) + + def test_readWaveformTrace_called(self): + data_time = [12512512, 12215125121] + readWaveformMSeed( + self.q330_waveform_file, self.q330_waveform_file.name, + self.station_id, self.channel_id, self.traces_info, data_time, + self.temp_dir.name) + self.assertTrue(self.mock_readWaveformTrace.called) + + def test_start_data_time_earlier_than_earliest_start_time(self): + self.mock_readWaveformTrace.return_value = { + 'startTmEpoch': 51251, + 'endTmEpoch': 2623623, + } + start_time = 0 + data_time = [start_time, 1625532949] + + readWaveformMSeed( + self.q330_waveform_file, self.q330_waveform_file.name, + self.station_id, self.channel_id, self.traces_info, data_time, + self.temp_dir.name) + self.assertEqual(data_time[0], start_time) + + def test_end_data_time_later_than_latest_start_time(self): + # End time set to be the last second of 9999 so as to be later than the + # end time of any test data. + self.mock_readWaveformTrace.return_value = { + 'startTmEpoch': 51251, + 'endTmEpoch': 2623623, + } + end_time = 253402326000 + data_time = [1625443222, end_time] + + readWaveformMSeed( + self.q330_waveform_file, self.q330_waveform_file.name, + self.station_id, self.channel_id, self.traces_info, data_time, + self.temp_dir.name) + self.assertEqual(data_time[1], end_time) + + def test_data_time_is_between_earliest_start_and_latest_end_time(self): + self.mock_readWaveformTrace.return_value = { + 'startTmEpoch': 51251, + 'endTmEpoch': 2623623, + } + start_time = 512579 + end_time = 2623616 + data_time = [start_time, end_time] + + expected_updated_data_time = [51251, 2623623] + readWaveformMSeed( + self.q330_waveform_file, self.q330_waveform_file.name, + self.station_id, self.channel_id, self.traces_info, data_time, + self.temp_dir.name) + self.assertEqual(data_time, expected_updated_data_time) + + # Test that dataTime is updated correctly +