1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
"""
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, tree_to_grid, _tree_grid_size)
_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/rowspan/colspan for each pane
if cfg.layout_mode == "free" and cfg.tree is not None:
gs = _tree_grid_size(cfg.tree)
grid_pos: dict = {}
tree_to_grid(cfg.tree, 0, 0, gs, gs, grid_pos)
# grid_pos[pane_idx] = (row, col, rowspan, colspan)
elif cfg.layout_mode == "sidebyside":
grid_pos = {i: (0, i, 1, 1) for i in range(n)}
elif cfg.layout_mode == "grid":
cols = max(1, cfg.grid_cols)
grid_pos = {i: (i // cols, i % cols, 1, 1) for i in range(n)}
else: # stacked (legacy default)
grid_pos = {i: (i, 0, 1, 1) for i in range(n)}
ref_plot = None
for idx, spec in enumerate(cfg.panes):
if idx not in grid_pos:
continue
row, col, rowspan, colspan = grid_pos[idx]
plot = self._gw.addPlot(row=row, col=col, rowspan=rowspan, colspan=colspan)
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 non-bottom panes in stacked legacy mode
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")
self._gw.ci.layout.setColumnStretchFactor(col, spec.weight)
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))
def on_channel_enabled_changed(self, dev_id: str, ch_id: str, enabled: bool):
key = (dev_id, ch_id)
for e in self._curves.get(key, []):
e["curve"].setVisible(enabled)
@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")
|
