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 (22)
Showing
with 305 additions and 140 deletions
Pegasus firmware was updated to record VEI in millivolts.
Previous versions of the firmware recorded VEI in microvolts.
We decided to apply VEI's conversion factor in newer version of Pegasus.
If user may see higher than expected values in the older version of Pegasus.
\ No newline at end of file
......@@ -249,3 +249,22 @@ def get_unit_bitweight(chan_db_info: Dict, bitweight_opt: str) -> str:
else:
unit_bitweight = "{}%s" % unit
return unit_bitweight
def get_disk_size_format(value: float) -> str:
"""
Break down unit in byte system.
:param value: size in KiB
:return: size with unit
"""
size = value * 1024.
if size >= 1024. ** 4:
return "%.2fTiB" % (size / 1024. ** 4)
elif size >= 1024. ** 3:
return "%.2fGiB" % (size / 1024. ** 3)
elif size >= 1024. ** 2:
return "%.2fMiB" % (size / 1024. ** 2)
elif size >= 1024.:
return "%.2fKiB" % (size / 1024.)
else:
return "%dB" % size
......@@ -7,7 +7,7 @@ import os
import json
import re
from pathlib import Path
from typing import List, Optional, Dict, Tuple
from typing import List, Optional, Dict, Tuple, Union
from PySide6.QtCore import QEventLoop, Qt
from PySide6.QtGui import QCursor
......@@ -167,7 +167,7 @@ def read_mseed_channels(tracking_box: QTextBrowser, list_of_dir: List[str],
return channel_info
def detect_data_type(list_of_dir: List[str]) -> Optional[str]:
def detect_data_type(list_of_dir: List[Union[str, Path]]) -> Optional[str]:
"""
Detect data type for the given directories using get_data_type_from_file
:param list_of_dir: list of directories selected by users
......@@ -183,6 +183,10 @@ def detect_data_type(list_of_dir: List[str]) -> Optional[str]:
dir_data_type_dict = {}
is_multiplex_dict = {}
for d in list_of_dir:
try:
d = d.as_posix()
except AttributeError:
pass
data_type = "Unknown"
is_multiplex = None
for path, subdirs, files in os.walk(d):
......@@ -264,10 +268,6 @@ def get_data_type_from_file(
'VP', 'VL', 'VL', 'VH',
'UN', 'UP', 'UL', 'UH']
if any(x in path2file.name for x in wf_chan_posibilities):
# Skip checking waveform files which aren't signature channels
return None, False
file = open(path2file, 'rb')
chans_in_stream = set()
data_type = None
......@@ -286,6 +286,9 @@ def get_data_type_from_file(
return
chan = record.record_metadata.channel
if any([wf_pattern in chan for wf_pattern in wf_chan_posibilities]):
# Skip checking waveform files which aren't signature channels
return None, False
if is_multiplex is None:
chans_in_stream.add(chan)
if len(chans_in_stream) > 1:
......
......@@ -18,18 +18,10 @@ def get_chan_plot_info(org_chan_id: str, data_type: str,
:return info of channel read from DB which is used for plotting
"""
chan = org_chan_id
if org_chan_id.startswith('EX'):
chan = 'EX?'
if org_chan_id.startswith('VM'):
chan = 'VM?'
if org_chan_id.startswith('MassPos'):
chan = 'MassPos?'
chan = convert_actual_channel_to_db_channel_w_question_mark(chan)
if org_chan_id.startswith('DS'):
chan = 'SEISMIC'
if org_chan_id.startswith('Event DS'):
chan = 'Event DS?'
if org_chan_id.startswith('Disk Usage'):
chan = 'Disk Usage?'
if dbConf['seisRE'].match(chan):
chan = 'SEISMIC'
# The valueColors for each color mode is stored in a separate column.
......@@ -74,7 +66,16 @@ def get_chan_plot_info(org_chan_id: str, data_type: str,
return chan_db_info[0]
def get_convert_factor(chan_id, data_type):
def get_convert_factor(chan_id: str, data_type: str) -> float:
"""
Get the convert factor to convert data from count (value read from file)
to actual value to display.
:param chan_id: actual channel name read from file
:param data_type: type of data in data set
:return: converting factor
"""
chan_id = convert_actual_channel_to_db_channel_w_question_mark(chan_id)
sql = f"SELECT convertFactor FROM Channels WHERE channel='{chan_id}' " \
f"AND dataType='{data_type}'"
ret = execute_db(sql)
......@@ -84,6 +85,24 @@ def get_convert_factor(chan_id, data_type):
return None
def convert_actual_channel_to_db_channel_w_question_mark(chan_id: str) -> str:
"""
The digit in channel end with a digit is represented with the question
mark '?' in DB. This function change the real channel name to DB
channel name with '?'.
:param chan_id: real channel name
:return chan_id: channel name with '?' at the end if available
"""
sql = "SELECT * FROM Channels WHERE channel like '%?'"
ret = execute_db(sql)
question_channels = [c[0][:-1] for c in ret]
if any(chan_id.startswith(x) for x in question_channels):
if chan_id[-1].isdigit():
# to prevent the case prefix similar to prefix of channel w/o ?
chan_id = chan_id[:-1] + '?'
return chan_id
def get_seismic_chan_label(chan_id):
"""
Get label for chan_id in which data stream can use chan_id for label while
......
No preview for this file type
......@@ -423,31 +423,30 @@ class MainWindow(QtWidgets.QMainWindow, UIMainWindow):
"TPS or RAW checkboxes checked.\nPlease clear the "
"selection of waveform if you don't want to display the data.")
if self.tps_check_box.isChecked() or self.raw_check_box.isChecked():
if self.all_wf_chans_check_box.isChecked():
req_mseed_wildcards = ['*']
req_dss = ['*'] # all data stream
else:
req_dss = []
req_mseed_wildcards = []
for idx, ds_checkbox in enumerate(self.ds_check_boxes):
if ds_checkbox.isChecked():
req_dss.append(idx + 1)
if self.mseed_wildcard_edit.text().strip() != "":
req_mseed_wildcards = self.mseed_wildcard_edit.text(
).split(",")
if self.data_type == 'RT130':
req_wf_chans = req_dss
if req_dss != ['*'] and req_mseed_wildcards != []:
msg = 'MSeed Wildcards will be ignored for RT130.'
self.processing_log.append((msg, LogType.WARNING))
else:
req_wf_chans = req_mseed_wildcards
if req_mseed_wildcards != ['*'] and req_dss != []:
msg = ('Checked data streams will be ignored for '
'none-RT130 data type.')
self.processing_log.append((msg, LogType.WARNING))
if self.all_wf_chans_check_box.isChecked():
req_mseed_wildcards = ['*']
req_dss = ['*'] # all data stream
else:
req_dss = []
req_mseed_wildcards = []
for idx, ds_checkbox in enumerate(self.ds_check_boxes):
if ds_checkbox.isChecked():
req_dss.append(idx + 1)
if self.mseed_wildcard_edit.text().strip() != "":
req_mseed_wildcards = self.mseed_wildcard_edit.text(
).split(",")
if self.data_type == 'RT130':
req_wf_chans = req_dss
if req_dss != ['*'] and req_mseed_wildcards != []:
msg = 'MSeed Wildcards will be ignored for RT130.'
self.processing_log.append((msg, LogType.WARNING))
else:
req_wf_chans = req_mseed_wildcards
if req_mseed_wildcards != ['*'] and req_dss != []:
msg = ('Checked data streams will be ignored for '
'none-RT130 data type.')
self.processing_log.append((msg, LogType.WARNING))
return req_wf_chans
def get_requested_soh_chan(self):
......@@ -894,8 +893,11 @@ class MainWindow(QtWidgets.QMainWindow, UIMainWindow):
self.is_plotting_waveform = False
self.is_plotting_tps = False
self.is_stopping = False
if len(self.data_object.keys) > 1:
self.plot_diff_key_button.setEnabled(True)
try:
if len(self.data_object.keys) > 1:
self.plot_diff_key_button.setEnabled(True)
except AttributeError:
pass
@QtCore.Slot()
def reset_is_plotting_waveform(self):
......
......@@ -190,7 +190,7 @@ class GPSDialog(QtWidgets.QWidget):
super().__init__()
self.parent = parent
self.data_path: Optional[Path] = None
self.data_path: Optional[Path, str] = None
# The directory the GPS will be exported to. By default, this will be
# the folder that contains the data set.
self.export_path: Path = Path.home()
......@@ -353,8 +353,10 @@ class GPSDialog(QtWidgets.QWidget):
msg = 'There is no GPS data to export.'
display_tracking_info(self.info_text_browser, msg, LogType.ERROR)
return
folder_name = self.data_path.name
try:
folder_name = self.data_path.name
except AttributeError:
folder_name = self.data_path.split(os.sep)[-1]
export_file_path = self.export_path / f'{folder_name}.gps.dat'
try:
......
......@@ -150,18 +150,19 @@ class Plotting:
has_min_max_lines=False)
val_cols = chan_db_info['valueColors'].split('|')
points_list = []
colors = []
# up/down has 2 values: 0, 1 which match with index of points_list
points_list = [[], []]
colors = [[], []]
for vc in val_cols:
v, c = vc.split(':')
val = get_val(v)
val = int(get_val(v))
points = []
for times, data in zip(c_data['times'], c_data['data']):
points += [times[i]
for i in range(len(data))
if data[i] == val]
points_list.append(points)
colors.append(c)
points_list[val] = points
colors[val] = c
# down dots
ax.plot(points_list[0], len(points_list[0]) * [-0.5], linestyle="",
......
......@@ -9,7 +9,7 @@ from matplotlib.backends.backend_qt5agg import (
FigureCanvasQTAgg as Canvas)
from sohstationviewer.controller.plotting_data import (
get_time_ticks, get_unit_bitweight)
get_time_ticks, get_unit_bitweight, get_disk_size_format)
from sohstationviewer.conf import constants
from sohstationviewer.view.util.color import clr
......@@ -299,6 +299,13 @@ class PlottingAxes:
:param max_y: maximum of y values
:param chan_db_info: info of channel from database
"""
if chan_db_info['channel'].startswith('Disk Usage'):
ax.set_yticks([org_min_y, org_max_y])
min_y_label = get_disk_size_format(org_min_y)
max_y_label = get_disk_size_format(org_max_y)
ax.set_yticklabels([min_y_label, max_y_label])
return
if chan_db_info['channel'] == 'GPS Lk/Unlk':
# to avoid case that the channel doesn't have Lk or Unlk
# preset min, max value so that GPS Clock Power is always in the
......
......@@ -21,7 +21,9 @@ from sohstationviewer.view.plotting.plotting_widget.plotting_axes import (
from sohstationviewer.view.plotting.plotting_widget.plotting import Plotting
from sohstationviewer.view.save_plot_dialog import SavePlotDialog
from sohstationviewer.controller.plotting_data import format_time
from sohstationviewer.controller.plotting_data import (
format_time, get_disk_size_format
)
from sohstationviewer.controller.util import display_tracking_info
......@@ -192,6 +194,10 @@ class PlottingWidget(QtWidgets.QScrollArea):
DataTypeModel.__init__.mass_pos_data[key]
"""
self.plotting_data2 = {}
# 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
# ----------------------------------------------------------------
QtWidgets.QScrollArea.__init__(self)
......@@ -303,52 +309,64 @@ class PlottingWidget(QtWidgets.QScrollArea):
"""
self.is_button_press_event_triggered_pick_event = True
artist = event.artist
if isinstance(artist, pl.Line2D):
ax = artist.axes
chan_id = ax.chan
if not isinstance(artist, pl.Line2D):
return
ax = artist.axes
chan_id = ax.chan
try:
chan_data = self.plotting_data1[chan_id]
except KeyError:
# in case of mass position
chan_data = self.plotting_data2[chan_id]
# list of x values of the plot
x_list = artist.get_xdata()
# list of y values of the plot
y_list = artist.get_ydata()
# index of the clicked point on the plot
click_plot_index = event.ind[0]
# time, val of the clicked point
clicked_time = x_list[click_plot_index]
clicked_val = y_list[click_plot_index]
real_idxes = get_index_from_data_picked(
chan_data, clicked_time, clicked_val)
if len(real_idxes) == 0:
display_tracking_info(self.tracking_box, "Point not found.")
return
clicked_data = chan_data['data'][0][real_idxes[0]]
if chan_id.startswith('Disk Usage'):
clicked_data = get_disk_size_format(clicked_data)
if hasattr(ax, 'unit_bw'):
clicked_data = ax.unit_bw.format(clicked_data)
formatted_clicked_time = format_time(
clicked_time, self.date_mode, 'HH:MM:SS')
info_str = (f"<pre>Channel: {chan_id} "
f"Point:{click_plot_index + 1} "
f"Time: {formatted_clicked_time} "
f"Value: {clicked_data}</pre>")
if self.rt130_log_data is not None:
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:
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:
chan_data = self.plotting_data1[chan_id]
except KeyError:
# in case of mass position
chan_data = self.plotting_data2[chan_id]
# list of x values of the plot
x_list = artist.get_xdata()
# list of y values of the plot
y_list = artist.get_ydata()
# index of the clicked point on the plot
click_plot_index = event.ind[0]
# time, val of the clicked point
clicked_time = x_list[click_plot_index]
clicked_val = y_list[click_plot_index]
real_idx = get_index_from_data_picked(
chan_data, clicked_time, clicked_val)
if real_idx is None:
display_tracking_info(self.tracking_box, "Point not found.")
return
clicked_data = chan_data['data'][0][real_idx]
if hasattr(ax, 'unit_bw'):
clicked_data = ax.unit_bw.format(clicked_data)
formatted_clicked_time = format_time(
clicked_time, self.date_mode, 'HH:MM:SS')
info_str = (f"<pre>Channel: {chan_id} "
f"Point:{click_plot_index + 1} "
f"Time: {formatted_clicked_time} "
f"Value: {clicked_data}</pre>")
display_tracking_info(self.tracking_box, info_str)
if 'logIdx' in chan_data.keys():
# For Reftek, need to hightlight the corresponding
# SOH message line based on the log_idx of the clicked point
self.parent.search_message_dialog.show()
clicked_log_idx = chan_data['logIdx'][0][real_idx]
try:
self.parent.search_message_dialog. \
show_log_entry_from_data_index(clicked_log_idx)
except ValueError as e:
QtWidgets.QMessageBox.warning(self, "Not found",
str(e))
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):
"""
......
from typing import List, Optional
from typing import Dict, Optional
import numpy as np
def get_index_from_data_picked(
chan_data: List[np.ndarray], tm: float, val: float) -> Optional[int]:
chan_data: Dict, tm: float, val: float) -> np.ndarray:
"""
Get index of data picked
:param chan_data: dict of data to plot that includes 'times', 'data' key
:param tm: epoch time of a clicked point
:param val: data value of a clicked point
:return section_idx: index of tm inside np.ndarray found
:return real_indexes: list index of data point inside np.ndarray found
"""
if chan_data['chan_db_info']['plotType'] == 'upDownDots':
# actual plotting has value -0.5 or 0.5;
......@@ -24,9 +24,7 @@ def get_index_from_data_picked(
else:
real_indexes = np.where((chan_data['times'][0] == tm) &
(chan_data['data'][0] == val))[0]
if len(real_indexes) != 1:
return
return real_indexes[0]
return real_indexes
def get_total_miny_maxy(
......
......@@ -33,6 +33,12 @@ class SOHWidget(MultiThreadedPlottingWidget):
self.data_object = d_obj
self.plotting_data1 = d_obj.soh_data[key] if key else {}
self.plotting_data2 = d_obj.mass_pos_data[key] if key else {}
self.rt130_log_data = None
if self.data_object.data_type == 'RT130':
try:
self.rt130_log_data = d_obj.log_data[key]['SOH'][0].split('\n')
except KeyError:
pass
channel_list = d_obj.soh_data[key].keys() if key else []
data_time = d_obj.data_time[key] if key else [0, 1]
ret = super().init_plot(d_obj, data_time, key, start_tm, end_tm,
......
......@@ -4,7 +4,8 @@ from pathlib import PosixPath, Path
from typing import Dict, List, Tuple, Callable, Union, Optional
from PySide6 import QtGui, QtCore, QtWidgets
from PySide6.QtWidgets import QStyle
from PySide6.QtWidgets import QStyle, QAbstractItemView
from PySide6.QtGui import QPalette
from sohstationviewer.view.search_message.highlight_delegate import (
HighlightDelegate)
......@@ -394,6 +395,14 @@ class SearchMessageDialog(QtWidgets.QWidget):
"""
# add 1 extra column to show scroll bar (+ 1)
table = QtWidgets.QTableWidget(rows, cols + 1)
# To prevent selected row not grey out. It still gets faded out, but
# the color remain blue which is better than grey out.
p = table.palette()
p.setBrush(QPalette.Inactive, QPalette.Highlight,
p.brush(QPalette.Highlight))
table.setPalette(p)
delegate = HighlightDelegate(table, self.display_color)
table.setItemDelegate(delegate)
# Hide header cells
......@@ -487,9 +496,47 @@ class SearchMessageDialog(QtWidgets.QWidget):
item : QTableWidgetItem
A valid QTableWidgetIem
"""
self.current_table.scrollToItem(item)
self.current_table.scrollToItem(item, QAbstractItemView.PositionAtTop)
self.current_table.setFocus()
def show_log_entry_from_log_indexes(self, log_indexes: List[int]):
"""
This is called when clicking a clickable data point on a SOH channel
of RT130, list of log row indexes will be passed to this method.
This method will:
+ set current tab to soh_table_dict['SOH']
+ scroll the first indexed row to top of table and highlight all
the row in log_indexes
Parameters
----
log_indexes : The list of indexes of log row to be selected
"""
if 'SOH' not in self.soh_tables_dict:
return
# switch to SOH tab
self.tab_widget.setCurrentWidget(self.soh_tables_dict['SOH'])
self.current_table.clearSelection()
# Allow to select multiple rows
self.current_table.setSelectionMode(
QAbstractItemView.MultiSelection)
# select all rows according to log_indexes
for idx in log_indexes:
self.current_table.selectRow(idx)
# scroll to the first index and place the row at top of the table
self.current_table.scrollToItem(
self.current_table.item(log_indexes[0], 1),
QAbstractItemView.PositionAtTop
)
# raise the message dialog on top of others
self.setWindowState(QtCore.Qt.WindowState.WindowActive)
self.raise_()
self.activateWindow()
self.current_table.setFocus() # focus row to not faded out
# return back to select single row
self.current_table.setSelectionMode(
QAbstractItemView.SingleSelection)
def show_log_entry_from_data_index(self, data_index: int):
"""
This is called when clicking a clickable data point on a SOH channel
......@@ -632,7 +679,8 @@ class SearchMessageDialog(QtWidgets.QWidget):
if ret is None:
return
self.selected_item, self.search_rowidx = ret
self.current_table.scrollToItem(self.selected_item)
self.current_table.scrollToItem(self.selected_item,
QAbstractItemView.PositionAtTop)
def _filter_lines_with_search_text_from_soh_messages(self):
"""
......
# UI and connectSignals for main_window
import configparser
from pathlib import Path
from typing import Union, List, Optional
from PySide6 import QtCore, QtGui, QtWidgets
......@@ -286,7 +285,7 @@ class UIMainWindow(object):
self.set_first_row(main_layout)
self.set_second_row(main_layout)
self.tracking_info_text_browser.setFixedHeight(60)
self.tracking_info_text_browser.setFixedHeight(80)
main_layout.addWidget(self.tracking_info_text_browser)
self.create_menu_bar(main_window)
self.connect_signals(main_window)
......
import unittest
from tempfile import TemporaryDirectory, NamedTemporaryFile
from pathlib import Path
......@@ -358,6 +359,7 @@ class TestGetDataTypeFromFile(TestCase):
Path(test_file.name), get_signature_channels())
self.assertEqual(ret, (None, False))
@unittest.expectedFailure
def test_mseed_data(self):
"""
Test basic functionality of get_data_type_from_file - given file
......
......@@ -6,11 +6,13 @@ from sohstationviewer.database.extract_data import (
get_signature_channels,
get_color_def,
get_color_ranges,
convert_actual_channel_to_db_channel_w_question_mark,
get_convert_factor
)
class TestExtractData(unittest.TestCase):
def test_get_chan_plot_info_good_soh_channel_and_data_type(self):
class TestGetChanPlotInfo(unittest.TestCase):
def test_good_soh_channel_and_data_type(self):
"""
Test basic functionality of get_chan_plot_info - channel and data type
combination exists in database table `Channels`
......@@ -27,7 +29,7 @@ class TestExtractData(unittest.TestCase):
self.assertDictEqual(get_chan_plot_info('SOH/Data Def', 'RT130'),
expected_result)
def test_get_chan_plot_info_masspos_channel(self):
def test_masspos_channel(self):
with self.subTest("Mass position 'VM'"):
expected_result = {'channel': 'VM1',
'plotType': 'linesMasspos',
......@@ -54,7 +56,7 @@ class TestExtractData(unittest.TestCase):
self.assertDictEqual(get_chan_plot_info('MassPos1', 'RT130'),
expected_result)
def test_get_chan_plot_info_seismic_channel(self):
def test_seismic_channel(self):
with self.subTest("RT130 Seismic"):
expected_result = {'channel': 'DS2',
'plotType': 'linesSRate',
......@@ -81,7 +83,7 @@ class TestExtractData(unittest.TestCase):
self.assertDictEqual(get_chan_plot_info('LHE', 'Q330'),
expected_result)
def test_get_chan_plot_info_data_type_is_unknown(self):
def test_data_type_is_unknown(self):
"""
Test basic functionality of get_chan_plot_info - data type is the
string 'Unknown'.
......@@ -113,7 +115,7 @@ class TestExtractData(unittest.TestCase):
get_chan_plot_info('LCE', 'Unknown'),
expected_result)
def test_get_chan_plot_info_bad_channel_or_data_type(self):
def test_bad_channel_or_data_type(self):
"""
Test basic functionality of get_chan_plot_info - channel and data type
combination does not exist in database table Channels and data type is
......@@ -159,7 +161,9 @@ class TestExtractData(unittest.TestCase):
self.assertDictEqual(get_chan_plot_info('SOH/Data Def', 'Q330'),
expected_result)
def test_get_seismic_chan_label_good_channel_id(self):
class TestGetSeismicChanLabel(unittest.TestCase):
def test_good_channel_id(self):
"""
Test basic functionality of get_seismic_chan_label - channel ID ends
in one of the keys in conf.dbSettings.dbConf['seisLabel'] or
......@@ -174,7 +178,7 @@ class TestExtractData(unittest.TestCase):
self.assertEqual(get_seismic_chan_label('DS-TEST-CHANNEL'),
'DS-TEST-CHANNEL')
def test_get_chan_label_bad_channel_id(self):
def test_bad_channel_id(self):
"""
Test basic functionality of get_seismic_chan_label - channel ID does
not end in one of the keys in conf.dbSettings.dbConf['seisLabel']
......@@ -182,16 +186,22 @@ class TestExtractData(unittest.TestCase):
"""
self.assertRaises(IndexError, get_seismic_chan_label, '')
class TestGetSignatureChannels(unittest.TestCase):
def test_get_signature_channels(self):
"""Test basic functionality of get_signature_channels"""
self.assertIsInstance(get_signature_channels(), dict)
class TestGetColorDef(unittest.TestCase):
def test_get_color_def(self):
"""Test basic functionality of get_color_def"""
colors = get_color_def()
expected_colors = ['K', 'U', 'C', 'G', 'Y', 'R', 'M', 'E']
self.assertListEqual(colors, expected_colors)
class TestGetColorRanges(unittest.TestCase):
def test_get_color_ranges(self):
"""Test basic functionality of get_color_ranges"""
names, all_counts, all_display_strings = get_color_ranges()
......@@ -212,3 +222,23 @@ class TestExtractData(unittest.TestCase):
# Check that each list of strings to display have enough items
for display_strings in all_display_strings:
self.assertEqual(len(display_strings), num_color_def + 1)
class TestConvertActualChannelToDBChannelWQuestionMark(unittest.TestCase):
def test_question_channel(self):
ret = convert_actual_channel_to_db_channel_w_question_mark('VM1')
self.assertEqual(ret, 'VM?')
def test_non_question_channel(self):
ret = convert_actual_channel_to_db_channel_w_question_mark('VCO')
self.assertEqual(ret, 'VCO')
class TestGetConvertFactor(unittest.TestCase):
def test_question_channel(self):
ret = get_convert_factor('VM1', 'Centaur')
self.assertEqual(ret, 10**(-6))
def test_non_question_channel(self):
ret = get_convert_factor('VEP', 'Q330')
self.assertEqual(ret, 0.15)
......@@ -5,13 +5,13 @@ from sohstationviewer.view.plotting.plotting_widget.plotting_widget_helper \
import get_total_miny_maxy, get_index_from_data_picked
class TestGetIndexFromTime(TestCase):
class TestGetIndexFromDataPicked(TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.plotting_data = {
'CH1': {
'times': [np.array([1, 2, 3, 4, 5, 6])],
'data': [np.array([1, 1, 0, 1, 1, 0])],
'times': [np.array([1, 2, 3, 4, 5, 6, 6])],
'data': [np.array([1, 1, 0, 1, 1, 0, 0])],
'chan_db_info': {'plotType': 'upDownDots'}
},
'CH2': {
......@@ -27,45 +27,51 @@ class TestGetIndexFromTime(TestCase):
}
def test_time_not_included(self):
real_idx = get_index_from_data_picked(
real_idxes = get_index_from_data_picked(
self.plotting_data['CH1'], 7, 1)
self.assertIsNone(real_idx)
self.assertEqual(len(real_idxes), 0)
def test_type_not_need_data_info(self):
# CH3 has plotType='dotForTime' in ["multiColorDots", "dotForTime"])
real_idx = get_index_from_data_picked(
def test_type_not_need_data_val(self):
# CH3 has plotType='dotForTime'
# which is in ["multiColorDots", "dotForTime"])
real_idxes = get_index_from_data_picked(
self.plotting_data['CH3'], 4, 4)
self.assertEqual(real_idx, 3)
self.assertEqual(real_idxes.tolist(), [3])
def test_type_need_data_info(self):
def test_type_need_data_val(self):
# CH2 has plotType='linesDots' not in ["multiColorDots", "dotForTime"])
with self.subTest('data not match time'):
real_idx = get_index_from_data_picked(
real_idxes = get_index_from_data_picked(
self.plotting_data['CH2'], 3, 5)
self.assertIsNone(real_idx)
self.assertEqual(len(real_idxes), 0)
with self.subTest('data match 1st value'):
real_idx = get_index_from_data_picked(
real_idxes = get_index_from_data_picked(
self.plotting_data['CH2'], 3, 7)
self.assertEqual(real_idx, 2)
self.assertEqual(real_idxes.tolist(), [2])
with self.subTest('data match 2nd value'):
real_idx = get_index_from_data_picked(
real_idxes = get_index_from_data_picked(
self.plotting_data['CH2'], 3, 4)
self.assertEqual(real_idx, 3)
self.assertEqual(real_idxes.tolist(), [3])
def test_type_up_down(self):
# CH1 has plotType='upDownDots'
with self.subTest('data not match time'):
real_idx = get_index_from_data_picked(
real_idxes = get_index_from_data_picked(
self.plotting_data['CH1'], 1, -0.5)
self.assertIsNone(real_idx)
self.assertEqual(len(real_idxes), 0)
with self.subTest('data=1 match time'):
real_idx = get_index_from_data_picked(
real_idxes = get_index_from_data_picked(
self.plotting_data['CH1'], 1, 0.5)
self.assertEqual(real_idx, 0)
self.assertEqual(real_idxes.tolist(), [0])
with self.subTest('data=0 match time'):
real_idx = get_index_from_data_picked(
real_idxes = get_index_from_data_picked(
self.plotting_data['CH1'], 3, -0.5)
self.assertEqual(real_idx, 2)
self.assertEqual(real_idxes.tolist(), [2])
def test_2_overlapped_points(self):
real_idxes = get_index_from_data_picked(
self.plotting_data['CH1'], 6, -0.5)
self.assertEqual(real_idxes.tolist(), [5, 6])
class TestGetTotalMinyMaxy(TestCase):
......