Implement extracting Q330 GPS data

sohstationviewer/model/data_type_model.py

@@ -3,12 +3,13 @@ import os
 
 from tempfile import mkdtemp
 import shutil
-from typing import List, Tuple, Dict, Optional, Union
+from typing import Optional,
 
 from PySide2 import QtCore
 
 from sohstationviewer.controller.util import displayTrackingInfo
 from sohstationviewer.conf import constants
+from sohstationviewer.model.gps_point import GPSPoint
 from sohstationviewer.view.util.enums import LogType
 from sohstationviewer.database.process_db import execute_db
 
@@ -234,6 +235,8 @@ class DataTypeModel():
         self._pauser = QtCore.QSemaphore()
         self.pause_response = None
 
+        self.gps_points: List[GPSPoint] = []
+
     def __del__(self):
         print("delete dataType Object")
         try:

sohstationviewer/model/mseed/mseed.py

 """
 MSeed object to hold and process MSeed data
 """
-
+import functools
+import operator
 import os
 from pathlib import Path
 from typing import Dict, Tuple, List, Set
@@ -11,6 +12,7 @@ from obspy.core import Stream
 from sohstationviewer.conf import constants
 from sohstationviewer.controller.util import validateFile
 from sohstationviewer.model.data_type_model import DataTypeModel, ThreadStopped
+from sohstationviewer.model.gps_point import GPSPoint
 from sohstationviewer.model.handling_data import (
     readWaveformMSeed, squash_gaps, checkWFChan, sortData, readSOHTrace,
 )
@@ -56,6 +58,8 @@ class MSeed(DataTypeModel):
         if len(self.reqWFChans) != 0:
             self.readWFFiles(self.selectedKey)
 
+        self.get_gps_data_q330()
+
     def read_soh_and_index_waveform(self, folder: str):
         """
             + read waveform data for filename associate with time range
@@ -310,6 +314,89 @@ class MSeed(DataTypeModel):
 
         sortData(self.waveformData)
 
+    def get_gps_data_q330(self):
+        """
+        Read LOG channel and extract GPS data stored in Q330 data set.
+        """
+        for station in self.logData:
+            if station == 'TEXT':
+                continue
+            # Q330 log data is composed of a list of string, so we combine them
+            # into one big string for ease of processing.
+            log_str = functools.reduce(
+                operator.iconcat, self.logData[station]['LOG'], ''
+            )
+
+            log_lines = [line for line in log_str.splitlines() if line != '']
+            for idx, line in enumerate(log_lines):
+                if line == "GPS Status":
+                    # We are assuming that a GPS status report is 12 lines
+                    # long, has a specific format, and is followed by a PLL
+                    # status report. This method call checks if these
+                    # preconditions hold and raise an error if not.
+                    self.check_q330_gps_status_format(log_lines[idx:idx+13])
+
+                    last_timemark = log_lines[idx+11][19:]
+
+                    fix_type = log_lines[idx+3][10:]
+                    # If fix type is off, there is no location data available
+                    # so we set them all to 0.
+                    if fix_type == 'OFF':
+                        self.gps_points.append(
+                            GPSPoint(last_timemark, fix_type, 0, 0, 0, 0)
+                        )
+                        continue
+                    if fix_type == 'NONE' and log_lines[idx + 1] == 'Time: ':
+                        self.gps_points.append(
+                            GPSPoint(last_timemark, fix_type, 0, 0, 0, 0)
+                        )
+                        continue
+
+                    num_sats_used = int(log_lines[idx+8].split(': ')[1])
+
+                    # Height is encoded as a float followed by the unit. We
+                    # don't know how many characters the unit is composed of,
+                    # so we have to loop through the height string backward
+                    # until we can detect the end of the height value.
+                    height_str: str = log_lines[idx+4].split(': ')[1]
+                    # Start pass the end of the string and look backward one
+                    # index every iteration so we don't have to add 1 to the
+                    # final index.
+                    i = len(height_str)
+                    current_char = height_str[i - 1]
+                    while current_char != '.' and not current_char.isnumeric():
+                        i -= 1
+                        current_char = height_str[i - 1]
+                    height = float(height_str[:i])
+
+                    # Latitude and longitude are encoded in the format
+                    # <degree><decimal minute><cardinal direction>. For
+                    # latitude, <degree> has two characters, while <degree> for
+                    # longitude has three.
+                    raw_latitude = log_lines[idx+5].split(': ')[1]
+                    lat_degree = int(raw_latitude[:2])
+                    lat_minute = float(raw_latitude[2:-1]) / 60
+                    latitude = lat_degree + lat_minute
+                    if raw_latitude[-1].lower() == 's':
+                        latitude = -latitude
+
+                    raw_longitude = log_lines[idx+6].split(': ')[1]
+                    long_degree = int(raw_longitude[:3])
+                    long_minute = float(raw_longitude[3:-1]) / 60
+                    longitude = long_degree + long_minute
+                    if raw_longitude[-1].lower() == 's':
+                        longitude = -longitude
+
+                    # Make latitude and longitude always nonnegative to make
+                    # plotting easier. Latitude ranges from -90 to 90, while
+                    # longitude ranges from -180 to 180.
+                    latitude += 90
+                    longitude += 180
+                    self.gps_points.append(
+                        GPSPoint(last_timemark, fix_type, num_sats_used,
+                                 height, latitude, longitude)
+                    )
+
     @staticmethod
     def check_q330_gps_status_format(gps_status_lines: List[str]):
         if gps_status_lines[12].lower() != 'pll status':