From d5acb04b88373d33b038bb59945fb5ab8b4f543b Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Mon, 20 Apr 2026 16:55:57 -0600 Subject: V8 --- ui/strip_chart.py | 329 +++++++++++++++++++++++++++++------------------------- 1 file changed, 180 insertions(+), 149 deletions(-) (limited to 'ui/strip_chart.py') diff --git a/ui/strip_chart.py b/ui/strip_chart.py index afc7cf7..7bc2a18 100644 --- a/ui/strip_chart.py +++ b/ui/strip_chart.py @@ -1,186 +1,217 @@ """ -ui/strip_chart.py +ui/strip_chart.py — Config-driven live chart. -Live scrolling strip chart. One pyqtgraph plot per device, -all time-axes linked. Per-channel colored traces with legend. +Consumes LayoutConfig from plot_window. +Supports stacked, side-by-side, and grid layouts. +X axis can be time or any channel (e.g. stress vs strain). """ import numpy as np -from typing import Dict +from typing import Dict, List, Optional, Tuple +from collections import deque from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, - QDoubleSpinBox, QPushButton, QSizePolicy, QComboBox, + QDoubleSpinBox, QPushButton, QSizePolicy, ) from PyQt6.QtCore import Qt, pyqtSlot +from PyQt6.QtGui import QFont try: import pyqtgraph as pg - pg.setConfigOptions(antialias=True, background="#0b0e13", foreground="#475569") _HAS_PG = True except ImportError: _HAS_PG = False +_THEME_DARK = {"bg": "#0b0e13", "fg": "#475569", "grid_alpha": 0.12, "legend_brush": "#161d2e", "legend_pen": "#2a3558", "tick_color": "#64748b"} +_THEME_LIGHT = {"bg": "#ffffff", "fg": "#334155", "grid_alpha": 0.18, "legend_brush": "#f8fafc", "legend_pen": "#cbd5e1", "tick_color": "#64748b"} +_current_theme = _THEME_DARK + +def set_chart_theme(theme: str): + global _current_theme + _current_theme = _THEME_DARK if theme == "dark" else _THEME_LIGHT + if _HAS_PG: + pg.setConfigOptions(antialias=True, + background=_current_theme["bg"], + foreground=_current_theme["fg"]) + +# Apply dark theme as default +if _HAS_PG: + pg.setConfigOptions(antialias=True, background=_THEME_DARK["bg"], foreground=_THEME_DARK["fg"]) + from devices.device_registry import DeviceRegistry from core.acquisition import AcquisitionEngine +from core.signal_processor import SignalProcessor +from ui.windows.plot_window import LayoutConfig, PaneSpec, TraceSpec, build_default_layout + +_STYLES = { + "solid": Qt.PenStyle.SolidLine, + "dash": Qt.PenStyle.DashLine, + "dot": Qt.PenStyle.DotLine, +} class StripChartWidget(QWidget): - def __init__(self, engine: AcquisitionEngine, registry: DeviceRegistry): + def __init__(self, engine: AcquisitionEngine, registry: DeviceRegistry, + processor: SignalProcessor): super().__init__() - self.engine = engine - self.registry = registry - self._window = 30.0 - self._paused = False - self._plots: Dict[str, Dict] = {} # dev_id -> {ch_id -> {curve, plot}} + self.engine = engine + self.registry = registry + self.processor = processor + self._paused = False + self._cfg: Optional[LayoutConfig] = None + # (dev_id, ch_id) → [{curve, plot, spec}] + self._curves: Dict[Tuple, List[dict]] = {} self._build() - self.refresh() - - # ── Build ──────────────────────────────────────────────────────────── + self.apply_layout(build_default_layout(registry, processor)) def _build(self): - layout = QVBoxLayout(self) - layout.setContentsMargins(6, 6, 6, 4) - layout.setSpacing(4) - - # Control bar + lay = QVBoxLayout(self); lay.setContentsMargins(6,6,6,4); lay.setSpacing(4) ctrl = QHBoxLayout() ctrl.addWidget(QLabel("Window:")) - - self._win_spin = QDoubleSpinBox() - self._win_spin.setRange(1.0, 3600.0) - self._win_spin.setValue(self._window) - self._win_spin.setSuffix(" s") - self._win_spin.valueChanged.connect(self._on_window_changed) - ctrl.addWidget(self._win_spin) - ctrl.addSpacing(16) - - ctrl.addWidget(QLabel("Y-Scale:")) - self._scale_cb = QComboBox() - self._scale_cb.addItems(["Auto", "Fixed"]) - self._scale_cb.currentTextChanged.connect(self._on_scale_changed) - ctrl.addWidget(self._scale_cb) - - ctrl.addStretch() - - self._pause_btn = QPushButton("⏸ Pause") - self._pause_btn.setCheckable(True) - self._pause_btn.setObjectName("pauseButton") - self._pause_btn.toggled.connect(self._on_pause) - ctrl.addWidget(self._pause_btn) - - layout.addLayout(ctrl) - + self._win = QDoubleSpinBox(); self._win.setRange(1,3600) + self._win.setSuffix(" s"); self._win.setValue(30.0) + self._win.valueChanged.connect(lambda v: self._cfg and setattr(self._cfg,"time_window_s",v)) + ctrl.addWidget(self._win); ctrl.addStretch() + self._pause = QPushButton("⏸ Pause"); self._pause.setCheckable(True) + self._pause.setObjectName("pauseButton"); self._pause.toggled.connect(self._on_pause) + ctrl.addWidget(self._pause); lay.addLayout(ctrl) if _HAS_PG: self._gw = pg.GraphicsLayoutWidget() - self._gw.setSizePolicy( - QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding - ) - layout.addWidget(self._gw) + self._gw.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) + lay.addWidget(self._gw) else: - layout.addWidget(QLabel( - "⚠ pyqtgraph not installed.\n\npip install pyqtgraph\n\nData is still acquired & logged.", - alignment=Qt.AlignmentFlag.AlignCenter, - )) - - # ── Refresh (rebuild plots after device list changes) ───────────────── - - def refresh(self): - if not _HAS_PG: - return - self._gw.clear() - self._plots.clear() - - devs = self.registry.all_instances() - n_devs = len(devs) - if n_devs == 0: - return - - ref_plot = None # for x-axis linking - - for row, dev in enumerate(devs): - plot = self._gw.addPlot(row=row, col=0) - plot.setLabel("left", f"{dev.info.icon} {dev.info.name}") - plot.showGrid(x=True, y=True, alpha=0.12) - plot.getAxis("left").setStyle(tickFont=self._mono_font()) - plot.getAxis("bottom").setStyle(tickFont=self._mono_font()) - - if row < n_devs - 1: + lay.addWidget(QLabel("⚠ pip install pyqtgraph", + alignment=Qt.AlignmentFlag.AlignCenter)) + + def apply_layout(self, cfg: LayoutConfig): + self._cfg = cfg + self._win.setValue(cfg.time_window_s) + if not _HAS_PG: return + self._gw.clear(); self._curves.clear() + if not cfg.panes: return + + mf = QFont("IBM Plex Mono", 8) + n = len(cfg.panes) + + # Determine row/col for each pane + if cfg.layout_mode == "sidebyside": + positions = [(0, c) for c in range(n)] + elif cfg.layout_mode == "grid": + cols = max(1, cfg.grid_cols) + positions = [(i // cols, i % cols) for i in range(n)] + else: # stacked + positions = [(r, 0) for r in range(n)] + + ref_plot = None + for idx, (spec, (row, col)) in enumerate(zip(cfg.panes, positions)): + plot = self._gw.addPlot(row=row, col=col) + plot.setLabel("left", spec.y_label or spec.title) + plot.setLabel("bottom", spec.x_label or "Elapsed (s)") + plot.showGrid(x=spec.grid, y=spec.grid, alpha=0.12) + plot.getAxis("left").setStyle(tickFont=mf) + plot.getAxis("bottom").setStyle(tickFont=mf) + + # Hide bottom axis labels for all but bottom row in stacked + if cfg.layout_mode == "stacked" and idx < n - 1: plot.getAxis("bottom").setStyle(showValues=False) plot.getAxis("bottom").setHeight(0) - else: - plot.setLabel("bottom", "Elapsed (s)") - if ref_plot is not None: + if cfg.link_x and ref_plot and spec.x_source == "time": plot.setXLink(ref_plot) - ref_plot = plot - - legend = plot.addLegend( - offset=(5, 5), - labelTextColor="#94a3b8", - brush=pg.mkBrush("#161d2e"), - pen=pg.mkPen("#2a3558"), - ) - - self._plots[dev.info.device_id] = {} - for ch in dev.info.channels: - if not ch.enabled: - continue - pen = pg.mkPen(color=ch.color, width=1.8) - curve = plot.plot([], [], pen=pen, name=ch.name) - self._plots[dev.info.device_id][ch.channel_id] = { - "curve": curve, - "plot": plot, - } - - @staticmethod - def _mono_font(): - from PyQt6.QtGui import QFont - f = QFont("IBM Plex Mono", 8) - return f - - # ── Slots ──────────────────────────────────────────────────────────── + if spec.x_source == "time": + ref_plot = plot + + if cfg.show_legend: + t = _current_theme + plot.addLegend(offset=(5,5), labelTextColor=t["tick_color"], + brush=pg.mkBrush(t["legend_brush"]), + pen=pg.mkPen(t["legend_pen"])) + + if not spec.y_auto: + plot.setYRange(spec.y_min, spec.y_max) + else: + plot.enableAutoRange(axis="y") + + # Set column stretch (relative weight) + self._gw.ci.layout.setColumnStretchFactor(col, spec.weight) + if cfg.layout_mode == "stacked": + self._gw.ci.layout.setRowStretchFactor(row, spec.weight) + + for tr in spec.traces: + if not tr.visible: continue + pen = pg.mkPen(color=tr.color, width=tr.width, + style=_STYLES.get(tr.style, Qt.PenStyle.SolidLine)) + curve = plot.plot([], [], pen=pen, name=tr.label or tr.channel_id) + key = (tr.device_id, tr.channel_id) + self._curves.setdefault(key, []).append( + {"curve": curve, "plot": plot, "spec": spec, "pane": spec}) + + def set_theme(self, theme: str): + """Hot-swap pyqtgraph colours when the user changes theme.""" + set_chart_theme(theme) + if _HAS_PG and self._gw is not None: + self._gw.setBackground(_current_theme["bg"]) + # Rebuild plots so legend/grid colours update + self.apply_layout(self._cfg or build_default_layout(self.registry, self.processor)) + + def refresh(self): + self.apply_layout(self._cfg or build_default_layout(self.registry, self.processor)) @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) - if dev_plots is None: - return - 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.window(self._window) - if not ts: - return - ts_arr = np.array(ts, dtype=np.float64) - vs_arr = np.array(vs, dtype=np.float64) - - entry["curve"].setData(ts_arr, vs_arr) - t_max = ts_arr[-1] - t_min = t_max - self._window - entry["plot"].setXRange(t_min, t_max, padding=0) - - if self._scale_cb.currentText() == "Auto": - entry["plot"].enableAutoRange(axis="y") - - def _on_window_changed(self, v: float): - self._window = v - - def _on_pause(self, checked: bool): - self._paused = checked - self._pause_btn.setText("▶ Resume" if checked else "⏸ Pause") - - def _on_scale_changed(self, text: str): - if not _HAS_PG: - return - if text == "Auto": - for dev_plots in self._plots.values(): - for entry in dev_plots.values(): - entry["plot"].enableAutoRange(axis="y") + def on_new_data(self, device_id: str, channel_id: str, ts: float, val: float): + if self._paused or not _HAS_PG or not self._cfg: return + key = (device_id, channel_id) + entries = self._curves.get(key) + if not entries: return + window = self._cfg.time_window_s + + # Get Y data + if device_id == "derived": + bufs = self.processor.get_derived_buffer(channel_id) + if bufs is None or len(bufs[0]) < 2: return + ts_l = list(bufs[0]); vs_l = list(bufs[1]) + else: + buf = self.engine.get_buffer(device_id, channel_id) + if buf is None or len(buf) < 2: return + ts_l, vs_l = buf.window(window) + if not ts_l: return + + for e in entries: + spec = e["pane"] + # Determine X axis data + if spec.x_source == "time": + cutoff = ts_l[-1] - window if ts_l else 0 + idx = next((i for i,t in enumerate(ts_l) if t >= cutoff), 0) + x_arr = np.array(ts_l[idx:], dtype=np.float64) + y_arr = np.array(vs_l[idx:], dtype=np.float64) + e["curve"].setData(x_arr, y_arr) + if x_arr.size: + e["plot"].setXRange(x_arr[-1]-window, x_arr[-1], padding=0) + else: + # X = another channel's processed values + parts = spec.x_source.split("/") + if len(parts) == 2: + x_dev, x_ch = parts + if x_dev == "derived": + xbufs = self.processor.get_derived_buffer(x_ch) + if xbufs and len(xbufs[0]) >= 2: + x_arr = np.array(list(xbufs[1]), dtype=np.float64) + y_arr = np.array(vs_l, dtype=np.float64) + mn = min(len(x_arr), len(y_arr)) + e["curve"].setData(x_arr[-mn:], y_arr[-mn:]) + else: + xbuf = self.engine.get_buffer(x_dev, x_ch) + if xbuf and len(xbuf) >= 2: + _, xvs = xbuf.all() + x_arr = np.array(xvs, dtype=np.float64) + y_arr = np.array(vs_l, dtype=np.float64) + mn = min(len(x_arr), len(y_arr)) + e["curve"].setData(x_arr[-mn:], y_arr[-mn:]) + + if spec.y_auto: + e["plot"].enableAutoRange(axis="y") + + def _on_pause(self, c: bool): + self._paused = c + self._pause.setText("▶ Resume" if c else "⏸ Pause") -- cgit v1.2.3