Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • software_public/passoft/sohstationviewer
1 result
Show changes
Commits on Source (2)
......@@ -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.
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)