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
Hide whitespace changes
Inline Side-by-side
sohstationviewer/view/ui/main_ui.py
View file @ 2a354444
......@@ -166,16 +166,16 @@ class UIMainWindow(object):
"""
self.all_soh_chans_check_box: Union[QCheckBox, None] = None
"""
prefer_soh_chan_button: Button to open self.channel_prefer_dialog to
view/add/edit/select a preferred IDs
select_pref_soh_list_button: Button to open self.channel_prefer_dialog
to view/add/edit/select a preferred IDs
"""
self.prefer_soh_chan_button: Union[QPushButton, None] = None
self.select_pref_soh_list_button: Union[QPushButton, None] = None
"""
curr_soh_ids_name_line_edit: textbox to display name of current IDs.
curr_pref_soh_list_name_txtbox: textbox to display name of current IDs.
When all_soh_chans_check_box is checked, no name show up, when
all_soh_chans_check_box is unchecked, current selected name show up
"""
self.curr_soh_ids_name_line_edit: Union[QLineEdit, None] = None
self.curr_pref_soh_list_name_txtbox: Union[QLineEdit, None] = None
"""
read_button: Button to read data from selected directory
from open_files_list and plot it
......@@ -498,15 +498,16 @@ class UIMainWindow(object):
self.all_soh_chans_check_box.setChecked(True)
chan_layout.addWidget(self.all_soh_chans_check_box)
self.prefer_soh_chan_button = QPushButton(
self.select_pref_soh_list_button = QPushButton(
'Pref', self.central_widget)
self.prefer_soh_chan_button.setFixedWidth(47)
chan_layout.addWidget(self.prefer_soh_chan_button)
self.select_pref_soh_list_button.setFixedWidth(47)
chan_layout.addWidget(self.select_pref_soh_list_button)
chan_layout.addWidget(QLabel('Cur'))
self.curr_soh_ids_name_line_edit = QLineEdit(
self.curr_pref_soh_list_name_txtbox = QLineEdit(
self.central_widget)
chan_layout.addWidget(self.curr_soh_ids_name_line_edit)
self.curr_pref_soh_list_name_txtbox.setReadOnly(True)
chan_layout.addWidget(self.curr_pref_soh_list_name_txtbox)
submit_layout = QHBoxLayout()
submit_layout.setSpacing(5)
......@@ -605,19 +606,20 @@ class UIMainWindow(object):
menu.addAction(self.add_positions_to_et_action)
date_format_menu = QMenu('Date Format:', main_window)
date_format_action_group = QActionGroup(main_window)
menu.addMenu(date_format_menu)
self.yyyy_doy_action = QAction(
'YYYY:DOY', main_window)
self.yyyy_doy_action.setCheckable(True)
date_format_menu.addAction(self.yyyy_doy_action)
self.yyyy_mm_dd_action = QAction(
'YYYY-MM-DD', main_window)
self.yyyy_mm_dd_action = QAction('YYYY-MM-DD', main_window)
self.yyyy_mm_dd_action.setCheckable(True)
date_format_menu.addAction(self.yyyy_mm_dd_action)
self.yyyymmmdd_action = QAction(
'YYYYMMMDD', main_window)
date_format_action_group.addAction(self.yyyy_mm_dd_action)
self.yyyy_doy_action = QAction('YYYY:DOY', main_window)
self.yyyy_doy_action.setCheckable(True)
date_format_menu.addAction(self.yyyy_doy_action)
date_format_action_group.addAction(self.yyyy_doy_action)
self.yyyymmmdd_action = QAction('YYYYMMMDD', main_window)
self.yyyymmmdd_action.setCheckable(True)
date_format_menu.addAction(self.yyyymmmdd_action)
date_format_action_group.addAction(self.yyyymmmdd_action)
def create_database_menu(self, main_window, menu):
"""
......@@ -693,11 +695,12 @@ class UIMainWindow(object):
main_window, 'mass_pos_volt_range_opt', 'trillium'))
self.yyyy_mm_dd_action.triggered.connect(
lambda: main_window.set_date_format('yyyy-MM-dd'))
lambda: main_window.set_date_format('YYYY-MM-DD'))
self.yyyymmmdd_action.triggered.connect(
lambda: main_window.set_date_format('yyyyMMMdd'))
lambda: main_window.set_date_format('YYYYMMMDD'))
self.yyyy_doy_action.triggered.connect(
lambda: main_window.set_date_format('yyyyDOY'))
lambda: main_window.set_date_format('YYYY:DOY'))
self.yyyy_mm_dd_action.trigger()
# Database
self.add_edit_data_type_action.triggered.connect(
......@@ -748,7 +751,7 @@ class UIMainWindow(object):
self.all_soh_chans_check_box.clicked.connect(
main_window.all_soh_chans_clicked)
self.prefer_soh_chan_button.clicked.connect(
self.select_pref_soh_list_button.clicked.connect(
main_window.open_channel_preferences)
self.read_button.clicked.connect(main_window.read_selected_files)
......
sohstationviewer/view/util/functions.py
View file @ 2a354444
......@@ -141,13 +141,13 @@ def get_soh_messages_for_view(
for msg_list in soh_messages['TEXT']:
for msg in msg_list.split('\n'):
soh_message_view['TEXT'].append(msg)
if key in soh_messages.keys():
for chan_id in soh_messages[key]:
soh_message_view[chan_id] = []
for chan_id in soh_messages[key]:
soh_message_view[chan_id] = []
for msg_lines in soh_messages[key][chan_id]:
for msg in msg_lines.split('\n'):
soh_message_view[chan_id].append(msg)
for msg_lines in soh_messages[key][chan_id]:
for msg in msg_lines.split('\n'):
soh_message_view[chan_id].append(msg)
return soh_message_view
......
tests/test_controller/test_plotting_data.py
View file @ 2a354444
......@@ -166,7 +166,7 @@ class TestGetTimeTicks(TestCase):
def setUp(self) -> None:
"""Set up text fixtures."""
self.label_cnt = 5
self.date_fmt = 'YYYYMMDD'
self.date_fmt = 'YYYY-MM-DD'
def test_expected_time_range(self):
"""
......@@ -180,8 +180,8 @@ class TestGetTimeTicks(TestCase):
expected = (
[1.0, 2.0, 3.0, 4.0],
[1.0, 2.0, 3.0, 4.0],
['19700101 00:00:01', '19700101 00:00:02',
'19700101 00:00:03', '19700101 00:00:04']
['1970-01-01 00:00:01', '1970-01-01 00:00:02',
'1970-01-01 00:00:03', '1970-01-01 00:00:04']
)
self.assertTupleEqual(
get_time_ticks(earliest, latest,
......@@ -195,8 +195,9 @@ class TestGetTimeTicks(TestCase):
expected = (
[60.0, 120.0, 180.0, 240.0, 300.0],
[60.0, 120.0, 180.0, 240.0, 300.0],
['19700101 00:01:00', '19700101 00:02:00', '19700101 00:03:00',
'19700101 00:04:00', '19700101 00:05:00']
['1970-01-01 00:01:00', '1970-01-01 00:02:00',
'1970-01-01 00:03:00', '1970-01-01 00:04:00',
'1970-01-01 00:05:00']
)
self.assertTupleEqual(
get_time_ticks(earliest, latest,
......@@ -210,8 +211,8 @@ class TestGetTimeTicks(TestCase):
expected = (
[3600.0, 7200.0, 10800.0, 14400.0, 18000.0, 21600.0, 25200.0],
[3600.0, 10800.0, 18000.0, 25200.0],
['19700101 01:00', '19700101 03:00', '19700101 05:00',
'19700101 07:00']
['1970-01-01 01:00', '1970-01-01 03:00', '1970-01-01 05:00',
'1970-01-01 07:00']
)
self.assertTupleEqual(
get_time_ticks(earliest, latest,
......@@ -227,7 +228,8 @@ class TestGetTimeTicks(TestCase):
[86400.0, 172800.0, 259200.0, 345600.0, 432000.0,
518400.0, 604800.0, 691200.0, 777600.0],
[86400.0, 259200.0, 432000.0, 604800.0, 777600.0],
['19700102', '19700104', '19700106', '19700108', '19700110']
['1970-01-02', '1970-01-04', '1970-01-06', '1970-01-08',
'1970-01-10']
)
self.assertTupleEqual(
get_time_ticks(earliest, latest,
......@@ -243,7 +245,8 @@ class TestGetTimeTicks(TestCase):
518400.0, 604800.0, 691200.0, 777600.0, 864000.0,
950400.0, 1036800.0, 1123200.0],
[86400.0, 345600.0, 604800.0, 864000.0, 1123200.0],
['19700102', '19700105', '19700108', '19700111', '19700114']
['1970-01-02', '1970-01-05', '1970-01-08', '1970-01-11',
'1970-01-14']
)
self.assertTupleEqual(
get_time_ticks(earliest, latest,
......@@ -257,7 +260,7 @@ class TestGetTimeTicks(TestCase):
latest = UTCDateTime(1970, 1, 31, 0, 0, 0).timestamp
expected = ([864000.0, 1728000.0],
[864000.0, 1728000.0],
['19700111', '19700121'])
['1970-01-11', '1970-01-21'])
self.assertTupleEqual(
get_time_ticks(earliest, latest,
self.date_fmt, self.label_cnt),
......@@ -271,7 +274,7 @@ class TestGetTimeTicks(TestCase):
[864000.0, 1728000.0, 2592000.0, 3456000.0, 4320000.0,
5184000.0, 6048000.0],
[864000.0, 2592000.0, 4320000.0, 6048000.0],
['19700111', '19700131', '19700220', '19700312']
['1970-01-11', '1970-01-31', '1970-02-20', '1970-03-12']
)
self.assertTupleEqual(
get_time_ticks(earliest, latest, self.date_fmt,
......@@ -349,7 +352,7 @@ class TestFormatTimeAndGetTitle(TestCase):
self.positive_epoch_time = 67567567
self.positive_formatted_dates = {
'YYYY-MM-DD': '1972-02-22',
'YYYYMMDD': '19720222',
'YYYYMMMDD': '1972FEB22',
'YYYY:DOY': '1972:053',
}
self.positive_formatted_time = '00:46:07'
......@@ -357,7 +360,7 @@ class TestFormatTimeAndGetTitle(TestCase):
self.negative_epoch_time = -67567567
self.negative_formatted_dates = {
'YYYY-MM-DD': '1967-11-10',
'YYYYMMDD': '19671110',
'YYYYMMMDD': '1967NOV10',
'YYYY:DOY': '1967:314',
}
self.negative_formatted_time = '23:13:53'
......@@ -371,12 +374,12 @@ class TestFormatTimeAndGetTitle(TestCase):
# formatter:off
test_name_to_date_mode_map = {
'test_year_month_day_format': 'YYYY-MM-DD',
'test_year_month_day_format_no_dash': 'YYYYMMDD',
'test_year_month_day_format_no_dash': 'YYYYMMMDD',
'test_day_of_year_format': 'YYYY:DOY'
}
zero_epoch_formatted = {
'test_year_month_day_format': '1970-01-01',
'test_year_month_day_format_no_dash': '19700101',
'test_year_month_day_format_no_dash': '1970JAN01',
'test_day_of_year_format': '1970:001',
}
# formatter:on
......@@ -401,12 +404,12 @@ class TestFormatTimeAndGetTitle(TestCase):
# formatter:off
test_name_to_date_mode_map = {
'test_year_month_day_format': 'YYYY-MM-DD',
'test_year_month_day_format_no_dash': 'YYYYMMDD',
'test_year_month_day_format_no_dash': 'YYYYMMMDD',
'test_day_of_year_format': 'YYYY:DOY',
}
zero_epoch_formatted = {
'test_year_month_day_format': '1970-01-01 00:00:00',
'test_year_month_day_format_no_dash': '19700101 00:00:00',
'test_year_month_day_format_no_dash': '1970JAN01 00:00:00',
'test_day_of_year_format': '1970:001 00:00:00',
}
# formatter:on
......@@ -441,7 +444,7 @@ class TestFormatTimeAndGetTitle(TestCase):
"""
test_name_to_date_mode_map = {
'test_year_month_day_format': 'YYYY-MM-DD',
'test_year_month_day_format_no_dash': 'YYYYMMDD',
'test_year_month_day_format_no_dash': 'YYYYMMMDD',
'test_day_of_year_format': 'YYYY:DOY',
}
utc_date_time = UTCDateTime(self.positive_epoch_time)
......@@ -458,7 +461,7 @@ class TestFormatTimeAndGetTitle(TestCase):
"""
test_name_to_date_mode_map = {
'test_year_month_day_format': 'YYYY-MM-DD',
'test_year_month_day_format_no_dash': 'YYYYMMDD',
'test_year_month_day_format_no_dash': 'YYYYMMMDD',
'test_day_of_year_format': 'YYYY:DOY',
}
test_time = UTCDateTime(self.positive_epoch_time)
......@@ -500,7 +503,7 @@ class TestFormatTimeAndGetTitle(TestCase):
supported.
"""
test_time = self.positive_epoch_time
date_format = 'YYYYMMDD'
date_format = 'YYYYMMMDD'
empty_format = ''
bad_format = 'bad_format'
......@@ -528,20 +531,20 @@ class TestFormatTimeAndGetTitle(TestCase):
def test_get_title(self):
"""Test basic functionality of get_title."""
date_mode = 'YYYYMMDD'
date_mode = 'YYYY-MM-DD'
min_time = 0
max_time = self.positive_epoch_time
formatted_max_time = (f'{self.positive_formatted_dates[date_mode]}'
f' {self.positive_formatted_time}')
with self.subTest('test_mseed'):
key = '3734'
expected = (f'3734 19700101 00:00:00 to '
expected = (f'3734 1970-01-01 00:00:00 to '
f'{formatted_max_time} (18768.77)')
self.assertEqual(get_title(key, min_time, max_time, date_mode),
expected)
with self.subTest('test_rt130'):
key = ('92EB', 25)
expected = (f"('92EB', 25) 19700101 00:00:00 to "
expected = (f"('92EB', 25) 1970-01-01 00:00:00 to "
f"{formatted_max_time} (18768.77)")
self.assertEqual(get_title(key, min_time, max_time, date_mode),
expected)
......@@ -551,7 +554,7 @@ class TestFormatTimeAndGetTitle(TestCase):
Test basic functionality of get_title - the given maximum time is
chronologically earlier than the given minimum time.
"""
date_mode = 'YYYYMMDD'
date_mode = 'YYYY-MM-DD'
min_time = self.positive_epoch_time
max_time = 0
formatted_max_time = (f'{self.positive_formatted_dates[date_mode]}'
......@@ -559,13 +562,13 @@ class TestFormatTimeAndGetTitle(TestCase):
with self.subTest('test_mseed'):
key = '3734'
expected = (f'3734 {formatted_max_time} to '
f'19700101 00:00:00 (-18768.77)')
f'1970-01-01 00:00:00 (-18768.77)')
self.assertEqual(get_title(key, min_time, max_time, date_mode),
expected)
with self.subTest('test_rt130'):
key = ('92EB', 25)
expected = (f"('92EB', 25) {formatted_max_time} to "
f"19700101 00:00:00 (-18768.77)")
f"1970-01-01 00:00:00 (-18768.77)")
self.assertEqual(get_title(key, min_time, max_time, date_mode),
expected)
......
tests/test_model/test_handling_data_trim_downsample.py
View file @ 2a354444
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Optional, Dict, Union, List
from unittest import TestCase
from unittest.mock import patch
......@@ -7,18 +8,27 @@ from unittest.mock import patch
from obspy.core import UTCDateTime
import numpy as np
import sohstationviewer.view.plotting.time_power_squared_processor
from sohstationviewer.conf import constants as const
from sohstationviewer.model.handling_data import (
trim_downsample_soh_chan,
trim_downsample_wf_chan,
trim_waveform_data,
downsample_waveform_data
downsample_waveform_data,
get_each_day_5_min_list,
)
from sohstationviewer.view.plotting.time_power_squared_processor import (
TimePowerSquaredProcessor,
)
from sohstationviewer.model.downsampler import downsample, chunk_minmax
ORIGINAL_CHAN_SIZE_LIMIT = const.CHAN_SIZE_LIMIT
ORIGINAL_RECAL_SIZE_LIMIT = const.RECAL_SIZE_LIMIT
ZERO_EPOCH_TIME = UTCDateTime(1970, 1, 1, 0, 0, 0).timestamp
TraceInfo = Dict[str, Union[float, str]]
ChannelData = Dict[str,
Union[float, List[np.ndarray], List[TraceInfo], Dict,
np.ndarray]]
class TestTrimWfData(TestCase):
......@@ -596,3 +606,335 @@ class TestTrimDownsampleWfChan(TestCase):
self.end_time, False)
self.assertTrue(mock_trim.called)
self.assertTrue(mock_downsample.called)
class TestGetTrimTpsData(TestCase):
def no_file_memmap(self, file_path: Path, *args, **kwargs):
"""
A mock of numpy.memmap. Reduce test run time significantly by making
sure that data access happens in memory and not on disk.
This method does not actually load the data stored on disk. Instead, it
constructs the array of data using the name of the given file. To do
so, this method requires the file name to be in the format
<prefix>_<index>. This method then constructs an array of
self.trace_size consecutive integers starting at
<index> * self.trace_size.
:param file_path: the path to a file used to construct the data array.
:param args: dummy arguments to make the API similar to numpy.memmap.
:param kwargs: dummy arguments to make the API similar to numpy.memmap.
:return: a numpy array constructed using file_path's name.
"""
file_idx = int(file_path.name.split('_')[-1])
start = file_idx * self.trace_size
end = start + self.trace_size
return np.arange(start, end)
def add_trace(self, start_time: float, idx: Optional[int] = None):
"""
Add a trace to the stored list of traces.
:param start_time: the start time of the trace to be added.
:param idx: the index to insert the trace into. If None, the new trace
will be appended to the list of traces
"""
trace = {}
trace['startTmEpoch'] = start_time
trace['endTmEpoch'] = start_time + self.trace_size - 1
trace['size'] = self.trace_size
file_idx = start_time // self.trace_size
times_file_name = Path(self.data_folder.name) / f'times_{file_idx}'
trace['times_f'] = times_file_name
data_file_name = Path(self.data_folder.name) / f'data_{file_idx}'
trace['data_f'] = data_file_name
if idx is not None:
self.traces_info.insert(idx, trace)
else:
self.traces_info.append(trace)
def setUp(self) -> None:
"""Set up text fixtures."""
memmap_patcher = patch.object(np, 'memmap',
side_effect=self.no_file_memmap)
self.addCleanup(memmap_patcher.stop)
memmap_patcher.start()
# Channel ID is only used when communicating with the main window.
# Seeing as we are testing the processing step here, we don't really
# need it.
channel_id = ''
self.channel_data: ChannelData = {'samplerate': 1}
self.traces_info = []
self.channel_data['tracesInfo'] = self.traces_info
self.data_folder = TemporaryDirectory()
self.trace_size = 1000
for i in range(100):
start_time = i * self.trace_size
self.add_trace(start_time)
self.start_time = 25000
self.end_time = 75000
self.each_day_5_mins_list = get_each_day_5_min_list(self.start_time,
self.end_time)
self.tps_processor = TimePowerSquaredProcessor(
channel_id, self.channel_data, self.start_time, self.end_time,
self.each_day_5_mins_list
)
local_TimePowerSquaredProcessor = (sohstationviewer.view.plotting.
time_power_squared_processor.
TimePowerSquaredProcessor)
# If object obj is instance of class A, then the method call obj.method1()
# translate to A.method1(obj) for Python. So, in order to mock method1 for
# obj, we mock it for the class A.
@patch.object(local_TimePowerSquaredProcessor, 'trim_waveform_data')
def test_data_is_trimmed(self, mock_trim_waveform_data):
"""Test that the data is trimmed."""
self.tps_processor.run()
self.assertTrue(mock_trim_waveform_data.called)
def test_appropriate_amount_of_5_mins_skipped(self):
"""Test that the trimmed part of the data is skipped over."""
self.tps_processor.run()
with self.subTest('test_skip_before_start_time'):
first_unskipped_idx = 83
skipped_tps_arr = (
self.channel_data['tps_data'][0][:first_unskipped_idx]
)
self.assertTrue((skipped_tps_arr == 0).all())
with self.subTest('test_skip_after_end_time'):
last_unskipped_idx = 252
skipped_tps_arr = (
self.channel_data['tps_data'][0][last_unskipped_idx + 1:]
)
self.assertTrue((skipped_tps_arr == 0).all())
def test_result_is_stored(self):
"""Test that the result of the TPS calculation is stored."""
self.tps_processor.run()
self.assertTrue('tps_data' in self.channel_data)
def test_formula_is_correct(self):
"""Test that the TPS calculation uses the correct formula."""
self.tps_processor.start_time = 50000
self.tps_processor.end_time = 52000
self.tps_processor.run()
first_unskipped_idx = 166
last_unskipped_idx = 175
tps_data = self.channel_data['tps_data'][0]
unskipped_tps_arr = (
tps_data[first_unskipped_idx:last_unskipped_idx + 1]
)
expected = np.array([
2.51497985e+09, 2.54515955e+09, 2.57551925e+09, 0.00000000e+00,
1.96222188e+09, 2.64705855e+09, 2.67801825e+09, 2.03969638e+09,
2.75095755e+09, 2.78251725e+09
])
self.assertTrue(np.allclose(unskipped_tps_arr, expected))
def test_one_tps_array_for_each_day_one_day_of_data(self):
"""
Test that there is one TPS array for each day of data.
Test the case where there is only one day of data.
"""
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 1)
def test_one_tps_array_for_each_day_multiple_days_of_data(self):
"""
Test that there is one TPS array for each dat of data.
Test the case where there are more than one day of data.
"""
# Currently, the data time goes from 0 to 100000, which is enough to
# cover two days (the start of the second positive day in epoch time is
# 86400). Thus, we only have to set the end time to the data end time
# to have two days of data.
self.tps_processor.end_time = 100000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time, self.tps_processor.end_time
)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
def test_data_has_gap_to_the_right_data_same_day_before_gap(self):
"""
Test that gaps in the data are skipped in TPS calculation by checking
that the elements in the TPS array corresponding to the gaps are
0.
Test the case where there are gaps to the right of the data and the
traces directly next to the gaps are in the same day.
"""
# Remove traces that go from 1000 to 24999 (traces 2 to 25) in order to
# create a gap on the right side of the data.
self.traces_info = [trace
for i, trace in enumerate(self.traces_info)
if not 0 < i < 25]
self.channel_data['tracesInfo'] = self.traces_info
with self.subTest('test_start_time_in_gap'):
self.tps_processor.start_time = 15000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 1)
tps_gap = slice(0, 50)
tps_data_in_gap = self.channel_data['tps_data'][0][tps_gap]
tps_data_in_gap_contains_zero = np.allclose(
tps_data_in_gap, np.zeros(tps_data_in_gap.size)
)
self.assertTrue(tps_data_in_gap_contains_zero)
with self.subTest('test_start_time_cover_all_traces'):
self.tps_processor.start_time = 500
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 1)
tps_gap = slice(2, 83)
tps_data_in_gap = self.channel_data['tps_data'][0][tps_gap]
tps_data_in_gap_contains_zero = np.allclose(
tps_data_in_gap, np.zeros(tps_data_in_gap.size)
)
self.assertTrue(tps_data_in_gap_contains_zero)
def test_data_has_gap_to_the_left_data_same_day_after_gap(self):
"""
Test that gaps in the data are skipped in TPS calculation by checking
that the elements in the TPS array corresponding to the gaps are
0.
Test the case where there are gaps to the left of the data and the
traces directly next to the gaps are in the same day.
"""
# Data end time is 100000, so we want a trace that starts after 100001
trace_start_time = 125000
self.add_trace(trace_start_time)
with self.subTest('test_end_time_in_gap'):
# Subject to change after Issue #37 is fixed
self.tps_processor.end_time = 110000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
tps_gaps = (slice(45, 128), slice(131, None))
tps_data_in_gaps = np.concatenate(
[self.channel_data['tps_data'][1][gap] for gap in tps_gaps]
)
tps_data_in_gaps_contains_zero = np.allclose(
tps_data_in_gaps, np.zeros(tps_data_in_gaps.size)
)
self.assertTrue(tps_data_in_gaps_contains_zero)
with self.subTest('test_end_time_cover_all_traces'):
self.tps_processor.end_time = trace_start_time + 50
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
tps_gaps = (slice(45, 128), slice(131, None))
tps_data_in_gaps = np.concatenate(
[self.channel_data['tps_data'][1][gap] for gap in tps_gaps]
)
tps_data_in_gaps_contains_zero = np.allclose(
tps_data_in_gaps, np.zeros(tps_data_in_gaps.size)
)
self.assertTrue(tps_data_in_gaps_contains_zero)
def test_data_has_gap_to_the_right_data_different_day_before_gap(self):
"""
Test that gaps in the data are skipped in TPS calculation by checking
that the elements in the TPS array corresponding to the gaps are
0.
Test the case where there are gaps to the right of the data and the
traces directly next to the gaps are in different days.
"""
trace_start_time = -50000
self.add_trace(trace_start_time, idx=0)
with self.subTest('test_start_time_in_gap'):
self.tps_processor.start_time = -25000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
tps_gap = slice(const.NO_5M_DAY)
tps_data_in_gap = self.channel_data['tps_data'][0][tps_gap]
tps_data_in_gap_contains_zero = np.allclose(
tps_data_in_gap, np.zeros(tps_data_in_gap.size)
)
self.assertTrue(tps_data_in_gap_contains_zero)
with self.subTest('test_start_time_cover_all_traces'):
self.tps_processor.start_time = -60000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
tps_gaps = (slice(0, 121), slice(124, None))
tps_data_in_gaps = np.concatenate(
[self.channel_data['tps_data'][0][gap] for gap in tps_gaps]
)
tps_data_in_gaps_contains_zero = np.allclose(
tps_data_in_gaps, np.zeros(tps_data_in_gaps.size)
)
self.assertTrue(tps_data_in_gaps_contains_zero)
def test_data_has_gap_to_the_left_data_different_day_after_gap(self):
"""
Test that gaps in the data are skipped in TPS calculation by checking
that the elements in the TPS array corresponding to the gaps are
0.
Test the case where there are gaps to the left of the data and the
traces directly next to the gaps are in different days.
"""
# The start of the third positive day in epoch time is 172800, so we
# want a trace that starts after 172801 so that the trace after the gap
# is of a different day than the previous ones
trace_start_time = 173100
self.add_trace(trace_start_time)
with self.subTest('test_end_time_same_day_as_second_to_last_trace'):
# Subject to change after Issue #37 is fixed
self.tps_processor.end_time = 125000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
with self.assertRaises(IndexError):
self.tps_processor.run()
with self.subTest('test_end_time_cover_all_traces'):
self.tps_processor.end_time = trace_start_time + 50
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 3)
tps_gap_day_2 = slice(45, None)
tps_gap_day_3 = slice(3, None)
tps_data_in_gaps = np.hstack(
(
self.channel_data['tps_data'][1][tps_gap_day_2],
self.channel_data['tps_data'][2][tps_gap_day_3]
)
)
tps_data_in_gaps_contains_zero = np.allclose(
tps_data_in_gaps, np.zeros(tps_data_in_gaps.size)
)
self.assertTrue(tps_data_in_gaps_contains_zero)
tests/test_model/test_mseed/test_gps.py
View file @ 2a354444
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)