import datetime
import warnings
import pyqtgraph as pg
import scipy.optimize.optimize
from PyQt5.QtWidgets import *
from PyQt5.QtGui import QColor
from PyQt5.QtCore import *
import pandas as pd
from pandas.api.types import is_numeric_dtype
from pyqtgraph.GraphicsScene.mouseEvents import MouseClickEvent
import utils
from config import Config
from dataprovider.exploprovider import ExploProvider
from dataprovider.icbktransitionprovider import IcbktransitionProvider
from gui.ui_mainwindow import Ui_MainWindow
from gui.stab_dialog import StabDialog
from cfatools.logreader.dataset import DatasetReader
from cfatools.tsanalyser import transition
class ExploUim(QObject):
sig_dataset_loaded = pyqtSignal(DatasetReader, name="dataset_loaded")
sig_cursor_moved = pyqtSignal(QTime)
# "Variables" tableWidget columns identifiers.
INSTRUMENT_COL = 0
VARIABLE_COL = 1
COLOR_COL = 2
OFFSET_COL = 3
MULT_COL = 4
TIMESHIFT_COL = 5
VISIBLE_COL = 6
XORIG_COL = 7
YORIG_COL = 8
DEFAULT_COLORS = [
QColor(228, 26, 28), # Red
QColor(55, 126, 184), # Blue
QColor(77, 175, 74), # Green
QColor(152, 78, 163), # Violet
QColor(255, 127, 0), # Orange
QColor(255, 255, 51), # Yellow
QColor(166, 86, 40), # Brown
QColor(247, 129, 191), # Pink
]
def __init__(
self,
explo_prvd: ExploProvider,
icbktransition_prvd: IcbktransitionProvider,
main_ui: Ui_MainWindow,
config: Config,
stab_dialog: StabDialog,
):
super(ExploUim, self).__init__()
self.main_ui = main_ui
self.explo_prvd = explo_prvd
self.icbktransition_prvd = icbktransition_prvd
self.config = config
self.stab_dialog = stab_dialog
self.dataset_readers = dict()
# The "var_id" is used to identify and match table's lines and plot's items (curves, etc.). Table's row id can
# not be used because it changes when previous rows are deleted.
self.current_variable_id = 0
self.__initialize_variable_table__()
self.__initialize_plot__()
# Internal signals
self.explo_prvd.sig_dataset_loaded.connect(self.__initialize_plot__)
self.explo_prvd.sig_dataset_loaded.connect(self.__reset_variable_table__)
self.explo_prvd.sig_dataset_loaded.connect(self.__refresh_existing_setups__)
self.icbktransition_prvd.sig_transitions_df_updated.connect(self.__update_icbk_transition__)
self.main_ui.explo_pushbutton_dataset_load.clicked.connect(self.__load_dataset__)
self.main_ui.explo_pushbutton_add_row.clicked.connect(self.__add_new_row_in_variable_table__)
self.main_ui.explo_tablewidget_variables.cellClicked.connect(self.__change_color__)
# Manual events display
self.main_ui.explo_checkbox_manualevent.stateChanged.connect(self.__update_manual_event__)
# Transitions vline display
self.main_ui.explo_checkbox_icbktransition.stateChanged.connect(self.__update_icbk_transition__)
# Export plot
self.main_ui.explo_pushbutton_export_plot.clicked.connect(self.__export_plot__)
# Save/load current setup
self.main_ui.explo_lineedit_setup_name.textChanged.connect(self.__check_setup_save_name__)
self.main_ui.explo_pushbutton_setup_save.clicked.connect(self.__save_setup__)
self.main_ui.explo_listwidget_setup_list.itemClicked.connect(
lambda: self.main_ui.explo_pushbutton_setup_load.setEnabled(True)
)
self.main_ui.explo_pushbutton_setup_load.clicked.connect(self.__load_setup__)
# Stabilization time analysis
self.__init_stab__()
self.main_ui.explo_pushbutton_stab.clicked.connect(self.__show_stab_dialog__)
self.region = None
self.rolling_avg_duration = 0
def __load_dataset__(self):
# Get the directory containing the dataset's files.
directory = utils.get_data_path(self.config, "datasets", "dataset")
if directory == "":
return
self.main_ui.explo_pushbutton_icbktransition.setEnabled(False)
self.main_ui.explo_checkbox_icbktransition.setEnabled(False)
try:
self.explo_prvd.get_dataset_reader(directory)
except ValueError as e:
utils.show_popup(str(e), "Could not load data set!")
return
self.main_ui.explo_lineedit_dataset.setText(self.explo_prvd.current_dataset.dataset_name)
# Enable control which are disabled by default, i.e. when no dataset is yet loaded.
self.main_ui.explo_pushbutton_export_plot.setEnabled(True)
self.main_ui.explo_pushbutton_add_row.setEnabled(True)
self.main_ui.explo_pushbutton_stab.setEnabled(True)
self.main_ui.explo_checkbox_manualevent.setEnabled(True)
self.main_ui.explo_lineedit_setup_name.setEnabled(True)
####################################################################################################################
# "Variables" table
def __initialize_variable_table__(self):
"""Initialize the table containing the to-be-displayed variables"""
# Set column widths
self.main_ui.explo_tablewidget_variables.setColumnWidth(self.INSTRUMENT_COL, 150)
self.main_ui.explo_tablewidget_variables.setColumnWidth(self.VARIABLE_COL, 130)
self.main_ui.explo_tablewidget_variables.setColumnWidth(self.COLOR_COL, 55)
self.main_ui.explo_tablewidget_variables.setColumnWidth(self.OFFSET_COL, 100)
self.main_ui.explo_tablewidget_variables.setColumnWidth(self.MULT_COL, 120)
self.main_ui.explo_tablewidget_variables.setColumnWidth(self.TIMESHIFT_COL, 100)
self.main_ui.explo_tablewidget_variables.setColumnWidth(self.VISIBLE_COL, 70)
self.main_ui.explo_tablewidget_variables.setColumnWidth(self.XORIG_COL, 90)
self.main_ui.explo_tablewidget_variables.setColumnWidth(self.YORIG_COL, 200)
def __reset_variable_table__(self):
self.main_ui.explo_tablewidget_variables.setRowCount(0)
self.__add_new_row_in_variable_table__()
def __add_new_row_in_variable_table__(self):
"""Add a new row in the calibration table. Initialize the cell's content and properties."""
table = self.main_ui.explo_tablewidget_variables
row_id = table.rowCount()
table.insertRow(row_id)
# Instruments
instrument_item = QComboBox()
table.setCellWidget(row_id, self.INSTRUMENT_COL, instrument_item)
# Variables
variable_item = QComboBox()
table.setCellWidget(row_id, self.VARIABLE_COL, variable_item)
table.cellWidget(row_id, self.VARIABLE_COL).currentTextChanged.connect(
lambda text, row_id=row_id: self.__apply_variable_change__(row_id=row_id, variable_name=text)
)
# Connect Instrument change to variables display
table.cellWidget(row_id, self.INSTRUMENT_COL).currentTextChanged.connect(
lambda text, row_id=row_id: self.__update_variables_combobox__(
combobox_text=text, variables_combobox=variable_item
)
)
# Color
color_item = QTableWidgetItem()
color_item.setBackground(self.DEFAULT_COLORS[row_id % len(self.DEFAULT_COLORS)])
table.setItem(row_id, self.COLOR_COL, color_item)
# Offset
offset_item = QDoubleSpinBox()
offset_item.setMinimum(-10000.0)
offset_item.setMaximum(10000.0)
offset_item.setDecimals(3)
table.setCellWidget(row_id, self.OFFSET_COL, offset_item)
table.cellWidget(row_id, self.OFFSET_COL).valueChanged.connect(
lambda value, row_id=row_id: self.__apply_variable_change__(row_id=row_id)
)
# Multiplicative factor
mult_item = QDoubleSpinBox()
mult_item.setValue(1.0)
mult_item.setMinimum(-10000.0)
mult_item.setMaximum(10000.0)
mult_item.setDecimals(5)
table.setCellWidget(row_id, self.MULT_COL, mult_item)
table.cellWidget(row_id, self.MULT_COL).valueChanged.connect(
lambda value, row_id=row_id: self.__apply_variable_change__(row_id=row_id)
)
# Time shift
timeshift_item = QDoubleSpinBox()
timeshift_item.setMinimum(-10000.0)
timeshift_item.setMaximum(10000.0)
table.setCellWidget(row_id, self.TIMESHIFT_COL, timeshift_item)
table.cellWidget(row_id, self.TIMESHIFT_COL).valueChanged.connect(
lambda value, row_id=row_id: self.__apply_variable_change__(row_id=row_id)
)
# Visible
visible_item = QCheckBox()
visible_item.setChecked(True)
table.setCellWidget(row_id, self.VISIBLE_COL, visible_item)
table.cellWidget(row_id, self.VISIBLE_COL).stateChanged.connect(
lambda state, row_id=row_id: self.__apply_variable_change__(row_id=row_id)
)
# X original
xorig_item = QTableWidgetItem()
xorig_item.setFlags(Qt.ItemIsEnabled) # Read only
table.setItem(row_id, self.XORIG_COL, xorig_item)
# Y original
yorig_item = QTableWidgetItem()
yorig_item.setFlags(Qt.ItemIsEnabled) # Read only
table.setItem(row_id, self.YORIG_COL, yorig_item)
self.__update_instruments_combobox__(
self.main_ui.explo_tablewidget_variables.cellWidget(row_id, self.INSTRUMENT_COL)
)
if hasattr(self, "plot_item"):
self.plot_item.getViewBox().enableAutoRange(enable=True)
def __update_instruments_combobox__(self, combobox: QComboBox):
combobox.clear()
for instrument_name in self.explo_prvd.current_dataset.get_instruments_names():
if instrument_name == "manual-event":
continue
combobox.addItem(instrument_name)
def __update_variables_combobox__(self, combobox_text: str, variables_combobox: QComboBox):
if combobox_text == "":
return
variables_combobox.clear()
instrument_name = combobox_text
variable_names = self.explo_prvd.get_instrument_variables(self.explo_prvd.current_dataset, instrument_name)
for variable_name in variable_names:
variables_combobox.addItem(variable_name)
def __change_color__(self, row: int, column: int):
if column != self.COLOR_COL:
return
color = QColorDialog.getColor()
self.main_ui.explo_tablewidget_variables.item(row, self.COLOR_COL).setBackground(color)
self.__apply_variable_change__(row)
def __get_row_dataframe__(self, row_id: int) -> pd.DataFrame:
table = self.main_ui.explo_tablewidget_variables
# Get instrument log
instrument_name = table.cellWidget(row_id, self.INSTRUMENT_COL).currentText()
variable_name = table.cellWidget(row_id, self.VARIABLE_COL).currentText()
timeseries = self.explo_prvd.get_timeseries(self.explo_prvd.current_dataset, instrument_name, variable_name)
return timeseries
def __apply_variable_change__(self, row_id: int, variable_name: str = None) -> None:
"""Read the information related to the row_id and call plot function
Parameters
----------
row_id
variable_name
"""
table = self.main_ui.explo_tablewidget_variables
# The variable combobox is cleared before variables of the newly-selected instrument are written.
# This function is called on variable-combobox edition, thus it is also called on combobox clear.
# So in this case, do nothing, wait for the call related to the filling of the combobox
if variable_name == "":
return
if table.cellWidget(row_id, self.VARIABLE_COL).currentText() == "":
return
timeseries = self.__get_row_dataframe__(row_id)
# Set tooltip
variable_combobox = table.cellWidget(row_id, self.VARIABLE_COL)
instrument_name = table.cellWidget(row_id, self.INSTRUMENT_COL).currentText()
if variable_name is None:
variable_name = variable_combobox.currentText()
variable_combobox = self.__add_tooltip_to_variable_combobox__(variable_combobox, instrument_name, variable_name)
# Get color
color = table.item(row_id, self.COLOR_COL).background().color()
# Get variable visibility
visible = table.cellWidget(row_id, self.VISIBLE_COL).isChecked()
# Get position adjustment variables (offset, multiplicative factor and time shift)
offset = table.cellWidget(row_id, self.OFFSET_COL).value()
mult = table.cellWidget(row_id, self.MULT_COL).value()
timeshift = table.cellWidget(row_id, self.TIMESHIFT_COL).value()
try:
self.__update_plot__(timeseries, row_id, color, offset, mult, timeshift, visible)
except TypeError:
self.main_ui.statusbar.showMessage("Failed to plot [" + variable_name + "]", msecs=3000)
def __update_xy_original__(self, instant: datetime.datetime):
table = self.main_ui.explo_tablewidget_variables
for row_id in range(table.rowCount()):
# Get X orig (original non-shifted datetime value)
timeshift_sec = table.cellWidget(row_id, self.TIMESHIFT_COL).value()
x_orig = instant - datetime.timedelta(seconds=timeshift_sec)
table.item(row_id, self.XORIG_COL).setText(x_orig.strftime("%H:%M:%S.%f")[:-5])
# Get Y orig (original non-shifted variable value)
df = self.__get_row_dataframe__(row_id).copy()
df = df[df.index <= instant]
if len(df.index) == 0:
y_orig = "Out of range"
else:
y_orig = df.iloc[-1]["value"]
try:
y_orig = float(y_orig)
except ValueError:
pass # y_orig is probably a string, which is normal for some variables (e.g. Ice core's name)
else:
y_orig = "{:.4f}".format(y_orig)
table.item(row_id, self.YORIG_COL).setText(y_orig)
def __add_tooltip_to_variable_combobox__(
self, combobox: QComboBox, instrument_name: str, variable_name: str
) -> QComboBox:
descr_dict = self.explo_prvd.get_variable_description(
self.explo_prvd.current_dataset, instrument_name, variable_name
)
tooltip_str = descr_dict["description"]
if descr_dict["unit"] != "":
tooltip_str += "\nUnits: " + descr_dict["unit"]
combobox.setToolTip(tooltip_str)
return combobox
####################################################################
# Plot
def __initialize_plot__(self) -> None:
self.plot_item = pg.PlotItem(axisItems={"bottom": utils.TimeAxisItem(orientation="bottom")})
self.step_curves = dict()
self.main_ui.explo_graphicsview_top.setCentralItem(self.plot_item)
self.lines_items = []
# Vertical lines for the manual events
self.event_vlines = []
self.event_texts = []
# Vertical lines for the transitions
self.transition_vlines = []
# Vertical line following the cursor
self.cursor_vline = pg.InfiniteLine(angle=90, movable=False, pen=pg.mkPen(style=Qt.DotLine))
self.plot_item.addItem(self.cursor_vline, ignoreBounds=True)
# Click-fixed vertical line to get measure
self.measure_vline = pg.InfiniteLine(angle=90, movable=False)
self.plot_item.addItem(self.measure_vline, ignoreBounds=True)
def __update_plot__(
self,
timeseries: pd.DataFrame,
row_id: int,
color: QColor,
offset: float,
mult: float,
timeshift_sec: float,
visible: bool,
) -> None:
# Get the to-be-modified curve
if row_id in self.step_curves:
step_curve = self.step_curves[row_id]
else:
step_curve = pg.PlotCurveItem()
self.step_curves[row_id] = step_curve
self.plot_item.addItem(step_curve)
step_curve.scene().sigMouseClicked.connect(lambda event: self.__mouse_clicked__(event))
step_curve.scene().sigMouseMoved.connect(lambda event: self.__mouse_moved__(event))
# Set curve visibility
if not visible:
step_curve.hide()
else:
step_curve.show()
# Set data to the plot
step_curve.setData(
x=self.__get_timeseries_x_values__(timeseries, timeshift_sec),
y=self.__get_timeseries_y_values__(timeseries, offset, mult),
pen=color,
stepMode=True,
)
def __reset_manual_event_checkbox__(self, dataset: DatasetReader):
# Enable/disable the possibility to display manual events, depending on the events log file availability.
has_manual_events = "manual-event" in dataset.get_instruments_names()
self.main_ui.explo_checkbox_manualevent.setEnabled(has_manual_events)
self.main_ui.explo_checkbox_manualevent.setChecked(False)
def __update_manual_event__(self, checked_state: int):
if checked_state == 2:
try:
events_df = self.explo_prvd.current_dataset.get_timeseries("manual-event", "event")
except KeyError:
# Can not find manual-event data: there is no manual event for this dataset
return
for event_date, event in events_df.iterrows():
x_pos = utils.pd_time_to_epoch_ms([event_date])[0]
# Vertical line
event_vline = pg.InfiniteLine(
pos=x_pos,
angle=90,
movable=False,
pen=pg.mkPen(color=QColor(255, 255, 255)),
)
self.event_vlines.append(event_vline)
self.plot_item.addItem(event_vline, ignoreBounds=True)
# Text
event_str = event["event"]
if event_str is None:
event_str = "[empty]"
event_str = event_str.replace("\n", "<br>")
text_item = pg.TextItem(
html='<div style="text-align: center"><span style="color: #FFF;">' + event_str + "</div>",
# anchor=(-0.3, 0.5),
rotateAxis=(0, 1),
# angle=60,
border="w",
fill=(0, 0, 255, 100),
)
self.plot_item.addItem(text_item, ignoreBounds=True)
text_item.setPos(x_pos, self.plot_item.getViewBox().viewRange()[1][0])
self.event_texts.append(text_item)
elif checked_state == 0:
for event_vline in self.event_vlines:
self.plot_item.removeItem(event_vline)
for text_item in self.event_texts:
self.plot_item.removeItem(text_item)
def __update_icbk_transition__(self):
self.main_ui.explo_checkbox_icbktransition.setEnabled(True)
checked = self.main_ui.explo_checkbox_icbktransition.isChecked()
for transition_vline in self.transition_vlines:
self.plot_item.removeItem(transition_vline)
self.transition_vlines = []
if checked:
transitions_df = self.icbktransition_prvd.transitions_df
transitions_df["timestamp"] = utils.pd_time_to_epoch_ms(transitions_df["datetime"])
for index, transition in transitions_df.iterrows():
transition_line = pg.InfiniteLine(
angle=90,
movable=False,
pen=pg.mkPen(
style=Qt.SolidLine,
width=1,
color=QColor(55, 126, 184), # Blue #377EB8
),
)
transition_line.setPos(transition["timestamp"])
self.transition_vlines.append(transition_line)
self.plot_item.addItem(transition_line, ignoreBounds=True)
def __mouse_clicked__(self, event: MouseClickEvent) -> None:
"""Function triggered when the user clicks on the plot. Display a vertical line under the mouse click and call
function ``__update_xy_original__``
Parameters
----------
event: pyqtgraph.MouseClickEvent
"""
pos = event.scenePos()
mouse_point = self.step_curves[0].getViewBox().mapSceneToView(pos)
instant = datetime.datetime.fromtimestamp(mouse_point.x(), tz=datetime.timezone.utc)
self.measure_vline.setPos(mouse_point.x())
self.__update_xy_original__(instant)
def __mouse_moved__(self, pos: QPointF) -> None:
"""Function triggered when the user's mouse cursor hovers over the plot. Display a vertical line where the
cursor is, and update time shift with click-fixed line, if any.
Parameters
----------
pos: PyQt5.QtCore.QPointF
"""
mouse_point = self.step_curves[0].getViewBox().mapSceneToView(pos)
self.cursor_vline.setPos(mouse_point.x())
if self.measure_vline.getPos() != [0, 0]:
timeshift_s = abs(self.measure_vline.getPos()[0] - self.cursor_vline.getPos()[0])
self.main_ui.explo_label_timeshift.setText("Time shift: " + "{:.2f}".format(timeshift_s) + "s")
pos_x = QDateTime()
pos_x.setMSecsSinceEpoch(round(mouse_point.x() * 1000))
pos_x = pos_x.toUTC()
self.sig_cursor_moved.emit(pos_x.time())
def __get_timeseries_x_values__(self, timeseries: pd.DataFrame, timeshift_sec: float = 0) -> list:
# As it is a _step_ curve, add a last datetime point to determine the end of the last step. This is the datetime
# of the last available data of the dataset, plus one second.
first, last = self.explo_prvd.current_dataset.get_data_timeframe()
last_datetime = last + datetime.timedelta(seconds=1)
x_values = timeseries.index.copy().to_series()
x_values = x_values.append(pd.Series([last_datetime]))
# Apply time shift
x_values = x_values + datetime.timedelta(seconds=timeshift_sec)
# Convert to epoch
x_values = utils.pd_time_to_epoch_ms(x_values)
return x_values
def __get_timeseries_y_values__(self, timeseries: pd.DataFrame, offset: float = 0.0, mult: float = 1.0) -> list:
# Get original value if it is a numeric, otherwise get its coded integer version.
if is_numeric_dtype(timeseries["value"]):
y_values = list(timeseries["value"])
else:
y_values = list(timeseries["value_int"])
# Apply multiplicative factor
y_values = [y * mult for y in y_values]
# Apply Y-axis offset
y_values = [y + offset for y in y_values]
return y_values
def __export_plot__(self):
directory = self.explo_prvd.current_dataset.dataset_path + "exported_plots/"
utils.export_plot(self.plot_item, self.main_ui.explo_graphicsview_top, base_dir=directory)
####################################################################################################################
# Save/load current setup
def __check_setup_save_name__(self, filename: str):
valid, error_msg = self.explo_prvd.setup_filename_is_valid(self.explo_prvd.current_dataset, filename)
if not valid:
self.main_ui.statusbar.showMessage(error_msg, 5000)
self.main_ui.explo_lineedit_setup_name.setStyleSheet("color: 'red';")
self.main_ui.explo_pushbutton_setup_save.setEnabled(False)
else:
self.main_ui.explo_pushbutton_setup_save.setEnabled(True)
self.main_ui.explo_lineedit_setup_name.setStyleSheet("color: 'black';")
def __save_setup__(self):
# File name
filename = self.main_ui.explo_lineedit_setup_name.text()
# Variables
variable_df = pd.DataFrame()
table = self.main_ui.explo_tablewidget_variables
for row_id in range(table.rowCount()):
row_dict = {
"instrument": table.cellWidget(row_id, self.INSTRUMENT_COL).currentText(),
"variable": table.cellWidget(row_id, self.VARIABLE_COL).currentText(),
"color": table.item(row_id, self.COLOR_COL).background().color().name(),
"offset": table.cellWidget(row_id, self.OFFSET_COL).value(),
"mult": table.cellWidget(row_id, self.MULT_COL).value(),
"timeshift": table.cellWidget(row_id, self.TIMESHIFT_COL).value(),
"visible": table.cellWidget(row_id, self.VISIBLE_COL).checkState() == 2,
}
variable_df = variable_df.append(row_dict, ignore_index=True)
# View range
view_range = self.plot_item.getViewBox().viewRange()
self.explo_prvd.save_setup(self.explo_prvd.current_dataset, filename, variable_df, view_range)
self.__refresh_existing_setups__(self.explo_prvd.current_dataset)
# Reset filename input widgets
self.main_ui.explo_lineedit_setup_name.setText(None)
self.main_ui.explo_pushbutton_setup_save.setEnabled(False)
def __load_setup__(self):
# Clear table and plot
self.main_ui.explo_tablewidget_variables.setRowCount(0)
self.__initialize_plot__()
# Get a dataframe containing the variables data
filename = self.main_ui.explo_listwidget_setup_list.selectedItems()[0].text()
variable_df, view_range_dict = self.explo_prvd.load_setup(self.explo_prvd.current_dataset, filename)
# Variables: table (and automatically: plot)
table = self.main_ui.explo_tablewidget_variables
for row_id, row in variable_df.iterrows():
self.__add_new_row_in_variable_table__()
# Instrument
instrument_index = table.cellWidget(row_id, self.INSTRUMENT_COL).findText(row["instrument"])
table.cellWidget(row_id, self.INSTRUMENT_COL).setCurrentIndex(instrument_index)
# Variable
variable_index = table.cellWidget(row_id, self.VARIABLE_COL).findText(row["variable"])
table.cellWidget(row_id, self.VARIABLE_COL).setCurrentIndex(variable_index)
# Color
color = QColor(row["color"])
table.item(row_id, self.COLOR_COL).setBackground(color)
# Offset
table.cellWidget(row_id, self.OFFSET_COL).setValue(row["offset"])
# Mult
table.cellWidget(row_id, self.MULT_COL).setValue(row["mult"])
# Timeshift
table.cellWidget(row_id, self.TIMESHIFT_COL).setValue(row["timeshift"])
# Visible
table.cellWidget(row_id, self.VISIBLE_COL).setChecked(row["visible"])
self.__apply_variable_change__(row_id) # Useful only for color change
# Plot view range
self.plot_item.getViewBox().enableAutoRange(enable=False)
if len(view_range_dict) > 0:
self.plot_item.getViewBox().setRange(
xRange=(view_range_dict["xmin"], view_range_dict["xmax"]),
yRange=(view_range_dict["ymin"], view_range_dict["ymax"]),
padding=0,
)
def __refresh_existing_setups__(self, dataset: DatasetReader):
files = self.explo_prvd.get_setup_saved_files(dataset)
self.main_ui.explo_listwidget_setup_list.clear()
for file in files:
self.main_ui.explo_listwidget_setup_list.addItem(file)
self.main_ui.explo_pushbutton_setup_load.setEnabled(False)
####################################################################################################################
# Stabilization analysis
def __init_stab__(self):
var_combo = self.stab_dialog.ui.stab_combobox_variable
self.stab_dialog.ui.stab_combobox_instrument.currentTextChanged.connect(
lambda inst_text, var_combo=var_combo: self.__update_variables_combobox__(inst_text, var_combo)
)
self.stab_dialog.ui.stab_combobox_variable.currentTextChanged.connect(self.__update_stab_plot__)
self.stab_dialog.ui.stab_spinbox_rolling_avg.valueChanged.connect(self.__update_rolling_avg)
def __show_stab_dialog__(self):
self.__init_stab_plot__()
self.__update_instruments_combobox__(self.stab_dialog.ui.stab_combobox_instrument)
self.stab_dialog.show()
def __init_stab_plot__(self):
self.stab_plot_item = pg.PlotItem(axisItems={"bottom": utils.TimeAxisItem(orientation="bottom")})
self.stab_dialog.ui.stab_graphicsview.setCentralItem(self.stab_plot_item)
# Base curve
self.stab_step_curve = pg.PlotCurveItem()
self.stab_plot_item.addItem(self.stab_step_curve)
# Region
self.stab_region = pg.LinearRegionItem(brush=pg.mkBrush(color=[0, 0, 255, 40]))
self.stab_region.setZValue(10)
self.stab_plot_item.addItem(self.stab_region, ignoreBounds=True)
self.stab_region.sigRegionChanged.connect(lambda region: self.__update_region__(region))
# Erf fit curve
self.stab_erf_curve = pg.PlotCurveItem(pen=pg.mkPen(color=[255, 0, 0]))
self.stab_plot_item.addItem(self.stab_erf_curve)
# Rolling_avg curve
self.stab_rolling_avg_curve = pg.PlotCurveItem(pen=pg.mkPen(color=[0, 255, 0]))
self.stab_plot_item.addItem(self.stab_rolling_avg_curve)
# Baseline
self.stab_baseline = pg.InfiniteLine(pen=pg.mkPen(color=[255, 0, 0], style=Qt.DotLine), angle=0)
self.stab_plot_item.addItem(self.stab_baseline)
# Amplitude
self.stab_amplitude = pg.InfiniteLine(pen=pg.mkPen(color=[255, 0, 0], style=Qt.DotLine), angle=0)
self.stab_plot_item.addItem(self.stab_amplitude)
# Center
self.stab_center = pg.InfiniteLine(pen=pg.mkPen(color=[255, 0, 0], style=Qt.DotLine), angle=90)
self.stab_plot_item.addItem(self.stab_center)
# Sigma
self.stab_sigma_left = pg.InfiniteLine(pen=pg.mkPen(color=[0, 255, 0], style=Qt.DotLine), angle=90)
self.stab_plot_item.addItem(self.stab_sigma_left)
self.stab_sigma_right = pg.InfiniteLine(pen=pg.mkPen(color=[0, 255, 0], style=Qt.DotLine), angle=90)
self.stab_plot_item.addItem(self.stab_sigma_right)
self.stab_timeseries = pd.DataFrame()
def __update_stab_plot__(self, var_combobox_text: str):
# The variable combobox is cleared before variables of the newly-selected instrument are written.
# This function is called on variable-combobox edition, thus it is also called on combobox clear.
# So in this case, do nothing, wait for the call related to the filling of the combobox
if var_combobox_text == "":
return
# Get data
instrument_name = self.stab_dialog.ui.stab_combobox_instrument.currentText()
variable_name = self.stab_dialog.ui.stab_combobox_variable.currentText()
self.stab_timeseries = self.explo_prvd.get_timeseries(
self.explo_prvd.current_dataset, instrument_name, variable_name
).copy()
# Update combobox tooltip
variable_combobox = self.stab_dialog.ui.stab_combobox_variable
variable_combobox = self.__add_tooltip_to_variable_combobox__(variable_combobox, instrument_name, variable_name)
# Set data to step curve
x_values = self.__get_timeseries_x_values__(self.stab_timeseries)
self.stab_step_curve.setData(
x=x_values,
y=self.__get_timeseries_y_values__(self.stab_timeseries),
stepMode=True,
)
# Convert data in a form more convenient for plot
self.stab_timeseries["datetime"] = utils.pd_time_to_epoch_ms(self.stab_timeseries.index)
if not is_numeric_dtype(self.stab_timeseries["value"]):
self.stab_timeseries["value"] = self.stab_timeseries["value_int"]
# If the regions are not yet placed, it means that this function is executed for stabilization window's
# initialization --> use view box of the main window's plot
if self.stab_region.getRegion() == (0, 1):
# Get the main windows view box
main_range = self.plot_item.getViewBox().viewRange()
# Place the 2 (right/left) regions
x_min = main_range[0][0]
x_max = main_range[0][1]
x_1_third = x_min + ((x_max - x_min) / 3)
x_2_third = x_min + 2 * ((x_max - x_min) / 3)
self.stab_region.setRegion([x_1_third, x_2_third])
# X range is the same as main window's plot X range
self.stab_plot_item.getViewBox().disableAutoRange()
self.stab_plot_item.getViewBox().setXRange(min=main_range[0][0], max=main_range[0][1], padding=0)
# Otherwise, the function is executed due to a variable change --> keep current viewbox and regions settings.
else:
pass
self.stab_plot_item.getViewBox().setYRange(
min=self.stab_timeseries["value"].min(),
max=self.stab_timeseries["value"].max(),
padding=0,
)
# Re-calculate the position of the left/right means
self.__update_region__(self.stab_region)
def __update_region__(self, region: pg.LinearRegionItem):
self.region = region
self.__update_statistics__()
def __update_rolling_avg(self, duration_sec: int):
self.rolling_avg_duration = duration_sec
self.__update_statistics__()
def __update_statistics__(self):
df = self.stab_timeseries.copy()
# Get the selected region's boundaries
xmin = self.region.getRegion()[0]
xmax = self.region.getRegion()[1]
df = df[(df["datetime"] >= xmin) & (df["datetime"] <= xmax)]
if len(df.index) == 0:
return
# Get data inside region's boundaries
inside_values = df["value"]
# Compute x_value's average and standard deviation over the selected region.
mean = inside_values.mean()
sd = inside_values.std()
duration_sec = xmax - xmin
duration_qtime = QTime(0, 0, 0)
duration_qtime = duration_qtime.addMSecs(duration_sec * 1000)
# Update UI's text zones
self.stab_dialog.ui.stab_doublespinbox_mean.setValue(mean)
self.stab_dialog.ui.stab_doublespinbox_sd.setValue(sd)
self.stab_dialog.ui.stab_timeedit_duration.setTime(duration_qtime)
self.__compute_erf_fitting__(df)
def __compute_erf_fitting__(self, df: pd.DataFrame):
if len(df.index) == 0:
return
if self.rolling_avg_duration > 0:
df["value_orig"] = df["value"]
df["value"] = (
df["value_orig"].rolling(str(self.rolling_avg_duration) + "s", min_periods=1, center=True).mean()
)
self.stab_rolling_avg_curve.setData(
x=utils.pd_time_to_epoch_ms(df.index.copy().to_series()),
y=self.__get_timeseries_y_values__(df),
stepMode=False,
)
self.stab_rolling_avg_curve.show()
else:
self.stab_rolling_avg_curve.hide()
# Make Scipy's OptimizeWarning as plain errors to be able to catch them to avoid trying to display erf fitting
# curves while fitting failed.
warnings.filterwarnings("error", category=scipy.optimize.optimize.OptimizeWarning)
try:
(
baseline,
amplitude,
center,
sigma,
erf_df,
) = transition.get_transition_duration_erf(df)
except RuntimeError:
self.__update_invalid_erf__()
return
except scipy.optimize.optimize.OptimizeWarning:
self.__update_invalid_erf__()
return
else:
self.stab_baseline.show()
self.stab_amplitude.show()
self.stab_center.show()
self.stab_sigma_left.show()
self.stab_sigma_right.show()
self.stab_erf_curve.show()
warnings.filterwarnings("default", category=scipy.optimize.optimize.OptimizeWarning)
# Plot lines
center_sec = pd.to_datetime(center).timestamp()
self.stab_baseline.setValue(baseline)
self.stab_amplitude.setValue(baseline + amplitude)
self.stab_center.setValue(center_sec)
self.stab_sigma_left.setValue(center_sec - sigma)
self.stab_sigma_right.setValue(center_sec + sigma)
# Plot Erf fitting
x_values = utils.pd_time_to_epoch_ms(erf_df.index.copy().to_series())
y_values = list(erf_df["erf"])
self.stab_erf_curve.setData(x=x_values, y=y_values)
# Update Spinboxes and TimeEdits
self.stab_dialog.ui.stab_doublespinbox_baseline.setValue(baseline)
self.stab_dialog.ui.stab_doublespinbox_amplitude.setValue(amplitude)
self.stab_dialog.ui.stab_doublespinbox_sigma.setValue(sigma)
self.stab_dialog.ui.stab_timeedit_center.setTime(
QTime(center.hour, center.minute, center.second, center.microsecond / 1000)
)
def __update_invalid_erf__(self):
"""Update plot lines and DoubleSpinBoxes and TimeEdits for the case when no Erf fitting is possible."""
# Hide InfiniteLines
self.stab_baseline.hide()
self.stab_amplitude.hide()
self.stab_center.hide()
self.stab_sigma_left.hide()
self.stab_sigma_right.hide()
# Hide Erf fit
self.stab_erf_curve.hide()
# Update Spinboxes and TimeEdits
self.stab_dialog.ui.stab_doublespinbox_baseline.setValue(-9999999999)
self.stab_dialog.ui.stab_doublespinbox_amplitude.setValue(-9999999999)
self.stab_dialog.ui.stab_doublespinbox_sigma.setValue(-9999999999)
self.stab_dialog.ui.stab_timeedit_center.setTime(QTime(0, 0))