diff --git a/tests/test_model/test_reftek/test_core.py b/tests/test_model/test_reftek/test_core.py
new file mode 100644
index 0000000000000000000000000000000000000000..ff05396c02a3cbf743f8a96b57dd2b5cd705f24e
--- /dev/null
+++ b/tests/test_model/test_reftek/test_core.py
@@ -0,0 +1,101 @@
+import os
+import unittest
+from pathlib import Path
+
+import numpy
+import obspy.core
+from numpy.testing import assert_array_equal
+
+from sohstationviewer.model.reftek.reftek_data.core import (
+ DiscontinuousTrace,
+ Reftek130,
+)
+from sohstationviewer.model.reftek.reftek_data.header import NotRT130FileError
+
+
+class TestDiscontinuousTrace(unittest.TestCase):
+ def setUp(self) -> None:
+ data = numpy.arange(1024)
+ stub_stats = obspy.core.Stats()
+ times = numpy.arange(1024)
+ self.trace = DiscontinuousTrace(data, stub_stats, times=times)
+
+ def test_times_argument_is_stored(self):
+ self.assertTrue(hasattr(self.trace, '_times'))
+
+ def test_times_utcdatetime(self):
+ with self.assertRaises(NotImplementedError):
+ self.trace.times('utcdatetime')
+
+ def test_times_matplotlib(self):
+ with self.assertRaises(NotImplementedError):
+ self.trace.times('matplotlib')
+
+ def test_times_relative(self):
+ with self.subTest('test_relative_to_start_time'):
+ # The default start time of a trace is 0 anyhow, but we write that
+ # down explicitly for clarity.
+ self.trace.stats.starttime = obspy.core.UTCDateTime(0)
+ expected = numpy.arange(1024)
+ assert_array_equal(self.trace.times('relative'), expected)
+
+ with self.subTest('test_relative_to_given_reftime'):
+ reftime = obspy.core.UTCDateTime(0)
+ expected = numpy.arange(1024)
+ assert_array_equal(self.trace.times('relative', reftime),
+ expected)
+
+ reftime = obspy.core.UTCDateTime(1024)
+ expected = numpy.arange(-1024, 0)
+ assert_array_equal(self.trace.times('relative', reftime),
+ expected)
+
+ reftime = obspy.core.UTCDateTime(-1024)
+ expected = numpy.arange(1024, 2048)
+ assert_array_equal(self.trace.times('relative', reftime),
+ expected)
+
+ def test_times_timestamp(self):
+ expected = numpy.arange(1024)
+ assert_array_equal(self.trace.times('timestamp'), expected)
+
+
+class TestReftek130FromFile(unittest.TestCase):
+ def setUp(self) -> None:
+ self.TEST_DATA_DIR = Path(os.getcwd()).joinpath('tests/test_data')
+ self.rt130_dir = self.TEST_DATA_DIR.joinpath(
+ 'RT130-sample/2017149.92EB/2017150/92EB'
+ )
+
+ def test_rt130_file(self):
+ file = self.rt130_dir.joinpath('0/000000000_00000000')
+ rt130 = Reftek130.from_file(file)
+ self.assertIsInstance(rt130, Reftek130)
+
+ def test_rt130_soh_file(self):
+ file = self.rt130_dir.joinpath('0/000000000_00000000')
+ rt130 = Reftek130.from_file(file)
+ # The most common SOH packet type looks to be SH, so we use that as
+ # the default.
+ self.assertIn(b'SH', rt130._data['packet_type'])
+
+ def test_rt130_raw_data_file(self):
+ file = self.rt130_dir.joinpath('1/000000015_0036EE80')
+ rt130 = Reftek130.from_file(file)
+ assert_array_equal(
+ numpy.unique(numpy.sort(rt130._data['packet_type'])),
+ numpy.sort([b'EH', b'DT', b'ET'])
+ )
+
+ def test_non_rt130_file(self):
+ with self.subTest('test_file_exist'):
+ test_file = self.TEST_DATA_DIR.joinpath(
+ 'Q330-sample/day_vols_AX08/AX08.XA..HHE.2021.186'
+ )
+ with self.assertRaises(NotRT130FileError):
+ Reftek130.from_file(test_file)
+
+ with self.subTest('test_file_does_not_exist'):
+ test_file = ''
+ with self.assertRaises(FileNotFoundError):
+ Reftek130.from_file(test_file)
diff --git a/tests/test_model/test_reftek/test_header.py b/tests/test_model/test_reftek/test_header.py
new file mode 100644
index 0000000000000000000000000000000000000000..b73e3d18356e39d2efd68aeeefa5782c71f7d829
--- /dev/null
+++ b/tests/test_model/test_reftek/test_header.py
@@ -0,0 +1,74 @@
+import unittest
+
+from sohstationviewer.model.reftek.reftek_data.header import (
+ parse_rt130_time,
+ get_rt130_packet_header, NotRT130FileError,
+)
+
+
+class TestParseRT130Time(unittest.TestCase):
+ def test_time_bytes_parsed_correctly(self):
+ time_bytes = b'\x36\x01\x15\x13\x51\x35'
+ year = 15
+ result = parse_rt130_time(year, time_bytes)
+ self.assertEqual(result.julday, 360)
+ self.assertEqual(result.day, 26)
+ self.assertEqual(result.month, 12)
+ self.assertEqual(result.hour, 11)
+ self.assertEqual(result.minute, 51)
+ self.assertEqual(result.second, 35)
+ self.assertEqual(result.microsecond, 135000)
+ self.assertEqual(result.ns, 1451130695135000000)
+
+ def test_year_1900s(self):
+ time_bytes = b'\x36\x01\x15\x13\x51\x35'
+ year = 71
+ result = parse_rt130_time(year, time_bytes)
+ self.assertEqual(result.year, 1971)
+
+ def test_year_2000s(self):
+ time_bytes = b'\x36\x01\x15\x13\x51\x35'
+ year = 12
+ result = parse_rt130_time(year, time_bytes)
+ self.assertEqual(result.year, 2012)
+
+ def test_year_threshold(self):
+ with self.subTest('test_year_is_49'):
+ time_bytes = b'\x36\x01\x15\x13\x51\x35'
+ year = 49
+ result = parse_rt130_time(year, time_bytes)
+ self.assertEqual(result.year, 2049)
+ with self.subTest('test_year_is_50'):
+ time_bytes = b'\x36\x01\x15\x13\x51\x35'
+ year = 50
+ result = parse_rt130_time(year, time_bytes)
+ self.assertEqual(result.year, 1950)
+
+
+
+class TestGetRT130PacketHeader(unittest.TestCase):
+ def test_header_extracted_correctly(self):
+ header = b'DT\x12\x15\x98\xe1\x36\x01\x15\x13\x51\x35\x05\x12\x01\x11'
+ packet = header + b' ' * 1008
+ result = get_rt130_packet_header(packet)
+ self.assertEqual(result.packet_type, 'DT')
+ self.assertEqual(result.experiment_number, 12)
+ self.assertEqual(result.unit_id, '98E1')
+ self.assertEqual(result.time.ns, 1451130695135000000)
+ self.assertEqual(result.byte_count, 512)
+ self.assertEqual(result.packet_sequence, 111)
+
+ def test_packet_type_cannot_be_parsed(self):
+ packet_type = b'\x01\x02'
+ header = packet_type + b'\x11' * 14
+ packet = header + b' ' * 1008
+ with self.assertRaises(NotRT130FileError):
+ get_rt130_packet_header(packet)
+
+ def test_packet_type_is_not_valid(self):
+ packet_type = b'AB'
+ header = packet_type + b'\x11' * 14
+ packet = header + b' ' * 1008
+ with self.assertRaises(NotRT130FileError):
+ get_rt130_packet_header(packet)
+
diff --git a/tests/test_model/test_reftek/test_packet_readers.py b/tests/test_model/test_reftek/test_packet_readers.py
new file mode 100644
index 0000000000000000000000000000000000000000..180e03b445f6f71e92ff4f3dca9a22ce5f4f5f5c
--- /dev/null
+++ b/tests/test_model/test_reftek/test_packet_readers.py
@@ -0,0 +1,184 @@
+import unittest
+from unittest.mock import patch
+
+from sohstationviewer.model.mseed_data.record_reader_helper import Unpacker
+from sohstationviewer.model.reftek.reftek_data.packet import \
+ eh_et_payload_end_in_packet
+from sohstationviewer.model.reftek.reftek_data.packet_readers import \
+ (
+ decode_uncompressed, decode_compressed, read_dt_packet, read_eh_et_packet,
+ read_soh_packet,
+)
+from sohstationviewer.model.reftek.reftek_data.packets import SOHExtendedHeader
+
+unpacker = Unpacker('>')
+
+
+class TestDecodeFunctions(unittest.TestCase):
+ def setUp(self) -> None:
+ self.header = b' ' * 24
+
+ def test_decode_uncompressed_format_16(self):
+ data_format = '16'
+ with self.subTest('test_positive_number'):
+ first_data_point_byte = b'\x06\x19'
+ data_filler = b' ' * 998
+ packet = self.header + first_data_point_byte + data_filler
+ actual = decode_uncompressed(packet, data_format, unpacker)
+ expected = 1561
+ self.assertEqual(actual, expected)
+ with self.subTest('test_negative_number'):
+ first_data_point_byte = b'\xf9\xe4'
+ data_filler = b' ' * 998
+ packet = self.header + first_data_point_byte + data_filler
+ actual = decode_uncompressed(packet, data_format, unpacker)
+ expected = -1564
+ self.assertEqual(actual, expected)
+
+ def test_decode_uncompressed_format_32(self):
+ data_format = '32'
+ with self.subTest('test_positive_number'):
+ first_data_point_byte = b'\x03\xc5\x4e\x9a'
+ data_filler = b' ' * 996
+ packet = self.header + first_data_point_byte + data_filler
+ actual = decode_uncompressed(packet, data_format, unpacker)
+ expected = 63262362
+ self.assertEqual(actual, expected)
+ with self.subTest('test_negative_number'):
+ first_data_point_byte = b'\xf6\xac\xba\x00'
+ data_filler = b' ' * 996
+ packet = self.header + first_data_point_byte + data_filler
+ actual = decode_uncompressed(packet, data_format, unpacker)
+ expected = -156452352
+ self.assertEqual(actual, expected)
+
+ def test_decode_uncompressed_format_33(self):
+ data_format = '33'
+ with self.subTest('test_positive_number'):
+ first_data_point_byte = b'\x03\xc5\x4e\x9a'
+ data_filler = b' ' * 996
+ packet = self.header + first_data_point_byte + data_filler
+ actual = decode_uncompressed(packet, data_format, unpacker)
+ expected = 63262362
+ self.assertEqual(actual, expected)
+ with self.subTest('test_negative_number'):
+ first_data_point_byte = b'\xf6\xac\xba\x00'
+ data_filler = b' ' * 996
+ packet = self.header + first_data_point_byte + data_filler
+ actual = decode_uncompressed(packet, data_format, unpacker)
+ expected = -156452352
+ self.assertEqual(actual, expected)
+
+ def test_decode_compressed(self):
+ data_format = 'C0'
+ filler = b' ' * 40
+ first_frame_code = b'\x00\x11\x11\x11'
+ start_data_point_byte = b'0000'
+ header = self.header + filler
+ bytes_before_data = header + first_frame_code + start_data_point_byte
+ with self.subTest('test_positive_number'):
+ end_point_byte = b'\x03\xc5\x4e\x9a'
+ data_filler = b' ' * 952
+ packet = bytes_before_data + end_point_byte + data_filler
+ actual = decode_compressed(packet, data_format, unpacker)
+ expected = 63262362
+ self.assertEqual(actual, expected)
+ with self.subTest('test_negative_number'):
+ end_point_byte = b'\xf6\xac\xba\x00'
+ data_filler = b' ' * 952
+ packet = bytes_before_data + end_point_byte + data_filler
+ actual = decode_compressed(packet, data_format, unpacker)
+ expected = -156452352
+ self.assertEqual(actual, expected)
+
+
+class TestReadDTPacket(unittest.TestCase):
+ def setUp(self) -> None:
+ self.header = b' ' * 16
+ # We only test if the correct method is used to extract the data point,
+ # so the data can be anything we want.
+ self.data = b' ' * 1000
+
+ uncompressed_patcher = patch(
+ 'sohstationviewer.model.reftek.reftek_data.packet_readers.'
+ 'decode_uncompressed'
+ )
+ compressed_patcher = patch(
+ 'sohstationviewer.model.reftek.reftek_data.packet_readers.'
+ 'decode_compressed'
+ )
+ self.mock_uncompressed = uncompressed_patcher.start()
+ self.mock_compressed = compressed_patcher.start()
+ self.addCleanup(uncompressed_patcher.stop)
+ self.addCleanup(compressed_patcher.stop)
+
+ def test_extended_header_is_extracted_correctly(self):
+ extended_header_bytes = b'\x01\x11\x01\x02\x05\x00\x00\xc0'
+ packet = self.header + extended_header_bytes + self.data
+ extended_header, _ = read_dt_packet(packet, unpacker)
+ self.assertEqual(extended_header.event_number, 111)
+ self.assertEqual(extended_header.data_stream_number, 1)
+ self.assertEqual(extended_header.channel_number, 2)
+ self.assertEqual(extended_header.number_of_samples, 500)
+ self.assertEqual(extended_header.flags, 0)
+ self.assertEqual(extended_header.data_format, 'C0')
+
+ def test_data_point_extracted_with_correct_method(self):
+ with self.subTest('test_uncompressed_packet'):
+ extended_header_bytes = b'\x01\x11\x01\x02\x05\x00\x00\x16'
+ packet = self.header + extended_header_bytes + self.data
+ read_dt_packet(packet, unpacker)
+ self.assertTrue(self.mock_uncompressed.called)
+ self.assertFalse(self.mock_compressed.called)
+
+ self.mock_uncompressed.reset_mock()
+ self.mock_compressed.reset_mock()
+
+ with self.subTest('test_compressed_packet'):
+ extended_header_bytes = b'\x01\x11\x01\x02\x05\x00\x00\xc0'
+ packet = self.header + extended_header_bytes + self.data
+ read_dt_packet(packet, unpacker)
+ self.assertTrue(self.mock_compressed.called)
+ self.assertFalse(self.mock_uncompressed.called)
+
+
+class TestReadEHETPacket(unittest.TestCase):
+ def setUp(self) -> None:
+ header = b' ' * 16
+ extended_header_bytes = b'\x01\x11\x01\x00\x00\x00\x00\xc0'
+ # We only care about the length of the payload (the content is dealt
+ # with somewhere else), and so it can contain dummy data.
+ payload = b' ' * 1000
+ self.packet = header + extended_header_bytes + payload
+
+ def test_extended_header_is_extracted_correctly(self):
+ extended_header, _ = read_eh_et_packet(self.packet, unpacker)
+ self.assertEqual(extended_header.event_number, 111)
+ self.assertEqual(extended_header.data_stream_number, 1)
+ self.assertEqual(extended_header.channel_number, 0)
+ self.assertEqual(extended_header.number_of_samples, 0)
+ self.assertEqual(extended_header.flags, 0)
+ self.assertEqual(extended_header.data_format, 'C0')
+
+ def test_payload_extracted_correctly(self):
+ _, payload = read_eh_et_packet(self.packet, unpacker)
+ self.assertEqual(len(payload), eh_et_payload_end_in_packet - 24)
+
+
+class TestReadSOHPacket(unittest.TestCase):
+ """
+ Test suite for packet_readers.read_soh_packet. We only test that the
+ function has the correct interface, seeing as the intended purpose of this
+ method is to be compatible with packet_readers.read_dt_packet and
+ packet_readers.read_eh_et_packet interface-wise.
+ """
+ def test_correct_interface(self):
+ packet = b' ' * 1024
+ extended_header, payload = read_soh_packet(packet, unpacker)
+ self.assertIsInstance(extended_header, SOHExtendedHeader)
+ self.assertIsInstance(payload, bytes)
+
+ def test_payload_has_correct_length(self):
+ packet = b' ' * 1024
+ extended_header, payload = read_soh_packet(packet, unpacker)
+ self.assertEqual(len(payload), 1000)
diff --git a/tests/test_model/test_reftek/test_reftek_helper.py b/tests/test_model/test_reftek/test_reftek_helper.py
new file mode 100644
index 0000000000000000000000000000000000000000..ecf16f80e5d43b44e737371b70d58296f2fab525
--- /dev/null
+++ b/tests/test_model/test_reftek/test_reftek_helper.py
@@ -0,0 +1,107 @@
+import os
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from obspy.io.reftek.packet import PACKET_FINAL_DTYPE
+
+from sohstationviewer.model.mseed_data.record_reader_helper import Unpacker
+from sohstationviewer.model.reftek.reftek_data.header import NotRT130FileError
+from sohstationviewer.model.reftek.reftek_data.packet_readers import \
+ (
+ read_eh_et_packet, read_dt_packet, read_soh_packet,
+)
+from sohstationviewer.model.reftek.reftek_data.packets import (
+ SOHPacket,
+ EHETPacket, DTPacket,
+)
+from sohstationviewer.model.reftek.reftek_data.reftek_helper import \
+ (
+ read_rt130_file, convert_packet_to_obspy_format,
+)
+
+unpacker = Unpacker('>')
+
+
+class TestReadRT130File(unittest.TestCase):
+ def setUp(self) -> None:
+ self.TEST_DATA_DIR = Path(os.getcwd()).joinpath('tests/test_data')
+ self.rt130_dir = self.TEST_DATA_DIR.joinpath(
+ 'RT130-sample/2017149.92EB/2017150/92EB'
+ )
+
+ eh_et_patcher = patch(
+ 'sohstationviewer.model.reftek.reftek_data.reftek_helper.'
+ 'read_eh_et_packet',
+ wraps=read_eh_et_packet
+ )
+ self.mock_read_eh_et = eh_et_patcher.start()
+ self.addCleanup(eh_et_patcher.stop)
+
+ dt_patcher = patch(
+ 'sohstationviewer.model.reftek.reftek_data.reftek_helper.'
+ 'read_dt_packet',
+ wraps=read_dt_packet
+ )
+ self.mock_read_dt = dt_patcher.start()
+ self.addCleanup(dt_patcher.stop)
+
+ soh_patcher = patch(
+ 'sohstationviewer.model.reftek.reftek_data.reftek_helper.'
+ 'read_soh_packet',
+ wraps=read_soh_packet
+ )
+ self.mock_read_soh = soh_patcher.start()
+ self.addCleanup(soh_patcher.stop)
+
+ def test_rt130_soh_file(self):
+ file = self.rt130_dir.joinpath('0/000000000_00000000')
+ packets = read_rt130_file(file, unpacker)
+ self.assertTrue(
+ all(isinstance(packet, SOHPacket) for packet in packets)
+ )
+ self.assertTrue(self.mock_read_soh.called)
+ self.assertFalse(self.mock_read_dt.called)
+ self.assertFalse(self.mock_read_eh_et.called)
+
+ def test_rt130_raw_data_file(self):
+ file = self.rt130_dir.joinpath('1/000000015_0036EE80')
+ packets = read_rt130_file(file, unpacker)
+ self.assertTrue(all(
+ isinstance(packet, EHETPacket) or isinstance(packet, DTPacket)
+ for packet in packets)
+ )
+ self.assertFalse(self.mock_read_soh.called)
+ self.assertTrue(self.mock_read_dt.called)
+ self.assertTrue(self.mock_read_eh_et.called)
+
+ def test_non_rt130_file(self):
+ with self.subTest('test_file_exist'):
+ file = self.TEST_DATA_DIR.joinpath(
+ 'Q330-sample/day_vols_AX08/AX08.XA..HHE.2021.186'
+ )
+ with self.assertRaises(NotRT130FileError):
+ read_rt130_file(file, unpacker)
+
+ with self.subTest('test_file_does_not_exist'):
+ file = ''
+ with self.assertRaises(FileNotFoundError):
+ read_rt130_file(file, unpacker)
+
+
+class TestConvertPacketToObspyFormat(unittest.TestCase):
+ def setUp(self) -> None:
+ TEST_DATA_DIR = Path(os.getcwd()).joinpath('tests/test_data')
+ rt130_dir = TEST_DATA_DIR.joinpath(
+ 'RT130-sample/2017149.92EB/2017150/92EB'
+ )
+ file = rt130_dir.joinpath('1/000000015_0036EE80')
+ self.packet = read_rt130_file(file, unpacker)[0]
+
+ def test_all_needed_fields_are_available(self):
+ converted_packet = convert_packet_to_obspy_format(
+ self.packet, unpacker
+ )
+
+ self.assertEqual(len(converted_packet), len(PACKET_FINAL_DTYPE))
+