From caf30f2ed68b716158c789d27daed273c981a86b Mon Sep 17 00:00:00 2001 From: andrey Date: Sun, 27 Sep 2026 02:47:24 +0300 Subject: [PATCH] Share waveform data and plot navigation and retain processing results --- python/set_devices/generator_data.py | 227 ++++++++++++++++++ .../set_devices/qt_ports/plot_navigation.py | 112 +++++++++ .../set_devices/qt_ports/plot_processing.py | 114 ++++++--- python/tests/test_generator_data.py | 37 +++ python/tests/test_plot_processing_qt.py | 75 +++++- 5 files changed, 534 insertions(+), 31 deletions(-) create mode 100644 python/set_devices/generator_data.py create mode 100644 python/set_devices/qt_ports/plot_navigation.py create mode 100644 python/tests/test_generator_data.py diff --git a/python/set_devices/generator_data.py b/python/set_devices/generator_data.py new file mode 100644 index 0000000..18db39c --- /dev/null +++ b/python/set_devices/generator_data.py @@ -0,0 +1,227 @@ +"""Built-in waveforms and CSV input for the point-based signal generator.""" +import csv +import io +import math +import json +from datetime import datetime +from set_devices.waveform import Waveform, generate as interpolate, load_recipe as load_interpolated_recipe +from set_devices.signal_reconstruction import dac12 + + +# TIM6 clock of the supported STM32F407 emulator firmware. The device protocol +# does not currently expose its timer clock; keep this hardware profile explicit. +F407_TIMER_HZ = 72_000_000 +F407_SAMPLE_RATES = tuple(sorted({ + rate + for divisor in range(1, math.isqrt(F407_TIMER_HZ) + 1) + if F407_TIMER_HZ % divisor == 0 + for rate in (divisor, F407_TIMER_HZ // divisor) + if 1 <= rate <= 1_000_000 +})) + + +def nearest_f407_sample_rate(rate): + """Nearest integer rate realizable by TIM6; ties prefer the lower rate.""" + return min(F407_SAMPLE_RATES, key=lambda candidate: (abs(candidate - rate), candidate)) + + +def generate(points, sample_rate=1000, vref=3.3, method="pchip", degree=2): + if method != "samples": + return interpolate(points, sample_rate, vref, method, degree) + if type(sample_rate) is not int or not 1 <= sample_rate <= 1000000: + raise ValueError("Частота отсчётов должна быть 1…1000000 Гц") + if not math.isfinite(vref) or not 0 < vref <= 3.6: + raise ValueError("Vref должен быть в диапазоне 0…3,6 В") + if not 2 <= len(points) <= 4096: + raise ValueError("Нужно 2…4096 отсчётов") + if any(not math.isfinite(x) or not math.isfinite(y) or not 0 <= y <= vref for x, y in points): + raise ValueError("В точках нужны конечные числа, напряжение 0…Vref") + samples = tuple((i * 1000 / sample_rate, y) for i, (_, y) in enumerate(points)) + return Waveform(samples, dac12([y for _, y in samples], vref), sample_rate, vref, + len(samples) * 1000 / sample_rate) + + +def load_recipe(text): + data = json.loads(text) + if not isinstance(data, dict) or data.get("method") != "samples": + return load_interpolated_recipe(text) + if data.get("version") != 1: + raise ValueError("Неизвестная версия задания генератора") + points = tuple((float(x), float(y)) for x, y in data["points_ms_volts"]) + wave = generate(points, data["sample_rate"], data["vref"], "samples", data["degree"]) + return wave.points, data + + +PRESET_NAMES = ( + "Синусоида", "Треугольник", "Пила вверх", "Пила вниз", "Меандр", + "Импульс 10%", "Трапеция", "Ступенчатый", "Выпрямленная синусоида", + "Затухающая синусоида", +) + + +def preset_points(index, vref, count=200): + """20 ms period, 200 samples at 10 kHz; finite one-sample pulse edges.""" + if not 0 <= index < len(PRESET_NAMES): + raise ValueError("Неизвестный сигнал") + shapes = ( + lambda t: .5 + .45 * math.sin(2 * math.pi * t), + lambda t: .05 + .9 * (1 - abs(2 * t - 1)), + lambda t: .05 + .9 * t, + lambda t: .95 - .9 * t, + lambda t: .95 if t < .5 else .05, + lambda t: .95 if .1 <= t < .2 else .05, + lambda t: .05 + .9 * max(0, min(1, (t - .1) / .2, (.9 - t) / .2)), + lambda t: .05 + .9 * min(4, int(t * 5)) / 4, + lambda t: .05 + .9 * abs(math.sin(2 * math.pi * t)), + lambda t: .5 + .45 * math.exp(-4 * t) * math.sin(8 * math.pi * t), + ) + points = [(i * 20 / count, vref * shapes[index](i / count)) for i in range(count)] + points.append((20., points[0][1])) + return points + + +def signal_csv_channels(text): + """Named channels in time-domain trend/processing exports.""" + first = next((line for line in text.lstrip('\ufeff').splitlines() if line.strip()), '') + delimiter = ';' if ';' in first else '\t' if '\t' in first else ',' + header = next(csv.reader([first], delimiter=delimiter), []) + normalized = [cell.strip().lower() for cell in header] + if normalized in (['time_ms', 'volts'], ['time_ms', 'volts', 'dac12']): + return [] + if header and header[0].strip().lower() in ('timestamp', 'время [мс]', 'time [s]', 'time [ms]', 'time_ms'): + return header[1:] + return [] + + +def _parse_trend_csv(text, channel): + first = next(line for line in text.splitlines() if line.strip()) + delimiter = ';' if ';' in first else '\t' if '\t' in first else ',' + rows = csv.reader(io.StringIO(text), delimiter=delimiter) + header = next(row for row in rows if any(c.strip() for c in row)) + names = header[1:] + if len(set(header)) != len(header): + raise ValueError('Повторяющиеся имена каналов CSV') + if channel is None: + if len(names) != 1: + raise ValueError('Выберите один канал CSV: ' + ', '.join(names)) + channel = names[0] + if channel not in names: + raise ValueError('Канал CSV не найден: ' + str(channel)) + column = header.index(channel) + kind = header[0].strip().lower() + points = [] + origin = None + previous = None + for number, row in enumerate(rows, 2): + if not any(c.strip() for c in row): + continue + try: + if len(row) != len(header): + raise ValueError('число колонок не соответствует заголовку') + if kind == 'timestamp': + stamp = datetime.fromisoformat(row[0].strip().replace('Z', '+00:00')) + else: + stamp = float(row[0].strip().replace(',', '.')) + if not math.isfinite(stamp): + raise ValueError('время должно быть конечным') + if previous is not None and stamp <= previous: + raise ValueError('время должно строго возрастать') + previous = stamp + if not row[column].strip(): + continue + value = float(row[column].strip().replace(',', '.')) + if not math.isfinite(value): + raise ValueError('значение должно быть конечным') + if origin is None: + origin = stamp + elapsed = ((stamp - origin).total_seconds() * 1000 if kind == 'timestamp' + else (stamp - origin) * (1000 if kind == 'time [s]' else 1)) + points.append((elapsed, value)) + except (ValueError, TypeError, OverflowError) as error: + raise ValueError(f'Строка {number}: {error}') from error + if len(points) < 2: + raise ValueError('Выбранный канал должен содержать минимум две точки') + return points, None + + +def resample_signal(points, rate): + """Fit a long trend to the DAC table using linear interpolation, no endpoint.""" + count = round(points[-1][0] * rate / 1000) + if not 2 <= count <= 4096: + raise ValueError('Таблица ЦАП должна содержать 2…4096 отсчётов') + result = [] + index = 0 + for i in range(count): + x = i * 1000 / rate + while index + 1 < len(points) - 1 and points[index + 1][0] < x: + index += 1 + x0, y0 = points[index] + x1, y1 = points[index + 1] + result.append((x, y0 + (y1 - y0) * (x - x0) / (x1 - x0))) + return result + + +def parse_signal_csv(text, raw=False, channel=None): + """Return points and optional inferred Fs for SETGUI's sampled CSV exports. + + Ordinary CSV: time in ms and volts; last row defines the period. + SETGUI export (time_ms,volts,dac12): uniform samples without endpoint. + """ + text = text.lstrip("\ufeff") + if signal_csv_channels(text): + return _parse_trend_csv(text, channel) + first = next((line for line in text.splitlines() if line.strip()), "") + delimiter = ";" if ";" in first else "\t" if "\t" in first else "," + rows = [(number, row) for number, row in enumerate( + csv.reader(io.StringIO(text), delimiter=delimiter), 1) if any(cell.strip() for cell in row)] + if not rows: + raise ValueError("CSV пуст") + def numeric(cell): + return float(cell.strip().replace(",", ".")) + header = [cell.strip().lower() for cell in rows[0][1]] + time_names = ("time_ms", "time", "время, мс", "время", "t") + volt_names = ("volts", "voltage", "напряжение, в", "напряжение", "v") + x_col = next((i for i, cell in enumerate(header) if cell in time_names), None) + y_col = next((i for i, cell in enumerate(header) if cell in volt_names), None) + sampled_export = x_col is not None and y_col is not None and "dac12" in header + if x_col is not None and y_col is not None: + rows = rows[1:] + else: + x_col, y_col = 0, 1 + if not 2 <= len(rows) <= 4096: + raise ValueError("CSV должен содержать от 2 до 4096 строк данных") + points = [] + for number, row in rows: + try: + x, y = numeric(row[x_col]), numeric(row[y_col]) + except (ValueError, IndexError) as error: + raise ValueError(f"Строка {number}: нужны числовые время в мс и напряжение в В") from error + if not math.isfinite(x) or not math.isfinite(y): + raise ValueError(f"Строка {number}: значения должны быть конечными") + if x < 0 or (points and x <= points[-1][0]): + raise ValueError(f"Строка {number}: время должно быть неотрицательным и строго возрастать") + points.append((x, y)) + if points[0][0] != 0: + raise ValueError("Первая временная метка должна быть 0 мс") + rate = None + if sampled_export: + step = points[1][0] + if step <= 0 or any(not math.isclose(x, i * step, rel_tol=1e-8, abs_tol=1e-9) + for i, (x, _) in enumerate(points)): + raise ValueError("В экспортированной таблице отсчёты должны идти с постоянным шагом") + rate = round(1000 / step) + if not 1 <= rate <= 1000000 or not math.isclose(rate * step, 1000, rel_tol=1e-8): + raise ValueError("Частота CSV должна быть целым числом от 1 до 1 000 000 отсчётов/с") + if raw: + return points, rate + points.append((len(points) * 1000 / rate, points[0][1])) + # A 4096-sample export needs an endpoint too. Remove only exactly redundant + # control points, preserving the sampled linear waveform without resampling. + if len(points) > 4096: + for i in range(1, len(points) - 1): + if math.isclose(points[i][1] * 2, points[i-1][1] + points[i+1][1], abs_tol=1e-12, rel_tol=0): + del points[i] + break + else: + raise ValueError("Для импорта этой таблицы с конечной точкой нужно более 4096 опорных точек") + return points, rate diff --git a/python/set_devices/qt_ports/plot_navigation.py b/python/set_devices/qt_ports/plot_navigation.py new file mode 100644 index 0000000..2c44f4c --- /dev/null +++ b/python/set_devices/qt_ports/plot_navigation.py @@ -0,0 +1,112 @@ +"""Common keyboard and wheel gestures for all plot canvases (Qt 5/6).""" +try: + from PySide6.QtCore import Qt + from PySide6.QtWidgets import QScrollArea +except ImportError: + from PySide2.QtCore import Qt + from PySide2.QtWidgets import QScrollArea + +NAVIGATION_HINT = ('Колесо и +/− — масштаб X; Ctrl + колесо и Ctrl + +/− — масштаб Y. ' + 'ЛКМ на свободном поле — перемещение; ←/→ — перемещение по X; Home/End — начало/конец.') + + +class PlotNavigation: + def setup_navigation(self): + self.setFocusPolicy(Qt.StrongFocus) + self.setToolTip(NAVIGATION_HINT) + + def keyPressEvent(self, event): + key = event.key() + modifiers = event.modifiers() + if modifiers & (Qt.AltModifier | Qt.MetaModifier): + return super().keyPressEvent(event) + if key in (Qt.Key_Plus, Qt.Key_Equal, Qt.Key_Minus): + self._navigation_zoom(1.25 if key != Qt.Key_Minus else .8, + bool(modifiers & Qt.ControlModifier)) + elif key in (Qt.Key_Left, Qt.Key_Right) and not (modifiers & Qt.ControlModifier): + self._navigation_pan(-.1 if key == Qt.Key_Left else .1) + elif key in (Qt.Key_Home, Qt.Key_End): + self._navigation_edge(key == Qt.Key_End) + else: + return super().keyPressEvent(event) + event.accept() + + def wheelEvent(self, event): + position = event.position() if hasattr(event, 'position') else event.posF() + rect = self.plot_rect() if hasattr(self, 'plot_rect') else self.rect() + if not rect.contains(position.toPoint()): + event.ignore() + return + delta = event.angleDelta() + ticks = delta.y() or delta.x() + if not ticks: + pixels = event.pixelDelta() + ticks = (pixels.y() or pixels.x()) * 2 + if ticks: + self.setFocus(Qt.MouseFocusReason) + self._navigation_zoom(1.25 ** max(-8, min(8, ticks / 120)), + bool(event.modifiers() & Qt.ControlModifier), position) + event.accept() + + +class ScrollPlotNavigation(PlotNavigation): + """Canvas navigation for waveforms whose viewport is a QScrollArea.""" + def scroll_area(self): + parent = self.parentWidget() + while parent is not None and not isinstance(parent, QScrollArea): + parent = parent.parentWidget() + return parent + + def _navigation_zoom(self, factor, vertical, position=None): + scroll = self.scroll_area() + size = self.height() if vertical else self.width() + if position is not None: + anchor = position.y() if vertical else position.x() + elif scroll is not None: + bar = scroll.verticalScrollBar() if vertical else scroll.horizontalScrollBar() + anchor = bar.value() + bar.pageStep() / 2 + else: + anchor = size / 2 + self.zoom_dragged.emit('y' if vertical else 'x', factor, anchor / max(1, size)) + + def _navigation_pan(self, fraction): + scroll = self.scroll_area() + if scroll is not None: + bar = scroll.horizontalScrollBar() + bar.setValue(round(bar.value() + fraction * bar.pageStep())) + self.navigation_panned() + + def _navigation_edge(self, end): + scroll = self.scroll_area() + if scroll is not None: + bar = scroll.horizontalScrollBar() + bar.setValue(bar.maximum() if end else bar.minimum()) + self.navigation_panned() + + def navigation_panned(self): + pass + + def begin_pan(self, event): + self.setFocus(Qt.MouseFocusReason) + self._pan_position = event.globalPosition() if hasattr(event, 'globalPosition') else event.globalPos() + self.setCursor(Qt.ClosedHandCursor) + + def move_pan(self, event): + previous = getattr(self, '_pan_position', None) + if previous is None: + return + position = event.globalPosition() if hasattr(event, 'globalPosition') else event.globalPos() + delta = position - previous + self._pan_position = position + scroll = self.scroll_area() + if scroll is not None: + for bar, amount in ((scroll.horizontalScrollBar(), delta.x()), (scroll.verticalScrollBar(), delta.y())): + bar.setValue(round(bar.value() - amount)) + if delta.x(): + self.navigation_panned() + + def end_pan(self, event): + self.move_pan(event) + self._pan_position = None + self.unsetCursor() + diff --git a/python/set_devices/qt_ports/plot_processing.py b/python/set_devices/qt_ports/plot_processing.py index 6cc83bf..4ab4faa 100644 --- a/python/set_devices/qt_ports/plot_processing.py +++ b/python/set_devices/qt_ports/plot_processing.py @@ -20,16 +20,20 @@ except ImportError: 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): + def __init__(self, signature, parent=None): super().__init__() self.signature = signature - self.signals = _ResultSignals() + # 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: @@ -49,10 +53,14 @@ class SignalProcessingPanel(QWidget): 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() @@ -81,11 +89,12 @@ class SignalProcessingPanel(QWidget): self.apply_button = QPushButton("Рассчитать") self.apply_button.clicked.connect(self.calculate) row.addWidget(self.apply_button) - self.clear_button = QPushButton("Убрать") + 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) @@ -97,9 +106,9 @@ class SignalProcessingPanel(QWidget): "PCHIP — сохранение формы без выбросов. Сплайн — гладкая кривая, возможны выбросы.\n" "Восстановление — оценка между измерениями; утраченные детали не определяются однозначно.") for widget in (self.channel, self.method): - widget.currentIndexChanged.connect(self.clear) + widget.currentIndexChanged.connect(self._parameters_changed) for widget in (self.degree, self.count): - widget.valueChanged.connect(self.clear) + widget.valueChanged.connect(self._parameters_changed) self._update_actions() def _signature(self): @@ -115,8 +124,7 @@ class SignalProcessingPanel(QWidget): self.count.value(), self.degree.value()) def set_snapshot(self, snapshot): - previous = ((self._result_signature or (self._job.signature if self._job is not None else None)) - if self.preserve_on_view_change else None) or self._signature() + 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())] @@ -133,9 +141,19 @@ class SignalProcessingPanel(QWidget): self._snapshot = snapshot if previous != self._current_request_for(previous): self._revision += 1 - if self._result is not None: - self._result = self._result_signature = None - self.status.setText("Данные или окно изменились — нажмите «Рассчитать» повторно.") + 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 + 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: @@ -155,11 +173,18 @@ class SignalProcessingPanel(QWidget): 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 calculate(self): if self._job is not None or self._snapshot.blocked_reason: return @@ -167,15 +192,17 @@ class SignalProcessingPanel(QWidget): if signature is None or len(signature.series.points) < 2: self.status.setText("Нужны минимум две точки аналогового канала.") return - self._job = _Calculation(signature) + self._job = _Calculation(signature, self) self._job_revision = self._revision - self._job.signals.done.connect(self._finished) + 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("Данные изменились во время расчёта. Остановите сбор и повторите расчёт.") @@ -183,19 +210,38 @@ class SignalProcessingPanel(QWidget): 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 [], {} - return [(self.RESULT_KEY, self._result.label, False, "#FF70D0", True)], {self.RESULT_KEY: self._result.points} + 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): @@ -204,7 +250,8 @@ class SignalProcessingPanel(QWidget): def export_csv(self): if self._result is None: return - path, _ = QFileDialog.getSaveFileName(self, "Сохранить расчётную кривую", "calculated-signal.csv", "CSV (*.csv)") + path, _ = QFileDialog.getSaveFileName(self, "Сохранить расчётную кривую", + self.export_path("calculated-signal.csv"), "CSV (*.csv)") if not path: return try: @@ -221,10 +268,11 @@ class PlotProcessingAttachment(QObject): 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): + 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 @@ -250,6 +298,12 @@ class PlotProcessingAttachment(QObject): 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 @@ -267,6 +321,8 @@ class PlotProcessingAttachment(QObject): 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) @@ -276,18 +332,22 @@ class PlotProcessingAttachment(QObject): self.dialog.activateWindow() def paint(self, painter, rect, project): - curve = self.curve - if curve is None: + curves = self.curves + if not curves: return painter.save() painter.setClipRect(rect) - path = QPainterPath() - for index, (x, y) in enumerate(curve.points): - point = project(x - (self._external_offset if self._external_curve is not None else 0), y, rect) - path.moveTo(point) if index == 0 else path.lineTo(point) - painter.setPen(QPen(QColor("#FF70D0"), 2, Qt.DashLine)) - painter.setBrush(Qt.NoBrush) - painter.drawPath(path) - painter.setPen(QColor("#FF70D0")) - painter.drawText(QPointF(rect.left() + 6, rect.bottom() - 8), curve.label) + for curve_index, curve in enumerate(curves): + path = QPainterPath() + for index, (x, y) in enumerate(curve.points): + point = project(x - (self._external_offset if self._external_curve is not None else 0), 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() diff --git a/python/tests/test_generator_data.py b/python/tests/test_generator_data.py new file mode 100644 index 0000000..1782f54 --- /dev/null +++ b/python/tests/test_generator_data.py @@ -0,0 +1,37 @@ +"""Waveform data helpers run independently of SETGUI and Qt.""" +import io +import json +import unittest +from set_devices.generator_data import ( + parse_signal_csv, signal_csv_channels, preset_points, generate, + load_recipe, resample_signal, nearest_f407_sample_rate, +) +from set_devices.waveform import recipe +from set_devices.plot_processing import Axis, Series, Snapshot, prepare, process, write_csv + + +class GeneratorDataTests(unittest.TestCase): + def test_processing_csv_roundtrip_into_generator(self): + request = prepare(Snapshot((Series('a', 'A', [(1000, 0), (1010, 2), (1020, 0)]),), + Axis('Время', 'мс', 'unix_ms')), 'a', 'linear', 21) + output = io.StringIO() + write_csv(process(request), output) + self.assertEqual(1, len(signal_csv_channels(output.getvalue()))) + points, rate = parse_signal_csv(output.getvalue()) + self.assertEqual((0, 0), points[0]) + self.assertEqual((20, 0), points[-1]) + self.assertIsNone(rate) + + def test_samples_recipe_and_resampling(self): + points = preset_points(0, 3.3) + samples = resample_signal(points, 10000) + self.assertEqual(200, len(samples)) + wave = generate(samples, 10000, method='samples') + restored, settings = load_recipe(json.dumps(recipe(wave.points, 10000, 3.3, 'samples', 2))) + self.assertEqual(wave.points, restored) + self.assertEqual('samples', settings['method']) + self.assertEqual(10000, nearest_f407_sample_rate(10001)) + + +if __name__ == '__main__': + unittest.main() diff --git a/python/tests/test_plot_processing_qt.py b/python/tests/test_plot_processing_qt.py index fac867e..040de12 100644 --- a/python/tests/test_plot_processing_qt.py +++ b/python/tests/test_plot_processing_qt.py @@ -36,14 +36,81 @@ class ProcessingQtTests(unittest.TestCase): self.assertIsNone(self.panel._job) self.assertIsNotNone(self.panel.curve, self.panel.status.text()) - def test_all_methods_are_available_and_parameter_changes_clear_result(self): - for method in ("polynomial", "linear", "pchip", "spline"): + def test_all_methods_accumulate_and_parameter_changes_keep_results(self): + for index, method in enumerate(("polynomial", "linear", "pchip", "spline")): self.panel.method.setCurrentIndex(self.panel.method.findData(method)) self.calculate() self.assertEqual(method, self.panel.curve.request.method) self.panel.count.setValue(self.panel.count.value() + 1) - self.assertIsNone(self.panel.curve) - self.assertFalse(self.panel.export_button.isEnabled()) + self.assertEqual(index + 1, len(self.panel.curves)) + self.assertTrue(self.panel.export_button.isEnabled()) + specs, data = self.panel.overlay() + self.assertEqual(4, len(specs)) + self.assertEqual(4, len(data)) + self.assertEqual(4, len({spec[3] for spec in specs})) + self.panel.clear() + self.assertEqual((), self.panel.curves) + self.assertIsNone(self.panel.curve) + + def test_repeated_request_does_not_duplicate_curve(self): + self.calculate() + self.calculate() + self.assertEqual(1, len(self.panel.curves)) + + def test_repeated_fft_workers_keep_gui_owned_signal_emitters(self): + self.panel.set_snapshot(replace(self.snapshot, axis=Axis("Frequency", "Hz"), source="FFT")) + for _ in range(5): + for method in ("linear", "pchip", "spline", "polynomial"): + self.panel.method.setCurrentIndex(self.panel.method.findData(method)) + self.panel.calculate() + self.assertIs(self.panel, self.panel._job.signals.parent()) + deadline = time.monotonic() + 5 + while self.panel._job is not None and time.monotonic() < deadline: + self.app.processEvents() + time.sleep(.002) + self.assertIsNone(self.panel._job) + self.assertEqual(method, self.panel.curve.request.method) + self.assertEqual(4, len(self.panel.curves)) + + def test_switching_channel_preserves_results_and_invalidates_only_changed_source(self): + snapshot = replace(self.snapshot, series=self.snapshot.series + + (Series("b", "B", [(0, 2), (1, 3), (2, 2)]),)) + self.panel.set_snapshot(snapshot) + self.calculate() + first = self.panel.curve + self.panel.channel.setCurrentIndex(1) + self.panel.set_snapshot(snapshot) + self.assertEqual((first,), self.panel.curves) + self.calculate() + second = self.panel.curve + self.assertEqual((first, second), self.panel.curves) + second_spec = self.panel.overlay()[0][1] + self.panel.set_snapshot(replace(snapshot, series=( + Series("a", "A", [(0, 5), (1, 6)]), snapshot.series[1]))) + self.assertEqual((second,), self.panel.curves) + self.assertEqual(second_spec, self.panel.overlay()[0][0]) + + def test_navigation_during_second_calculation_preserves_both_results(self): + self.panel.preserve_on_view_change = True + self.calculate() + first = self.panel.curve + self.panel.method.setCurrentIndex(self.panel.method.findData("linear")) + with patch("set_devices.qt_ports.plot_processing.QThreadPool"): + self.panel.calculate() + job = self.panel._job + self.panel.set_snapshot(replace(self.snapshot, x_range=(0, 1))) + self.panel._finished(job.signature, process(job.signature), "") + self.assertEqual(2, len(self.panel.curves)) + self.assertIs(first, self.panel.curves[0]) + + def test_failed_calculation_keeps_previous_curve(self): + self.calculate() + previous = self.panel.curve + self.panel.count.setValue(123) + with patch("set_devices.qt_ports.plot_processing.QThreadPool"): + self.panel.calculate() + self.panel._finished(self.panel._job.signature, None, "test error") + self.assertEqual((previous,), self.panel.curves) def test_source_units_and_blocking_invalidate_result(self): for changed in (replace(self.snapshot, axis=Axis("Frequency", "Hz")),