Add tests

tests/test_model/test_reftek/test_core.py

+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)

tests/test_model/test_reftek/test_header.py

+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)
+

tests/test_model/test_reftek/test_packet_readers.py

+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)

tests/test_model/test_reftek/test_reftek_helper.py

+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))
+