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 (7)
Showing
with 293 additions and 92 deletions
No preview for this file type
......@@ -909,14 +909,15 @@ def trim_downsample_wf_chan(chan: Dict, start_tm: float, end_tm: float,
chan['data'] = np.hstack(downsampled_data)
def get_each_day_5_min_list(start_tm: float, end_tm: float) -> np.ndarray:
def get_start_5mins_of_diff_days(start_tm: float, end_tm: float) -> np.ndarray:
"""
Get the list of all five minute for every day start from the day of startTm
and end at the day of endTm.
Get the list of the start time of all five minutes for each day start from
the day of startTm and end at the day of endTm.
:param start_tm: float - start time
:param end_tm: float - end time
:return every_day_5_min_list: [[288 of floats], ] - the list of all start
of five minutes for every day in which each day has 288 of 5 minutes.
:return start_5mins_of_diff_days: [[288 of floats], ] - the list of
start of all five minutes of days specified by start_tm and end_tm in
which each day has 288 of 5 minutes.
"""
exact_day_tm = (start_tm // const.SEC_DAY) * const.SEC_DAY
exact_day_tm_list = []
......@@ -928,43 +929,57 @@ def get_each_day_5_min_list(start_tm: float, end_tm: float) -> np.ndarray:
exact_day_tm_list.append(exact_day_tm)
exact_day_tm += const.SEC_DAY
# list of start/end 5m in each day: every_day_5_min_list
# list of start/end 5m in each day: start_5mins_of_diff_days
for idx, start_day_tm in enumerate(exact_day_tm_list):
a_day_5_min = np.arange(start_day_tm,
start_day_tm + const.SEC_DAY,
const.SEC_5M)
start_5mins_of_day = np.arange(start_day_tm,
start_day_tm + const.SEC_DAY,
const.SEC_5M)
if idx == 0:
every_day_5_min_list = np.array([a_day_5_min])
start_5mins_of_diff_days = np.array([start_5mins_of_day])
else:
every_day_5_min_list = np.vstack(
(every_day_5_min_list, a_day_5_min))
return every_day_5_min_list
start_5mins_of_diff_days = np.vstack(
(start_5mins_of_diff_days, start_5mins_of_day))
return start_5mins_of_diff_days
def find_tps_tm(given_tm: float, each_day_5_min_list: List[List[float]]
) -> Tuple[float, float]:
def find_tps_tm_idx(
given_tm: float, start_5mins_of_diff_days: List[List[float]]) \
-> Tuple[float, float]:
"""
Find the position of the given time (given_tm) in time-power-squared plot
:param given_tm: float - given time
:param each_day_5_min_list: [[288 of floats], ] - the list of all start
of five minutes for every day in which each day has 288 of 5 minutes.
:return x_idx: int - index of time in the each_day_5_min_list
:return y_idx: int - index of day plotted
(look in TimePowerSquaredWidget.getZoomData())
:param start_5mins_of_diff_days: [[288 of floats], ] - the list of
start of all five minutes of some specific days in which each day has
288 of 5 minutes.
:return x_idx: int - index of 5m section
:return y_idx: int - index of the day the given time belong to in plotting
"""
x_idx = None
y_idx = None
for day_idx, day in enumerate(each_day_5_min_list):
for tm_idx, tm in enumerate(day):
if tm > given_tm:
for day_idx, a_day_5mins in enumerate(start_5mins_of_diff_days):
for start_5m_idx, start_5m in enumerate(a_day_5mins):
if start_5m > given_tm:
# index of day start from 0 to negative because day is plotted
# from top to bottom
y_idx = - day_idx
x_idx = tm_idx - 1
if x_idx < 0:
x_idx = start_5m_idx - 1
if start_5m_idx == 0:
# if the start_5m_idx == 0, the given time belong to the
# last 5m of the previous day
y_idx = -(day_idx - 1)
x_idx = len(day) - 1
x_idx = const.NO_5M_DAY - 1
break
if x_idx is not None:
break
if x_idx is None:
# x_idx == None happens when the given time fall into the last 5m of
# the last day. Although the time 24:00 of the last day belongs
# to the next days of other cases, but since there is no more days to
# plot it, it is no harm to set it at the last 5m of the last day.
x_idx = const.NO_5M_DAY - 1
y_idx = - (len(start_5mins_of_diff_days) - 1)
return x_idx, y_idx
......
......@@ -80,6 +80,7 @@ class Plotting:
if c == '_':
prev_val = val
continue
points = []
for times, data in zip(c_data['times'], c_data['data']):
if v.startswith('+'):
points = [times[i]
......@@ -258,7 +259,6 @@ class Plotting:
for cStr in color_parts:
obj, c = cStr.split(':')
colors[obj] = c
l_color = 'G'
has_dot = False
if 'L' in colors:
......@@ -266,12 +266,15 @@ class Plotting:
if 'D' in colors:
d_color = colors['D']
has_dot = True
for x, y in zip(x_list, y_list):
if not has_dot:
ax.myPlot = ax.plot(x, y,
# set marker to be able to click point for info
# but marker's size is small to not show dot.
ax.myPlot = ax.plot(x, y, marker='o', markersize=0.01,
linestyle='-', linewidth=0.7,
color=clr[l_color])
zorder=constants.Z_ORDER['LINE'],
color=clr[l_color],
picker=True, pickradius=2)
else:
ax.myPlot = ax.plot(x, y, marker='s', markersize=1.5,
linestyle='-', linewidth=0.7,
......@@ -280,6 +283,7 @@ class Plotting:
mfc=clr[d_color],
mec=clr[d_color],
picker=True, pickradius=3)
if linked_ax is None:
ax.x_list = x_list
ax.y_list = y_list
......
......@@ -2,14 +2,15 @@
Class of which object is used to plot data
"""
from typing import List
import numpy as np
from PySide2.QtCore import QTimer
from matplotlib import pyplot as pl
from PySide2 import QtCore, QtWidgets
from sohstationviewer.conf import constants
from sohstationviewer.view.util.color import set_color_mode
from sohstationviewer.view.util.functions import get_total_miny_maxy
from sohstationviewer.view.util.functions import (
get_total_miny_maxy, get_index_from_time
)
from sohstationviewer.view.plotting.plotting_widget.plotting_axes import (
PlottingAxes)
from sohstationviewer.view.plotting.plotting_widget.plotting import Plotting
......@@ -279,18 +280,24 @@ class PlottingWidget(QtWidgets.QScrollArea):
chan_data = self.plotting_data1[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 value of the clicked point
# time, val of the clicked point
clicked_time = x_list[click_plot_index]
# indexes of the clicked time in data (one value only)
clicked_indexes = np.where(chan_data['times'] == clicked_time)
"""
clicked_indexes and click_plot_index can be different if there
are different plots for a channel.
"""
clicked_data = chan_data['data'][clicked_indexes][0]
clicked_val = y_list[click_plot_index]
list_idx, section_idx = get_index_from_time(
chan_data, clicked_time, clicked_val)
if list_idx is None:
display_tracking_info(self.tracking_box, "Point not found.")
return
clicked_data = chan_data['data'][list_idx][section_idx]
if hasattr(ax, 'unit_bw'):
clicked_data = ax.unit_bw.format(clicked_data)
formatted_clicked_time = format_time(
......@@ -300,12 +307,17 @@ class PlottingWidget(QtWidgets.QScrollArea):
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'][clicked_indexes][0]
self.parent.search_message_dialog. \
show_log_entry_from_data_index(clicked_log_idx)
clicked_log_idx = chan_data['logIdx'][0][section_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))
def on_button_press_event(self, event):
"""
......
......@@ -16,7 +16,7 @@ from sohstationviewer.database.extract_data import (
get_color_def, get_color_ranges, get_chan_label,
)
from sohstationviewer.model.handling_data import (
get_each_day_5_min_list, find_tps_tm,
get_start_5mins_of_diff_days, find_tps_tm_idx,
)
from sohstationviewer.view.util.enums import LogType
from sohstationviewer.view.plotting.plotting_widget import plotting_widget
......@@ -54,7 +54,7 @@ class TimePowerSquaredWidget(plotting_widget.PlottingWidget):
of five minutes for every day in which each day has 288 of
5 minutes.
"""
self.each_day5_min_list = []
self.start_5mins_of_diff_days = []
"""
tps_t: float - prompt's time on tps's chart to help rulers on other
plotting widgets to identify their location
......@@ -121,15 +121,15 @@ class TimePowerSquaredWidget(plotting_widget.PlottingWidget):
self.min_x = max(data_time[0], start_tm)
self.max_x = min(data_time[1], end_tm)
self.each_day5_min_list = get_each_day_5_min_list(self.min_x,
self.max_x)
self.start_5mins_of_diff_days = get_start_5mins_of_diff_days(
self.min_x, self.max_x)
for chan_id in self.plotting_data1:
c_data = self.plotting_data1[chan_id]
if 'tps_data' not in c_data:
self.channels.append(chan_id)
channel_processor = TimePowerSquaredProcessor(
chan_id, c_data, self.min_x, self.max_x,
self.each_day5_min_list
self.start_5mins_of_diff_days
)
channel_processor.signals.finished.connect(self.channel_done)
channel_processor.signals.stopped.connect(self.channel_done)
......@@ -382,7 +382,8 @@ class TimePowerSquaredWidget(plotting_widget.PlottingWidget):
five_min_index = xdata
try:
# identify time for rulers on other plotting widget
self.tps_t = self.each_day5_min_list[day_index, five_min_index]
self.tps_t = self.start_5mins_of_diff_days[
day_index, five_min_index]
format_t = format_time(self.tps_t, self.date_mode, 'HH:MM:SS')
info_str += f"<pre>{format_t}:"
for chan_id in self.plotting_data1:
......@@ -408,7 +409,7 @@ class TimePowerSquaredWidget(plotting_widget.PlottingWidget):
:param xdata: float - time value in other plot
"""
self.zoom_marker1_shown = False
x_idx, y_idx = find_tps_tm(xdata, self.each_day5_min_list)
x_idx, y_idx = find_tps_tm_idx(xdata, self.start_5mins_of_diff_days)
for rl in self.rulers:
rl.set_data(x_idx, y_idx)
......@@ -446,10 +447,12 @@ class TimePowerSquaredWidget(plotting_widget.PlottingWidget):
y index (which day) of self.min_x and self.min_y, and set data for
all markers in self.zoom_marker1s and self.zoom_marker2s.
"""
x_idx, y_idx = find_tps_tm(self.min_x, self.each_day5_min_list)
x_idx, y_idx = find_tps_tm_idx(self.min_x,
self.start_5mins_of_diff_days)
for zm1 in self.zoom_marker1s:
zm1.set_data(x_idx, y_idx)
x_idx, y_idx = find_tps_tm(self.max_x, self.each_day5_min_list)
x_idx, y_idx = find_tps_tm_idx(self.max_x,
self.start_5mins_of_diff_days)
for zm2 in self.zoom_marker2s:
zm2.set_data(x_idx, y_idx)
......
......@@ -13,13 +13,13 @@ class TimePowerSquaredProcessorSignal(QtCore.QObject):
class TimePowerSquaredProcessor(QtCore.QRunnable):
def __init__(self, channel_id: str, channel_data: dict, start_time: float,
end_time: float, each_day_5_mins_list: np.ndarray):
end_time: float, start_5mins_of_diff_days: np.ndarray):
super().__init__()
self.channel_id = channel_id
self.channel_data = channel_data
self.start_time = start_time
self.end_time = end_time
self.each_day_5_mins_list = each_day_5_mins_list
self.start_5mins_of_diff_days = start_5mins_of_diff_days
self.signals = TimePowerSquaredProcessorSignal()
# Flag to indicate whether the processor should stop running and clean
# up.
......@@ -79,7 +79,8 @@ class TimePowerSquaredProcessor(QtCore.QRunnable):
trimmed_traces_list = self.trim_waveform_data()
# preset all 0 for all 5 minutes for each day
tps_data = np.zeros((len(self.each_day_5_mins_list), const.NO_5M_DAY))
tps_data = np.zeros((len(self.start_5mins_of_diff_days),
const.NO_5M_DAY))
spr = self.channel_data['samplerate']
self.channel_data['tps_data'] = []
......@@ -110,8 +111,8 @@ class TimePowerSquaredProcessor(QtCore.QRunnable):
# identify index in case of overlaps or gaps
index = np.where(
(self.each_day_5_mins_list <= times[start_index]) &
(self.each_day_5_mins_list + const.SEC_5M > times[start_index])
(self.start_5mins_of_diff_days <= times[start_index]) &
(self.start_5mins_of_diff_days + const.SEC_5M > times[start_index]) # noqa: E501
)
curr_row = index[0][0]
curr_col = index[1][0]
......
......@@ -299,7 +299,9 @@ class SearchMessageDialog(QtWidgets.QWidget):
count = 0
for log_line in self.soh_dict[chan_id]:
self.add_line(
self.soh_tables_dict[chan_id], text1=count, text2=log_line)
self.soh_tables_dict[chan_id],
text1=str(count),
text2=log_line)
count += 1
def add_nav_button(self, nav: QtWidgets.QToolBar,
......@@ -510,7 +512,7 @@ class SearchMessageDialog(QtWidgets.QWidget):
self.search_rowidx = 0
ret = self.search(col=0, start_search=True)
if ret is None:
raise ValueError(f'Not found line: ({data_index})')
raise ValueError(f'Not found line: {data_index}')
it, r = ret
self.on_log_entry_clicked(it)
self._show_log_message(it)
......
......@@ -292,5 +292,38 @@ def extract_netcodes(data_obj):
return '\n\t'.join(net_info_list)
def get_index_from_time(chan_data: List[np.ndarray], tm: float, val: float) \
-> Tuple[int, int]:
"""
Get index of tm in chan_data['time'] which is a list of np.ndarray
:param chan_data: dict of data to plot that includes 'times' key
:param tm: epoch time of a clicked point
:param val: data value of a clicked point
:return list_idx: index of the np.ndarray in the list
:return section_idx: index of tm inside np.ndarray found
"""
list_idx = None
section_idx = None
for i in range(len(chan_data['times'])):
section_indexes = np.where(chan_data['times'][i] == tm)[0]
if len(section_indexes) != 0:
if (chan_data['chan_db_info']['plotType'] not in
['linesDots', 'linesSRate']):
# don't check val because the plotting value of some plot
# is for displaying only, not actual value. And this type
# of data isn't under the effect of overlap.
section_idx = section_indexes[0]
list_idx = i
break
else:
# to prevent 2 values with the same times in case of overlap
for j in section_indexes:
if chan_data['data'][i][j] == val:
list_idx = i
section_idx = j
break
return list_idx, section_idx
if __name__ == '__main__':
create_table_of_content_file(Path('../../../documentation'))
from unittest import TestCase
from obspy import UTCDateTime
from sohstationviewer.model.handling_data import (
get_start_5mins_of_diff_days, find_tps_tm_idx
)
class TestGetEachDay5MinList(TestCase):
def test_start_in_midle_end_exact(self):
"""
Start in the middle of a day and end at the exact end of a day
"""
with self.subTest("start, end in different day"):
start = UTCDateTime("2012-09-07T12:15:00").timestamp
end = UTCDateTime("2012-09-09T00:00:00").timestamp
start_5mins_of_diff_days = get_start_5mins_of_diff_days(
start, end
)
self.assertEqual(len(start_5mins_of_diff_days), 2)
self.assertEqual(len(start_5mins_of_diff_days[0]), 288)
self.assertEqual(len(start_5mins_of_diff_days[1]), 288)
with self.subTest("start, end in same day"):
start = UTCDateTime("2012-09-07T12:15:00").timestamp
end = UTCDateTime("2012-09-08T00:00:00").timestamp
start_5mins_of_diff_days = get_start_5mins_of_diff_days(
start, end
)
self.assertEqual(len(start_5mins_of_diff_days), 1)
self.assertEqual(len(start_5mins_of_diff_days[0]), 288)
def test_start_exact_end_in_middle(self):
"""
Start at the very beginning of a day and end in the middle of a day
"""
with self.subTest("start, end in different day"):
start = UTCDateTime("2012-09-07T00:00:00").timestamp
end = UTCDateTime("2012-09-08T12:15:00").timestamp
start_5mins_of_diff_days = get_start_5mins_of_diff_days(
start, end
)
self.assertEqual(len(start_5mins_of_diff_days), 2)
self.assertEqual(len(start_5mins_of_diff_days[0]), 288)
self.assertEqual(len(start_5mins_of_diff_days[1]), 288)
with self.subTest("start, end in same day"):
start = UTCDateTime("2012-09-0700:00:00").timestamp
end = UTCDateTime("2012-09-07T12:13:00").timestamp
start_5mins_of_diff_days = get_start_5mins_of_diff_days(
start, end
)
self.assertEqual(len(start_5mins_of_diff_days), 1)
self.assertEqual(len(start_5mins_of_diff_days[0]), 288)
class TestFindTPSTmIdx(TestCase):
@classmethod
def setUpClass(cls) -> None:
start = UTCDateTime("2012-09-07T12:15:00").timestamp
end = UTCDateTime("2012-09-09T00:00:00").timestamp
# cover 2 days: 2012/09/07 and 2012/09/08
cls.start_5mins_of_diff_days = get_start_5mins_of_diff_days(
start, end
)
def test_given_time_beginning_of_first_day(self):
tm = UTCDateTime("2012-09-07T00:00:00").timestamp
tps_tm_idx = find_tps_tm_idx(tm, self.start_5mins_of_diff_days)
self.assertEqual(tps_tm_idx, (0, 0))
def test_given_time_middle_of_day(self):
tm = UTCDateTime("2012-09-07T12:13:00").timestamp
tps_tm_idx = find_tps_tm_idx(tm, self.start_5mins_of_diff_days)
self.assertEqual(tps_tm_idx, (146, 0))
def test_given_time_beginning_of_day_in_middle(self):
tm = UTCDateTime("2012-09-08T00:00:00").timestamp
tps_tm_idx = find_tps_tm_idx(tm, self.start_5mins_of_diff_days)
self.assertEqual(tps_tm_idx, (0, -1))
def test_given_time_very_end_of_last_day(self):
tm = UTCDateTime("2012-09-09T00:00:00").timestamp
start_tps_tm_idx = find_tps_tm_idx(tm, self.start_5mins_of_diff_days)
self.assertEqual(start_tps_tm_idx, (287, -1))
......@@ -15,7 +15,7 @@ from sohstationviewer.model.handling_data import (
trim_downsample_wf_chan,
trim_waveform_data,
downsample_waveform_data,
get_each_day_5_min_list,
get_start_5mins_of_diff_days,
)
from sohstationviewer.view.plotting.time_power_squared_processor import (
TimePowerSquaredProcessor,
......@@ -682,11 +682,11 @@ class TestGetTrimTpsData(TestCase):
self.add_trace(start_time)
self.start_time = 25000
self.end_time = 75000
self.each_day_5_mins_list = get_each_day_5_min_list(self.start_time,
self.end_time)
self.start_5mins_of_diff_days = get_start_5mins_of_diff_days(
self.start_time, self.end_time)
self.tps_processor = TimePowerSquaredProcessor(
channel_id, self.channel_data, self.start_time, self.end_time,
self.each_day_5_mins_list
self.start_5mins_of_diff_days
)
local_TimePowerSquaredProcessor = (sohstationviewer.view.plotting.
......@@ -761,9 +761,9 @@ class TestGetTrimTpsData(TestCase):
# 86400). Thus, we only have to set the end time to the data end time
# to have two days of data.
self.tps_processor.end_time = 100000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time, self.tps_processor.end_time
)
self.tps_processor.start_5mins_of_diff_days = \
get_start_5mins_of_diff_days(self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
......@@ -785,9 +785,9 @@ class TestGetTrimTpsData(TestCase):
with self.subTest('test_start_time_in_gap'):
self.tps_processor.start_time = 15000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.start_5mins_of_diff_days = \
get_start_5mins_of_diff_days(self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 1)
tps_gap = slice(0, 50)
......@@ -799,9 +799,9 @@ class TestGetTrimTpsData(TestCase):
with self.subTest('test_start_time_cover_all_traces'):
self.tps_processor.start_time = 500
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.start_5mins_of_diff_days = \
get_start_5mins_of_diff_days(self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 1)
tps_gap = slice(2, 83)
......@@ -827,9 +827,9 @@ class TestGetTrimTpsData(TestCase):
with self.subTest('test_end_time_in_gap'):
# Subject to change after Issue #37 is fixed
self.tps_processor.end_time = 110000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.start_5mins_of_diff_days = \
get_start_5mins_of_diff_days(self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
tps_gaps = (slice(45, 128), slice(131, None))
......@@ -843,9 +843,9 @@ class TestGetTrimTpsData(TestCase):
with self.subTest('test_end_time_cover_all_traces'):
self.tps_processor.end_time = trace_start_time + 50
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.start_5mins_of_diff_days = \
get_start_5mins_of_diff_days(self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
tps_gaps = (slice(45, 128), slice(131, None))
......@@ -871,9 +871,9 @@ class TestGetTrimTpsData(TestCase):
with self.subTest('test_start_time_in_gap'):
self.tps_processor.start_time = -25000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.start_5mins_of_diff_days = \
get_start_5mins_of_diff_days(self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
tps_gap = slice(const.NO_5M_DAY)
......@@ -885,9 +885,9 @@ class TestGetTrimTpsData(TestCase):
with self.subTest('test_start_time_cover_all_traces'):
self.tps_processor.start_time = -60000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.start_5mins_of_diff_days = \
get_start_5mins_of_diff_days(self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 2)
tps_gaps = (slice(0, 121), slice(124, None))
......@@ -919,17 +919,17 @@ class TestGetTrimTpsData(TestCase):
with self.subTest('test_end_time_same_day_as_second_to_last_trace'):
# Subject to change after Issue #37 is fixed
self.tps_processor.end_time = 125000
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.start_5mins_of_diff_days = \
get_start_5mins_of_diff_days(self.tps_processor.start_time,
self.tps_processor.end_time)
with self.assertRaises(IndexError):
self.tps_processor.run()
with self.subTest('test_end_time_cover_all_traces'):
self.tps_processor.end_time = trace_start_time + 50
self.tps_processor.each_day_5_mins_list = get_each_day_5_min_list(
self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.start_5mins_of_diff_days = \
get_start_5mins_of_diff_days(self.tps_processor.start_time,
self.tps_processor.end_time)
self.tps_processor.run()
self.assertEqual(len(self.channel_data['tps_data']), 3)
tps_gap_day_2 = slice(45, None)
......
......@@ -9,7 +9,7 @@ from sohstationviewer.view.util.functions import (
get_soh_messages_for_view, log_str, is_doc_file,
create_search_results_file, create_table_of_content_file,
check_chan_wildcards_format, check_masspos, get_total_miny_maxy,
extract_netcodes
extract_netcodes, get_index_from_time
)
from sohstationviewer.view.util.enums import LogType
......@@ -455,3 +455,49 @@ class TestExtractNetcodes(TestCase):
data_obj = MockObj({'3734': {'XX', 'NA'}})
ret = extract_netcodes(data_obj)
self.assertEqual(ret, "NA,XX")
class TestGetIndexFromTime(TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.plotting_data = {
'CH1': {
'times': [np.array([1, 2, 3]), np.array([4, 5, 6])],
'data': [np.array([6, 9, 7]), np.array([4, 3, 9])],
'chan_db_info': {'plotType': 'upDownDots'}
},
'CH2': {
'times': [np.array([1, 2, 3]), np.array([3, 5, 6])],
'data': [np.array([6, 9, 7]), np.array([4, 3, 9])],
'chan_db_info': {'plotType': 'linesDots'}
}
}
def test_time_not_included(self):
list_idx, section_idx = get_index_from_time(
self.plotting_data['CH1'], 7, 6)
self.assertIsNone(list_idx)
def test_type_not_need_data_info(self):
# CH1 has plotType='upDownDots' not in ['linesDots', 'linesSRate']
list_idx, section_idx = get_index_from_time(
self.plotting_data['CH1'], 4, 4)
self.assertEqual(list_idx, 1)
self.assertEqual(section_idx, 0)
def test_type_need_data_info(self):
# CH2 has plotType='linesDots in ['linesDots', 'linesSRate']
with self.subTest('data not match time'):
list_idx, section_idx = get_index_from_time(
self.plotting_data['CH2'], 3, 5)
self.assertIsNone(list_idx)
with self.subTest('data match 1st value'):
list_idx, section_idx = get_index_from_time(
self.plotting_data['CH2'], 3, 7)
self.assertEqual(list_idx, 0)
self.assertEqual(section_idx, 2)
with self.subTest('data match 2nd value'):
list_idx, section_idx = get_index_from_time(
self.plotting_data['CH2'], 3, 4)
self.assertEqual(list_idx, 1)
self.assertEqual(section_idx, 0)