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)
Showing
with 456 additions and 282 deletions
......@@ -9,13 +9,18 @@ def get_chan_plot_info(org_chan_id: str, data_type: str,
color_mode: ColorMode = 'B') -> Dict:
"""
Given chanID read from raw data file and detected dataType
Return plotting info from DB for that channel
Return plotting info from DB for that channel.
:param org_chan_id: channel name read from data source
:param chan_info: info of the channel read from data source
:param data_type: type of data
:param color_mode: B/W
:return info of channel read from DB which is used for plotting
:return chan_db_info[0]: info of channel read from DB which is used for
plotting. In which,
+ Key 'dbChannel' keeps channel's name in DB
+ Key 'channel' keeps channel's name read from data
+ Key 'dbLabel' keeps label value in DB
+ Key 'label' keeps label to be displayed in the plotting
"""
chan = org_chan_id
chan = convert_actual_channel_to_db_channel_w_question_mark(chan)
......@@ -28,8 +33,8 @@ def get_chan_plot_info(org_chan_id: str, data_type: str,
# Seeing as we only need one of these columns for a color mode, we only
# pull the needed valueColors column from the database.
value_colors_column = 'valueColors' + color_mode
o_sql = (f"SELECT channel, plotType, height, unit,"
f" convertFactor, label, fixPoint, "
o_sql = (f"SELECT C.param as param, channel as dbChannel, plotType,"
f" height, unit, convertFactor, label as dbLabel, fixPoint, "
f"{value_colors_column} AS valueColors "
f"FROM Channels as C, Parameters as P")
if data_type == 'Unknown':
......@@ -42,13 +47,17 @@ def get_chan_plot_info(org_chan_id: str, data_type: str,
if len(chan_db_info) == 0:
chan_db_info = execute_db_dict(
f"{o_sql} WHERE channel='DEFAULT' and C.param=P.param")
chan_db_info[0]['channel'] = chan_db_info[0]['dbChannel']
else:
if chan_db_info[0]['channel'] == 'SEISMIC':
if chan_db_info[0]['dbChannel'] == 'SEISMIC':
seismic_label = get_seismic_chan_label(org_chan_id)
chan_db_info[0]['channel'] = org_chan_id
# add plotLabel key to be used in plotting.
# the original key label is unchanged to help when editing the channel's.
chan_db_info[0]['label'] = (
'' if chan_db_info[0]['label'] is None else chan_db_info[0]['label'])
'' if chan_db_info[0]['dbLabel'] is None
else chan_db_info[0]['dbLabel'])
chan_db_info[0]['unit'] = (
'' if chan_db_info[0]['unit'] is None else chan_db_info[0]['unit'])
chan_db_info[0]['fixPoint'] = (
......@@ -63,8 +72,8 @@ def get_chan_plot_info(org_chan_id: str, data_type: str,
elif seismic_label is not None:
chan_db_info[0]['label'] = seismic_label
else:
chan_db_info[0]['label'] = '-'.join([chan_db_info[0]['channel'],
chan_db_info[0]['label']])
chan_db_info[0]['label'] = '-'.join(
[chan_db_info[0]['channel'], chan_db_info[0]['label']])
if chan_db_info[0]['label'].strip() == 'DEFAULT':
chan_db_info[0]['label'] = 'DEFAULT-' + org_chan_id
return chan_db_info[0]
......@@ -179,14 +188,6 @@ def get_params():
return sorted([d[0] for d in param_rows])
def get_channel_info(chan_id: str, data_type: str):
# get channel info from DB
sql = f"SELECT * FROM Channels " \
f"WHERE channel='{chan_id}' AND dataType='{data_type}'"
chan_info = execute_db_dict(sql)[0]
return chan_info
def get_param_info(param: str):
# get all info of a param from DB
sql = f"SELECT * FROM Parameters WHERE param='{param}'"
......
......@@ -15,8 +15,7 @@ from sohstationviewer.view.plotting.gps_plot.gps_point import GPSPoint
from sohstationviewer.view.util.enums import LogType
from sohstationviewer.model.general_data.general_data_helper import \
retrieve_data_time_from_data_dict, retrieve_gaps_from_data_dict, \
combine_data, sort_data, squash_gaps, apply_convert_factor_to_data_dict, \
reset_data
combine_data, sort_data, squash_gaps, reset_data
from sohstationviewer.view.create_muti_buttons_dialog import (
create_multi_buttons_dialog)
......@@ -247,7 +246,6 @@ class GeneralData():
self.sort_all_data()
self.combine_all_data()
self.apply_convert_factor_to_data_dicts()
self.retrieve_gaps_from_data_dicts()
self.retrieve_data_time_from_data_dicts()
......@@ -408,19 +406,6 @@ class GeneralData():
if data_set_id not in self.log_data:
self.log_data[data_set_id] = {}
def apply_convert_factor_to_data_dicts(self):
"""
Applying convert_factor to avoid using flags to prevent double
applying convert factor when plotting
"""
for data_set_id in self.data_set_ids:
apply_convert_factor_to_data_dict(
data_set_id, self.soh_data, self.data_type)
apply_convert_factor_to_data_dict(
data_set_id, self.mass_pos_data, self.data_type)
apply_convert_factor_to_data_dict(
data_set_id, self.waveform_data, self.data_type)
def reset_all_selected_data(self):
"""
Remove all data_set_ids created in the plotting process.
......
......@@ -3,8 +3,6 @@ import numpy as np
import os
from pathlib import Path
from sohstationviewer.database.extract_data import get_convert_factor
def _check_related_gaps(min1: float, max1: float,
min2: float, max2: float,
......@@ -172,25 +170,6 @@ def combine_data(selected_data_set_id: Union[str, Tuple[str, str]],
}]
def apply_convert_factor_to_data_dict(
selected_data_set_id: Union[str, Tuple[str, str]],
data_dict: Dict, data_type: str) -> None:
"""
Traverse through traces in each channel to convert data according to
convert_factor got from DB
:param selected_data_set_id: the key of the selected data set
:param data_dict: dict of data
:param data_type: type of data
"""
selected_data_dict = data_dict[selected_data_set_id]
for chan_id in selected_data_dict:
channel = selected_data_dict[chan_id]
convert_factor = get_convert_factor(chan_id, data_type)
if convert_factor is not None and convert_factor != 1:
for tr in channel['tracesInfo']:
tr['data'] = convert_factor * tr['data']
def reset_data(selected_data_set_id: Union[str, Tuple[str, str]],
data_dict: Dict):
"""
......
......@@ -132,7 +132,6 @@ class RT130(GeneralData):
self.sort_all_data()
self.combine_all_data()
self.apply_convert_factor_to_data_dicts()
retrieve_gaps_from_stream_header(
self.stream_header_by_data_set_id_chan,
......
......@@ -3,14 +3,19 @@ import platform
import os
from typing import Optional, Dict
from matplotlib.axes import Axes
from PySide6 import QtWidgets, QtGui
from PySide6.QtWidgets import QWidget, QDialog
from sohstationviewer.database.process_db import execute_db
from sohstationviewer.database.extract_data import (
get_params, get_channel_info, create_assign_string_for_db_query
get_params, get_chan_plot_info
)
from sohstationviewer.view.plotting.plotting_widget.plotting import Plotting
from sohstationviewer.view.plotting.plotting_widget.plotting_axes import \
PlottingAxes
from sohstationviewer.view.db_config.edit_single_param_dialog import \
EditSingleParamDialog
......@@ -33,23 +38,25 @@ class AddEditSingleChannelDialog(QDialog):
"""
Dialog to add info for channel not in database or edit the existing channel
"""
def __init__(self, parent: Optional[QWidget],
chan_id: str, data_type: str):
def __init__(self, parent: Optional[QWidget], plotting: Plotting,
chan_id: str, data_type: str, ax: Axes):
"""
:param parent: the parent widget
:param plotting: object with plotting functions
:param chan_id: name of channel to be added/edited
:param data_type: type of the data being processed
:param ax: current axes to plot the channel the dialog working on
"""
self.parent = parent
# name of the channel
self.chan_id = chan_id
# data_type of the channel
self.data_type = data_type
self.ax = ax
self.plotting = plotting
# param of the channel
self.param: str = 'Default'
# True if this channel isn't in DB yet
self.is_new_db_channel: bool = False
# To skip on_param_chkbox_changed() when param is changed by the
# program at the beginning
self.param_changed_by_signal: bool = False
......@@ -99,8 +106,6 @@ class AddEditSingleChannelDialog(QDialog):
self.edit_param_btn = QtWidgets.QPushButton("EDIT PARAMETER", self)
# button to save changes to DB
self.save_btn = QtWidgets.QPushButton("SAVE CHANNEL", self)
# button to save changes and replot channel
self.save_replot_btn = QtWidgets.QPushButton("SAVE & REPLOT", self)
# button to close dialog without doing anything
self.cancel_btn = QtWidgets.QPushButton('CANCEL', self)
......@@ -109,7 +114,8 @@ class AddEditSingleChannelDialog(QDialog):
self.connect_signals()
def setup_ui(self) -> None:
dlg_type = 'Add' if 'DEFAULT' in self.chan_id else 'Edit'
dlg_type = ('Add' if 'DEFAULT' == self.ax.chan_db_info['channel']
else 'Edit')
self.setWindowTitle(f"{dlg_type} channel {self.chan_id}"
f" - {self.data_type}")
......@@ -146,36 +152,28 @@ class AddEditSingleChannelDialog(QDialog):
channel_layout.addWidget(QtWidgets.QLabel('Parameter'), 6, 0, 1, 1)
channel_layout.addWidget(self.param_cbobox, 6, 1, 1, 1)
channel_layout.addWidget(self.save_btn, 7, 0, 1, 1)
channel_layout.addWidget(self.save_replot_btn, 7, 1, 1, 1)
channel_layout.addWidget(self.edit_param_btn, 7, 1, 1, 1)
channel_layout.addWidget(self.edit_param_btn, 8, 1, 1, 1)
channel_layout.addWidget(self.save_btn, 8, 1, 1, 1)
channel_layout.addWidget(self.cancel_btn, 8, 0, 1, 1)
self.save_replot_btn.setFocus()
self.save_btn.setFocus()
def connect_signals(self) -> None:
self.param_cbobox.currentTextChanged.connect(
self.on_param_cbobox_changed)
self.cancel_btn.clicked.connect(self.close)
self.save_btn.clicked.connect(self.on_save)
self.save_replot_btn.clicked.connect(self.on_save_replot)
self.edit_param_btn.clicked.connect(self.on_edit_param)
def set_channel_info(self):
"""
Add all Channel related info according to information got from DB.
In case Channel isn't in the DB, use the info of DEFAULT channel.
Call set_param_info to set Parameter related info.
"""
try:
self.channel_info = get_channel_info(self.chan_id, self.data_type)
except IndexError:
self.is_new_db_channel = True
self.channel_info = get_channel_info('DEFAULT', 'Default')
self.channel_info = self.ax.chan_db_info
self.channel_name_lnedit.setText(self.chan_id)
self.label_lnedit.setText(self.channel_info['label'])
self.label_lnedit.setText(self.channel_info['dbLabel'])
self.conversion_lnedit.setText(
str(float(self.channel_info['convertFactor'])))
......@@ -212,7 +210,7 @@ class AddEditSingleChannelDialog(QDialog):
self.param_changed_by_signal = True
self.param_cbobox.setCurrentText(self.param)
return
if not self.is_new_db_channel:
if self.channel_info['param'] != 'Default':
msg = ("ARE YOU SURE YOU WANT TO CHANGE PARAMETER FOR CHANNEL "
f"'{self.chan_id}'?")
result = QtWidgets.QMessageBox.question(
......@@ -227,29 +225,25 @@ class AddEditSingleChannelDialog(QDialog):
self.param = new_param
self.set_buttons_enabled()
def save(self):
def on_save(self):
"""
Save info from GUI to DB
Save info from GUI to DB and replot according to new parameters
except for change in height.
"""
if self.is_new_db_channel:
if self.channel_info['channel'] == 'DEFAULT':
self.insert_channel_info()
else:
self.update_channel_info()
def on_save(self):
"""
Save new channel info to DB
"""
self.save()
self.close()
def on_save_replot(self):
"""
Save new channel info to DB
TODO: Replot the channel in the plotting area
"""
self.save()
print("Do REPLOT")
self.ax.c_data['chan_db_info'] = get_chan_plot_info(
self.channel_name_lnedit.text(),
self.data_type,
self.parent.color_mode
)
PlottingAxes.clean_axes(self.ax)
self.plotting.plot_channel(self.ax.c_data,
self.channel_name_lnedit.text(),
self.ax)
self.close()
def set_buttons_enabled(self):
......@@ -260,11 +254,9 @@ class AddEditSingleChannelDialog(QDialog):
if self.param == 'Default':
self.edit_param_btn.setEnabled(False)
self.save_btn.setEnabled(False)
self.save_replot_btn.setEnabled(False)
else:
self.edit_param_btn.setEnabled(True)
self.save_btn.setEnabled(True)
self.save_replot_btn.setEnabled(True)
def on_edit_param(self):
"""
......@@ -284,28 +276,12 @@ class AddEditSingleChannelDialog(QDialog):
win = EditSingleParamDialog(self, self.param_cbobox.currentText())
win.exec()
def update_para_info(self, param):
"""
Save parameter related info to Parameters table
:param param: param condition string
"""
plot_type = create_assign_string_for_db_query(
'plotType', self.plot_type_cbo_box.currentText())
value_colorb = create_assign_string_for_db_query(
'valueColorsB', self.value_colorb_widget.text())
value_colorw = create_assign_string_for_db_query(
'valueColorsW', self.value_colorw_widget.text())
height = f"height={self.height_spnbox.value()}"
sql = (f"UPDATE Parameters SET {plot_type}, {value_colorb}, "
f"{value_colorw}, {height} WHERE {param}")
execute_db(sql)
def insert_channel_info(self):
sql = ("INSERT INTO Channels VALUES ("
f"'{self.channel_name_lnedit.text()}', "
f"'{self.label_lnedit.text()}', "
f"'{self.param_cbobox.currentText()}', "
f"NULL, " # linkedChan for RT130 only and won't be changed
f"NULL, " # linkedChan won't be used anymore
f"{self.conversion_lnedit.text()}, "
f"'{self.unit_lnedit.text()}', "
f"{self.fix_point_spnbox.value()}, "
......@@ -320,9 +296,8 @@ class AddEditSingleChannelDialog(QDialog):
convert_factor = f"convertFactor={self.conversion_lnedit.text()}"
unit = f"unit='{self.unit_lnedit.text()}'"
fix_point = f"fixPoint={self.fix_point_spnbox.value()}"
data_type = f"dataType='{self.data_type_lnedit.text()}'"
sql = (f"UPDATE Channels SET {label}, {param}, {linked_chan}, "
f"{convert_factor}, {unit}, {fix_point}, {data_type} "
f"{convert_factor}, {unit}, {fix_point} "
f"WHERE {channel}")
execute_db(sql)
......
......@@ -57,6 +57,11 @@ class EditSingleParamDialog(QDialog):
self.height_spnbox.setMinimum(0)
self.height_spnbox.setMaximum(8)
self.height_spnbox.setToolTip("Relative height of the plot")
self.height_warning_label = QtWidgets.QLabel(
"(Height setting will only be applied after RePlot is clicked.)")
self.height_warning_label.setStyleSheet(
"QLabel {color: red; font-size: 10; font-style: italic;}"
)
# button to save change to DB
self.save_param_btn = QtWidgets.QPushButton(
......@@ -91,8 +96,9 @@ class EditSingleParamDialog(QDialog):
param_layout.addWidget(QtWidgets.QLabel('Height'), 3, 0, 1, 1)
param_layout.addWidget(self.height_spnbox, 3, 1, 1, 1)
param_layout.addWidget(self.cancel_btn, 4, 0, 1, 1)
param_layout.addWidget(self.save_param_btn, 4, 1, 1, 1)
param_layout.addWidget(self.height_warning_label, 4, 0, 1, 2)
param_layout.addWidget(self.cancel_btn, 5, 0, 1, 1)
param_layout.addWidget(self.save_param_btn, 5, 1, 1, 1)
def connect_signals(self) -> None:
self.plot_type_cbo_box.currentTextChanged.connect(self.set_plot_type)
......
......@@ -143,16 +143,30 @@ class PlottingAxes:
fontsize=const.FONTSIZE + 1)
timestamp_bar.set_xlim(self.parent.min_x, self.parent.max_x)
def create_axes(self, plot_b, plot_h, has_min_max_lines=True):
@staticmethod
def clean_axes(ax: Axes):
"""
Remove texts and plots on the given axes ax
:param ax: axes that has texts and plots to be removed
"""
for chan_plot in ax.chan_plots:
chan_plot.remove()
for text in ax.texts:
text.remove()
ax.chan_plots = []
def create_axes(self, plot_b: float, plot_h: float,
has_min_max_lines: bool = True) -> Axes:
"""
Create axes to plot a channel.
:param plot_b: float - bottom of the plot
:param plot_h: float - height of the plot
:param has_min_max_lines: bool - flag showing if the plot need min/max
lines
:return ax: matplotlib.axes.Axes - axes of a channel
:param plot_b: bottom of the plot
:param plot_h: height of the plot
:param has_min_max_lines: flag showing if the plot need min/max lines
:return ax: axes created
"""
ax = self.fig.add_axes(
[const.PLOT_LEFT_NORMALIZE, plot_b,
const.PLOT_WIDTH_NORMALIZE, plot_h],
......@@ -178,6 +192,8 @@ class PlottingAxes:
labelsize=const.FONTSIZE)
# transparent background => self.fig will take care of background
ax.patch.set_alpha(0)
# prepare chan_plots list to be reference for the plotted lines/dots
ax.chan_plots = []
return ax
def create_sample_no_label(self, ax: Axes, pos_y: float,
......@@ -221,7 +237,8 @@ class PlottingAxes:
:param sample_no_colors: list of color to display sample numbers
:param sample_no_pos: list of position to display sample numbers
[0.05, 0.5, 0.95] are the basic positions.
:param label: title of the plot. If None, show chan_db_info['label']
:param label: title of the plot. If label is None, use
chan_db_info['label']
:param info: additional info to show in sub title which is
smaller and under title on the left side
:param y_list: y values of the channel for min/max labels, lines
......@@ -256,6 +273,9 @@ class PlottingAxes:
size=const.FONTSIZE + 1
)
if not sample_no_pos:
ax.set_yticklabels([])
return
# set samples' total on right side
# bottom
ax.bottom_total_point_lbl = self.create_sample_no_label(
......
from typing import List, Union, Tuple, Dict
import numpy as np
from copy import copy
from sohstationviewer.view.util.enums import LogType
from sohstationviewer.conf import constants
from sohstationviewer.controller.util import get_val
from sohstationviewer.view.util.color import clr
from sohstationviewer.database.extract_data import get_convert_factor
# TODO: put this in DB
mass_pos_volt_ranges = {"regular": [0.5, 2.0, 4.0, 7.0],
......@@ -187,3 +190,20 @@ def get_colors_sizes_for_abs_y_from_value_colors(
# The last value color
colors[i] = clr[c]
return colors, sizes
def apply_convert_factor(data: List[np.ndarray], chan_id: str, data_type: str
) -> np.ndarray:
"""
Convert data according to convert_factor got from DB
:param data: list of value array
:param chan_id: name of channel
:param data_type: type of data
"""
convert_factor = get_convert_factor(chan_id, data_type)
if convert_factor is not None and convert_factor != 1:
new_data = [convert_factor * copy(data[0])]
return new_data
else:
return data
......@@ -172,6 +172,11 @@ class PlottingWidget(QtWidgets.QScrollArea):
# List of SOH message lines in RT130 to display in info box when
# there're more than 2 lines for one data point clicked
self.rt130_log_data: Optional[List[str]] = None
"""
log_idxes: line index of RT130's log messages
"""
self.log_idxes = None
# ----------------------------------------------------------------
QtWidgets.QScrollArea.__init__(self)
......@@ -342,6 +347,7 @@ class PlottingWidget(QtWidgets.QScrollArea):
"""
self.is_button_press_event_triggered_pick_event = True
artist = event.artist
self.log_idxes = None
if not isinstance(artist, pl.Line2D):
return
ax = artist.axes
......@@ -381,25 +387,16 @@ class PlottingWidget(QtWidgets.QScrollArea):
f"Time: {formatted_clicked_time} "
f"Value: {clicked_data}</pre>")
if 'logIdx' in chan_data.keys():
log_idxes = [chan_data['logIdx'][0][idx]
for idx in real_idxes]
self.log_idxes = [chan_data['logIdx'][0][idx]
for idx in real_idxes]
if len(real_idxes) > 1:
info_str = info_str.replace(
"</pre>", f" ({len(real_idxes)} lines)")
for idx in log_idxes:
for idx in self.log_idxes:
info_str += (
"<pre> " + self.rt130_log_data[idx] + "</pre>")
display_tracking_info(self.tracking_box, info_str)
if 'logIdx' in chan_data.keys():
# For Reftek, need to hightlight the corresponding
# SOH message lines based on the log_idxes of the clicked point
self.parent.search_message_dialog.show()
try:
self.parent.search_message_dialog. \
show_log_entry_from_log_indexes(log_idxes)
except ValueError as e:
QtWidgets.QMessageBox.warning(self, "Not found",
str(e))
def on_button_press_event(self, event):
"""
......
# Drawing State-Of-Health channels and mass position
from typing import Tuple, Union, Dict
from sohstationviewer.view.util.plot_func_names import plot_functions
from typing import Tuple, Union, Dict, Optional
from matplotlib.axes import Axes
from matplotlib.backend_bases import MouseButton
from PySide6 import QtWidgets, QtCore
from sohstationviewer.model.general_data.general_data import GeneralData
from sohstationviewer.view.util.enums import LogType
from sohstationviewer.view.plotting.plotting_widget.\
multi_threaded_plotting_widget import MultiThreadedPlottingWidget
from sohstationviewer.view.db_config.add_edit_single_channel_dialog import \
AddEditSingleChannelDialog
class SOHWidget(MultiThreadedPlottingWidget):
......@@ -18,6 +21,78 @@ class SOHWidget(MultiThreadedPlottingWidget):
def __init__(self, *args, **kwargs):
MultiThreadedPlottingWidget.__init__(self, *args, **kwargs)
"""
curr_ax: current axes to be edited
"""
self.curr_ax: Optional[Axes] = None
def on_button_press_event(self, event):
"""
When right-clicking on a plot with no Keyboard pressed,
set self.curr_ax to the ax of that plot.
"""
modifiers = event.guiEvent.modifiers()
if modifiers != QtCore.Qt.KeyboardModifier.NoModifier:
return super().on_button_press_event(event)
x = event.xdata
if x is None:
# when clicking outside of the plots
self.curr_ax = None
else:
if event.button == MouseButton.RIGHT:
# RIGHT click
self.curr_ax = event.inaxes
self.parent.raise_()
else:
# LEFT click
self.curr_ax = None
if self.log_idxes is not None:
# For Reftek, need to hightlight the corresponding
# SOH message lines based on the log_idxes of the clicked
# point
self.parent.search_message_dialog.show()
try:
self.parent.search_message_dialog. \
show_log_entry_from_log_indexes(self.log_idxes)
except ValueError as e:
QtWidgets.QMessageBox.warning(self, "Not found",
str(e))
else:
self.parent.raise_()
def contextMenuEvent(self, event):
"""
Create menu showing up when right click mouse to add/edit channel
"""
try:
if '?' in self.curr_ax.chan_db_info['dbChannel']:
warning_action_str = (
f"Channel '{self.curr_ax.chan_db_info['channel']} '"
"can't be edited because it has '?' in its DB name, "
f"{self.curr_ax.chan_db_info['dbChannel']}.")
elif 'DEFAULT' in self.curr_ax.chan_db_info['label']:
add_edit_action_str = f"Add new channel {self.curr_ax.chan}"
else:
add_edit_action_str = f"Edit channel {self.curr_ax.chan}"
except AttributeError:
return
context_menu = QtWidgets.QMenu(self)
try:
add_edit_chan_action = context_menu.addAction(add_edit_action_str)
add_edit_chan_action.triggered.connect(self.add_edit_channel)
except UnboundLocalError:
pass
try:
context_menu.addAction(warning_action_str)
except UnboundLocalError:
pass
context_menu.exec_(self.mapToGlobal(event.pos()))
self.curr_ax = None # to make sure curr_ax is clear
def init_plot(self, d_obj: GeneralData,
data_set_id: Union[str, Tuple[str, str]],
......@@ -70,13 +145,19 @@ class SOHWidget(MultiThreadedPlottingWidget):
data_structure.MD
:param chan_id: name of channel
"""
if len(c_data['times']) == 0:
return
chan_db_info = c_data['chan_db_info']
plot_type = chan_db_info['plotType']
ax = getattr(
self.plotting, plot_functions[plot_type]['plot_function'])(
c_data, chan_db_info, chan_id)
ax = self.plotting.plot_channel(c_data, chan_id, None)
c_data['ax'] = ax
ax.c_data = c_data
ax.chan = chan_id
self.axes.append(ax)
def add_edit_channel(self):
win = AddEditSingleChannelDialog(
self.parent,
self.plotting,
self.curr_ax.chan,
self.parent.data_type,
self.curr_ax
)
win.exec()
self.draw()
......@@ -5,7 +5,6 @@ from PySide6 import QtCore, QtWidgets
from sohstationviewer.model.general_data.general_data import GeneralData
from sohstationviewer.view.util.plot_func_names import plot_functions
from sohstationviewer.view.plotting.plotting_widget.\
multi_threaded_plotting_widget import MultiThreadedPlottingWidget
......@@ -47,15 +46,9 @@ class WaveformWidget(MultiThreadedPlottingWidget):
'times' and 'data'. Refer to general_data/data_structures.MD
:param chan_id: name of channel
"""
if len(c_data['times']) == 0:
return
chan_db_info = c_data['chan_db_info']
plot_type = chan_db_info['plotType']
# refer to doc string for mass_pos_data to know the reason for 'ax_wf'
ax = getattr(
self.plotting, plot_functions[plot_type]['plot_function'])(
c_data, chan_db_info, chan_id)
ax = self.plotting.plot_channel(c_data, chan_id, None)
# 'ax_wf' is the ax to plot mass position in WaveformWidget.
# Refer to data_structures.MD for more explanation
c_data['ax_wf'] = ax
ax.chan = chan_id
self.axes.append(ax)
......
......@@ -16,11 +16,14 @@ class TestGetChanPlotInfo(BaseTestCase):
Test basic functionality of get_chan_plot_info - channel and data type
combination exists in database table `Channels`
"""
expected_result = {'channel': 'SOH/Data Def',
expected_result = {'param': 'SOH data definitions',
'dbChannel': 'SOH/Data Def',
'channel': 'SOH/Data Def',
'plotType': 'upDownDots',
'height': 2,
'unit': '',
'convertFactor': 1,
'dbLabel': None,
'label': 'SOH/Data Def',
'fixPoint': 0,
'valueColors': '0:W|1:C'}
......@@ -29,11 +32,14 @@ class TestGetChanPlotInfo(BaseTestCase):
def test_masspos_channel(self):
with self.subTest("Mass position 'VM'"):
expected_result = {'channel': 'VM1',
expected_result = {'param': 'Mass position',
'dbChannel': 'VM?',
'channel': 'VM1',
'plotType': 'linesMasspos',
'height': 4,
'unit': 'V',
'convertFactor': 0.1,
'dbLabel': 'MassPos',
'label': 'VM1-MassPos',
'fixPoint': 1,
'valueColors': None}
......@@ -41,11 +47,14 @@ class TestGetChanPlotInfo(BaseTestCase):
expected_result)
with self.subTest("Mass position 'MassPos'"):
expected_result = {'channel': 'MassPos1',
expected_result = {'param': 'Mass position',
'dbChannel': 'MassPos?',
'channel': 'MassPos1',
'plotType': 'linesMasspos',
'height': 4,
'unit': 'V',
'convertFactor': 1,
'dbLabel': None,
'label': 'MassPos1',
'fixPoint': 1,
'valueColors': None}
......@@ -54,26 +63,32 @@ class TestGetChanPlotInfo(BaseTestCase):
def test_seismic_channel(self):
with self.subTest("RT130 Seismic"):
expected_result = {'channel': 'DS2',
expected_result = {'param': 'Seismic data',
'dbChannel': 'SEISMIC',
'channel': 'DS2',
'plotType': 'linesSRate',
'height': 8,
'unit': '',
'convertFactor': 1,
'label': 'DS2',
'dbLabel': None,
'fixPoint': 0,
'valueColors': None}
'valueColors': None,
'label': 'DS2'}
self.assertDictEqual(get_chan_plot_info('DS2', 'RT130'),
expected_result)
with self.subTest("MSeed Seismic"):
expected_result = {'channel': 'LHE',
expected_result = {'param': 'Seismic data',
'dbChannel': 'SEISMIC',
'channel': 'LHE',
'plotType': 'linesSRate',
'height': 8,
'unit': '',
'convertFactor': 1,
'label': 'LHE-EW',
'dbLabel': 'SeismicData',
'fixPoint': 0,
'valueColors': None}
'valueColors': None,
'label': 'LHE-EW'}
self.assertDictEqual(get_chan_plot_info('LHE', 'Q330'),
expected_result)
......@@ -83,11 +98,14 @@ class TestGetChanPlotInfo(BaseTestCase):
string 'Unknown'.
"""
# Channel does not exist in database
expected_result = {'channel': 'DEFAULT',
expected_result = {'param': 'Default',
'dbChannel': 'DEFAULT',
'channel': 'DEFAULT',
'plotType': 'linesDots',
'height': 2,
'unit': '',
'convertFactor': 1,
'dbLabel': '',
'label': 'DEFAULT-Bad Channel ID',
'fixPoint': 0,
'valueColors': None}
......@@ -95,11 +113,14 @@ class TestGetChanPlotInfo(BaseTestCase):
expected_result)
# Channel exist in database
expected_result = {'channel': 'LCE',
expected_result = {'param': 'Clock phase error',
'dbChannel': 'LCE',
'channel': 'LCE',
'plotType': 'linesDots',
'height': 3,
'unit': 'us',
'convertFactor': 1,
'dbLabel': 'PhaseError',
'label': 'LCE-PhaseError',
'fixPoint': 0,
'valueColors': 'L:W|D:Y'}
......@@ -114,12 +135,15 @@ class TestGetChanPlotInfo(BaseTestCase):
not the string 'Unknown'.
"""
# noinspection PyDictCreation
expected_result = {'channel': 'DEFAULT',
expected_result = {'param': 'Default',
'dbChannel': 'DEFAULT',
'channel': 'DEFAULT',
'plotType': 'linesDots',
'height': 2,
'unit': '',
'convertFactor': 1,
'label': None, # Change for each test case
'dbLabel': '',
'label': 'DEFAULT-SOH/Data Def',
'fixPoint': 0,
'valueColors': None}
......
import numpy as np
from pathlib import Path
from unittest.mock import patch
from sohstationviewer.model.general_data.general_data_helper import (
_check_related_gaps, squash_gaps, sort_data,
retrieve_data_time_from_data_dict, retrieve_gaps_from_data_dict,
combine_data, apply_convert_factor_to_data_dict, read_text
combine_data, read_text
)
from tests.base_test_case import BaseTestCase
......@@ -302,22 +300,3 @@ class TestCombineData(BaseTestCase):
self.assertListEqual(
data_dict['STA1']['CH1']['tracesInfo'][0]['times'].tolist(),
[5, 8, 11, 15, 25, 29, 33, 36, 40])
class TestApplyConvertFactorToDataDict(BaseTestCase):
def setUp(self) -> None:
self.data_dict = {
'STA1': {
'CH1': {'tracesInfo': [{'data': np.array([1, 2, 2, -1])}]}
}
}
self.expected_data = [0.1, 0.2, 0.2, -0.1]
@patch('sohstationviewer.model.general_data.general_data_helper.'
'get_convert_factor')
def test_convert_factor(self, mock_get_convert_factor):
mock_get_convert_factor.return_value = 0.1
apply_convert_factor_to_data_dict('STA1', self.data_dict, 'Q330')
self.assertEqual(
self.data_dict['STA1']['CH1']['tracesInfo'][0]['data'].tolist(),
self.expected_data)
import numpy as np
from numpy import testing as nptesting
from unittest.mock import patch
from sohstationviewer.view.plotting.plotting_widget.plotting_helper import (
get_masspos_value_colors,
get_categorized_data_from_value_color_equal_on_upper_bound,
get_categorized_data_from_value_color_equal_on_lower_bound,
get_colors_sizes_for_abs_y_from_value_colors
get_colors_sizes_for_abs_y_from_value_colors,
apply_convert_factor
)
from sohstationviewer.view.util.color import clr
from tests.base_test_case import BaseTestCase
......@@ -163,3 +168,14 @@ class TestGetColorsSizesForAbsYFromValueColors(BaseTestCase):
]
)
self.assertEqual(sizes, [1.5] * len(y))
class TestApplyConvertFactor(BaseTestCase):
@patch('sohstationviewer.view.plotting.plotting_widget.plotting_helper.'
'get_convert_factor')
def test_convert_factor(self, mock_get_convert_factor):
mock_get_convert_factor.return_value = 0.1
test_data = [np.array([1, 2, 2, -1])]
expected_data = [np.array([0.1, 0.2, 0.2, -0.1])]
result_data = apply_convert_factor(test_data, 'CHA', 'Q330')
nptesting.assert_array_equal(result_data, expected_data)