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)
......@@ -28,7 +28,7 @@ when reading the data set.
Click button "Pref" to open "SOH Channel Preferences" to create/edit/select a
list of SOH channel to read.
<br />
<img alt="SOH Channel Preferences Dialog" src="images/select_soh/soh_channel_preferences.png" height="300" />
<img alt="SOH Channel Preferences Dialog" src="images/select_soh/soh_channel_preferences.png" height="500" />
<br />
### SOH channel list
......@@ -59,14 +59,21 @@ click on button EDIT (5) to open a separated dialog for editing easier.
### Bottom buttons
<br />
<img alt="Buttons" src="images/select_soh/buttons.png" height="30" />
<img alt="Buttons" src="images/select_soh/buttons.png" height="120" />
<br />
**Beside adding preferred channels manually, users have two other options to start the list:**
+ "Add DB channels": All SOH channels for the selected Data Type in the
database will be added to Preferred SOH List box.
+ "Scan Channel from Data Source": SOHView will scan for all available SOH
channels in the data set.
+ "Scan Channel from MSeed Data": SOHView will scan for all available SOH
channels in the data set. At the bottom of the dialog there will be radio
buttons representing data keys. Switching between data keys will display all info
related to the key like SOH channels with sample rate less than or equal to 1 in
preferred channels box (4); mass position channels, waveform channels, soh channels
with sample rate greater than 1 in tracking box; data time at the bottom of the dialog.
If the data is multiplexed, both start time and end time will be displayed. However,
if the data isn't multiplexed, only start time will be displayed because SOHView only
read the first record of a file to speed up the performance.
**For saving Preferred SOH List**
+ "Save": Save any changes to database
......
documentation/images/select_soh/buttons.png

19.4 KiB | W: 0px | H: 0px

documentation/images/select_soh/buttons.png

40.7 KiB | W: 0px | H: 0px

documentation/images/select_soh/buttons.png
documentation/images/select_soh/buttons.png
documentation/images/select_soh/buttons.png
documentation/images/select_soh/buttons.png
  • 2-up
  • Swipe
  • Onion skin
documentation/images/select_soh/soh_channel_preferences.png

319 KiB | W: 0px | H: 0px

documentation/images/select_soh/soh_channel_preferences.png

312 KiB | W: 0px | H: 0px

documentation/images/select_soh/soh_channel_preferences.png
documentation/images/select_soh/soh_channel_preferences.png
documentation/images/select_soh/soh_channel_preferences.png
documentation/images/select_soh/soh_channel_preferences.png
  • 2-up
  • Swipe
  • Onion skin
......@@ -7,12 +7,13 @@ import os
import json
import re
from pathlib import Path
from typing import List, Set, Optional, Dict, Tuple
from typing import List, Optional, Dict, Tuple
from PySide2.QtCore import QEventLoop, Qt
from PySide2.QtGui import QCursor
from PySide2.QtWidgets import QTextBrowser, QApplication
from obspy.io import reftek
from obspy import UTCDateTime
from sohstationviewer.model.mseed_data.mseed_reader import \
move_to_next_record
......@@ -32,8 +33,9 @@ from sohstationviewer.controller.util import (
from sohstationviewer.view.util.enums import LogType
def _read_mseed_chanids(path2file: Path, is_multiplex) \
-> Tuple[Set[str], Set[str], Set[str], Set[str]]:
def _read_mseed_chanids(path2file: Path,
is_multiplex: bool,
channel_info: Dict) -> None:
"""
from the given file get set of channel ids for soh, mass position,
waveform, soh with sample rate greater than 1 which are
......@@ -41,15 +43,9 @@ def _read_mseed_chanids(path2file: Path, is_multiplex) \
:param path2file: path to file
:param is_multiplex: flag that tell if data has more than one channel in
a file
:return soh_chan_ids: list of chan_ids for SOH
:return mass_pos_chan_ids: list of chan_ids for mass position
:return wf_chan_ids: list of chan_ids for waveform
:return spr_gr_1_chan_ids: list of chan_ids for SOH with sample rate > 1
:param channel_info: Dict of channels' name categorized into soh, masspos,
waveform, soh with sample rate greater than one for each station.
"""
soh_chan_ids = set()
mass_pos_chan_ids = set()
wf_chan_ids = set()
spr_gr_1_chan_ids = set()
file = open(path2file, 'rb')
while 1:
is_eof = (file.read(1) == b'')
......@@ -68,27 +64,41 @@ def _read_mseed_chanids(path2file: Path, is_multiplex) \
if not is_multiplex:
break
continue
sta_id = record.record_metadata.station
if sta_id not in channel_info:
channel_info[sta_id] = {'start_time': UTCDateTime(),
'end_time': UTCDateTime(0),
'mass_pos': set(),
'waveform': set(),
'soh': set(),
'soh_spr_gr_1': set()}
if chan_type == 'MP':
mass_pos_chan_ids.add(chan_id)
channel_info[sta_id]['mass_pos'].add(chan_id)
elif chan_type == 'SOH':
sample_rate = record.record_metadata.sample_rate
if sample_rate <= 1:
soh_chan_ids.add(chan_id)
channel_info[sta_id]['soh'].add(chan_id)
else:
spr_gr_1_chan_ids.add(chan_id)
channel_info[sta_id]['soh_spr_gr_1'].add(chan_id)
else:
channel_info[sta_id]['waveform'].add(chan_id)
start_time = channel_info[sta_id]['start_time']
channel_info[sta_id]['start_time'] = min(
record.record_metadata.start_time, start_time)
if is_multiplex:
end_time = channel_info[sta_id]['end_time']
channel_info[sta_id]['end_time'] = max(
record.record_metadata.end_time, end_time)
else:
wf_chan_ids.add(chan_id)
if not is_multiplex:
break
move_to_next_record(file, current_record_start, record)
file.close()
return soh_chan_ids, mass_pos_chan_ids, wf_chan_ids, spr_gr_1_chan_ids
def read_mseed_channels(tracking_box: QTextBrowser, list_of_dir: List[str],
is_multiplex: bool,
on_unittest: bool = False
) -> Set[str]:
) -> Dict[str, Dict]:
"""
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
......@@ -99,12 +109,10 @@ def read_mseed_channels(tracking_box: QTextBrowser, list_of_dir: List[str],
:param list_of_dir: list of directories selected by users
:param is_multiplex: flag that tell if data is multiplex
:param on_unittest: flag to avoid cursor code to display tracking_info
:return data_object.channels: set of channels present in listofDir
:return channel_info: Dict of channels' name categorized into soh, masspos,
waveform, soh with sample rate greater than one for each station.
"""
soh_chan_ids = set()
mass_pos_chan_ids = set()
wf_chan_ids = set()
spr_gr_1_chan_ids = set()
channel_info = {}
count = 0
if not on_unittest:
QApplication.setOverrideCursor(QCursor(Qt.WaitCursor))
......@@ -135,24 +143,30 @@ def read_mseed_channels(tracking_box: QTextBrowser, list_of_dir: List[str],
f"RT130 file: {path2file}",
LogType.ERROR
)
return [], [], [], []
return channel_info
try:
# skip when data type isn't mseed
ret = _read_mseed_chanids(path2file, is_multiplex)
soh_chan_ids.update(ret[0])
mass_pos_chan_ids.update(ret[1])
wf_chan_ids.update(ret[2])
spr_gr_1_chan_ids.update(ret[3])
_read_mseed_chanids(
path2file, is_multiplex, channel_info)
except Exception:
display_tracking_info(
tracking_box, f'Skip non-mseed file: {file_name}',
LogType.WARNING
)
pass
if not on_unittest:
QApplication.restoreOverrideCursor()
return sorted(list(soh_chan_ids)), sorted(list(mass_pos_chan_ids)), \
sorted(list(wf_chan_ids)), sorted(list(spr_gr_1_chan_ids))
for sta_id in channel_info:
channel_info[sta_id]['mass_pos'] = sorted(list(
channel_info[sta_id]['mass_pos']))
channel_info[sta_id]['waveform'] = sorted(list(
channel_info[sta_id]['waveform']))
channel_info[sta_id]['soh'] = sorted(list(
channel_info[sta_id]['soh']))
channel_info[sta_id]['soh_spr_gr_1'] = sorted(list(
channel_info[sta_id]['soh_spr_gr_1']))
return channel_info
def detect_data_type(list_of_dir: List[str]) -> Optional[str]:
......
from typing import Dict, List, Union
from pathlib import Path
from PySide2 import QtWidgets, QtCore
from PySide2.QtWidgets import QDialogButtonBox, QDialog, QPlainTextEdit
from PySide2.QtWidgets import QDialogButtonBox, QDialog, QPlainTextEdit, \
QMainWindow
from sohstationviewer.database.process_db import (
execute_db, trunc_add_db, execute_db_dict)
from sohstationviewer.controller.processing import (
read_mseed_channels, detect_data_type
)
from sohstationviewer.controller.processing import read_mseed_channels
from sohstationviewer.controller.util import display_tracking_info
from sohstationviewer.controller.plotting_data import format_time
from sohstationviewer.view.util.enums import LogType
from sohstationviewer.view.util.one_instance_at_a_time import \
......@@ -46,17 +48,16 @@ class InputDialog(QDialog):
class ChannelPreferDialog(OneWindowAtATimeDialog):
def __init__(self, parent, dir_names):
def __init__(self, parent: QMainWindow, dir_names: List[Path]):
"""
Dialog to create lists of preferred SOH channels that users want to
select for plotting.
User can either fill in manually, get all channels for selected data
type from DB or scan the current selected folder for the list.
:param parent: QWidget/QMainWindow - widget that calls this plotting
:param parent: widget that calls this plotting
widget
:param dir_names: str - absolute path to the folder that contains
data
:param dir_names: list of absolute paths of data sets
"""
super(ChannelPreferDialog, self).__init__()
self.parent = parent
......@@ -138,19 +139,51 @@ class ChannelPreferDialog(OneWindowAtATimeDialog):
row_sel_radio_btns: list of radio buttons to select rows
"""
self.row_sel_radio_btns: List[QtWidgets.QRadioButton] = []
"""
channel_info: dictionary that stores chan_ids scanned from files for
different categories and stores data time, following each data key
and keeps radio buttons to switch between key.
"""
self.channel_info = {}
# ============================ GUI ===============================
"""
chan_ids_row_radio_button_group: group for radio buttons of chan_ids
rows
"""
self.chan_ids_row_radio_button_group = QtWidgets.QButtonGroup(self)
"""
data_set_key_radio_button_group: group for radio buttons to select
data set key
"""
self.data_set_key_radio_button_group = QtWidgets.QButtonGroup(self)
"""
data_set_key_radio_button_dict: radio buttons to select data set key
"""
self.data_set_key_radio_button_dict: Dict[
str, QtWidgets.QRadioButton] = {}
self.create_soh_list_table_widget()
main_layout.addWidget(self.soh_list_table_widget, 1)
button_layout = self.create_buttons_section()
main_layout.addLayout(button_layout)
"""
tracking_info_text_browser: - to display tracking info
tracking_info_text_browser: to display tracking info
when scanning for available channel list from given folder
"""
self.tracking_info_text_browser: QtWidgets.QTextBrowser = \
QtWidgets.QTextBrowser(self)
self.tracking_info_text_browser.setFixedHeight(80)
main_layout.addWidget(self.tracking_info_text_browser)
"""
keys_layout: layout for radio buttons representing data set keys
"""
self.data_set_key_layout = QtWidgets.QHBoxLayout()
self.data_set_key_layout.addWidget(QtWidgets.QLabel("Data Key:"))
main_layout.addLayout(self.data_set_key_layout)
"""
data_time_label: label for displaying data time.
"""
self.data_time_Label = QtWidgets.QLabel("Data Time:")
main_layout.addWidget(self.data_time_Label)
def create_buttons_section(self):
"""
......@@ -158,13 +191,15 @@ class ChannelPreferDialog(OneWindowAtATimeDialog):
"""
h_layout = QtWidgets.QHBoxLayout()
self.add_db_chan_btn = QtWidgets.QPushButton(
self, text='Add DB Channels')
self, text='Add Channels from DB')
self.add_db_chan_btn.clicked.connect(self.add_db_channels)
h_layout.addWidget(self.add_db_chan_btn)
self.scan_chan_btn = QtWidgets.QPushButton(
self, text='Scan Channels from Data Source')
self, text='Scan Channels from MSeed Data')
self.scan_chan_btn.clicked.connect(self.scan_channels)
if self.dir_names == [] or self.parent.data_type == "RT130":
self.scan_chan_btn.setEnabled(False)
h_layout.addWidget(self.scan_chan_btn)
self.save_btn = QtWidgets.QPushButton(self, text='Save')
......@@ -222,7 +257,9 @@ class ChannelPreferDialog(OneWindowAtATimeDialog):
:param row_idx: row index
"""
self.row_sel_radio_btns.append(QtWidgets.QRadioButton(self))
self.row_sel_radio_btns.append(QtWidgets.QRadioButton())
self.chan_ids_row_radio_button_group.addButton(
self.row_sel_radio_btns[row_idx])
self.row_sel_radio_btns[row_idx].clicked.connect(
lambda checked: self.curr_sel_changed(row_idx))
self.soh_list_table_widget.setCellWidget(
......@@ -402,14 +439,8 @@ class ChannelPreferDialog(OneWindowAtATimeDialog):
"""
if not self.validate_row():
return
if self.parent.rt130_das_dict != {}:
data_type = 'RT130'
else:
try:
data_type, is_multiplex = detect_data_type(self.dir_names)
except Exception as e:
QtWidgets.QMessageBox.warning(self, "Scan Channels", str(e))
return
data_type, is_multiplex = (self.parent.data_type,
self.parent.is_multiplex)
if data_type in self.avail_data_types:
self.data_type_combobox.setCurrentText(data_type)
else:
......@@ -417,8 +448,8 @@ class ChannelPreferDialog(OneWindowAtATimeDialog):
if data_type == 'RT130':
self.scan_chan_btn.setEnabled(False)
msg = ("Data type of the current data set is RT130 of which "
"Scan Channel from Data Source isn't available.\n\n"
msg = ("Data type of the current data set is RT130 for which "
"Scan Channel from MSeed Data isn't available.\n\n"
"Do you want to add channels from DB?")
result = QtWidgets.QMessageBox.question(
self, "Add Channels from DB for RT130", msg,
......@@ -426,32 +457,75 @@ class ChannelPreferDialog(OneWindowAtATimeDialog):
if result == QtWidgets.QMessageBox.Ok:
self.add_db_channels()
else:
self.scan_chan_btn.setEnabled(True)
ret = read_mseed_channels(self.tracking_info_text_browser,
self.dir_names, is_multiplex)
if ret == ([], [], [], []):
# clear all key radio buttons
for i in reversed(range(self.data_set_key_layout.count())):
widget = self.data_set_key_layout.takeAt(i).widget()
if widget is not None:
widget.setParent(None)
self.channel_info = read_mseed_channels(
self.tracking_info_text_browser, self.dir_names, is_multiplex)
if len(self.channel_info) == 0:
msg = "No data can be read from " + ', '.join(self.dir_names)
return QtWidgets.QMessageBox.warning(self, "No data", msg)
self.scan_chan_btn.setEnabled(True)
self.data_set_key_layout.addWidget(QtWidgets.QLabel("Data Key:"))
sta_ids = sorted(list(self.channel_info.keys()))
set_checked = True
self.data_set_key_radio_button_dict = {}
for sta_id in sta_ids:
# set new key radio buttons
self.data_set_key_radio_button_dict[sta_id] = \
QtWidgets.QRadioButton(sta_id)
self.data_set_key_radio_button_group.addButton(
self.data_set_key_radio_button_dict[sta_id])
self.data_set_key_radio_button_dict[sta_id].toggled.connect(
lambda: self.change_station(is_multiplex))
self.data_set_key_layout.addWidget(
self.data_set_key_radio_button_dict[sta_id])
if set_checked:
self.data_set_key_radio_button_dict[sta_id].setChecked(
True)
set_checked = False
self.data_set_key_layout.addStretch(1)
self.change_station(is_multiplex)
(soh_chan_ids, mass_pos_chan_ids,
wf_chan_ids, spr_gt_1_chan_ids) = ret
self.soh_list_item.setText(','.join(soh_chan_ids))
msg = ""
if mass_pos_chan_ids:
msg += f"Mass position channels (MP): " \
f"{','.join(mass_pos_chan_ids)}\n"
if wf_chan_ids:
msg += f"Waveform channels (WF): {','.join(wf_chan_ids)}\n"
if spr_gt_1_chan_ids:
msg += (
f"Non-WF with spr>1: "
f"{','.join(spr_gt_1_chan_ids)}\nNon-WF with spr>1 can be "
f"added to SOH channel list at your own risk because they "
f"slow down the performance significantly.")
if msg != "":
msg = f"<pre>{msg}</pre>"
display_tracking_info(self.tracking_info_text_browser, msg)
@QtCore.Slot()
def change_station(self, is_multiplex):
sel_sta_id = [
sta_id for sta_id in self.channel_info
if self.data_set_key_radio_button_dict[sta_id].isChecked()][0]
self.soh_list_item.setText(
','.join(self.channel_info[sel_sta_id]['soh']))
msg = ""
if self.channel_info[sel_sta_id]['mass_pos']:
msg += f"Mass position channels (MP): " \
f"{','.join(self.channel_info[sel_sta_id]['mass_pos'])}\n"
if self.channel_info[sel_sta_id]['waveform']:
msg += f"Waveform channels (WF): " \
f"{','.join(self.channel_info[sel_sta_id]['waveform'])}\n"
if self.channel_info[sel_sta_id]['soh_spr_gr_1']:
msg += (
f"Non-WF with spr>1: "
f"{','.join(self.channel_info[sel_sta_id]['soh_spr_gr_1'])}\n"
f"Non-WF with spr>1 can be added to SOH channel list "
f"manually. They are not added automatically because they are "
f"generally not SOH channels.")
if msg != "":
msg = f"<pre>{msg}</pre>"
display_tracking_info(self.tracking_info_text_browser, msg)
start_time = format_time(self.channel_info[sel_sta_id]['start_time'],
'YYYY-MM-DD',
'HH:MM:SS')
if not is_multiplex:
time_msg = f"Start Time: {start_time}"
else:
end_time = format_time(self.channel_info[sel_sta_id]['end_time'],
'YYYY-MM-DD',
'HH:MM:SS')
time_msg = f"Data Time: {start_time} - {end_time}"
self.data_time_Label.setText(time_msg)
@QtCore.Slot()
def save(self):
......
......@@ -58,7 +58,7 @@ class MainWindow(QtWidgets.QMainWindow, UIMainWindow):
super().__init__(parent)
self.setup_ui(self)
"""
dir_names: list of absolute path of data set directory
dir_names: list of absolute paths of data sets
"""
self.dir_names: List[Path] = []
"""
......@@ -279,8 +279,12 @@ class MainWindow(QtWidgets.QMainWindow, UIMainWindow):
try:
self.read_from_file_list()
except Exception as e:
QtWidgets.QMessageBox.warning(self, "Select directory", str(e))
return
msg = f"{str(e)}!!!\n\nDo you want to continue?"
result = QtWidgets.QMessageBox.question(
self, "Confirmation", msg,
QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No)
if result == QtWidgets.QMessageBox.No:
return
try:
win = ChannelPreferDialog(self, self.dir_names)
win.show()
......
......@@ -4,6 +4,8 @@ from pathlib import Path
from unittest import TestCase
from unittest.mock import patch
from obspy import UTCDateTime
from sohstationviewer.controller.processing import (
read_mseed_channels,
detect_data_type,
......@@ -47,10 +49,14 @@ class TestReadMSeedChannel(TestCase):
q330_spr_gt_1 = []
ret = read_mseed_channels(self.widget_stub, [q330_dir],
False, True)
self.assertListEqual(ret[0], q330_soh_channels)
self.assertListEqual(ret[1], q330_mass_pos_channels)
self.assertListEqual(ret[2], q330_wf_channels)
self.assertListEqual(ret[3], q330_spr_gt_1)
self.assertListEqual(list(ret.keys()), ['AX08'])
self.assertListEqual(ret['AX08']['soh'], q330_soh_channels)
self.assertListEqual(ret['AX08']['mass_pos'],
q330_mass_pos_channels)
self.assertListEqual(ret['AX08']['waveform'], q330_wf_channels)
self.assertListEqual(ret['AX08']['soh_spr_gr_1'], q330_spr_gt_1)
self.assertEqual(ret['AX08']['start_time'], 1625443228.34)
self.assertEqual(ret['AX08']['end_time'], UTCDateTime(0))
with self.subTest(("centaur - multiplex - is_multiplex=True")):
centaur_soh_channels = sorted(
......@@ -61,10 +67,14 @@ class TestReadMSeedChannel(TestCase):
centaur_spr_gt_1 = []
ret = read_mseed_channels(self.widget_stub, [centaur_dir],
True, True)
self.assertListEqual(ret[0], centaur_soh_channels)
self.assertListEqual(ret[1], centaur_mass_pos_channels)
self.assertListEqual(ret[2], centaur_wf_channels)
self.assertListEqual(ret[3], centaur_spr_gt_1)
self.assertListEqual(list(ret.keys()), ['3734'])
self.assertListEqual(ret['3734']['soh'], centaur_soh_channels)
self.assertListEqual(ret['3734']['mass_pos'],
centaur_mass_pos_channels)
self.assertListEqual(ret['3734']['waveform'], centaur_wf_channels)
self.assertListEqual(ret['3734']['soh_spr_gr_1'], centaur_spr_gt_1)
self.assertEqual(ret['3734']['start_time'], 1534464000.0)
self.assertEqual(ret['3734']['end_time'], 1534550400.0)
with self.subTest("centaur - multiplex - is_multiplex=False"):
# not all channels detected if is_multiplex isn't set
......@@ -74,10 +84,14 @@ class TestReadMSeedChannel(TestCase):
centaur_spr_gt_1 = []
ret = read_mseed_channels(self.widget_stub, [centaur_dir],
False, True)
self.assertListEqual(ret[0], centaur_soh_channels)
self.assertListEqual(ret[1], centaur_mass_pos_channels)
self.assertListEqual(ret[2], centaur_wf_channels)
self.assertListEqual(ret[3], centaur_spr_gt_1)
self.assertListEqual(list(ret.keys()), ['3734'])
self.assertListEqual(ret['3734']['soh'], centaur_soh_channels)
self.assertListEqual(ret['3734']['mass_pos'],
centaur_mass_pos_channels)
self.assertListEqual(ret['3734']['waveform'], centaur_wf_channels)
self.assertListEqual(ret['3734']['soh_spr_gr_1'], centaur_spr_gt_1)
self.assertEqual(ret['3734']['start_time'], 1534464000.0)
self.assertEqual(ret['3734']['end_time'], UTCDateTime(0))
with self.subTest("pegasus - non multiplex"):
pegasus_soh_channels = sorted(['VDT', 'VE1'])
......@@ -86,10 +100,14 @@ class TestReadMSeedChannel(TestCase):
pegasus_spr_gt_1 = []
ret = read_mseed_channels(self.widget_stub, [pegasus_dir],
False, True)
self.assertListEqual(ret[0], pegasus_soh_channels)
self.assertListEqual(ret[1], pegasus_mass_pos_channels)
self.assertListEqual(ret[2], pegasus_wf_channels)
self.assertListEqual(ret[3], pegasus_spr_gt_1)
self.assertListEqual(list(ret.keys()), ['KC01'])
self.assertListEqual(ret['KC01']['soh'], pegasus_soh_channels)
self.assertListEqual(ret['KC01']['mass_pos'],
pegasus_mass_pos_channels)
self.assertListEqual(ret['KC01']['waveform'], pegasus_wf_channels)
self.assertListEqual(ret['KC01']['soh_spr_gr_1'], pegasus_spr_gt_1)
self.assertEqual(ret['KC01']['start_time'], 1588978560.0)
self.assertEqual(ret['KC01']['end_time'], UTCDateTime(0))
with self.subTest("q330 - multiplex - is_multiplex=True"):
multiplex_soh_channels = ['LOG']
......@@ -104,10 +122,17 @@ class TestReadMSeedChannel(TestCase):
'SS1', 'SS2', 'SS3', 'SS4', 'SS5', 'SS6'])
ret = read_mseed_channels(self.widget_stub, [multiplex_dir],
True, True)
self.assertListEqual(ret[0], multiplex_soh_channels)
self.assertListEqual(ret[1], multiplex_mass_pos_channels)
self.assertListEqual(ret[2], multiplex_wf_channels)
self.assertListEqual(ret[3], multiplex_spr_gt_1)
self.assertListEqual(list(ret.keys()), ['3203'])
self.assertListEqual(ret['3203']['soh'], multiplex_soh_channels)
self.assertListEqual(ret['3203']['mass_pos'],
multiplex_mass_pos_channels)
self.assertListEqual(ret['3203']['waveform'],
multiplex_wf_channels)
self.assertListEqual(ret['3203']['soh_spr_gr_1'],
multiplex_spr_gt_1)
self.assertEqual(ret['3203']['start_time'],
1671729934.9000392)
self.assertEqual(ret['3203']['end_time'], 1671733530.57)
with self.subTest("q330 - multiplex - is_multiplex=False"):
# not all channels detected if is_multiplex isn't set
......@@ -117,10 +142,17 @@ class TestReadMSeedChannel(TestCase):
multiplex_spr_gt_1 = sorted([])
ret = read_mseed_channels(self.widget_stub, [multiplex_dir],
False, True)
self.assertListEqual(ret[0], multiplex_soh_channels)
self.assertListEqual(ret[1], multiplex_mass_pos_channels)
self.assertListEqual(ret[2], multiplex_wf_channels)
self.assertListEqual(ret[3], multiplex_spr_gt_1)
self.assertListEqual(list(ret.keys()), ['3203'])
self.assertListEqual(ret['3203']['soh'], multiplex_soh_channels)
self.assertListEqual(ret['3203']['mass_pos'],
multiplex_mass_pos_channels)
self.assertListEqual(ret['3203']['waveform'],
multiplex_wf_channels)
self.assertListEqual(ret['3203']['soh_spr_gr_1'],
multiplex_spr_gt_1)
self.assertEqual(ret['3203']['start_time'],
1671730004.145029)
self.assertEqual(ret['3203']['end_time'], UTCDateTime(0))
def test_read_channels_rt130_dir(self):
"""
......@@ -128,7 +160,7 @@ class TestReadMSeedChannel(TestCase):
RT130 data.
"""
ret = read_mseed_channels(self.widget_stub, [rt130_dir], True, True)
self.assertEqual(ret, ([], [], [], []))
self.assertEqual(ret, {})
def test_read_mseed_channels_no_dir(self):
"""
......@@ -137,7 +169,7 @@ class TestReadMSeedChannel(TestCase):
"""
no_dir = []
ret = read_mseed_channels(self.widget_stub, no_dir, True, True)
self.assertEqual(ret, ([], [], [], []))
self.assertEqual(ret, {})
def test_read_mseed_channels_dir_does_not_exist(self):
"""
......@@ -146,12 +178,12 @@ class TestReadMSeedChannel(TestCase):
"""
empty_name_dir = ['']
ret = read_mseed_channels(self.widget_stub, empty_name_dir, True, True)
self.assertEqual(ret, ([], [], [], []))
self.assertEqual(ret, {})
non_existent_dir = ['non_existent_dir']
ret = read_mseed_channels(self.widget_stub, non_existent_dir,
True, True)
self.assertEqual(ret, ([], [], [], []))
self.assertEqual(ret, {})
def test_read_mseed_channels_empty_dir(self):
"""
......@@ -161,7 +193,7 @@ class TestReadMSeedChannel(TestCase):
with TemporaryDirectory() as empty_dir:
ret = read_mseed_channels(self.widget_stub, [empty_dir],
True, True)
self.assertEqual(ret, ([], [], [], []))
self.assertEqual(ret, {})
def test_read_mseed_channels_empty_data_dir(self):
"""
......@@ -172,7 +204,7 @@ class TestReadMSeedChannel(TestCase):
with TemporaryDirectory(dir=outer_dir):
ret = read_mseed_channels(self.widget_stub, [outer_dir],
True, True)
self.assertEqual(ret, ([], [], [], []))
self.assertEqual(ret, {})
class TestDetectDataType(TestCase):
......