406 lines
16 KiB
Python
406 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
轻量滚动曲线控件(QPainter 绘制, 不引入 matplotlib/pyqtgraph)
|
||
================================================================
|
||
N 通道时间序列: add(ch, value) 追加点(内部取相对时间), deque 限长滚动丢弃;
|
||
y 轴自动缩放(全平数据回落 ±1)或 set_yrange 固定量程; 颜色/文字取自当前
|
||
palette, 明暗主题自动跟随。每通道 >1000 点时抽稀绘制, 3000 点 × 4 通道
|
||
重绘开销可控。
|
||
"""
|
||
|
||
import bisect
|
||
import time
|
||
from collections import deque
|
||
|
||
from PySide6.QtCore import QPointF, QRect, Qt
|
||
from PySide6.QtGui import QColor, QFont, QPainter, QPainterPath, QPen, QPalette
|
||
from PySide6.QtWidgets import QSizePolicy, QWidget
|
||
|
||
|
||
def fmt_val(v):
|
||
"""数值可读显示: 普通小数带千分位(最多 4 位小数, 去尾零), 达 10 亿量级
|
||
才退回科学计数 —— 调试现场 2.821e+07 这类写法不直观"""
|
||
if v != v or v in (float('inf'), float('-inf')):
|
||
return str(v)
|
||
if abs(v) >= 1e9:
|
||
return f'{v:.4g}'
|
||
s = f'{v:,.4f}' if abs(v) < 1 else f'{v:,.3f}'
|
||
return s.rstrip('0').rstrip('.') or '0'
|
||
|
||
|
||
class PlotWidget(QWidget):
|
||
|
||
COLORS = ('#0f6cbd', '#d83b01', '#107c10', '#8764b8')
|
||
MAXLEN = 3000 # 每通道缓冲点数(超限滚动丢弃)
|
||
M_LEFT, M_TOP, M_RIGHT, M_BOTTOM = 52, 10, 12, 22 # 绘图区内边距
|
||
|
||
def __init__(self, channels=4, parent=None):
|
||
super().__init__(parent)
|
||
n = channels
|
||
self._times = [deque(maxlen=self.MAXLEN) for _ in range(n)]
|
||
self._vals = [deque(maxlen=self.MAXLEN) for _ in range(n)]
|
||
self._names = [''] * n
|
||
self._latest = [None] * n
|
||
self._paused = False
|
||
self._yrange = None # None=自动 或 (min, max)
|
||
self._span = None # 可见时间窗宽(秒); None=跟随全部
|
||
self._offset = 0.0 # 右端距最新数据的秒数(0=实时)
|
||
self._frozen = False # 暂停渲染: 继续收数但不重绘
|
||
self._drag_y = None # Y 轴拖动基准(仅手动量程)
|
||
self.on_yrange_changed = None # 拖动改 Y 量程回调(主界面同步控件)
|
||
self.on_span_changed = None # 滚轮改时间窗回调(主界面同步下拉)
|
||
self._hover_pos = None # 悬停光标位置(None=不在绘图区)
|
||
self._hover_lines = [] # 悬停读值(顶部小框内容)
|
||
self._t0 = time.perf_counter()
|
||
self.setMinimumHeight(150)
|
||
self.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
|
||
self.setMouseTracking(True) # 无按键也接收 mouseMove(悬停读值)
|
||
|
||
# --- 数据接口(主线程调用) ---
|
||
|
||
def set_channel_name(self, ch, name):
|
||
self._names[ch] = name
|
||
self._touch()
|
||
|
||
def set_yrange(self, yr):
|
||
self._yrange = yr
|
||
self._touch()
|
||
|
||
def set_view(self, span):
|
||
"""设置可见时间窗宽(秒); None=跟随全部数据"""
|
||
self._span = None if span is None else max(float(span), 1.0)
|
||
self._touch()
|
||
|
||
def set_offset(self, offset):
|
||
"""可见窗右端距最新数据的秒数(0=实时跟随最新)"""
|
||
self._offset = max(float(offset), 0.0)
|
||
self._touch()
|
||
|
||
def view_offset(self):
|
||
return self._offset
|
||
|
||
def view_span(self):
|
||
return self._span
|
||
|
||
def view_range(self):
|
||
"""(全部数据 tmin, tmax, 可回看的最大滞后秒数) —— 供时间滑条换算"""
|
||
t_all = [t for ts in self._times if ts for t in (ts[0], ts[-1])]
|
||
tmin = min(t_all) if t_all else 0.0
|
||
tmax = max(t_all) if t_all else tmin + 5.0
|
||
span = self._span or max(tmax - tmin, 5.0)
|
||
return tmin, tmax, max(tmax - tmin - span, 0.0)
|
||
|
||
def visible_domain(self, tmin, tmax):
|
||
"""按时间窗/滞后换算可见横轴域(纯计算, 供单测)。
|
||
不做数据边界收敛: 选 5min 窗而数据只有 23s 时, 显示完整的 5min 轴,
|
||
数据贴右端随采集逐渐填满(示波器式); 滞后超出全部历史则为空窗"""
|
||
if self._span is None:
|
||
return tmin, tmax
|
||
right = tmax - self._offset
|
||
left = right - self._span
|
||
if right <= left: # 滞后超出全部历史 -> 空窗保宽
|
||
right = left + self._span
|
||
return left, right
|
||
|
||
def auto_yrange(self):
|
||
"""自动 Y 量程: 可见数据的 min/max 加 8% 边距(全平数据回落 ±1)"""
|
||
visible = [v for ch in range(len(self._vals)) for v in self._vals[ch]]
|
||
if not visible:
|
||
return -1.0, 1.0
|
||
ylo, yhi = min(visible), max(visible)
|
||
if ylo == yhi:
|
||
return ylo - 1.0, yhi + 1.0
|
||
pad = (yhi - ylo) * 0.08
|
||
return ylo - pad, yhi + pad
|
||
|
||
def set_paused(self, paused):
|
||
self._paused = bool(paused)
|
||
|
||
def is_paused(self):
|
||
return self._paused
|
||
|
||
def set_frozen(self, frozen):
|
||
"""暂停渲染(弹窗用): 继续接收入库但不重绘; 解冻后一次性渲染全部积累。
|
||
冻结是绝对状态: 覆写 update() 吞掉一切重绘请求, 任何代码路径都无法
|
||
在冻结期打破(悬停/滑条等), 只有 set_frozen(False) 恢复"""
|
||
self._frozen = bool(frozen)
|
||
if not self._frozen:
|
||
super().update()
|
||
|
||
def is_frozen(self):
|
||
return self._frozen
|
||
|
||
def update(self):
|
||
if self._frozen:
|
||
return
|
||
super().update()
|
||
|
||
def _touch(self):
|
||
self.update()
|
||
|
||
def add(self, ch, value):
|
||
if self._paused or value is None:
|
||
return
|
||
self._times[ch].append(time.perf_counter() - self._t0)
|
||
self._vals[ch].append(float(value))
|
||
self._latest[ch] = float(value)
|
||
self._touch()
|
||
|
||
def clear(self):
|
||
for t in self._times:
|
||
t.clear()
|
||
for v in self._vals:
|
||
v.clear()
|
||
self._latest = [None] * len(self._latest)
|
||
self._touch()
|
||
|
||
# --- 悬停读值 ---
|
||
|
||
def _nearest(self, ch, t, tol):
|
||
"""通道 ch 在时刻 t 附近(±tol 秒)最近的采样点 -> (名称, 值) 或 None"""
|
||
ts, vs = self._times[ch], self._vals[ch]
|
||
if not ts:
|
||
return None
|
||
lst = list(ts) # deque 无随机 bisect, 拷贝代价可接受
|
||
i = bisect.bisect_left(lst, t)
|
||
best = None
|
||
for j in (i - 1, i):
|
||
if 0 <= j < len(lst) and abs(lst[j] - t) <= tol:
|
||
if best is None or abs(lst[j] - t) < abs(lst[best] - t):
|
||
best = j
|
||
if best is None:
|
||
return None
|
||
return self._names[ch] or f'CH{ch + 1}', vs[best]
|
||
|
||
def _plot_geometry(self):
|
||
"""绘图区几何与可见横轴域: (left, top, pw, ph, tmin, tmax)"""
|
||
left, top = self.M_LEFT, self.M_TOP
|
||
pw = self.width() - left - self.M_RIGHT
|
||
ph = self.height() - top - self.M_BOTTOM
|
||
t_all = [t for ts in self._times if ts for t in (ts[0], ts[-1])]
|
||
if t_all:
|
||
tmin, tmax = min(t_all), max(t_all)
|
||
else:
|
||
tmin, tmax = 0.0, 5.0
|
||
tmin, tmax = self.visible_domain(tmin, tmax)
|
||
return left, top, pw, ph, tmin, tmax
|
||
|
||
def wheelEvent(self, ev):
|
||
"""滚轮缩放时间窗: 上滚放大(窗宽变短), 下滚缩小; 滞后按比例缩放"""
|
||
delta = ev.angleDelta().y()
|
||
if delta:
|
||
factor = 0.8 if delta > 0 else 1.25
|
||
old = self._span
|
||
if old is None: # 跟随模式 -> 以当前数据跨度为基准
|
||
_tmin, tmax, _mp = self.view_range()
|
||
old = max(tmax - _tmin, 5.0)
|
||
new = max(min(old * factor, 900.0), 2.0)
|
||
if self._offset and old:
|
||
self._offset = self._offset * new / old
|
||
self.set_view(new)
|
||
if self.on_span_changed is not None:
|
||
self.on_span_changed()
|
||
super().wheelEvent(ev)
|
||
|
||
def mousePressEvent(self, ev):
|
||
if ev.button() == Qt.LeftButton and self._yrange is not None:
|
||
self._drag_y = ev.position().toPoint().y()
|
||
super().mousePressEvent(ev)
|
||
|
||
def mouseReleaseEvent(self, ev):
|
||
if ev.button() == Qt.LeftButton:
|
||
self._drag_y = None
|
||
super().mouseReleaseEvent(ev)
|
||
|
||
def mouseMoveEvent(self, ev):
|
||
pos = ev.position().toPoint()
|
||
if self._drag_y is not None and self._yrange is not None:
|
||
_left, _top, _pw, ph, *_rest = self._plot_geometry()
|
||
if ph > 40:
|
||
ylo, yhi = self._yrange
|
||
shift = (yhi - ylo) * (pos.y() - self._drag_y) / ph
|
||
yr = (ylo - shift, yhi - shift) # 下拖数据下移, 上拖上移
|
||
self.set_yrange(yr)
|
||
if self.on_yrange_changed is not None:
|
||
self.on_yrange_changed(yr)
|
||
self._drag_y = pos.y()
|
||
self._hover_pos = pos
|
||
self.update()
|
||
super().mouseMoveEvent(ev)
|
||
|
||
def leaveEvent(self, ev):
|
||
self._hover_pos = None
|
||
self.update()
|
||
super().leaveEvent(ev)
|
||
|
||
def _hover_lines_at(self, pos):
|
||
"""光标位置 -> 读值框内容; 不在绘图区/无命中返回空列表。
|
||
由 paintEvent 每帧调用(而非仅在鼠标移动时), 鼠标停住不动,
|
||
实时数据到来时读值依然持续刷新(用户实测反馈)"""
|
||
left, top, pw, ph, tmin, tmax = self._plot_geometry()
|
||
span = max(tmax - tmin, 0.001)
|
||
if pw < 40 or ph < 40 or not (left <= pos.x() <= left + pw) \
|
||
or not (top <= pos.y() <= top + ph):
|
||
return []
|
||
t = tmin + (pos.x() - left) / pw * span
|
||
tol = span / max(pw, 1) * 10 # 光标 ±10px 内的采样算命中
|
||
lines = [f'T = {t:.2f}s']
|
||
for ch in range(len(self._vals)):
|
||
got = self._nearest(ch, t, tol)
|
||
if got:
|
||
lines.append(f'{got[0]} = {fmt_val(got[1])}')
|
||
return lines if len(lines) > 1 else []
|
||
|
||
# --- 绘制 ---
|
||
|
||
def paintEvent(self, _ev):
|
||
p = QPainter(self)
|
||
p.setRenderHint(QPainter.Antialiasing)
|
||
rect = self.rect()
|
||
pal = self.palette()
|
||
p.fillRect(rect, pal.color(QPalette.Base))
|
||
text = pal.color(QPalette.WindowText)
|
||
dim = QColor(text)
|
||
dim.setAlpha(110)
|
||
grid = QColor(text)
|
||
grid.setAlpha(38)
|
||
|
||
left, top = self.M_LEFT, self.M_TOP
|
||
right, bottom = self.M_RIGHT, self.M_BOTTOM
|
||
pw, ph = rect.width() - left - right, rect.height() - top - bottom
|
||
if pw < 40 or ph < 40:
|
||
return
|
||
|
||
p.setFont(QFont(self.font().family(), 8))
|
||
p.setPen(dim)
|
||
|
||
# y 量程: 固定 或 自动(auto_yrange: 8% 边距, 全平数据回落 ±1)
|
||
if self._yrange is not None:
|
||
ylo, yhi = self._yrange
|
||
if ylo >= yhi:
|
||
yhi = ylo + 1.0
|
||
else:
|
||
ylo, yhi = self.auto_yrange()
|
||
|
||
def my(y): # 值 -> 像素 y
|
||
return top + (yhi - y) / (yhi - ylo) * ph
|
||
|
||
# 横向参考线(4 等分)与每条线上的纵轴数值刻度
|
||
p.setPen(QPen(grid, 1, Qt.DashLine))
|
||
for i in range(1, 4):
|
||
y = top + ph * i / 4
|
||
p.drawLine(left, int(y), left + pw, int(y))
|
||
for i in range(1, 6):
|
||
x = left + pw * i / 6
|
||
p.drawLine(int(x), top, int(x), top + ph)
|
||
p.setPen(dim)
|
||
for i in range(5):
|
||
v = yhi - (yhi - ylo) * i / 4
|
||
y = top + ph * i / 4
|
||
p.drawText(QRect(2, int(y) - 9, left - 8, 18),
|
||
Qt.AlignRight | Qt.AlignVCenter, fmt_val(v))
|
||
|
||
# x 范围: 时间窗(span)内、按滞后(offset)取可见域; 无窗宽则跟随全部数据
|
||
t_all = [t for ch in range(len(self._times)) if self._times[ch]
|
||
for t in (self._times[ch][0], self._times[ch][-1])]
|
||
if t_all:
|
||
tmin, tmax = min(t_all), max(t_all)
|
||
else:
|
||
tmin, tmax = 0.0, 5.0
|
||
data_span = tmax - tmin
|
||
tmin, tmax = self.visible_domain(tmin, tmax)
|
||
span = max(tmax - tmin, 0.001)
|
||
|
||
def mx(t): # 时间 -> 像素 x
|
||
return left + (t - tmin) / span * pw
|
||
|
||
empty = True
|
||
for ch in range(len(self._vals)):
|
||
ts, vs = self._times[ch], self._vals[ch]
|
||
n = len(vs)
|
||
if n == 0:
|
||
continue
|
||
empty = False
|
||
color = QColor(self.COLORS[ch % len(self.COLORS)])
|
||
p.setPen(QPen(color, 1.6))
|
||
if n == 1:
|
||
p.drawEllipse(QPointF(mx(ts[0]), my(vs[0])), 2.2, 2.2)
|
||
continue
|
||
step = max(1, n // 1000)
|
||
path = QPainterPath(QPointF(mx(ts[0]), my(vs[0])))
|
||
for i in range(step, n, step):
|
||
path.lineTo(mx(ts[i]), my(vs[i]))
|
||
path.lineTo(mx(ts[-1]), my(vs[-1]))
|
||
p.drawPath(path)
|
||
|
||
p.setPen(dim)
|
||
tag = '窗口' if self._span is not None else '最近'
|
||
if self._span is not None and data_span < self._span - 0.5:
|
||
label = f'已采集 {data_span:.0f}s / 窗口 {self._span:.0f}s'
|
||
else:
|
||
label = f'{tag} {span:.0f}s' if span >= 10 else f'{tag} {span:.1f}s'
|
||
p.drawText(rect.adjusted(0, rect.height() - 18, -4, 0),
|
||
Qt.AlignRight | Qt.AlignVCenter, label)
|
||
if self._hover_pos is not None:
|
||
self._hover_lines = self._hover_lines_at(self._hover_pos)
|
||
if self._hover_lines:
|
||
self._hover_box(p)
|
||
if empty:
|
||
p.drawText(rect, Qt.AlignCenter, '等待数据…(勾选通道并使能接收)')
|
||
elif self._paused:
|
||
p.drawText(rect.adjusted(0, top, 0, 0), Qt.AlignHCenter, '已暂停')
|
||
elif self._frozen:
|
||
p.drawText(rect.adjusted(0, top, 0, 0), Qt.AlignHCenter,
|
||
'已冻结渲染(采集继续)')
|
||
self._legend(p, rect, text)
|
||
|
||
def _hover_box(self, p):
|
||
"""顶部居中的悬停读值小框(反色底), 内容在 mouseMove 里持续刷新"""
|
||
p.setFont(QFont(self.font().family(), 9))
|
||
fm = p.fontMetrics()
|
||
w = max(fm.horizontalAdvance(s) for s in self._hover_lines) + 16
|
||
h = fm.height() * len(self._hover_lines) + 10
|
||
x = self.M_LEFT + (self.width() - self.M_LEFT - self.M_RIGHT - w) // 2
|
||
y = self.M_TOP + 6
|
||
bg = QColor(self.palette().color(QPalette.WindowText))
|
||
bg.setAlpha(210)
|
||
p.setPen(Qt.NoPen)
|
||
p.setBrush(bg)
|
||
p.drawRoundedRect(x, y, w, h, 4, 4)
|
||
p.setPen(self.palette().color(QPalette.Base))
|
||
for i, s in enumerate(self._hover_lines):
|
||
p.drawText(x + 8, y + 5 + i * fm.height(), w - 16, fm.height(),
|
||
Qt.AlignLeft | Qt.AlignVCenter, s)
|
||
p.setBrush(Qt.NoBrush)
|
||
|
||
def _legend(self, p, rect, text):
|
||
"""右上角图例: 色块 + 通道名 + 最新值"""
|
||
lines = []
|
||
for ch in range(len(self._vals)):
|
||
if self._latest[ch] is None:
|
||
continue
|
||
name = self._names[ch] or f'CH{ch + 1}'
|
||
lines.append((ch, f'{name} = {fmt_val(self._latest[ch])}'))
|
||
if not lines:
|
||
return
|
||
p.setFont(QFont(self.font().family(), 8))
|
||
fm = p.fontMetrics()
|
||
w = max(fm.horizontalAdvance(s) for _c, s in lines) + 26
|
||
h = fm.height() * len(lines) + 8
|
||
x1, y1 = rect.width() - 16 - w, 8
|
||
bg = QColor(text)
|
||
bg.setAlpha(18)
|
||
p.setPen(Qt.NoPen)
|
||
p.setBrush(bg)
|
||
p.drawRoundedRect(x1, y1, w, h, 4, 4)
|
||
p.setBrush(Qt.NoBrush)
|
||
for i, (ch, s) in enumerate(lines):
|
||
y = y1 + 4 + i * fm.height()
|
||
p.setPen(Qt.NoPen)
|
||
p.setBrush(QColor(self.COLORS[ch % len(self.COLORS)]))
|
||
p.drawRect(x1 + 6, y + fm.ascent() - 8, 8, 8)
|
||
p.setPen(self.palette().color(QPalette.WindowText))
|
||
p.drawText(x1 + 20, y, w - 22, fm.height(),
|
||
Qt.AlignLeft | Qt.AlignVCenter, s)
|