""" ui/strip_chart.py — Config-driven live chart. 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, List, Optional, Tuple from collections import deque from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QLabel, QDoubleSpinBox, QPushButton, QSizePolicy, ) from PyQt6.QtCore import Qt, pyqtSlot from PyQt6.QtGui import QFont try: import pyqtgraph as pg _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, processor: SignalProcessor): super().__init__() 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.apply_layout(build_default_layout(registry, processor)) def _build(self): lay = QVBoxLayout(self); lay.setContentsMargins(6,6,6,4); lay.setSpacing(4) ctrl = QHBoxLayout() ctrl.addWidget(QLabel("Window:")) 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) lay.addWidget(self._gw) else: 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) if cfg.link_x and ref_plot and spec.x_source == "time": plot.setXLink(ref_plot) 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, 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")