291 lines
13 KiB
Python
291 lines
13 KiB
Python
import pandas as pd
|
||
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QLabel
|
||
|
||
|
||
import pyqtgraph as pg
|
||
import numpy as np
|
||
from typing import Optional, Any
|
||
from PyQt5.QtOpenGL import QGLWidget
|
||
|
||
from utils.base.base import BasePlotWidget
|
||
|
||
class ProcessStage():
|
||
mean_value:int
|
||
start_index:int
|
||
finish_index:int
|
||
|
||
class PlotWidget(BasePlotWidget):
|
||
|
||
def _create_navigator(self,
|
||
time_region:tuple[float, float],
|
||
main_plot: pg.PlotWidget,
|
||
dataframe: pd.DataFrame,
|
||
real_signals: list[dict[str, Any]]) -> list[pg.PlotWidget, pg.LinearRegionItem]:
|
||
"""
|
||
Создаёт график-навигатор, отображающий все данные в уменьшенном масштабе.
|
||
"""
|
||
navigator = pg.PlotWidget(title="Navigator")
|
||
navigator.setFixedHeight(100)
|
||
navigator.setBackground('d')
|
||
|
||
for signal in real_signals:
|
||
if signal["name"] in dataframe.columns:
|
||
x = dataframe["time"]
|
||
y = dataframe[signal["name"]]
|
||
|
||
x_downsampled, y_downsampled = self._downsample_data(x, y, max_points=1000)
|
||
navigator.plot(x_downsampled, y_downsampled, pen=signal["pen"], name=signal["name"])
|
||
|
||
ROI_region = pg.LinearRegionItem(values=time_region, movable=True, brush=pg.mkBrush(0, 0, 255, 50))
|
||
navigator.addItem(ROI_region)
|
||
|
||
# Связываем изменение региона навигатора с обновлением области просмотра основного графика
|
||
ROI_region.sigRegionChanged.connect(lambda: self._sync_main_plot_with_navigator(main_plot, ROI_region))
|
||
|
||
return navigator, ROI_region
|
||
|
||
def _downsample_data(self, x, y, max_points=5000):
|
||
"""
|
||
Понижает разрешение данных до заданного количества точек для улучшения производительности навигатора.
|
||
"""
|
||
if len(x) > max_points:
|
||
factor = len(x) // max_points
|
||
x_downsampled = x[::factor]
|
||
y_downsampled = y[::factor]
|
||
return x_downsampled, y_downsampled
|
||
return x, y
|
||
|
||
def _sync_main_plot_with_navigator(self,
|
||
main_plot: pg.PlotWidget,
|
||
region: pg.LinearRegionItem):
|
||
"""
|
||
Синхронизирует область просмотра основного графика с регионом навигатора.
|
||
"""
|
||
x_min, x_max = region.getRegion()
|
||
if main_plot:
|
||
main_plot.blockSignals(True)
|
||
main_plot.setXRange(x_min, x_max, padding=0)
|
||
main_plot.blockSignals(False)
|
||
|
||
def _create_curve_ideal(self,
|
||
signal: dict[str, Any],
|
||
ideal_data: pd.DataFrame,
|
||
start_timestamp: float,
|
||
finish_timestamp: float) -> Optional[pg.PlotDataItem]:
|
||
"""
|
||
Создаёт идеальную кривую для сигнала, если заданы корректные временные рамки.
|
||
"""
|
||
if start_timestamp is not None and finish_timestamp is not None:
|
||
return pg.PlotDataItem(
|
||
x=start_timestamp + ideal_data["time"],
|
||
y=ideal_data[signal["name"]],
|
||
pen=signal["pen"]
|
||
)
|
||
return None
|
||
|
||
def _create_stage_region(self,
|
||
stage: str,
|
||
start_timestamp: float,
|
||
finish_timestamp: float,
|
||
transparency: int) -> Optional[pg.LinearRegionItem]:
|
||
"""
|
||
Создает регион для определённого этапа, если заданы временные рамки.
|
||
"""
|
||
if start_timestamp is not None and finish_timestamp is not None:
|
||
region = pg.LinearRegionItem([start_timestamp, finish_timestamp], movable=False)
|
||
color = self._stage_colors.get(stage, [100, 100, 100, 100])
|
||
region.setBrush(pg.mkBrush(color[:3] + [transparency]))
|
||
return region
|
||
return None
|
||
|
||
@staticmethod
|
||
def _init_plot_widget(title: str) -> tuple[pg.PlotWidget, pg.LegendItem]:
|
||
plot_widget = pg.PlotWidget(title=title)
|
||
# Оптимизация отображения графиков
|
||
plot_widget.setDownsampling(auto=True, mode='peak')
|
||
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 _add_stage_regions(self,
|
||
plot_widget: pg.PlotWidget,
|
||
point_events: dict[str, list[float]],
|
||
dataframe_headers: list[str],
|
||
transparency: int = 75) -> None:
|
||
"""
|
||
Добавляет регионы для реальных этапов, если все стадии есть в заголовках датафрейма.
|
||
"""
|
||
stages = point_events.keys()
|
||
if all(stage in dataframe_headers for stage in stages):
|
||
for stage in stages:
|
||
start_t, end_t = point_events[stage]
|
||
region = self._create_stage_region(stage, start_t, end_t, transparency)
|
||
if region is not None:
|
||
plot_widget.addItem(region)
|
||
|
||
def _add_ideal_stage_regions(self,
|
||
plot_widget: pg.PlotWidget,
|
||
ideal_data: dict[str, Any],
|
||
point_events: dict[str, list[float]],
|
||
transparency: int = 125) -> None:
|
||
"""
|
||
Добавляет регионы для идеальных этапов.
|
||
"""
|
||
ideal_timings = ideal_data["Ideal timings"]
|
||
stages = list(point_events.keys())
|
||
for i, stage in enumerate(stages):
|
||
start_t = point_events[stage][0]
|
||
end_t = start_t + ideal_timings[i]
|
||
region = self._create_stage_region(stage, start_t, end_t, transparency)
|
||
if region:
|
||
plot_widget.addItem(region)
|
||
|
||
def _add_ideal_signals(self,
|
||
plot_widget: pg.PlotWidget,
|
||
ideal_data: dict[str, Any],
|
||
point_events: dict[str, list[float]],
|
||
ideal_signals: list[dict[str, Any]]) -> None:
|
||
"""
|
||
Добавляет идеальные сигналы для каждого этапа.
|
||
"""
|
||
for stage in point_events.keys():
|
||
for signal in ideal_signals:
|
||
curve = self._create_curve_ideal(
|
||
signal,
|
||
ideal_data[stage],
|
||
point_events[stage][0],
|
||
point_events[stage][1]
|
||
)
|
||
if curve:
|
||
plot_widget.addItem(curve)
|
||
|
||
def _add_real_signals(self,
|
||
plot_widget: pg.PlotWidget,
|
||
dataframe: pd.DataFrame,
|
||
real_signals: list[dict[str, Any]],
|
||
legend: pg.LegendItem) -> None:
|
||
"""
|
||
Добавляет реальные сигналы из dataframe на виджет.
|
||
"""
|
||
dataframe_headers = dataframe.columns.tolist()
|
||
for signal in real_signals:
|
||
if signal["name"] in dataframe_headers:
|
||
plot = plot_widget.plot(dataframe["time"], dataframe[signal["name"]], pen=signal["pen"], fast=True)
|
||
legend.addItem(plot, signal["name"])
|
||
|
||
def _add_performance_label(self,
|
||
layout: QVBoxLayout,
|
||
TWC_time: float,
|
||
ideal_time: float,
|
||
tesla_time: float) -> None:
|
||
"""
|
||
Добавляет QLabel с информацией о производительности.
|
||
"""
|
||
tesla_TWC = round((1 - TWC_time/tesla_time)*100, 2) if tesla_time else 0.0
|
||
tesla_ideal = round((1 - ideal_time/tesla_time)*100, 2) if tesla_time else 0.0
|
||
TWC_ideal = round((ideal_time/TWC_time)*100, 2) if TWC_time else 0.0
|
||
|
||
performance_label = QLabel(
|
||
f"Сокращение длительности: фактическое = {tesla_TWC} %, "
|
||
f"идеальное = {tesla_ideal} %; КДИП = {TWC_ideal}%"
|
||
)
|
||
layout.addWidget(performance_label)
|
||
performance_label.update()
|
||
|
||
def _build_widget(self, data: list[Any]) -> QWidget:
|
||
"""
|
||
Собирает графический виджет для одного набора данных.
|
||
Параметр `data` предполагается списком: [dataframe, points_pocket, tesla_time].
|
||
"""
|
||
widget = QGLWidget()
|
||
layout = QVBoxLayout(widget)
|
||
|
||
dataframe, points_pocket, tesla_time = data
|
||
dataframe_headers = dataframe.columns.tolist()
|
||
|
||
for widget_num, (channel, description) in enumerate(self._plt_channels.items()):
|
||
plot_widget, legend = self._init_plot_widget(title=channel)
|
||
settings = description["Settings"]
|
||
|
||
TWC_time = 0.0
|
||
ideal_time = 0.0
|
||
worst_perf = 2
|
||
|
||
# Итерация по точкам
|
||
for cur_point, point_data in enumerate(points_pocket):
|
||
# point_data структура: [point_timeframe, ideal_data, point_events]
|
||
point_timeframe, ideal_data, point_events = point_data
|
||
|
||
# Добавляем реальные стадии
|
||
if settings["stages"]:
|
||
self._add_stage_regions(plot_widget, point_events, dataframe_headers, transparency=75)
|
||
|
||
# Добавляем идеальные стадии и идеальные сигналы
|
||
if settings["ideals"]:
|
||
self._add_ideal_stage_regions(plot_widget, ideal_data, point_events)
|
||
self._add_ideal_signals(plot_widget, ideal_data, point_events, description["Ideal_signals"])
|
||
|
||
# Подсчёт производительности
|
||
if settings["performance"]:
|
||
is_last_point = (cur_point == len(points_pocket) - 1)
|
||
if is_last_point:
|
||
TWC_delta = sum([point_events[stage][1] - point_events[stage][0] for stage in ["Closing", "Squeeze", "Welding"]])
|
||
ideal_delta = sum(ideal_data["Ideal timings"][0:3])
|
||
else:
|
||
TWC_delta = point_timeframe[1] - point_timeframe[0]
|
||
ideal_delta = ideal_data["Ideal cycle"]
|
||
TWC_time += TWC_delta
|
||
ideal_time += ideal_delta
|
||
curr_perf = ideal_delta/TWC_delta if TWC_delta != 0 else 1
|
||
if curr_perf < worst_perf:
|
||
worst_perf = curr_perf
|
||
worst_timeframe = point_timeframe
|
||
|
||
# Добавляем реальные сигналы
|
||
self._add_real_signals(plot_widget, dataframe, description["Real_signals"], legend)
|
||
if widget_num == 0:
|
||
main_plot = plot_widget
|
||
else:
|
||
# Связываем остальные графики с основным графиком
|
||
plot_widget.setXLink(main_plot)
|
||
|
||
# Если есть настройка производительности, добавляем label
|
||
if settings["performance"]:
|
||
self._add_performance_label(layout, TWC_time, ideal_time, tesla_time)
|
||
navigator, ROI_region = self._create_navigator(worst_timeframe, main_plot, dataframe, description["Real_signals"])
|
||
|
||
layout.addWidget(plot_widget)
|
||
|
||
layout.addWidget(navigator)
|
||
self._sync_main_plot_with_navigator(main_plot, ROI_region)
|
||
main_plot.sigXRangeChanged.connect(lambda _, plot=main_plot, region=ROI_region: self._sync_navigator_with_main(main_plot=plot, region=region))
|
||
|
||
widget.setLayout(layout)
|
||
return widget
|
||
|
||
def _sync_navigator_with_main(self, main_plot: pg.PlotWidget, region:pg.LinearRegionItem):
|
||
"""
|
||
Синхронизирует регион навигатора с областью просмотра основного графика.
|
||
"""
|
||
if region:
|
||
x_min, x_max = main_plot
|
||
region.blockSignals(True) # Предотвращаем рекурсию
|
||
region.setRegion([x_min, x_max])
|
||
region.blockSignals(False)
|
||
|
||
def build(self, data: list[list[Any]]) -> None:
|
||
"""
|
||
Создает набор виджетов по предоставленному списку данных.
|
||
Предполагается, что data — это список элементов вида:
|
||
[
|
||
[dataframe, points_pocket, tesla_time],
|
||
[dataframe, points_pocket, tesla_time],
|
||
...
|
||
]
|
||
"""
|
||
widgets = [self._build_widget(data_sample) for data_sample in data]
|
||
self._mediator.notify(self, widgets)
|
||
|
||
|