diff options
| author | Christian Kolset <christian.kolset@gmail.com> | 2026-04-13 15:22:56 -0600 |
|---|---|---|
| committer | Christian Kolset <christian.kolset@gmail.com> | 2026-04-13 15:22:56 -0600 |
| commit | 2ed9d37da7b27d25173535550fb92702225ac14e (patch) | |
| tree | d6ff360a656e577629d3cefdaa1ae02af7255864 /daq_system/ui/strip_chart.py | |
| parent | 8d6acf3a8ea4b37f86b321dbf430be5be01b1267 (diff) | |
Removed QtPy5 snippet and fixed directories
Diffstat (limited to 'daq_system/ui/strip_chart.py')
| -rw-r--r-- | daq_system/ui/strip_chart.py | 140 |
1 files changed, 0 insertions, 140 deletions
diff --git a/daq_system/ui/strip_chart.py b/daq_system/ui/strip_chart.py deleted file mode 100644 index 46d73fa..0000000 --- a/daq_system/ui/strip_chart.py +++ /dev/null @@ -1,140 +0,0 @@ -""" -ui/strip_chart.py - -Live scrolling strip chart using pyqtgraph. -Each channel gets its own colored trace. Window duration is adjustable. -""" - -from typing import Dict, List -import numpy as np - -from PyQt6.QtWidgets import ( - QWidget, QVBoxLayout, QHBoxLayout, QLabel, - QComboBox, QDoubleSpinBox, QCheckBox, QScrollArea, - QPushButton, QSizePolicy -) -from PyQt6.QtCore import Qt, pyqtSlot - -try: - import pyqtgraph as pg - pg.setConfigOptions(antialias=True, background="#0d1117", foreground="#c9d1d9") - HAS_PG = True -except ImportError: - HAS_PG = False - -from devices.device_registry import DeviceRegistry -from core.acquisition import AcquisitionEngine - - -class StripChartWidget(QWidget): - def __init__(self, engine: AcquisitionEngine, registry: DeviceRegistry): - super().__init__() - self.engine = engine - self.registry = registry - self._plots: Dict[str, Dict] = {} # device_id -> {channel_id -> curve} - self._window_s = 30.0 - self._paused = False - self._build_ui() - self.refresh() - - def _build_ui(self): - layout = QVBoxLayout(self) - layout.setContentsMargins(4, 4, 4, 4) - layout.setSpacing(4) - - # Control bar - ctrl = QHBoxLayout() - ctrl.addWidget(QLabel("Window:")) - - self.window_spin = QDoubleSpinBox() - self.window_spin.setRange(1, 600) - self.window_spin.setValue(self._window_s) - self.window_spin.setSuffix(" s") - self.window_spin.valueChanged.connect(lambda v: setattr(self, "_window_s", v)) - ctrl.addWidget(self.window_spin) - - ctrl.addStretch() - - self.pause_btn = QPushButton("⏸ Pause") - self.pause_btn.setCheckable(True) - self.pause_btn.setObjectName("pauseButton") - self.pause_btn.toggled.connect(lambda c: setattr(self, "_paused", c)) - self.pause_btn.toggled.connect(lambda c: self.pause_btn.setText("▶ Resume" if c else "⏸ Pause")) - ctrl.addWidget(self.pause_btn) - - layout.addLayout(ctrl) - - if HAS_PG: - self._build_pg_chart(layout) - else: - layout.addWidget(QLabel( - "pyqtgraph not installed.\nRun: pip install pyqtgraph\n\nData is still being acquired.", - alignment=Qt.AlignmentFlag.AlignCenter - )) - - def _build_pg_chart(self, parent_layout): - self.plot_widget = pg.GraphicsLayoutWidget() - self.plot_widget.setSizePolicy( - QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding - ) - parent_layout.addWidget(self.plot_widget) - - def refresh(self): - if not HAS_PG: - return - self.plot_widget.clear() - self._plots.clear() - - devices = self.registry.all_instances() - num = len(devices) - if num == 0: - return - - for row_idx, dev in enumerate(devices): - plot = self.plot_widget.addPlot(row=row_idx, col=0) - plot.setLabel("left", f"{dev.info.icon} {dev.info.name}") - plot.showGrid(x=True, y=True, alpha=0.15) - plot.getAxis("bottom").setStyle(showValues=(row_idx == num - 1)) - - if row_idx < num - 1: - plot.getAxis("bottom").setHeight(0) - else: - plot.setLabel("bottom", "Time (s)") - - self._plots[dev.info.device_id] = {} - for ch in dev.info.channels: - if not ch.enabled: - continue - color = ch.color - pen = pg.mkPen(color=color, width=1.5) - curve = plot.plot([], [], pen=pen, name=ch.name) - self._plots[dev.info.device_id][ch.channel_id] = { - "curve": curve, - "plot": plot, - } - - @pyqtSlot(str, str, float, float) - def on_new_data(self, device_id: str, channel_id: str, timestamp: float, value: float): - if self._paused or not HAS_PG: - return - dev_plots = self._plots.get(device_id, {}) - entry = dev_plots.get(channel_id) - if entry is None: - return - - buf = self.engine.get_buffer(device_id, channel_id) - if buf is None or len(buf) < 2: - return - - ts, vs = buf.all() - ts = np.array(ts, dtype=np.float64) - vs = np.array(vs, dtype=np.float64) - - t_max = ts[-1] - t_min = t_max - self._window_s - mask = ts >= t_min - ts = ts[mask] - vs = vs[mask] - - entry["curve"].setData(ts, vs) - entry["plot"].setXRange(t_min, t_max, padding=0) |
