sohstationviewer/model/mseed/mseed.py

@@ -5,9 +5,10 @@ import functools
 import operator
 import os
 from pathlib import Path
-from typing import Dict, Tuple, List, Set
+from typing import Dict, Tuple, List, Set, Optional
 
-from obspy.core import Stream
+import numpy as np
+from obspy.core import Stream, UTCDateTime
 
 from sohstationviewer.conf import constants
 from sohstationviewer.controller.util import validate_file
@@ -25,6 +26,7 @@ class MSeed(DataTypeModel):
     read and process mseed file into object with properties can be used to
     plot SOH data, mass position data, waveform data and gaps
     """
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         # channels: available channels got from files among
@@ -456,3 +458,107 @@ class MSeed(DataTypeModel):
                 'Q330 log data is malformed: '
                 'Last GPS timemark is not at expected position.'
             )
+
+    def get_gps_channel_prefix(self, data_type: str) -> Optional[str]:
+        """
+        Determine the first letter of the GPS channel name for the current data
+        set. Only applicable to Centaur and Pegasus data sets. Q330 data sets
+        store GPS data in the LOG channel.
+
+        Centaur and Pegasus data sets share the last two character for GPS
+        channels. However, Pegasus GPS channels start with an 'V', while
+        Centaur GPS channels all start with a 'G'.
+
+        :param data_type: the type of data contained in the current data set
+        :return: the first character of the GPS channels. Is 'V' if data_type
+             is 'Pegasus', or 'G' if data_type is 'Centaur'
+        """
+        gps_prefix = None
+        if data_type == 'Pegasus':
+            gps_prefix = 'V'
+        elif data_type == 'Centaur':
+            gps_prefix = 'G'
+        else:
+            gps_suffixes = {'NS', 'LA', 'LO', 'EL'}
+            pegasus_gps_channels = {'V' + suffix for suffix in gps_suffixes}
+            centaur_gps_channels = {'G' + suffix for suffix in gps_suffixes}
+
+            # Determine the GPS channels by checking if the current data set
+            # has all the GPS channels of a data type.
+            if pegasus_gps_channels & self.channels == pegasus_gps_channels:
+                gps_prefix = 'V'
+            elif centaur_gps_channels & self.channels == centaur_gps_channels:
+                gps_prefix = 'G'
+            else:
+                msg = "Can't detect GPS channels."
+                self.track_info(msg, LogType.ERROR)
+        return gps_prefix
+
+    def get_chan_soh_trace_as_dict(self, chan: str) -> Dict[float, float]:
+        """
+        Get the data of a channel as a dictionary mapping a time to its
+        corresponding data.
+        :param chan: the channel name
+        :return: a dict that maps the times of channel chan to its data
+        """
+        chan_data = self.soh_data[self.selected_key][chan]
+        data = chan_data['orgTrace']['data']
+        data = data[~data.mask]
+        times = chan_data['orgTrace']['times']
+        times = times[~times.mask]
+        data_dict = {time: data
+                     for time, data in np.column_stack((times, data))}
+        return data_dict
+
+    def get_gps_data_pegasus_centaur(self, data_type: str):
+        """
+        Extract GPS data of the current data set and store it in
+        self.gps_points. Only applicable to Centaur and Pegasus data sets. Q330
+        data sets store GPS data in the LOG channel.
+
+        :param data_type: data type of the current data set
+        """
+        # Caching GPS data in dictionaries for faster access. In the algorithm
+        # below, we need to access the data associated with a time. If we leave
+        # the times and data in arrays, we will need to search for the index of
+        # the specified time in the times array, which takes O(N) time. The
+        # algorithm then repeats this step n times, which gives us a total
+        # complexity of O(n^2). Meanwhile, if we cache the times and data in
+        # a dictionary, we only need to spend O(n) time building the cache and
+        # O(n) time accessing the cache, which amounts to O(n) time in total.
+        gps_prefix = self.get_gps_channel_prefix(data_type)
+
+        ns_dict = self.get_chan_soh_trace_as_dict(gps_prefix + 'NS')
+        la_dict = self.get_chan_soh_trace_as_dict(gps_prefix + 'LA')
+        lo_dict = self.get_chan_soh_trace_as_dict(gps_prefix + 'LO')
+        el_dict = self.get_chan_soh_trace_as_dict(gps_prefix + 'EL')
+
+        for time, num_sats_used in ns_dict.items():
+            # There is no channel for GPS fix type in Pegasus and Centaur data,
+            # so we are storing it as not available.
+            fix_type = 'N/A'
+            lat = la_dict.get(time, None)
+            long = lo_dict.get(time, None)
+            height = el_dict.get(time, None)
+            if lat is None or long is None or height is None:
+                continue
+            # Convert the location data to the appropriate unit. Latitude and
+            # longitude are being converted from microdegree to degree, while
+            # height is being converted from centimeter to meter.
+            lat = lat / 1e6
+            long = long / 1e6
+            height = height / 100
+            height_unit = 'M'
+            formatted_time = UTCDateTime(time).strftime('%Y-%m-%d %H:%M:%S')
+            gps_point = GPSPoint(formatted_time, fix_type, num_sats_used, lat,
+                                 long, height, height_unit)
+            self.gps_points.append(gps_point)
+        # We only need to loop through one dictionary. If a time is not
+        # available for a channel, the GPS data point at that time would be
+        # invalid (it is missing a piece of data). Once we loop through a
+        # channel's dictionary, we know that any time not contained in that
+        # dictionary is not available for the channel. As a result, any time
+        # we pass through in the other channels after the first loop would
+        # result in an invalid GPS data point. Because we discard any invalid
+        # point, there is no point in looping through the dictionary of other
+        # channels.

tests/test_model/test_mseed/test_gps.py

-import unittest
+from unittest import TestCase
 import math
 
 from sohstationviewer.model.mseed.mseed import MSeed
+from sohstationviewer.view.util.enums import LogType
 
 
-class TestCheckGPSStatusFormatQ330(unittest.TestCase):
+class TestCheckGPSStatusFormatQ330(TestCase):
     def setUp(self) -> None:
         self.status_lines = [
             'GPS Status',
@@ -74,7 +75,7 @@ class TestCheckGPSStatusFormatQ330(unittest.TestCase):
             MSeed.check_gps_status_format_q330(self.status_lines)
 
 
-class TestExtractGPSPointQ330(unittest.TestCase):
+class TestExtractGPSPointQ330(TestCase):
     def setUp(self) -> None:
         self.gps_lines = ['GPS Status',
                           'Time: 03:37:39',
@@ -195,3 +196,58 @@ class TestExtractGPSPointQ330(unittest.TestCase):
         self.assertEqual(result.latitude, 0)
         self.assertEqual(result.longitude, 0)
         self.assertEqual(result.num_satellite_used, 0)
+
+
+class MockMSeed(MSeed):
+    """
+    This class mocks out some methods of MSeed that are used in
+    MSeed.get_gps_channel_prefix but which would make testing it more
+    cumbersome. The methods mocked out either run very long or change the GUI
+    and/or terminal in some way.
+    """
+    def __init__(self):  # noqa
+        self.notification_signal = None
+        self.tmp_dir = ''
+
+    def track_info(self, text: str, type: LogType) -> None:
+        print(text)
+
+    def __del__(self):
+        pass
+
+
+class TestGetGPSChannelPrefix(TestCase):
+    def setUp(self) -> None:
+        self.mseed_obj = MockMSeed()
+        self.mseed_obj.channels = set()
+
+    def test_pegasus_data_type(self):
+        data_type = 'Pegasus'
+        expected = 'V'
+        result = self.mseed_obj.get_gps_channel_prefix(data_type)
+        self.assertEqual(result, expected)
+
+    def test_centaur_data_type(self):
+        data_type = 'Centaur'
+        expected = 'G'
+        result = self.mseed_obj.get_gps_channel_prefix(data_type)
+        self.assertEqual(result, expected)
+
+    def test_unknown_data_type_pegasus_gps_channels(self):
+        data_type = 'Unknown'
+        self.mseed_obj.channels = {'VNS', 'VLA', 'VLO', 'VEL'}
+        expected = 'V'
+        result = self.mseed_obj.get_gps_channel_prefix(data_type)
+        self.assertEqual(expected, result)
+
+    def test_unknown_data_type_centaur_gps_channels(self):
+        data_type = 'Unknown'
+        self.mseed_obj.channels = {'GNS', 'GLA', 'GLO', 'GEL'}
+        expected = 'G'
+        result = self.mseed_obj.get_gps_channel_prefix(data_type)
+        self.assertEqual(expected, result)
+
+    def test_unknown_data_type_channels_do_not_match_either_data_type(self):
+        data_type = 'Unknown'
+        result = self.mseed_obj.get_gps_channel_prefix(data_type)
+        self.assertIsNone(result)