Andrey
41e24629ce
Добавить общее ядро DEIF, эмуляцию приборов и переносимые панели Python/Qt. Расширить загрузчики F2812 и F28335, обновление Flash и профиль STM Invertor. Добавить альтернативный МНК Eigen, модели неисправностей и диагностику сигналов. Обновить справочники устройств, проекты CCS и регрессионные проверки.
90 lines
6.0 KiB
Python
90 lines
6.0 KiB
Python
"""Debounced viewport statistics shared by desktop charts."""
|
|
import time
|
|
from threading import Event # Сигнал отмены позволяет фоновому расчёту прекращать обработку старого окна при продолжении масштабирования.
|
|
from dataclasses import replace
|
|
try:
|
|
from PySide6.QtCore import QObject, QTimer, QThreadPool, Qt
|
|
except ImportError:
|
|
from PySide2.QtCore import QObject, QTimer, QThreadPool, Qt
|
|
from .plot_workers import NavigationSignals as _NavigationSignals, NavigationJob as _NavigationJob
|
|
from set_devices.oscilloscope import time_label
|
|
def number(value):
|
|
return f"{value:g}"
|
|
|
|
class ViewportNavigation(QObject):
|
|
def __init__(self, chart):
|
|
super().__init__(chart)
|
|
self.chart = chart
|
|
self._navigation_revision = 0
|
|
self._navigation_busy = False
|
|
self._navigation_cancel = Event() # Отдельное потокобезопасное событие относится к текущему фоновому расчёту измерений.
|
|
self._navigation_emitter = _NavigationSignals(self)
|
|
self._navigation_emitter.ready.connect(self._navigation_ready, Qt.ConnectionType.QueuedConnection)
|
|
self._navigation_timer = QTimer(self)
|
|
self._navigation_timer.setSingleShot(True)
|
|
self._navigation_timer.setInterval(180)
|
|
self._navigation_timer.timeout.connect(self._navigation_finished)
|
|
|
|
def needs_worker(self): # Общая нагрузка всех каналов определяет допустимость синхронного расчёта, а не длина самого длинного ряда.
|
|
return sum(len(series.values) for series in self.chart.plot._series) > 20000 # Бюджет 20000 исходных значений сохраняет быстрый путь малых записей и отправляет многоканальные окна в фон.
|
|
|
|
def _navigation_changed(self):
|
|
self._navigation_revision += 1
|
|
self._navigation_cancel.set() # Сразу отменяем статистику предыдущего окна, чтобы она не конкурировала с новой отрисовкой миллионов отсчётов.
|
|
if not self.needs_worker(): # Проверяем суммарный объём тем же правилом, что используют панель измерений и легенда.
|
|
self.chart._scope_measurements()
|
|
self.chart._refresh_processing()
|
|
self.chart._refresh_legend()
|
|
return
|
|
self._navigation_timer.start()
|
|
|
|
def _navigation_finished(self):
|
|
if False:
|
|
return
|
|
if not self.needs_worker(): # После задержки повторно проверяем текущие данные: источник мог измениться во время ожидания.
|
|
self.chart._scope_measurements()
|
|
self.chart._refresh_processing()
|
|
return
|
|
if self._navigation_busy:
|
|
self._navigation_timer.start()
|
|
return
|
|
self._navigation_busy = True
|
|
self._navigation_cancel = Event() # После окончания перемещения создаём новый независимый сигнал отмены для актуального окна.
|
|
bounds = self.chart.plot.visible_bounds
|
|
panel = self.chart.processing
|
|
offset = self.chart._plot_time_offset if not self.chart.fft.isChecked() else 0.
|
|
# Navigation changes the next calculation's domain; retained curves keep
|
|
# their original domain. Avoid preparing millions of inputs on the UI.
|
|
panel._snapshot = replace(panel._snapshot, x_range=(bounds.left + offset, bounds.right + offset))
|
|
series = [(s.key, s.timestamps, s.values) for s in self.chart.plot._series if s.visible and not s.discrete] # Скрытые каналы не занимают процессор повторным расчётом статистики при перемещении графика.
|
|
QThreadPool.globalInstance().start(_NavigationJob(self._navigation_revision, series, bounds, # Фоновый расчёт использует полные исходные значения только актуального видимого окна.
|
|
self._navigation_emitter, self._navigation_cancel)) # Передаём событие отмены вместе с существующими ревизией и каналом результата.
|
|
|
|
def _navigation_ready(self, revision, stats, bounds):
|
|
self._navigation_busy = False
|
|
if False or revision != self._navigation_revision:
|
|
return
|
|
for row in range(self.chart.legend.rowCount()):
|
|
item = self.chart.legend.item(row, 0)
|
|
if item is None:
|
|
continue
|
|
result, _ = stats.get(item.data(Qt.ItemDataRole.UserRole), (None, 0))
|
|
for column, key in ((4, "rms"), (5, "maximum"), (6, "minimum")):
|
|
cell = self.chart.legend.item(row, column)
|
|
if cell is not None:
|
|
cell.setText("—" if result is None else "%g" % result[key])
|
|
scope = self.chart.scope
|
|
result, count = stats.get(scope.channel.currentData(), (None, 0))
|
|
if result is None:
|
|
scope.readings.setText("Измерения: —")
|
|
return
|
|
text = " · ".join(f"{title} {number(result[key])}" for title, key in (
|
|
("Min", "minimum"), ("Max", "maximum"), ("Размах", "peak_to_peak"),
|
|
("Среднее", "mean"), ("RMS", "rms")))
|
|
if not self.chart.fft.isChecked():
|
|
frequency = result["frequency"]
|
|
text += " · f ≈ " + (f"{number(frequency)} Гц" if frequency is not None else "—")
|
|
text += f" · {time_label((bounds.right - bounds.left) / 10)}/дел"
|
|
scope.readings.setText(text + f" · N={count}")
|
|
|