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 (14)
Showing
with 376 additions and 60 deletions
# Selecting Mass Position
<br />
Users can either select mass position according to group of channels 1, 2, 3 or 4, 5, 6 or both by checking the checkboxes as displayed.
<br />
<img alt="Select Mass Position" src="images/select_masspos/select_masspos.png" height="30" />
<br />
---------------------------
---------------------------
## Warning when the selected mass position channels not found
If the selected mass position channel cannot be found, a warning will be created in Processing Logs.
<br />
<img alt="Select Mass Position" src="images/select_masspos/warning_req_masspos_not_exist.png" height="60" />
<br />
\ No newline at end of file
documentation/images/select_masspos/select_masspos.png

4.07 KiB

documentation/images/select_masspos/warning_req_masspos_not_exist.png

26 KiB

......@@ -178,7 +178,7 @@ def get_time_ticks(earliest: float, latest: float, date_format: str,
times = []
time_labels = []
time += interval
while time < latest:
while time <= latest:
times.append(time)
time_label = format_time(time, date_format, 'HH:MM:SS')
if mode == "DD" or mode == "D":
......
......@@ -75,8 +75,11 @@ def load_data(data_type: str, tracking_box: QTextBrowser, dir_list: List[str],
def read_channels(tracking_box: QTextBrowser, list_of_dir: List[str]
) -> Set[str]:
"""
Scan available channels (to be used in channel preferences dialog). Since
channels for RT130 is hard code, this function won't be applied for it.
Scan available for SOH channels (to be used in channel preferences dialog).
Since channels for RT130 is hard code, this function won't be applied
for it.
Note that Mass position channels are excluded because the default
include_mp123 and include_mp456 for MSeed are False
:param tracking_box: widget to display tracking info
:param list_of_dir: list of directories selected by users
:return data_object.channels: set of channels present in listofDir
......
......@@ -27,7 +27,9 @@ class DataLoaderWorker(QtCore.QObject):
def __init__(self, data_type: str, tracking_box: QtWidgets.QTextBrowser,
folder: str, req_wf_chans: Union[List[str], List[int]] = [],
req_soh_chans: List[str] = [], read_start: float = 0,
read_end: float = constants.HIGHEST_INT, parent_thread=None):
read_end: float = constants.HIGHEST_INT,
include_mp123: bool = False, include_mp456: bool = False,
parent_thread=None):
super().__init__()
self.data_type = data_type
self.tracking_box = tracking_box
......@@ -36,6 +38,8 @@ class DataLoaderWorker(QtCore.QObject):
self.req_soh_chans = req_soh_chans
self.read_start = read_start
self.read_end = read_end
self.include_mp123 = include_mp123
self.include_mp456 = include_mp456
self.parent_thread = parent_thread
# display_tracking_info updates a QtWidget, which can only be done in
# the read. Since self.run runs in a background thread, we need to use
......@@ -60,8 +64,11 @@ class DataLoaderWorker(QtCore.QObject):
data_object.__init__(
self.tracking_box, self.folder,
req_wf_chans=self.req_wf_chans,
req_soh_chans=self.req_soh_chans, read_start=self.read_start,
read_end=self.read_end, creator_thread=self.parent_thread,
req_soh_chans=self.req_soh_chans,
read_start=self.read_start, read_end=self.read_end,
include_mp123=self.include_mp123,
include_mp456=self.include_mp456,
creator_thread=self.parent_thread,
notification_signal=self.notification,
pause_signal=self.button_dialog
)
......@@ -98,7 +105,9 @@ class DataLoader(QtCore.QObject):
list_of_dir: List[Union[str, Path]],
req_wf_chans: Union[List[str], List[int]] = [],
req_soh_chans: List[str] = [], read_start: float = 0,
read_end: float = constants.HIGHEST_INT):
read_end: float = constants.HIGHEST_INT,
include_mp123: bool = False,
include_mp456: bool = False):
"""
Initialize the data loader. Construct the thread and worker and connect
them together. Separated from the actual loading of the data to allow
......@@ -112,6 +121,8 @@ class DataLoader(QtCore.QObject):
:param req_soh_chans: list of requested SOH channel
:param read_start: the time before which no data is read
:param read_end: the time after which no data is read
:param include_mp123: if mass position channels 1,2,3 are requested
:param include_mp456: if mass position channels 4,5,6 are requested
"""
if self.running:
# TODO: implement showing an error window
......@@ -128,6 +139,8 @@ class DataLoader(QtCore.QObject):
req_soh_chans=req_soh_chans,
read_start=read_start,
read_end=read_end,
include_mp123=include_mp123,
include_mp456=include_mp456,
parent_thread=self.thread
)
......
......@@ -36,6 +36,8 @@ class DataTypeModel():
req_wf_chans: Union[List[str], List[int]] = [],
req_soh_chans: List[str] = [], read_start: float = 0,
read_end: float = constants.HIGHEST_INT,
include_mp123: bool = False,
include_mp456: bool = False,
creator_thread: Optional[QtCore.QThread] = None,
notification_signal: Optional[QtCore.Signal] = None,
pause_signal: Optional[QtCore.Signal] = None,
......@@ -50,6 +52,8 @@ class DataTypeModel():
:param req_soh_chans: requested SOH channel list
:param read_start: requested start time to read
:param read_end: requested end time to read
:param include_mp123: if mass position channels 1,2,3 are requested
:param include_mp456: if mass position channels 4,5,6 are requested
:param creator_thread: the thread the current DataTypeModel instance is
being created in. If None, the DataTypeModel instance is being
created in the main thread
......@@ -65,6 +69,8 @@ class DataTypeModel():
self.read_chan_only = read_chan_only
self.read_start = read_start
self.read_end = read_end
self.include_mp123 = include_mp123
self.include_mp456 = include_mp456
if creator_thread is None:
err_msg = (
'A signal is not None while running in main thread'
......@@ -281,7 +287,8 @@ class DataTypeModel():
def create_data_object(cls, data_type, tracking_box, folder,
read_chan_only=False, req_wf_chans=[],
req_soh_chans=[],
read_start=0, read_end=constants.HIGHEST_INT):
read_start=0, read_end=constants.HIGHEST_INT,
include_mp123=False, include_mp456=False):
"""
Create a DataTypeModel object, with the concrete class being based on
data_type. Run on the same thread as its caller, and so will block the
......@@ -296,6 +303,8 @@ class DataTypeModel():
:param req_soh_chans: [str,] - requested soh channel list
:param read_start: [float,] - start time of read data
:param read_end: [float,] - finish time of read data
:param include_mp123: bool - if mass pos 1,2,3 are requested
:param include_mp456: bool - if mass pos 4,5,6 are requested
:return: DataTypeModel - object that keep the data read from
folder
"""
......@@ -304,13 +313,15 @@ class DataTypeModel():
data_object = RT130(
tracking_box, folder, readChanOnly=read_chan_only,
reqWFChans=req_wf_chans, reqSOHChans=req_soh_chans,
readStart=read_start, readEnd=read_end)
readStart=read_start, readEnd=read_end,
include_mp123=include_mp123, include_mp456=include_mp456)
else:
from sohstationviewer.model.mseed.mseed import MSeed
data_object = MSeed(
tracking_box, folder, readChanOnly=read_chan_only,
reqWFChans=req_wf_chans, reqSOHChans=req_soh_chans,
readStart=read_start, readEnd=read_end)
readStart=read_start, readEnd=read_end,
include_mp123=include_mp123, include_mp456=include_mp456)
return data_object
def pause(self) -> None:
......
......@@ -23,6 +23,7 @@ from sohstationviewer.view.util.enums import LogType
def read_soh_mseed(path2file: Path, file_name: str,
soh_streams: Dict[str, Dict[str, Stream]],
log_data: Dict[str, List[str]],
include_mp123: bool, include_mp456: bool,
nets_in_file: Dict[Tuple[str, ...], str],
track_info: Callable[[str, LogType], None]
) -> Dict[str, List]:
......@@ -37,6 +38,8 @@ def read_soh_mseed(path2file: Path, file_name: str,
:param file_name: name of mseed file
:param soh_streams: holder of SOH mseed streams
:param log_data: holder of info from log
:param include_mp123: if mass position channels 1,2,3 are requested
:param include_mp456: if mass position channels 4,5,6 are requested
:param nets_in_file: holder of dict with key are
nets of file and value is user-selected net so the rule will be
applied for other files if their nets is subset of a key
......@@ -67,11 +70,17 @@ def read_soh_mseed(path2file: Path, file_name: str,
nets_in_file[tuple(nets)] = new_net_id
for trace in stream:
chan_id = trace.stats['channel'].strip()
if chan_id.startswith('VM'):
if not include_mp123 and chan_id[-1] in ['1', '2', '3']:
continue
if not include_mp456 and chan_id[-1] in ['4', '5', '6']:
continue
if new_net_id is not None:
# use one net ID for all traces in a file
trace.stats['network'] = new_net_id
net_id = trace.stats['network'].strip()
chan_id = trace.stats['channel'].strip()
sta_id = trace.stats['station'].strip()
stats.add(sta_id)
chan_ids.add(chan_id)
......
......@@ -421,21 +421,22 @@ def read_hdrs(path2file: Path, file_name: str,
soh_streams: Dict[str, Dict[str, Stream]],
log_data: Dict[str, Union[List[str], Dict[str, List[str]]]],
req_soh_chans: List[str], req_wf_chans: List[str],
include_mp123: bool, include_mp456: bool,
nets_in_file: Dict[Tuple[str, ...], str],
track_info: Callable[[str, LogType], None]
) -> Optional[Dict[str, Union[float, str, bool, List[str]]]]:
"""
read headers of a given file build dictionary for quick access
:param path2file: str - path to file
:param file_name: str - name of file
:param soh_streams: dict - holder for different sets of soh mseed stream
:param log_data: dict - holder for logging messages
:param req_soh_chans: list of string - requested SOH channels sent
from Main Window
:param req_wf_chans: list of string - requested waveform channel sent from
Main Window
:param nets_in_file: dict - holder for all network
:param track_info: function - to do process tracking
:param path2file: path to file
:param file_name: name of file
:param soh_streams: holder for different sets of soh mseed stream
:param log_data: holder for logging messages
:param req_soh_chans: requested SOH channels sent from Main Window
:param req_wf_chans: requested waveform channel sent from Main Window
:param include_mp123: if mass position channels 1,2,3 are requested
:param include_mp456: if mass position channels 4,5,6 are requested
:param nets_in_file: holder for all network
:param track_info: to do process tracking
:return:
+ if file is mseed but cannot read: raise error
+ if file is mseed but chan_type isn't requested, do nothing
......@@ -466,8 +467,8 @@ def read_hdrs(path2file: Path, file_name: str,
return
if chan_type == 'SOH':
read_soh_mseed(path2file, file_name, soh_streams,
log_data, nets_in_file, track_info)
read_soh_mseed(path2file, file_name, soh_streams, log_data,
include_mp123, include_mp456, nets_in_file, track_info)
return
if req_wf_chans == []:
return
......
......@@ -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
......@@ -111,6 +113,7 @@ class MSeed(DataTypeModel):
ret = read_hdrs(
path2file, file_name, soh_streams, self.log_data,
self.req_soh_chans, self.req_wf_chans,
self.include_mp123, self.include_mp456,
self.nets_in_file, self.track_info)
if ret is None:
......@@ -281,6 +284,8 @@ class MSeed(DataTypeModel):
"""
count = 0
if sta_id not in self.waveform_data.keys():
return
for chan_id in self.waveform_data[sta_id]['filesInfo']:
traces_info = self.waveform_data[sta_id][
'readData'][chan_id]['tracesInfo']
......@@ -454,3 +459,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.
......@@ -30,6 +30,7 @@ class Reftek130Exception(ObsPyException):
class Reftek130(obspy_rt130_core.Reftek130):
def to_stream(self, network="", location="", component_codes=None,
include_mp123=False, include_mp456=False,
headonly=False, verbose=False,
sort_permuted_package_sequence=False):
"""
......@@ -211,7 +212,11 @@ class Reftek130(obspy_rt130_core.Reftek130):
if DS != 9:
tr.stats.channel = "DS%s-%s" % (DS, channel_number + 1)
else:
tr.stats.channel = "MP-%s" % (channel_number + 1)
if not include_mp123 and channel_number < 3:
continue
if not include_mp456 and channel_number > 2:
continue
tr.stats.channel = "MP%s" % (channel_number + 1)
# check if endtime of trace is consistent
t_last = packets_[-1]['time']
npts_last = packets_[-1]['number_of_samples']
......
......@@ -243,6 +243,8 @@ class RT130(DataTypeModel):
stream = core.Reftek130.to_stream(
rt130,
include_mp123=self.include_mp123,
include_mp456=self.include_mp456,
headonly=False,
verbose=False,
sort_permuted_package_sequence=True)
......
......@@ -8,7 +8,6 @@ class CalendarWidget(QtWidgets.QCalendarWidget):
"""
def __init__(self, parent):
super().__init__(parent)
self.setup_ui()
"""
toggle_day_of_year: QCheckBox: for user to choose to show day of year
or not
......@@ -20,6 +19,7 @@ class CalendarWidget(QtWidgets.QCalendarWidget):
"""
self._show_day_of_year = getattr(
self, 'toggle_day_of_year', None) is None
self.setup_ui()
def setup_ui(self):
"""
......@@ -48,6 +48,8 @@ class CalendarWidget(QtWidgets.QCalendarWidget):
self.toggle_day_of_year.show()
self.set_show_day_of_year(False)
def show_day_of_year(self):
"""
Get the value of self._show_day_of_year
......
......@@ -29,7 +29,8 @@ from sohstationviewer.view.plotting.state_of_health_widget import SOHWidget
from sohstationviewer.view.help_view import HelpBrowser
from sohstationviewer.view.ui.main_ui import UIMainWindow
from sohstationviewer.view.util.enums import LogType
from sohstationviewer.view.util.functions import check_chan_wildcards_format
from sohstationviewer.view.util.functions import (
check_chan_wildcards_format, check_masspos)
from sohstationviewer.view.channel_prefer_dialog import ChannelPreferDialog
from sohstationviewer.controller.processing import detect_data_type
......@@ -282,8 +283,11 @@ class MainWindow(QtWidgets.QMainWindow, UIMainWindow):
fd.setFileMode(QtWidgets.QFileDialog.Directory)
fd.setDirectory(self.curr_dir_line_edit.text())
fd.exec_()
new_path = fd.selectedFiles()[0]
self.set_current_directory(new_path)
try:
new_path = fd.selectedFiles()[0]
self.set_current_directory(new_path)
except IndexError:
pass
def get_requested_wf_chans(self) -> List[Union[str, int]]:
"""
......@@ -396,13 +400,17 @@ class MainWindow(QtWidgets.QMainWindow, UIMainWindow):
self.start_tm = datetime.strptime(start_tm_str, TM_FORMAT).timestamp()
self.end_tm = datetime.strptime(end_tm_str, TM_FORMAT).timestamp()
self.data_loader.init_loader(self.data_type,
self.tracking_info_text_browser,
self.dir_names,
req_wf_chans=self.req_wf_chans,
req_soh_chans=self.req_soh_chans,
read_start=self.start_tm,
read_end=self.end_tm)
self.data_loader.init_loader(
self.data_type,
self.tracking_info_text_browser,
self.dir_names,
req_wf_chans=self.req_wf_chans,
req_soh_chans=self.req_soh_chans,
read_start=self.start_tm,
read_end=self.end_tm,
include_mp123=self.mass_pos_123_check_box.isChecked(),
include_mp456=self.mass_pos_456_check_box.isChecked()
)
self.data_loader.worker.finished.connect(self.data_loaded)
self.data_loader.worker.stopped.connect(self.problem_happened)
......@@ -501,8 +509,18 @@ class MainWindow(QtWidgets.QMainWindow, UIMainWindow):
soh_data = deepcopy(do.soh_data[sel_key])
soh_chans = list(soh_data.keys())
mp_data = deepcopy(do.mass_pos_data[sel_key])
try:
check_masspos(mp_data, sel_key,
self.mass_pos_123_check_box.isChecked(),
self.mass_pos_456_check_box.isChecked())
except Exception as e:
self.processing_log.append((str(e), LogType.WARNING))
if len(self.req_wf_chans) != 0:
wf_data = deepcopy(do.waveform_data[sel_key]['readData'])
try:
wf_data = deepcopy(do.waveform_data[sel_key]['readData'])
except KeyError:
wf_data = {}
else:
wf_data = {}
try:
......@@ -677,6 +695,10 @@ class MainWindow(QtWidgets.QMainWindow, UIMainWindow):
self.data_loader.thread.quit()
self.data_loader.thread.wait()
# close all remaining windows
for window in QtWidgets.QApplication.topLevelWidgets():
window.close()
def delete_old_temp_data_folder(self) -> None:
"""
Delete temp_data_folder which is used for keeping memmap files in case
......
......@@ -204,5 +204,32 @@ def check_chan_wildcards_format(wildcards: str):
)
def check_masspos(mp_data: Dict[str, Dict], sel_key: Union[tuple, str],
include_mp123: bool, include_mp456: bool) -> None:
"""
Check mass positions channels to raise Exception if requested channels
aren't included.
:param mp_data: mass position channels of the selected key
:param sel_key: selected to be used in the error message
:param include_mp123: if mass position channels 1,2,3 are requested
:param include_mp456: if mass position channels 4,5,6 are requested
"""
included_mp = []
for chan_id in mp_data.keys():
included_mp.append(chan_id[-1])
req_mp = []
if include_mp123:
req_mp += ['1', '2', '3']
if include_mp456:
req_mp += ['4', '5', '6']
not_included_mp = [mp for mp in req_mp if mp not in included_mp]
if not_included_mp != []:
raise Exception(f"Data set {sel_key} doesn't include mass position "
f"{','.join(not_included_mp)}")
if __name__ == '__main__':
create_table_of_content_file(Path('../../../documentation'))
......@@ -178,10 +178,11 @@ class TestGetTimeTicks(TestCase):
earliest = UTCDateTime(1970, 1, 1, 0, 0, 0).timestamp
latest = UTCDateTime(1970, 1, 1, 0, 0, 5).timestamp
expected = (
[1.0, 2.0, 3.0, 4.0],
[1.0, 2.0, 3.0, 4.0],
[1.0, 2.0, 3.0, 4.0, 5.0],
[1.0, 2.0, 3.0, 4.0, 5.0],
['1970-01-01 00:00:01', '1970-01-01 00:00:02',
'1970-01-01 00:00:03', '1970-01-01 00:00:04']
'1970-01-01 00:00:03', '1970-01-01 00:00:04',
'1970-01-01 00:00:05']
)
self.assertTupleEqual(
get_time_ticks(earliest, latest,
......@@ -226,7 +227,7 @@ class TestGetTimeTicks(TestCase):
latest = UTCDateTime(1970, 1, 11, 0, 0, 0).timestamp
expected = (
[86400.0, 172800.0, 259200.0, 345600.0, 432000.0,
518400.0, 604800.0, 691200.0, 777600.0],
518400.0, 604800.0, 691200.0, 777600.0, 864000.0],
[86400.0, 259200.0, 432000.0, 604800.0, 777600.0],
['1970-01-02', '1970-01-04', '1970-01-06', '1970-01-08',
'1970-01-10']
......@@ -258,9 +259,9 @@ class TestGetTimeTicks(TestCase):
# Exactly 30 days
earliest = UTCDateTime(1970, 1, 1, 0, 0, 0).timestamp
latest = UTCDateTime(1970, 1, 31, 0, 0, 0).timestamp
expected = ([864000.0, 1728000.0],
[864000.0, 1728000.0],
['1970-01-11', '1970-01-21'])
expected = ([864000.0, 1728000.0, 2592000.0],
[864000.0, 1728000.0, 2592000.0],
['1970-01-11', '1970-01-21', '1970-01-31'])
self.assertTupleEqual(
get_time_ticks(earliest, latest,
self.date_fmt, self.label_cnt),
......@@ -289,14 +290,13 @@ class TestGetTimeTicks(TestCase):
time ranges make get_time_ticks returns a tuple that contains empty
lists.
"""
expected = ([], [], [])
with self.subTest('test_exactly_one_second'):
earliest = UTCDateTime(1970, 1, 1, 0, 0, 0).timestamp
latest = UTCDateTime(1970, 1, 1, 0, 0, 1).timestamp
self.assertTupleEqual(
get_time_ticks(earliest, latest,
self.date_fmt, self.label_cnt),
expected
([1.0], [1.0], ['1970-01-01 00:00:01'])
)
with self.subTest('test_exactly_one_minute'):
earliest = UTCDateTime(1970, 1, 1, 0, 0, 0).timestamp
......@@ -304,7 +304,7 @@ class TestGetTimeTicks(TestCase):
self.assertTupleEqual(
get_time_ticks(earliest, latest,
self.date_fmt, self.label_cnt),
expected
([60.0], [60.0], ['1970-01-01 00:01:00'])
)
with self.subTest('test_exactly_one_hour'):
earliest = UTCDateTime(1970, 1, 1, 0, 0, 0).timestamp
......@@ -312,7 +312,7 @@ class TestGetTimeTicks(TestCase):
self.assertTupleEqual(
get_time_ticks(earliest, latest,
self.date_fmt, self.label_cnt),
expected
([3600.0], [3600.0], ['1970-01-01 01:00'])
)
def test_earliest_time_later_than_latest_time(self):
......
......@@ -152,17 +152,16 @@ class TestLoadDataAndReadChannels(TestCase):
Test basic functionality of load_data - the given directory contains
MSeed data.
"""
q330_channels = {'VKI', 'VM1'}
q330_channels = {'VKI'}
self.assertSetEqual(read_channels(self.widget_stub, [q330_dir]),
q330_channels)
centaur_channels = {'VDT', 'VM3', 'EX3', 'GEL', 'VEC', 'EX2', 'LCE',
centaur_channels = {'VDT', 'EX3', 'GEL', 'VEC', 'EX2', 'LCE',
'EX1', 'GLA', 'LCQ', 'GPL', 'GNS', 'GST', 'VCO',
'GAN', 'GLO', 'VPB', 'VEI', 'VM2', 'VM1'}
self.assertSetEqual(read_channels(self.widget_stub, [centaur_dir]),
'GAN', 'GLO', 'VPB', 'VEI'}
ret = read_channels(self.widget_stub, [centaur_dir])
self.assertSetEqual(ret,
centaur_channels)
pegasus_channels = {'VDT', 'VM1', 'VE1'}
pegasus_channels = {'VDT', 'VE1'}
self.assertSetEqual(read_channels(self.widget_stub, [pegasus_dir]),
pegasus_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)
......@@ -7,7 +7,7 @@ from unittest import TestCase
from sohstationviewer.view.util.functions import (
get_soh_messages_for_view, log_str, is_doc_file,
create_search_results_file, create_table_of_content_file,
check_chan_wildcards_format
check_chan_wildcards_format, check_masspos
)
from sohstationviewer.view.util.enums import LogType
......@@ -373,3 +373,43 @@ class TestCheckChanWildcardsFormat(TestCase):
str(context.exception),
f"Request '{wc}' has length={len(wc)} > 3 which isn't allowed."
)
class TestCheckMassPos(TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.mp_data = {'MP1': {'chan_id': 'MP1', 'samplerate': 1},
'MP3': {'chan_id': 'MP3', 'samplerate': 1},
'MP4': {'chan_id': 'MP4', 'samplerate': 1}}
cls.sel_key = '1378'
def test_include_mp123(self):
with self.assertRaises(Exception) as context:
check_masspos(self.mp_data, self.sel_key,
include_mp123=True, include_mp456=False)
self.assertEqual(
str(context.exception),
f"Data set {self.sel_key} doesn't include mass position 2")
def test_include_mp456(self):
with self.assertRaises(Exception) as context:
check_masspos(self.mp_data, self.sel_key,
include_mp123=False, include_mp456=True)
self.assertEqual(
str(context.exception),
f"Data set {self.sel_key} doesn't include mass position 5,6")
def test_include_mp123456(self):
with self.assertRaises(Exception) as context:
check_masspos(self.mp_data, self.sel_key,
include_mp123=True, include_mp456=True)
self.assertEqual(
str(context.exception),
f"Data set {self.sel_key} doesn't include mass position 2,5,6")
def test_not_include_mp(self):
try:
check_masspos(self.mp_data, self.sel_key,
include_mp123=False, include_mp456=False)
except Exception:
self.fail("check_masspos() raise Exception unexpectedly")