import pandas as pd from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel, QGraphicsRectItem import pyqtgraph as pg import numpy as np from numpy import floating from typing import Optional, Any, NamedTuple from src.utils.base.base import BasePlotWidget class ProcessStage(NamedTuple): mean_value: floating[Any] start_index: int finish_index: int class PlotWidget(BasePlotWidget): def _create_curve_ideal(self, signal: str, ideal_data: pd.DataFrame, start_timestamp: float, finish_timestamp: float) -> Optional[pg.PlotDataItem]: if start_timestamp and finish_timestamp: plot = pg.PlotDataItem(x=start_timestamp+ideal_data["time"], y=ideal_data[signal["name"]], pen=signal["pen"]) return plot return None def _create_stage_region(self, stage: str, start_timestamp: float, finish_timestamp: float) -> Optional[pg.LinearRegionItem]: if start_timestamp and finish_timestamp: region = pg.LinearRegionItem([start_timestamp, finish_timestamp], movable=False) region.setBrush(pg.mkBrush(self._stage_colors[stage])) return region return None @staticmethod def _init_plot_widget(title: str) -> tuple[pg.PlotWidget, pg.LegendItem]: plot_widget = pg.PlotWidget(title=title) plot_widget.showGrid(x=True, y=True) legend = pg.LegendItem((80, 60), offset=(70, 20)) legend.setParentItem(plot_widget.graphicsItem()) return plot_widget, legend def get_stage_info(self, stage: str, dataframe: pd.DataFrame, signal_name: str) -> Optional[ProcessStage]: stage_diff = np.diff(dataframe[stage]) start_index = np.where(stage_diff == 1)[0] finish_index = np.where(stage_diff == -1)[0] data = dataframe[signal_name] if signal_name in dataframe.columns.tolist() else [] if data.size and start_index.size: start = start_index[0] finish = finish_index[0] if finish_index.size else (len(data) - 1) data_slice = data[start:finish] mean = np.mean(data_slice) return ProcessStage(mean_value=mean, start_index=int(start), finish_index=int(finish)) return None def _build_widget(self, data: list[pd.DataFrame, dict, dict, list]) -> QWidget: widget = QWidget() layout = QVBoxLayout() for channel, description in self._plt_channels.items(): performance_list = [] df_continuous = pd.DataFrame({}) plot_widget, legend = self._init_plot_widget(title=channel) settings = description["Settings"] for dataframe, ideal_data, events, tesla_data in data: df_continuous = pd.concat([df_continuous, dataframe], axis=0) dataframe_headers = dataframe.columns.tolist() stages = events.keys() if settings["stages"] and all([stage in dataframe_headers for stage in stages]): for stage in stages: start_t, end_t = events[stage] region = self._create_stage_region(stage, start_t, end_t) if region: plot_widget.addItem(region) if settings["ideals"]: for stage in stages: for signal in description["Ideal_signals"]: curve = self._create_curve_ideal(signal, ideal_data[stage], events[stage][0], events[stage][1]) if curve: plot_widget.addItem(curve) if settings["performance"]: ideal_delta = ideal_data["Ideal cycle"] delta = 0 for stage in stages: delta += events[stage][1] - events[stage][0] performance = ideal_delta/delta*100 performance_list.append(performance) if settings["zoom"]: pass """if max(time_axis) < 5.0: stages = [self.get_stage_info("Welding", dataframe, signal["name"]) for signal in description["Real_signals"]] if stages: means_raw = [stage.mean_value for stage in stages] mean = max(means_raw) start = time_axis[stages[0].start_index] finish = time_axis[stages[0].finish_index] overshoot = pg.BarGraphItem(x0=0, y0=mean - mean * 0.05, height=mean * 0.05 * 2, width=start, brush=pg.mkBrush([0, 250, 0, 100])) plot_widget.addItem(overshoot) stable = pg.BarGraphItem(x0=start, y0=mean - mean * 0.015, height=mean * 0.015 * 2, width=finish - start, brush=pg.mkBrush([0, 250, 0, 100])) plot_widget.addItem(stable) plot_widget.setYRange(mean - 260, mean + 260) plot_widget.setInteractive(False) else: max_value = min([max(dataframe[signal["name"]]) for signal in description["Real_signals"]]) region = pg.LinearRegionItem([max_value - max_value * 0.015, max_value + max_value * 0.015], movable=False, orientation="horizontal") region.setBrush(pg.mkBrush([0, 250, 0, 100])) plot_widget.setYRange(max_value - 200, max_value + 200) plot_widget.setXRange(3.5, 4.5) plot_widget.addItem(region) plot_widget.setInteractive(False)""" for signal in description["Real_signals"]: if signal["name"] in dataframe_headers: plot = plot_widget.plot(df_continuous["time"], df_continuous[signal["name"]], pen=signal["pen"]) legend.addItem(plot, signal["name"]) if settings["performance"]: performance_list = np.array(performance_list) performance_label = QLabel(f"""Performance: best = {round(performance_list.max(),2)} %, worse = {round(performance_list.min(),2)} %, average = {round(performance_list.mean(),2)}%""") TWC_start = 0 TWC_len = df_continuous["time"].max() TWC_region = QGraphicsRectItem(TWC_start, -1000, TWC_start+TWC_len, 100) TWC_region.setBrush(pg.mkBrush(0, 255, 0, 255)) tesla_start = 0 tesla_len = 35.488 tesla_reginon = QGraphicsRectItem(tesla_start, -1100, tesla_start+tesla_len ,100) tesla_reginon.setBrush(pg.mkBrush(255, 0, 0, 255)) time_reduction = round((1 - TWC_len/tesla_len)*100, 2) time_reduction_label = QLabel(f"Time reduction: {time_reduction} %") layout.addWidget(performance_label) layout.addWidget(time_reduction_label) plot_widget.addItem(TWC_region) plot_widget.addItem(tesla_reginon) layout.addWidget(plot_widget) widget.setLayout(layout) return widget def build(self, data: list[pd.DataFrame]) -> None: widgets = [self._build_widget(data_sample) for data_sample in data] self._mediator.notify(self, widgets)