396 lines
19 KiB
Python
396 lines
19 KiB
Python
# Qt-панель обработки графиков с выполнением расчётов в рабочей задаче. Снимок входных данных
|
|
# отделён от живого графика; сигналы доставляют результат обратно в UI для добавления
|
|
# самостоятельной вычисленной кривой.
|
|
|
|
"""Reusable Qt processing panel and plot attachment; no SETGUI dependency."""
|
|
from __future__ import annotations
|
|
from dataclasses import replace
|
|
|
|
try:
|
|
from PySide6.QtCore import QObject, QPointF, QRunnable, QThreadPool, QSignalBlocker, QTimer, Qt, Signal, Slot
|
|
from PySide6.QtGui import QColor, QPainterPath, QPen
|
|
from PySide6.QtWidgets import (
|
|
QComboBox, QDialog, QFileDialog, QHBoxLayout, QLabel, QPushButton, QSpinBox,
|
|
QVBoxLayout, QWidget,
|
|
)
|
|
except ImportError:
|
|
from PySide2.QtCore import QObject, QPointF, QRunnable, QThreadPool, QSignalBlocker, QTimer, Qt, Signal, Slot
|
|
from PySide2.QtGui import QColor, QPainterPath, QPen
|
|
from PySide2.QtWidgets import (
|
|
QComboBox, QDialog, QFileDialog, QHBoxLayout, QLabel, QPushButton, QSpinBox,
|
|
QVBoxLayout, QWidget,
|
|
)
|
|
|
|
from set_devices.signal_reconstruction import METHODS
|
|
from set_devices.plot_processing import Snapshot, prepare, process, write_csv
|
|
|
|
RESULT_COLORS = ("#FF70D0", "#50D5FF", "#FFD166", "#83E377", "#B99AFF", "#FF9870")
|
|
|
|
|
|
class _ResultSignals(QObject):
|
|
done = Signal(object, object, str)
|
|
|
|
|
|
class _Calculation(QRunnable):
|
|
def __init__(self, signature, parent=None):
|
|
super().__init__()
|
|
self.signature = signature
|
|
# Qt owns the emitter on the GUI thread, independently of QRunnable's
|
|
# auto-deletion by the pool (important for PySide2 wrapper lifetimes).
|
|
self.signals = _ResultSignals(parent)
|
|
|
|
def run(self):
|
|
try:
|
|
result = process(self.signature)
|
|
except Exception as error:
|
|
self.signals.done.emit(self.signature, None, str(error))
|
|
else:
|
|
self.signals.done.emit(self.signature, result, "")
|
|
|
|
|
|
class SignalProcessingPanel(QWidget):
|
|
changed = Signal()
|
|
RESULT_KEY = "__calculated_trend__"
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self._snapshot = Snapshot()
|
|
self._result = None
|
|
self._result_signature = None
|
|
self._results = []
|
|
self._result_ids = {}
|
|
self._next_result_id = 0
|
|
self._job = None
|
|
self._revision = 0
|
|
self._job_revision = 0
|
|
self.preserve_on_view_change = False
|
|
self.export_path = lambda filename: filename
|
|
layout = QVBoxLayout(self)
|
|
layout.setContentsMargins(0, 0, 0, 0)
|
|
row = QHBoxLayout()
|
|
row.addWidget(QLabel("Обработка"))
|
|
self.channel = QComboBox()
|
|
self.channel.setMinimumWidth(140)
|
|
row.addWidget(self.channel)
|
|
self.method = QComboBox()
|
|
for key, title in METHODS.items():
|
|
self.method.addItem(title, key)
|
|
self.method.setCurrentIndex(2)
|
|
row.addWidget(self.method)
|
|
row.addStretch(1)
|
|
layout.addLayout(row)
|
|
row = QHBoxLayout()
|
|
row.addWidget(QLabel("Степень"))
|
|
self.degree = QSpinBox()
|
|
self.degree.setRange(1, 5)
|
|
self.degree.setValue(2)
|
|
row.addWidget(self.degree)
|
|
row.addWidget(QLabel("Точек результата"))
|
|
self.count = QSpinBox()
|
|
self.count.setRange(2, 10000)
|
|
self.count.setValue(1000)
|
|
row.addWidget(self.count)
|
|
self.apply_button = QPushButton("Рассчитать")
|
|
self.apply_button.clicked.connect(self.calculate)
|
|
row.addWidget(self.apply_button)
|
|
self.clear_button = QPushButton("Убрать все")
|
|
self.clear_button.clicked.connect(self.clear)
|
|
row.addWidget(self.clear_button)
|
|
self.export_button = QPushButton("Результат CSV…")
|
|
self.export_button.clicked.connect(self.export_csv)
|
|
self.export_button.setToolTip("Сохранить последний рассчитанный результат")
|
|
row.addWidget(self.export_button)
|
|
row.addStretch(1)
|
|
layout.addLayout(row)
|
|
self.status = QLabel("Выберите аналоговый канал и метод. Расчёт по переданным точкам, без экстраполяции.")
|
|
self.status.setWordWrap(True)
|
|
layout.addWidget(self.status)
|
|
self.method.setToolTip(
|
|
"Полином МНК — сглаживание шумных данных. Линейная — отрезки между точками.\n"
|
|
"PCHIP — сохранение формы без выбросов. Сплайн — гладкая кривая, возможны выбросы.\n"
|
|
"Восстановление — оценка между измерениями; утраченные детали не определяются однозначно.")
|
|
for widget in (self.channel, self.method):
|
|
widget.currentIndexChanged.connect(self._parameters_changed)
|
|
for widget in (self.degree, self.count):
|
|
widget.valueChanged.connect(self._parameters_changed)
|
|
self._update_actions()
|
|
|
|
def _signature(self):
|
|
return prepare(self._snapshot, self.channel.currentData(), self.method.currentData(),
|
|
self.count.value(), self.degree.value())
|
|
|
|
def _current_request_for(self, signature):
|
|
if not self.preserve_on_view_change or signature is None:
|
|
return self._signature()
|
|
# Validate the original calculation domain, independently of navigation.
|
|
return prepare(replace(self._snapshot, x_range=signature.x_range),
|
|
self.channel.currentData(), self.method.currentData(),
|
|
self.count.value(), self.degree.value())
|
|
|
|
def set_snapshot(self, snapshot):
|
|
previous = self._job.signature if self._job is not None else self._signature()
|
|
was_blocked = bool(self._snapshot.blocked_reason)
|
|
choices = [(signal.key, signal.label) for signal in snapshot.analogs]
|
|
current = [(self.channel.itemData(i), self.channel.itemText(i)) for i in range(self.channel.count())]
|
|
if choices != current:
|
|
selected = self.channel.currentData()
|
|
blocker = QSignalBlocker(self.channel)
|
|
self.channel.clear()
|
|
for key, title in choices:
|
|
self.channel.addItem(title, key)
|
|
index = self.channel.findData(selected)
|
|
if index >= 0:
|
|
self.channel.setCurrentIndex(index)
|
|
del blocker
|
|
self._snapshot = snapshot
|
|
if previous != self._current_request_for(previous):
|
|
self._revision += 1
|
|
retained = []
|
|
for result in self._results:
|
|
request = result.request
|
|
source = replace(snapshot, x_range=request.x_range) if self.preserve_on_view_change else snapshot
|
|
# Visibility selects inputs for NEW calculations; existing results
|
|
# depend on samples/units, not the source curve's checkbox.
|
|
source = replace(source, series=tuple(replace(series, visible=True)
|
|
for series in source.series))
|
|
if prepare(source, request.series.key, request.method, request.output_count, request.degree) == request:
|
|
retained.append(result)
|
|
if len(retained) != len(self._results):
|
|
self._results = retained
|
|
self._result_ids = {id(curve): self._result_ids[id(curve)] for curve in retained}
|
|
if not any(curve is self._result for curve in retained):
|
|
self._result = retained[-1] if retained else None
|
|
self._result_signature = self._result.request if self._result is not None else None
|
|
self.status.setText("Данные или окно изменились — нажмите «Рассчитать» повторно.")
|
|
if snapshot.blocked_reason:
|
|
self.status.setText(snapshot.blocked_reason)
|
|
elif not choices:
|
|
self.status.setText("Нет видимых аналоговых каналов для обработки.")
|
|
elif was_blocked:
|
|
self.status.setText("Выберите канал и метод, затем нажмите «Рассчитать».")
|
|
self._update_actions()
|
|
|
|
def _update_actions(self):
|
|
self.degree.setEnabled(self.method.currentData() == "polynomial")
|
|
request = self._signature()
|
|
self.apply_button.setEnabled(self._job is None and request is not None and
|
|
len(request.series.points) >= 2)
|
|
self.export_button.setEnabled(self._result is not None and not self._snapshot.blocked_reason)
|
|
self.clear_button.setEnabled(self._result is not None or self._job is not None)
|
|
self.apply_button.setToolTip("Обработка выбранной кривой в её единицах X/Y. Для живого потока остановите сбор.")
|
|
|
|
def clear(self, *_):
|
|
self._revision += 1
|
|
self._results.clear()
|
|
self._result_ids.clear()
|
|
self._result = self._result_signature = None
|
|
self.status.setText("Расчёт по текущему окну. Исходные измерения сохраняются; кривая — оценка между точками.")
|
|
self._update_actions()
|
|
self.changed.emit()
|
|
|
|
def _parameters_changed(self, *_):
|
|
self._revision += 1
|
|
self.status.setText("Нажмите «Рассчитать», чтобы добавить кривую. Предыдущие результаты сохраняются.")
|
|
self._update_actions()
|
|
|
|
def remove_curve(self, key):
|
|
for curve in self._results:
|
|
index = self._result_ids[id(curve)]
|
|
curve_key = self.RESULT_KEY if index == 0 else f"{self.RESULT_KEY}_{index}"
|
|
if curve_key != key:
|
|
continue
|
|
if self._job is not None and self._job.signature == curve.request:
|
|
self._revision += 1
|
|
self._results.remove(curve)
|
|
del self._result_ids[id(curve)]
|
|
if self._result is curve:
|
|
self._result = self._results[-1] if self._results else None
|
|
self._result_signature = self._result.request if self._result is not None else None
|
|
self.status.setText(f"Расчётная кривая удалена. Осталось: {len(self._results)}.")
|
|
self._update_actions()
|
|
self.changed.emit()
|
|
return True
|
|
return False
|
|
|
|
def calculate(self):
|
|
if self._job is not None or self._snapshot.blocked_reason:
|
|
return
|
|
signature = self._signature()
|
|
if signature is None or len(signature.series.points) < 2:
|
|
self.status.setText("Нужны минимум две точки аналогового канала.")
|
|
return
|
|
self._job = _Calculation(signature, self)
|
|
self._job_revision = self._revision
|
|
self._job.signals.done.connect(self._finished, Qt.QueuedConnection)
|
|
self.status.setText("Расчёт…")
|
|
self._update_actions()
|
|
QThreadPool.globalInstance().start(self._job)
|
|
|
|
@Slot(object, object, str)
|
|
def _finished(self, signature, result, error):
|
|
if self._job is not None:
|
|
self._job.signals.deleteLater()
|
|
self._job = None
|
|
if signature != self._current_request_for(signature) or self._job_revision != self._revision:
|
|
self.status.setText("Данные изменились во время расчёта. Остановите сбор и повторите расчёт.")
|
|
elif error:
|
|
self.status.setText("Расчёт не выполнен: " + error)
|
|
else:
|
|
self._result, self._result_signature = result, signature
|
|
# Repeating exactly the same request replaces its identical overlay.
|
|
existing = next((curve for curve in self._results if curve.request == signature), None)
|
|
if existing is not None:
|
|
self._result_ids[id(result)] = self._result_ids.pop(id(existing))
|
|
self._results[self._results.index(existing)] = result
|
|
else:
|
|
self._result_ids[id(result)] = self._next_result_id
|
|
self._next_result_id += 1
|
|
self._results.append(result)
|
|
duplicates = result.input_count - result.unique_count
|
|
self.status.setText(
|
|
f"{self.channel.currentText()}: {result.unique_count} исходных → {len(result.points)} расчётных точек; "
|
|
f"СКО на измерениях: {result.rmse:.6g}. "
|
|
+ (f"Повторы X усреднены: {duplicates}. " if duplicates else "")
|
|
+ f"Расчётных кривых: {len(self._results)}. Исходные данные сохранены.")
|
|
self._update_actions()
|
|
self.changed.emit()
|
|
|
|
def overlay(self):
|
|
if self._result is None or self._snapshot.blocked_reason:
|
|
return [], {}
|
|
specs, data = [], {}
|
|
for curve in self.curves:
|
|
index = self._result_ids[id(curve)]
|
|
key = self.RESULT_KEY if index == 0 else f"{self.RESULT_KEY}_{index}"
|
|
specs.append((key, curve.label, False, RESULT_COLORS[index % len(RESULT_COLORS)], True))
|
|
data[key] = curve.points
|
|
return specs, data
|
|
|
|
@property
|
|
def curves(self):
|
|
return tuple(self._results) if not self._snapshot.blocked_reason else ()
|
|
|
|
@property
|
|
def curve(self):
|
|
return self._result
|
|
|
|
def export_csv(self):
|
|
if self._result is None:
|
|
return
|
|
path, _ = QFileDialog.getSaveFileName(self, "Сохранить расчётную кривую",
|
|
self.export_path("calculated-signal.csv"), "CSV (*.csv)")
|
|
if not path:
|
|
return
|
|
try:
|
|
with open(path, "w", newline="", encoding="utf-8-sig") as stream:
|
|
write_csv(self._result, stream)
|
|
except (OSError, ValueError, OverflowError) as error:
|
|
self.status.setText("Не удалось сохранить CSV: " + str(error))
|
|
|
|
|
|
class PlotProcessingAttachment(QObject):
|
|
"""Attach to any Qt plot via snapshot/project/repaint callbacks.
|
|
|
|
Call source_changed when raw data, units or visible channel selection change.
|
|
Paint after raw analog curves with the same rect/projection. No channel or
|
|
history mutation, no dependency on the caller's renderer or acquisition API.
|
|
"""
|
|
def __init__(self, parent, snapshot, repaint, export_path=None):
|
|
super().__init__(parent)
|
|
self.snapshot = snapshot
|
|
self.repaint = repaint
|
|
self.export_path = export_path
|
|
self.dialog = None
|
|
self.panel = None
|
|
self._external_curve = None
|
|
self._external_offset = 0.0
|
|
self.button = QPushButton("Обработка…", parent)
|
|
self.button.setToolTip("Аппроксимация, интерполяция и восстановление аналоговой кривой")
|
|
self.button.clicked.connect(self.open)
|
|
self._refresh_timer = QTimer(self)
|
|
self._refresh_timer.setSingleShot(True)
|
|
self._refresh_timer.timeout.connect(self.refresh)
|
|
self._dirty = False
|
|
|
|
@property
|
|
def curve(self):
|
|
# Hide stale geometry immediately, before coalesced refresh runs.
|
|
if self._external_curve is not None:
|
|
return self._external_curve
|
|
return self.panel.curve if self.panel is not None and not self._dirty else None
|
|
|
|
def set_external_curve(self, curve, x_offset=0.0):
|
|
"""Embedded panels own snapshot invalidation; display rebasing is explicit."""
|
|
self._external_curve = curve
|
|
self._external_offset = x_offset
|
|
self.repaint()
|
|
|
|
@property
|
|
def curves(self):
|
|
if self._external_curve is not None:
|
|
return (self._external_curve,)
|
|
return self.panel.curves if self.panel is not None and not self._dirty else ()
|
|
|
|
def source_changed(self):
|
|
if self.panel is not None:
|
|
self._dirty = True
|
|
self._refresh_timer.start(0)
|
|
|
|
def formula_entries(self, x, origin=0., unit='', window=None):
|
|
from set_devices.plot_formula import window_piece
|
|
entries = []
|
|
offset = self._external_offset if self._external_curve is not None else 0.
|
|
for curve in self.curves:
|
|
piece = (window_piece(curve, x + offset, window[0] + offset, window[1] + offset)
|
|
if window is not None else curve.piece_at(x + offset))
|
|
if piece is not None:
|
|
color_index = self.panel._result_ids[id(curve)] if self._external_curve is None else 0
|
|
text = curve.label + '\n' + piece.text(origin + offset, unit)
|
|
entries.append((RESULT_COLORS[color_index % len(RESULT_COLORS)], text))
|
|
return entries
|
|
|
|
def refresh(self):
|
|
if self.panel is not None:
|
|
self.panel.set_snapshot(self.snapshot())
|
|
self._dirty = False
|
|
self.repaint()
|
|
|
|
def open(self):
|
|
if self.dialog is None:
|
|
self.dialog = QDialog(self.parent())
|
|
self.dialog.setWindowTitle("Обработка графика")
|
|
layout = QVBoxLayout(self.dialog)
|
|
self.panel = SignalProcessingPanel(self.dialog)
|
|
if self.export_path is not None:
|
|
self.panel.export_path = self.export_path
|
|
self.panel.changed.connect(self.repaint)
|
|
layout.addWidget(self.panel)
|
|
self.dialog.resize(760, 240)
|
|
self.refresh()
|
|
self.dialog.show()
|
|
self.dialog.raise_()
|
|
self.dialog.activateWindow()
|
|
|
|
def paint(self, painter, rect, project, project_series=None):
|
|
curves = self.curves
|
|
if not curves:
|
|
return
|
|
painter.save()
|
|
painter.setClipRect(rect)
|
|
for curve_index, curve in enumerate(curves):
|
|
path = QPainterPath()
|
|
for index, (x, y) in enumerate(curve.points):
|
|
stamp = x - (self._external_offset if self._external_curve is not None else 0)
|
|
point = (project_series(curve.request.series, stamp, y, rect) if project_series
|
|
else project(stamp, y, rect))
|
|
path.moveTo(point) if index == 0 else path.lineTo(point)
|
|
color_index = self.panel._result_ids[id(curve)] if self._external_curve is None else 0
|
|
color = QColor(RESULT_COLORS[color_index % len(RESULT_COLORS)])
|
|
painter.setPen(QPen(color, 2, Qt.DashLine))
|
|
painter.setBrush(Qt.NoBrush)
|
|
painter.drawPath(path)
|
|
painter.setPen(color)
|
|
painter.drawText(QPointF(rect.left() + 6,
|
|
rect.bottom() - 8 - curve_index * painter.fontMetrics().height()), curve.label)
|
|
painter.restore()
|