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
...@@ -26,8 +26,8 @@ class WaveformWidget(plotting_widget.PlottingWidget): ...@@ -26,8 +26,8 @@ class WaveformWidget(plotting_widget.PlottingWidget):
finished = QtCore.Signal() finished = QtCore.Signal()
def __init__(self, parent, tracking_box, *args): def __init__(self, parent, tracking_box, name):
super().__init__(parent, tracking_box) super().__init__(parent, tracking_box, name)
self.data_processors: List[WaveformChannelProcessor] = [] self.data_processors: List[WaveformChannelProcessor] = []
...@@ -83,12 +83,20 @@ class WaveformWidget(plotting_widget.PlottingWidget): ...@@ -83,12 +83,20 @@ class WaveformWidget(plotting_widget.PlottingWidget):
self.min_x = max(data_time[0], start_tm) self.min_x = max(data_time[0], start_tm)
self.max_x = min(data_time[1], end_tm) self.max_x = min(data_time[1], end_tm)
self.plot_total = len(self.plotting_data1) + len(self.plotting_data2) self.plot_total = len(self.plotting_data1) + len(self.plotting_data2)
title = get_title(key, self.min_x, self.max_x, self.date_mode) if waveform_data == {}:
title = "NO WAVEFORM DATA TO DISPLAY."
self.processing_log.append("No waveform data to display.")
else:
title = get_title(key, self.min_x, self.max_x, self.date_mode)
self.plotting_bot = const.BOTTOM self.plotting_bot = const.BOTTOM
self.plotting_bot_pixel = const.BOTTOM_PX self.plotting_bot_pixel = const.BOTTOM_PX
self.axes = [] self.axes = []
self.timestamp_bar_top = self.plotting_axes.add_timestamp_bar(0.003) self.timestamp_bar_top = self.plotting_axes.add_timestamp_bar(0.003)
self.plotting_axes.set_title(title) self.plotting_axes.set_title(title)
if waveform_data == {}:
return False
else:
return True
def plot_channels(self, start_tm, end_tm, key, def plot_channels(self, start_tm, end_tm, key,
data_time, time_ticks_total, data_time, time_ticks_total,
...@@ -115,9 +123,12 @@ class WaveformWidget(plotting_widget.PlottingWidget): ...@@ -115,9 +123,12 @@ class WaveformWidget(plotting_widget.PlottingWidget):
self.is_working = True self.is_working = True
start_msg = 'Plotting waveform data...' start_msg = 'Plotting waveform data...'
display_tracking_info(self.tracking_box, start_msg, 'info') display_tracking_info(self.tracking_box, start_msg, 'info')
self.init_plot(start_tm, end_tm, key, ret = self.init_plot(start_tm, end_tm, key,
data_time, time_ticks_total, data_time, time_ticks_total,
waveform_data, mass_pos_data) waveform_data, mass_pos_data)
if not ret:
self.draw()
return
self.create_waveform_channel_processors() self.create_waveform_channel_processors()
self.process_channel() self.process_channel()
...@@ -354,7 +365,7 @@ class WaveformDialog(QtWidgets.QWidget): ...@@ -354,7 +365,7 @@ class WaveformDialog(QtWidgets.QWidget):
mass position channel mass position channel
""" """
self.plotting_widget = WaveformWidget( self.plotting_widget = WaveformWidget(
self, self.info_text_browser) self, self.info_text_browser, 'RAWWidget')
self.plotting_widget.finished.connect(self.plot_finished) self.plotting_widget.finished.connect(self.plot_finished)
main_layout.addWidget(self.plotting_widget, 2) main_layout.addWidget(self.plotting_widget, 2)
......
This diff is collapsed.
import re
from pathlib import Path from pathlib import Path
from typing import Dict, List, Tuple, Union from typing import Dict, List, Tuple, Union
from sohstationviewer.view.util.enums import LogType from sohstationviewer.view.util.enums import LogType
from sohstationviewer.conf import constants as const from sohstationviewer.conf import constants as const
from sohstationviewer.conf.constants import (WF_1ST, WF_2ND, WF_3RD)
def is_doc_file(file: Path, include_table_of_contents: bool = False)\ def is_doc_file(file: Path, include_table_of_contents: bool = False)\
...@@ -160,5 +163,46 @@ def log_str(log_info: Tuple[str, LogType]) -> str: ...@@ -160,5 +163,46 @@ def log_str(log_info: Tuple[str, LogType]) -> str:
return f"{log_type.name}: {log_text}" return f"{log_type.name}: {log_text}"
def check_chan_wildcards_format(wildcards: str):
"""
Check format of channel's wildcards. Raise exception if invalid.
:param wildcards: wildcards that are separated with ','.
:type wildcards: str
"""
for wc in wildcards.split():
wc = wc.strip()
if len(wc) == 1 and wc != '*':
raise Exception(
f"Request '{wc}' has length=1 which must be '*'.")
if '**' in wc:
raise Exception(
f"Request '{wc}' includes '**' which isn't allowed.")
if len(wc) > 3:
raise Exception(
f"Request '{wc}' has length={len(wc)} > 3 which isn't allowed."
)
if len(wc) == 2 and '*' not in wc:
raise Exception(
f"Request '{wc}' has length=2 which required one '*'."
)
pattern = f"[{WF_1ST}*]"
if not re.compile(pattern).match(wc[0]):
raise Exception(
f"Request '{wc}' has first character not match {pattern}."
)
pattern = f"[{WF_3RD}*]"
if not re.compile(pattern).match(wc[-1]):
raise Exception(
f"Request '{wc}' has last character not match {pattern}."
)
if len(wc) == 3:
pattern = f"[{WF_2ND}*]"
if not re.compile(pattern).match(wc[1]):
raise Exception(
f"Request '{wc}' has second character not match {pattern}."
)
if __name__ == '__main__': if __name__ == '__main__':
create_table_of_content_file(Path('../../../documentation')) create_table_of_content_file(Path('../../../documentation'))
from unittest import TestCase
from unittest.mock import patch
from sohstationviewer.model.handling_data import (
check_chan,
check_wf_chan,
check_soh_chan
)
class TestCheckChan(TestCase):
def setUp(self) -> None:
self.req_soh_chans = ['LCE', 'LCQ']
self.req_wf_chans = ['LHE', 'HHE']
check_soh_chan_patcher = patch(
'sohstationviewer.model.handling_data.check_soh_chan',
wraps=check_soh_chan
)
self.addCleanup(check_soh_chan_patcher.stop)
self.mock_check_soh_chan = check_soh_chan_patcher.start()
check_wf_chan_patcher = patch(
'sohstationviewer.model.handling_data.check_wf_chan',
wraps=check_wf_chan
)
self.addCleanup(check_wf_chan_patcher.stop)
self.mock_check_wf_chan = check_wf_chan_patcher.start()
def test_channel_is_waveform_and_is_requested(self):
waveform_channel = 'LHE'
ret = check_chan(waveform_channel, self.req_soh_chans,
self.req_wf_chans)
self.assertEqual(ret, 'WF')
self.assertTrue(self.mock_check_wf_chan.called)
def test_channel_is_waveform_but_not_requested(self):
waveform_channel = 'HH1'
ret = check_chan(waveform_channel, self.req_soh_chans,
self.req_wf_chans)
self.assertFalse(ret)
self.assertTrue(self.mock_check_wf_chan.called)
def test_channel_is_soh_and_is_requested(self):
with self.subTest('test_normal_channel'):
soh_channel = 'LCE'
ret = check_chan(
soh_channel, self.req_soh_chans, self.req_wf_chans)
self.assertEqual(ret, 'SOH')
self.assertTrue(self.mock_check_soh_chan.called)
self.mock_check_soh_chan.reset_mock()
with self.subTest('test_mass_position_channel'):
soh_channel = 'VM1'
ret = check_chan(
soh_channel, self.req_soh_chans, self.req_wf_chans)
self.assertEqual(ret, 'SOH')
self.assertTrue(self.mock_check_soh_chan.called)
def test_channel_is_soh_but_not_requested(self):
soh_channel = 'VKI'
ret = check_chan(soh_channel, self.req_soh_chans, self.req_wf_chans)
self.assertFalse(ret)
self.assertTrue(self.mock_check_soh_chan.called)
class TestCheckSohChan(TestCase):
def setUp(self) -> None:
self.req_soh_chans = ['LCE', 'LCQ', 'EX?']
self.sample_channel_ids = ['ODV', 'QGA', 'NF4', 'OLY', 'UZM']
def test_all_channels_requested(self):
self.req_soh_chans = []
for channel_id in self.sample_channel_ids:
self.assertTrue(check_soh_chan(channel_id, self.req_soh_chans))
def test_channel_is_requested(self):
with self.subTest('test_normal_channels'):
channel_id = 'LCE'
self.assertTrue(check_soh_chan(channel_id, self.req_soh_chans))
with self.subTest('test_mass_position_channels'):
base_channel_id = 'VM'
channel_suffixes = ['0', '1', '2', '3', '4', '5', '6']
for suffix in channel_suffixes:
channel_id = base_channel_id + suffix
self.assertTrue(check_soh_chan(channel_id, self.req_soh_chans))
with self.subTest('test_external_soh_channels'):
base_channel_id = 'EX'
channel_suffixes = ['1', '2', '3']
for suffix in channel_suffixes:
channel_id = base_channel_id + suffix
self.assertTrue(check_soh_chan(channel_id, self.req_soh_chans))
def test_channel_not_requested(self):
for channel_id in self.sample_channel_ids:
self.assertFalse(check_soh_chan(channel_id, self.req_soh_chans))
class TestCheckWfChan(TestCase):
def test_channel_is_requested(self):
req_wf_chans = ['LHE', 'HHE']
channel_id = 'LHE'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', True)
)
def test_channel_not_requested(self):
req_wf_chans = ['LHE', 'HHE']
with self.subTest('test_waveform_channel'):
channel_id = 'AHE'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', False)
)
with self.subTest('test_non_waveform_channel'):
channel_id = 'Not a waveform channel'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('', False)
)
def test_len1_wildcard_request(self):
req_wf_chans = ['*']
sample_channel_ids = ['LHE', 'HHE', 'AL2', 'MNZ', 'VNN']
with self.subTest('test_waveform_channel'):
for channel_id in sample_channel_ids:
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', True)
)
with self.subTest('test_non_waveform_channel'):
channel_id = 'Not a waveform channel'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('', False)
)
def test_len2_wildcard_request(self):
req_wf_chans = ['H*']
with self.subTest('Requested start char'):
channel_id = 'HHE'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', True)
)
with self.subTest('None-requested start char'):
channel_id = 'LHE'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', False)
)
req_wf_chans = ['*E']
with self.subTest('Requested end char'):
channel_id = 'HHE'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', True)
)
with self.subTest('None-requested end char'):
channel_id = 'LHN'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', False)
)
def test_len3_wildcard_request(self):
req_wf_chans = ['HH*']
with self.subTest('Requested start 2 chars'):
channel_id = 'HHE'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', True)
)
with self.subTest('None-requested start 2 chars'):
channel_id = 'LHE'
ret = check_wf_chan(channel_id, req_wf_chans)
self.assertTupleEqual(
ret,
('WF', False)
)
req_wf_chans = ['H*E']
with self.subTest('Requested start or end char'):
channel_id = 'HHE'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', True)
)
with self.subTest('None-requested start or end char'):
channel_id = 'LHN'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', False)
)
req_wf_chans = ['*HE']
with self.subTest('Requested 2 end chars'):
channel_id = 'HHE'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', True)
)
with self.subTest('None-requested 2 end chars'):
channel_id = 'LHN'
self.assertTupleEqual(
check_wf_chan(channel_id, req_wf_chans),
('WF', False)
)
...@@ -6,10 +6,13 @@ from unittest import TestCase ...@@ -6,10 +6,13 @@ from unittest import TestCase
from sohstationviewer.view.util.functions import ( from sohstationviewer.view.util.functions import (
get_soh_messages_for_view, log_str, is_doc_file, get_soh_messages_for_view, log_str, is_doc_file,
create_search_results_file, create_table_of_content_file) create_search_results_file, create_table_of_content_file,
check_chan_wildcards_format
)
from sohstationviewer.view.util.enums import LogType from sohstationviewer.view.util.enums import LogType
from sohstationviewer.conf import constants as const from sohstationviewer.conf import constants as const
from sohstationviewer.conf.constants import (WF_1ST, WF_2ND, WF_3RD)
class TestGetSOHMessageForView(TestCase): class TestGetSOHMessageForView(TestCase):
...@@ -208,3 +211,165 @@ class TestCreateTableOfContentFile(TestCase): ...@@ -208,3 +211,165 @@ class TestCreateTableOfContentFile(TestCase):
"+ [file1](file1.help.md)\n\n", "+ [file1](file1.help.md)\n\n",
) )
) )
class TestCheckChanWildcardsFormat(TestCase):
def test_len1(self):
with self.subTest("Wildcard is *"):
wc = '*'
try:
check_chan_wildcards_format(wc)
except Exception:
self.fail(f"Wildcard '{wc}' raise Exception unexpectedly")
with self.subTest("Wildcard isn't *"):
wc = 'L'
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' has length=1 which must be '*'."
)
def test_len2(self):
with self.subTest("Wildcard with first char matched"):
wc = 'L*'
try:
check_chan_wildcards_format(wc)
except Exception:
self.fail(f"Wildcard '{wc}' raise Exception unexpectedly")
with self.subTest("Wildcard with first char not matched"):
wc = 'W*'
pattern = f"[{WF_1ST}*]"
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' has first character not match {pattern}."
)
with self.subTest("Wildcard with last char matched"):
wc = '*N'
try:
check_chan_wildcards_format(wc)
except Exception:
self.fail(f"Wildcard '{wc}' raise Exception unexpectedly")
with self.subTest("Wildcard with last char not matched"):
wc = '*L'
pattern = f"[{WF_3RD}*]"
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' has last character not match {pattern}."
)
with self.subTest("Wildcard is **"):
wc = '**'
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' includes '**' which isn't allowed."
)
def test_len3(self):
with self.subTest("Wildcard with first char matched"):
wc = 'HL*'
try:
check_chan_wildcards_format(wc)
except Exception:
self.fail(f"Wildcard '{wc}' raise Exception unexpectedly")
with self.subTest("Wildcard with first char is a '*'"):
wc = '*L*'
try:
check_chan_wildcards_format(wc)
except Exception:
self.fail(f"Wildcard '{wc}' raise Exception unexpectedly")
with self.subTest("Wildcard with first char not matched"):
wc = 'WL*'
pattern = f"[{WF_1ST}*]"
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' has first character not match {pattern}."
)
with self.subTest("Wildcard with last char matched"):
wc = '*LN'
try:
check_chan_wildcards_format(wc)
except Exception:
self.fail(f"Wildcard '{wc}' raise Exception unexpectedly")
with self.subTest("Wildcard with last char is a '*'"):
wc = 'HL*'
try:
check_chan_wildcards_format(wc)
except Exception:
self.fail(f"Wildcard '{wc}' raise Exception unexpectedly")
with self.subTest("Wildcard with last char not matched"):
wc = '*LL'
pattern = f"[{WF_3RD}*]"
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' has last character not match {pattern}."
)
with self.subTest("Wildcard with second char matched"):
wc = 'HHN'
try:
check_chan_wildcards_format(wc)
except Exception:
self.fail(f"Wildcard '{wc}' raise Exception unexpectedly")
with self.subTest("Wildcard with second char is a star"):
wc = 'H*N'
try:
check_chan_wildcards_format(wc)
except Exception:
self.fail(f"Wildcard '{wc}' raise Exception unexpectedly")
with self.subTest("Wildcard with second char not matched"):
wc = 'HWE'
pattern = f"[{WF_2ND}*]"
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' has second character not match {pattern}."
)
with self.subTest("Wildcard start with '**'"):
wc = '**E'
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' includes '**' which isn't allowed."
)
with self.subTest("Wildcard end with '**'"):
wc = 'H**'
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' includes '**' which isn't allowed."
)
def test_len_gt_3(self):
wc = 'HHLL'
with self.assertRaises(Exception) as context:
check_chan_wildcards_format(wc)
self.assertEqual(
str(context.exception),
f"Request '{wc}' has length={len(wc)} > 3 which isn't allowed."
)