from typing import Dict, List, Union
from PySide2 import QtWidgets, QtCore
from PySide2.QtWidgets import QDialogButtonBox, QDialog, QPlainTextEdit
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.util import display_tracking_info
from sohstationviewer.view.util.enums import LogType
from sohstationviewer.view.util.one_instance_at_a_time import \
OneWindowAtATimeDialog
INSTRUCTION = """
Select the list of SOH channels to be used in plotting.\n
Edit lists of channels to be read in the Preferred SOH List field.\n
Click "Save - Add to Main" to save the changes and add the selected list to \
main window.
"""
TOTAL_ROW = 20
COL = {'sel': 0, 'name': 1, 'dataType': 2, 'IDs': 3, 'edit': 4, 'clr': 5}
class InputDialog(QDialog):
def __init__(self, parent=None, text=''):
super().__init__(parent)
self.resize(500, 300)
self.text_box = QPlainTextEdit(self)
self.text_box.setPlainText(text)
button_box = QDialogButtonBox(
QDialogButtonBox.Ok | QDialogButtonBox.Cancel, self)
layout = QtWidgets.QFormLayout(self)
layout.addRow(self.text_box)
layout.addWidget(button_box)
button_box.accepted.connect(self.accept)
button_box.rejected.connect(self.reject)
def get_input(self):
return self.text_box.toPlainText()
class ChannelPreferDialog(OneWindowAtATimeDialog):
def __init__(self, parent, dir_names):
"""
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
widget
:param dir_names: str - absolute path to the folder that contains
data
"""
super(ChannelPreferDialog, self).__init__()
self.parent = parent
self.dir_names = dir_names
self.setWindowTitle("SOH Channel Preferences")
self.setGeometry(100, 100, 1100, 800)
main_layout = QtWidgets.QVBoxLayout()
main_layout.setContentsMargins(7, 7, 7, 7)
self.setLayout(main_layout)
main_layout.addWidget(QtWidgets.QLabel(INSTRUCTION))
# ===================== soh_list_table_widget =========================
"""
soh_list_table_widget: the table to display/edit different
preferred channel id list
"""
self.soh_list_table_widget: Union[QtWidgets.QTableWidget, None] = None
"""
avail_data_types: list of available data types in DB
"""
self.avail_data_types: List[str] = []
"""
curr_row: current row
"""
self.curr_row: int = -1
# ========================== buttons ============================
"""
add_db_chan_btn: Button to add all channels for selected
data type from DB to add to the current row
"""
self.add_db_chan_btn: Union[QtWidgets.QPushButton, None] = None
"""
scan_chan_btn: Button to scan through all channels from dir_names
folder for the list of channels and data type to add
to the current row
"""
self.scan_chan_btn: Union[QtWidgets.QPushButton, None] = None
"""
save_btn: Button to save the selected preferred channel list to DB
"""
self.save_btn: Union[QtWidgets.QPushButton, None] = None
"""
save_add_main_btn: Button to save the selected preferred channel list
to DB, close the dialog and add to
parent.curr_pref_soh_list_name_txtbox
"""
self.save_add_main_btn: Union[QtWidgets.QPushButton, None] = None
"""
close_btn: button to close the dialog and do nothing
"""
self.close_btn: Union[QtWidgets.QPushButton, None] = None
self.changed = False
# ======================= pre-define ==========================
"""
soh_list_name_item: text box widget for name of preferred channel
list
"""
self.soh_list_name_item: Union[QtWidgets.QLineEdit, None] = None
"""
data_type_combobox: dropdown list to select datatype for
preferred channel list
"""
self.data_type_combobox: Union[QtWidgets.QComboBox, None] = None
"""
soh_list_item: text box for preferred channel list
"""
self.soh_list_item: Union[QtWidgets.QLineEdit, None] = None
"""
clear_btn: QPushButton - button to clear the preferred channel
"""
self.clear_btn: Union[QtWidgets.QPushButton, None] = None
"""
data_type: str - data_type of the current preferred channel
"""
self.data_type: Union[str, None] = None
"""
row_sel_radio_btns: list of radio buttons to select rows
"""
self.row_sel_radio_btns: List[QtWidgets.QRadioButton] = []
# ============================ GUI ===============================
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
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)
def create_buttons_section(self):
"""
Create the row of buttons
"""
h_layout = QtWidgets.QHBoxLayout()
self.add_db_chan_btn = QtWidgets.QPushButton(
self, text='Add DB Channels')
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.scan_chan_btn.clicked.connect(self.scan_channels)
h_layout.addWidget(self.scan_chan_btn)
self.save_btn = QtWidgets.QPushButton(self, text='Save')
self.save_btn.clicked.connect(self.save)
h_layout.addWidget(self.save_btn)
self.save_add_main_btn = QtWidgets.QPushButton(
self, text='Save - Add to Main')
self.save_add_main_btn.clicked.connect(self.save_add_to_main_and_close)
h_layout.addWidget(self.save_add_main_btn)
self.close_btn = QtWidgets.QPushButton(self, text='Cancel')
self.close_btn.clicked.connect(self.close)
h_layout.addWidget(self.close_btn)
return h_layout
def create_soh_list_table_widget(self):
"""
Create self.soh_list_table_widget to view and edit different lists of
preferred channels
"""
self.soh_list_table_widget = QtWidgets.QTableWidget(self)
self.soh_list_table_widget.cellClicked.connect(self.cell_clicked)
self.soh_list_table_widget.itemChanged.connect(self.input_changed)
self.soh_list_table_widget.setColumnCount(6)
col_headers = ['', 'Name', 'DataType', 'Preferred SOH List',
'Edit', 'Clear']
self.soh_list_table_widget.setHorizontalHeaderLabels(col_headers)
header = self.soh_list_table_widget.horizontalHeader()
header.setSectionResizeMode(QtWidgets.QHeaderView.ResizeToContents)
header.setSectionResizeMode(3, QtWidgets.QHeaderView.Stretch)
self.soh_list_table_widget.setRowCount(TOTAL_ROW)
self.avail_data_types = self.get_data_types()
for row_idx in range(TOTAL_ROW):
self.add_row(row_idx)
self.update_data_table_widget_items()
def cell_clicked(self, row, col):
"""
Turn the radio button of row to checked and set the row selected
:param row: row id
:type row: int
:param col: column id
:type col: int
"""
self.row_sel_radio_btns[row].setChecked(True)
self.curr_sel_changed(row)
@QtCore.Slot()
def add_row(self, row_idx):
"""
Add a row to self.soh_list_table_widget
:param row_idx: row index
"""
self.row_sel_radio_btns.append(QtWidgets.QRadioButton(self))
self.row_sel_radio_btns[row_idx].clicked.connect(
lambda checked: self.curr_sel_changed(row_idx))
self.soh_list_table_widget.setCellWidget(
row_idx, COL['sel'], self.row_sel_radio_btns[row_idx])
name_item = QtWidgets.QTableWidgetItem()
self.soh_list_table_widget.setItem(row_idx, COL['name'], name_item)
data_type_combo_box = QtWidgets.QComboBox(self)
data_type_combo_box.textActivated.connect(
lambda: self.cell_clicked(row_idx, 2))
data_type_combo_box.currentIndexChanged.connect(self.input_changed)
data_type_combo_box.addItems(['Unknown'] + self.avail_data_types)
data_type_combo_box.setCurrentIndex(-1)
self.soh_list_table_widget.setCellWidget(
row_idx, COL['dataType'], data_type_combo_box)
soh_list_item = QtWidgets.QTableWidgetItem()
self.soh_list_table_widget.setItem(row_idx, COL['IDs'], soh_list_item)
edit_button = QtWidgets.QPushButton(self, text='EDIT')
edit_button.clicked.connect(lambda arg: self.edit_soh_list(row_idx))
self.soh_list_table_widget.setCellWidget(
row_idx, COL['edit'], edit_button)
del_button = QtWidgets.QPushButton(self, text='CLR')
del_button.clicked.connect(lambda arg: self.clear_soh_list(row_idx))
self.soh_list_table_widget.setCellWidget(
row_idx, COL['clr'], del_button)
@QtCore.Slot()
def input_changed(self):
"""
Set self.changed to True if there is any info have been edit to
display a warning message be for processing DB
"""
self.changed = True
def update_data_table_widget_items(self):
"""
Update the content of self.soh_list_table_widget based on data read
from DB
"""
soh_list_rows = self.get_soh_list_rows()
# first row
self.soh_list_table_widget.cellWidget(0, COL['sel']).setChecked(True)
self.curr_sel_changed(0)
count = 0
for r in soh_list_rows:
self.soh_list_table_widget.cellWidget(
count, COL['sel']).setChecked(
True if r['current'] == 1 else False)
self.soh_list_table_widget.item(
count, COL['name']).setText(r['name'])
self.soh_list_table_widget.cellWidget(
count, COL['dataType']).setCurrentText(r['dataType'])
self.soh_list_table_widget.item(
count, COL['IDs']).setText(r['IDs'])
if r['current'] == 1:
self.curr_sel_changed(count)
self.soh_list_table_widget.selectRow(count)
count += 1
self.update()
def get_row(self, row_idx):
"""
Get content of a self.soh_list_table_widget's row
:param row_idx: int - index of a row
"""
soh_list_name_item = self.soh_list_table_widget.item(
row_idx, COL['name'])
data_type_combobox = self.soh_list_table_widget.cellWidget(
row_idx, COL['dataType'])
soh_list_item = self.soh_list_table_widget.item(
row_idx, COL['IDs'])
clear_widget = self.soh_list_table_widget.cellWidget(
row_idx, COL['clr'])
return (soh_list_name_item,
data_type_combobox,
soh_list_item,
clear_widget)
@QtCore.Slot()
def curr_sel_changed(self, row_idx):
"""
When check the radio button of a row, set self.curr_row with new
row_idx and get content of the row.
:param row_idx: int - index of a row
"""
if self.row_sel_radio_btns[row_idx].isChecked():
self.curr_row = row_idx
(self.soh_list_name_item, self.data_type_combobox,
self.soh_list_item, self.clear_btn) = self.get_row(row_idx)
@QtCore.Slot()
def edit_soh_list(self, row_idx):
soh_list_item = self.soh_list_table_widget.item(row_idx, COL['IDs'])
edit_dialog = InputDialog(text=soh_list_item.text())
if edit_dialog.exec_():
soh_list_item.setText(edit_dialog.get_input())
@QtCore.Slot()
def clear_soh_list(self, row_idx):
"""
When 'Clear' button at the end of a row is click. Confirm with user and
then clear the content of soh_list_item in that row.
:param row_idx: int - index of a row
"""
(soh_list_name_item, data_type_combobox,
soh_list_item, clear_widget) = self.get_row(row_idx)
if soh_list_item.text().strip() != "":
msg = ("Are you sure you want to delete the SOH channel list of "
"row #%s?" % (row_idx + 1))
result = QtWidgets.QMessageBox.question(
self, "Confirmation", msg,
QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No)
if result == QtWidgets.QMessageBox.No:
return
self.changed = True
if soh_list_name_item.text() != '':
self.changed = True
soh_list_name_item.setText('')
if data_type_combobox.currentText != '':
self.changed = True
data_type_combobox.setCurrentIndex(-1)
if soh_list_item.text() != '':
self.changed = True
soh_list_item.setText('')
def validate_row(self, check_data_type=False):
"""
Make sure there is a row selected by checking the radio button at the
beginning of a row.
Check if data_type is selected if required.
:param check_data_type: bool - flag to require checking data type
"""
if self.curr_row == -1:
msg = "Please select a row."
QtWidgets.QMessageBox.information(self, "Select row", msg)
return False
if check_data_type:
self.data_type = self.data_type_combobox.currentText()
if self.data_type not in self.avail_data_types:
msg = ("Please select a data type that isn't 'Unknown' for "
"the selected row.")
QtWidgets.QMessageBox.information(
self, "Select data type", msg)
return False
return True
@QtCore.Slot()
def add_db_channels(self):
"""
Add channels from DB to preferred channel list.
"""
if not self.validate_row(check_data_type=True):
return
db_channels = self.get_db_channels(self.data_type)
self.soh_list_item.setText(','.join(db_channels))
@QtCore.Slot()
def scan_channels(self):
"""
Scan for all available channels in folder self.dir_names to add to
preferred channel list.
For RT130, all SOH channels are kept in a log file. It will be more
reasonable to get channels from DB because the task of getting
channels from log files is like reading data from it.
"""
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
if data_type in self.avail_data_types:
self.data_type_combobox.setCurrentText(data_type)
else:
self.data_type_combobox.setCurrentText('Unknown')
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"
"Do you want to add channels from DB?")
result = QtWidgets.QMessageBox.question(
self, "Add Channels from DB for RT130", msg,
QtWidgets.QMessageBox.Ok | QtWidgets.QMessageBox.Cancel)
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 == ([], [], [], []):
msg = "No data can be read from " + ', '.join(self.dir_names)
return QtWidgets.QMessageBox.warning(self, "No data", msg)
(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 save(self):
"""
Save changes to DB by doing the following steps:
+ If there are any changes to inform user before save.
+ Create insert sql for each row
+ If there is nothing in preference list reset parent's setting for
preferred channels and finish.
+ Do truncate and execute insert sqls above
+ Set parent's setting for preferred channels with info from the
current selected row.
"""
if not self.validate_row():
return
if self.changed:
msg = ("All IDs in the database will be overwritten with "
"current IDs in the dialog.\nClick Cancel to stop updating "
"database.")
result = QtWidgets.QMessageBox.question(
self, "Confirmation", msg,
QtWidgets.QMessageBox.Ok | QtWidgets.QMessageBox.Cancel)
if result == QtWidgets.QMessageBox.Cancel:
return False
sql_list = []
for row_idx in range(TOTAL_ROW):
sql = self.create_save_row_sql(row_idx)
if sql is not None:
sql_list.append(sql)
if len(sql_list) == 0:
self.parent.pref_soh_list = []
self.parent.pref_soh_list_name = ''
self.parent.pref_soh_list_data_type = 'Unknown'
return True
ret = trunc_add_db('ChannelPrefer', sql_list)
if ret is not True:
display_tracking_info(self.parent, ret, LogType.ERROR)
self.parent.pref_soh_list = [
t.strip() for t in self.soh_list_item.text().split(',')]
self.parent.pref_soh_list_name = self.soh_list_name_item.text().strip()
self.parent.pref_soh_list_data_type = \
self.data_type_combobox.currentText()
self.changed = False
return True
def create_save_row_sql(self, row_idx):
"""
Read info from the row with index row_idx to create insert sql.
:param row_idx: int - index of a row
:return: str - insert sql with the row's info
"""
current = 1 if self.soh_list_table_widget.cellWidget(
row_idx, COL['sel']).isChecked() else 0
name = self.soh_list_table_widget.item(
row_idx, COL['name']).text()
data_type = self.soh_list_table_widget.cellWidget(
row_idx, COL['dataType']).currentText()
idx = self.soh_list_table_widget.item(
row_idx, COL['IDs']).text()
if idx.strip() == '':
return
if name.strip() == '' and idx.strip() != '':
msg = f"Please add Name for row {row_idx}."
QtWidgets.QMessageBox.information(self, "Missing info", msg)
return
return (f"INSERT INTO ChannelPrefer (name, IDs, dataType, current)"
f"VALUES ('{name}', '{idx}', '{data_type}', {current})")
@QtCore.Slot()
def save_add_to_main_and_close(self):
"""
Save changes to DB and reset main window's preferred channels with info
of the selected rows.
"""
if not self.save():
return
self.parent.curr_pref_soh_list_name_txtbox.setText(
self.parent.pref_soh_list_name)
self.parent.all_soh_chans_check_box.setChecked(False)
self.close()
@staticmethod
def get_data_types():
"""
Get list of data types from DB.
:return: [str, ] - list of data types
"""
data_type_rows = execute_db(
'SELECT * FROM DataTypes ORDER BY dataType ASC')
return [d[0] for d in data_type_rows]
@staticmethod
def get_db_channels(data_type) -> List[str]:
"""
Get all channels defined in DB table Channels for the given data type.
:param data_type: str - the given data type
"""
channel_rows = execute_db(
f"SELECT channel FROM CHANNELS"
f" WHERE dataType='{data_type}'"
f" AND param NOT IN ('Seismic data', 'Mass position')"
f" ORDER BY dataType ASC")
return sorted([c[0] for c in channel_rows])
@staticmethod
def get_soh_list_rows() -> List[Dict]:
"""
Get all data from DB table ChannelPrefer to fill up
self.soh_list_table_widget
:return id_rows: [dict,] - list of data for each row
"""
id_rows = execute_db_dict(
"SELECT name, IDs, dataType, current FROM ChannelPrefer "
" ORDER BY name ASC")
return id_rows