hulk/FilmForge
1
0
Fork 0
generated from hulk/gd32f4xx_template_cmake_vscode
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity
Files
000781e2f5883e31e63e5dab0451f95cb7d4b5eb
FilmForge/GUI/app_fluent.py
T
hulk 000781e2f5 GUI v0.2.13: chart Y-axis range control (right-edge dual-handle slider) 
- TemperatureChart gains a right-edge slider on a FIXED -20..120C rail
  (user-specified): top handle = upper bound, bottom handle = lower bound,
  drag middle band = pan keeping span, click free rail = snap that bound.
  First drag engages manual mode (yManualEngaged -> sync button state +
  config); release quantizes to 0.1C via yRangeChanged, min span 0.5C.
  Bound values during drag must come from live display range, not the
  stale paint cache (two drags without repaint made pan jump the range).
- Auto/manual toggle moved into the card header (was a wrapped second
  row); pause/clear merged into header too; legend moved to top-left.
- Value labels next to handles only visible on hover/drag (constant
  labels covered the curve per user feedback).
- Out-of-range curves clipped back to the plot area (setClipRect);
  current-value marker suppressed outside the manual range.
- Draft textbox rows from v0.2.10 superseded; versions v0.2.10..v0.2.13.
- Fixed "no data yet" hint never disappearing: set_source cached a
  stale _has_data bool once at build time while data flows through the
  same list references; now evaluated live at paint time.
- config: y_axis key persists mode/bounds; invalid stored values fall
  back to auto. selftest settings isolated to a temp dir (user json with
  stored manual state broke the fresh-default assertions).
- selftest_gui: slider hit/drag/clamp/pan/clipping regression + hint
  gating; 83 checks all pass.
2026-10-15 21:30:21 +08:00

1701 lines
72 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""FilmForge Fluent UI — 热压模具温控闭环监视台。
协议与串口模块(filmforge_rs485)保持不变, 本文件只做呈现层:
PySide6-Fluent-Widgets 卡片式双栏仪表盘 + 自绘滚动温度曲线(无第三方图表依赖)。
闭环监视 = 实测温度曲线 + 目标温度阶梯线同图 + 加热/上报/传感器状态联动;
温度监控走固件 M2 周期上报帧(M2S<ms> 开启), M1 仅作手动单点查询。
"""
import ctypes
import math
import os
import queue
import sys
import time
from collections import deque
# PyInstaller keeps Shiboken's ABI DLLs in a sibling directory. Register both
# Qt and Shiboken before importing QtCore; this is required by one-file builds
# on Windows/Python 3.14 where the normal package import hook is unavailable.
if getattr(sys, "frozen", False) and os.name == "nt":
_frozen_dll_dirs = []
for _dll_dir in (os.path.join(getattr(sys, "_MEIPASS", ""), "PySide6"),
os.path.join(getattr(sys, "_MEIPASS", ""), "shiboken6")):
if os.path.isdir(_dll_dir):
_frozen_dll_dirs.append(_dll_dir)
try:
os.add_dll_directory(_dll_dir)
except (AttributeError, OSError):
pass
# Python 3.14 extension loading may ignore AddDllDirectory for a pyd's
# transitive dependencies; PATH keeps QtCore.pyd compatible with both
# Python 3.11-3.14 and PyInstaller one-file extraction.
if _frozen_dll_dirs:
os.environ["PATH"] = os.pathsep.join(_frozen_dll_dirs + [os.environ.get("PATH", "")])
# Preload the two libraries used by QtCore.pyd. This also covers Windows
# loader configurations that do not apply PATH/AddDllDirectory to a pyd's
# transitive dependencies after PyInstaller extraction.
try:
ctypes.WinDLL(os.path.join(getattr(sys, "_MEIPASS", ""), "shiboken6",
"shiboken6.abi3.dll"))
ctypes.WinDLL(os.path.join(getattr(sys, "_MEIPASS", ""), "PySide6",
"Qt6Core.dll"))
except (AttributeError, OSError):
pass
from PySide6.QtCore import QPointF, QRectF, Qt, QTimer, Signal
from PySide6.QtGui import QColor, QCursor, QFont, QIcon, QPainter, QPainterPath, QPen, QPalette
from PySide6.QtWidgets import (
QApplication, QHBoxLayout, QLabel, QMainWindow, QMessageBox, QPlainTextEdit,
QScrollArea, QSplitter, QVBoxLayout, QWidget,
)
from qfluentwidgets import (
BodyLabel, CardWidget, ComboBox, LineEdit, PrimaryPushButton, PushButton,
Slider, StrongBodyLabel, Theme, setCustomStyleSheet, setTheme,
)
import config
import filmforge_rs485 as ff
APP_NAME = "FilmForge"
APP_VERSION = "0.2.13"
REPO_URL = "https://gitea.hulk.wang/hulk/FilmForge"
MONO = QFont("Consolas", 10)
CHIP_IDLE = "#666666"
CHIP_OK = "#2FA574"
CHIP_ERR = "#D9534F"
CHIP_WARN = "#E8833A"
PENDING_TIMEOUT = 1.2 # 应答超时(秒): 协议建议 300ms 重发, 监视台不自动重发只提示
RESPONSE_FRESH = 0.5 # 应答新鲜度: 该窗口内的温度帧优先按“查询应答”派发
MAX_FAULTS = 1000 # 故障标记内存上限(条): NTC 抖动场景防无限增长
CHART_COLORS_DARK = dict(bg="#242424", grid="#3a3a3a", text="#c9c9c9",
measured="#00c8d7", target=CHIP_WARN, fault=CHIP_ERR)
CHART_COLORS_LIGHT = dict(bg="#ffffff", grid="#e3e8ee", text="#5a6572",
measured="#0f6cbd", target="#d96f1a", fault=CHIP_ERR)
X_TICK_STEPS = (1, 2, 5, 10, 15, 30, 60, 120, 300, 600, 900, 1800, 3600)
RIGHT_MARGIN = 12 # 主绘图区右缘留白(px)
GUTTER_W = 18 # 右侧 Y 轴滑条栏宽(px)
HANDLE_H = 9 # 滑条把手厚(px)
HANDLE_HIT = 12 # 把手命中半径(px)
MIN_Y_SPAN = 0.5 # 手动 Y 量程最小间距(℃)
Y_RAIL_LO = -20.0 # 滑条固定参考域(℃, 用户拍板): 把手位置=℃ 读数
Y_RAIL_HI = 120.0 # (覆盖工艺全程 0~120 与冷启动低温, 不随数据漂移)
DANGER_QSS = ("PushButton, PushButton:hover, PushButton:pressed, PushButton:disabled "
"{ background-color: #D9534F; color: white; border: none; }")
class TemperatureChart(QWidget):
"""自绘滚动温度曲线(QWidget + QPainter, 不引第三方图表库)。
数据由 App 持有, 经 set_source() 传入引用; 视窗/冻结/配色/Y 轴模式独立设置。
曲线语义: 实测=折线(NTC 故障处 NaN 断开+底部红标), 目标=阶梯横线(橙虚线)。
Y 轴(auto)=按可见数据自适应; Y 轴(manual)=右缘滑条双把手固定量程
(上把手=上限/下把手=下限/拖中带=整段平移, 曲线裁剪回绘图区)。
"""
yManualEngaged = Signal() # 首次拖拽滑条: 从 auto 切入 manual
yRangeChanged = Signal(float, float) # 松手: 量化 0.1℃ 后的范围
MIN_Y_SPAN = MIN_Y_SPAN # 手动量程最小间距(℃, 供测试断言)
def __init__(self, parent=None):
super().__init__(parent)
self.setMinimumHeight(300)
self._samples_t = []
self._samples_v = []
self._target_hist = []
self._faults = []
self._window_sec = config.DEFAULT_WINDOW_SEC
self._frozen_at = None # None=实时跟随; 否则冻结该时刻
self._colors = dict(CHART_COLORS_DARK)
self._y_mode = "auto" # 'auto'=按可见数据自适应; 'manual'=固定范围
self._y_manual = None # 手动范围 (ymin, ymax); None=未设
self._y_drag = None # 拖拽态: dict(kind/y0/lo/hi/h/ref) 或 None
self._drag_cursor = False # 拖拽中 override 光标(保证 release 还原)
self._y_hover = False # 鼠标在滑条栏上: 值标签仅此时显示(不常驻遮曲线)
self._paint_area = None # 最近绘制的布局缓存: 命中/拖拽复用免重算
self._paint_range = (-5.0, 55.0)
self._paint_ref = None
self.setMouseTracking(True) # 悬停把手换光标(mouseMove 无按键也来)
self.last_auto_range = (-5.0, 55.0) # 最近绘制的自适应范围(切手动时预填)
# ---- 数据/状态注入 ----
def set_source(self, samples_t, samples_v, target_hist, faults):
"""注入数据列表引用(原地变更可见)。仅在构建时调用一次: 是否有数据
必须由绘制时实时判定, 旧版在这里缓存 _has_data 布尔, 数据经引用
追加后永不刷新 → "暂无数据"提示文字不消失(实测踩坑)。"""
self._samples_t, self._samples_v = samples_t, samples_v
self._target_hist, self._faults = target_hist, faults
self.update()
def set_window(self, sec):
self._window_sec = int(sec)
self.update()
def set_frozen(self, at):
self._frozen_at = at
self.update()
def set_colors(self, colors):
self._colors = dict(colors)
self.update()
def set_y_mode(self, mode):
self._y_mode = mode if mode in ("auto", "manual") else "auto"
self.update()
def set_y_manual(self, lo, hi):
lo, hi = float(lo), float(hi)
if not lo < hi:
return False
self._y_manual = (lo, hi)
self.update()
return True
# ---- 视窗 ----
def _view(self):
t1 = self._frozen_at if self._frozen_at is not None else time.time()
if self._window_sec > 0:
return t1 - self._window_sec, t1
ts = self._samples_t
t0 = min(ts[0], self._target_hist[0][0]) if ts and self._target_hist \
else (ts[0] if ts else t1 - 60)
return min(t0, t1 - 30), t1
def _visible(self, t0, t1):
"""视窗内样本 (t 列表, v 列表); 超过 4000 点按步进抽稀(保 NaN 断口)"""
ts, vs = self._samples_t, self._samples_v
if not ts:
return [], []
lo, hi = 0, len(ts) # 左边界二分(留一点前文保证连线)
while lo < hi:
m = (lo + hi) // 2
if ts[m] < t0:
lo = m + 1
else:
hi = m
lo = max(lo - 1, 0)
end, hi = lo, len(ts) # 右边界二分: 暂停(冻结)后不切冻结尾后新样本
while end < hi:
m = (end + hi) // 2
if ts[m] <= t1:
end = m + 1
else:
hi = m
t_out, v_out = ts[lo:end], vs[lo:end]
if len(t_out) > 4000:
stride = math.ceil(len(t_out) / 4000)
gt, gv = [], []
for i in range(0, len(t_out), stride):
chunk = v_out[i:i + stride]
gt.append(t_out[i])
gv.append(float("nan") if any(v != v for v in chunk) else chunk[0])
t_out, v_out = gt, gv
return t_out, v_out
def _target_points(self, t0, t1):
"""目标阶梯线: [(t,v)...] 每段从自身时刻起垂直落变并水平延伸至下一变更,
最后一段延伸到右缘。旧实现对"最后一次变更"只画终点, 会从旧值斜拉一条
直线到右缘(实测 57→41 被画成 2 分钟斜坡), 已改为纯阶梯。"""
hist = self._target_hist
if not hist:
return []
pts = []
for i, (ti, vi) in enumerate(hist):
tn = hist[i + 1][0] if i + 1 < len(hist) else t1
if tn < t0 or ti > t1:
continue
start, end = max(ti, t0), min(tn, t1)
if not pts:
pts.append((start, vi))
else:
if pts[-1][0] != start:
pts.append((start, pts[-1][1])) # 空窗: 先延续旧值(防御)
pts.append((start, vi)) # 与上一段终点同 x → 垂直落变
if end > start:
pts.append((end, vi))
if pts and pts[-1][0] < t1:
pts.append((t1, pts[-1][1]))
return pts
# ---- 绘制 ----
def _layout(self):
"""一次算齐布局(转换区/S 轴范围), paint 与滑条命中/拖拽共用, 保证一致"""
area = QRectF(self.rect()).adjusted(52, 10, -(RIGHT_MARGIN + GUTTER_W), -26)
t0, t1 = self._view()
vis_t, vis_v = self._visible(t0, t1)
tgt = self._target_points(t0, t1)
ymin, ymax = self._resolve_y_range(vis_v, tgt)
return area, t0, t1, vis_t, vis_v, tgt, ymin, ymax
def paintEvent(self, _event):
p = QPainter(self)
c = self._colors
p.fillRect(self.rect(), QColor(c["bg"]))
area, t0, t1, vis_t, vis_v, tgt, ymin, ymax = self._layout()
self._paint_area = area
self._paint_range = (ymin, ymax)
self._paint_ref = self._y_ref()
sy = self._scale_y(area, ymin, ymax)
sx = self._scale_x(area, t0, t1)
self._draw_grid(p, area, t0, t1, ymin, ymax, sx, sy)
p.save()
p.setClipRect(area) # 手动范围下数据可越界, 裁剪回绘图区
self._draw_target(p, tgt, sx, sy)
self._draw_measured(p, vis_t, vis_v, sx, sy)
p.restore()
self._draw_marker(p, area, vis_t, vis_v, sx, sy)
self._draw_faults(p, area, t0, t1)
self._draw_y_gutter(p, area, ymin, ymax)
self._draw_legend(p, area)
if not (self._samples_t or self._faults): # 实时判定: 数据到位即消失
p.setPen(QColor(c["text"]))
p.drawText(self.rect(), Qt.AlignCenter,
"暂无数据 — 连接设备后开启周期上报(M2S<ms>)即可实时推送; 单点查询用 M1")
def _resolve_y_range(self, vis_v, tgt_pts):
"""本次绘制使用的 Y 轴范围: manual 模式用固定值, 否则按可见数据自适应;
空数据时自适应值是缺省 (-5,55), 不得覆盖 last_auto_range(切手动预填用)"""
auto = self._y_range(vis_v, tgt_pts)
if self._y_mode == "manual" and self._y_manual is not None:
return self._y_manual
pts = tgt_pts or []
if [v for v in vis_v if v == v] or [v for _t, v in pts]:
self.last_auto_range = auto
return auto
# ---- 右缘 Y 轴滑条(手动量程) ----
def _y_ref(self):
"""滑条固定参考域(用户拍板 -20~120℃): 把手像素位置直接对应 ℃ 读数,
空数据/冷机不再出现随数据漂移的误导位置"""
return Y_RAIL_LO, Y_RAIL_HI
def _gutter(self):
w = self.width()
return w - RIGHT_MARGIN - GUTTER_W, w - RIGHT_MARGIN
def _cached_layout(self):
"""命中/拖拽用绘制缓存; 从未绘制过(如测试直接调)则补算一次"""
if self._paint_area is None:
area, _t0, _t1, _vt, _vv, _tgt, ymin, ymax = self._layout()
self._paint_area = area
self._paint_range = (ymin, ymax)
self._paint_ref = self._y_ref()
return self._paint_area, self._paint_range, self._paint_ref
def _rail_px(self, area, ref, v):
ref_lo, ref_hi = ref
v = min(max(v, ref_lo), ref_hi)
return area.bottom() - (v - ref_lo) / max(ref_hi - ref_lo, 1e-9) * area.height()
def _rail_value(self, area, ref, y):
ref_lo, ref_hi = ref
frac = min(max((area.bottom() - y) / max(area.height(), 1e-6), 0.0), 1.0)
return ref_lo + frac * (ref_hi - ref_lo)
def _gutter_hit(self, x, y):
"""右栏命中: 就近把手优先, 轨道空白区视为就近边界点击, 上下把手间为整段平移"""
x0, x1 = self._gutter()
if not x0 - 3 <= x <= x1 + 3:
return None
area, (lo, hi), ref = self._cached_layout()
y_hi = self._rail_px(area, ref, hi)
y_lo = self._rail_px(area, ref, lo)
d_hi, d_lo = abs(y - y_hi), abs(y - y_lo)
if min(d_hi, d_lo) <= HANDLE_HIT:
return "max" if d_hi <= d_lo else "min"
if y_hi <= y <= y_lo:
return "win"
return "max" if y < y_hi else "min"
def start_y_drag(self, kind, y):
"""按下滑条(测试可直接调): auto 模式以当前显示范围起坡并切手动。
几何(绘图区/参考域)可用绘制缓存, 界值必须取实时显示值——拖拽间存在
未重绘的域间修改(实测: pan 用陈旧界值把量程拖跳)。
kind: 'min'(下把手)=下限 / 'max'(上把手)=上限 / 'win'=整段平移"""
area, rng, ref = self._cached_layout()
if self._y_mode != "manual":
lo, hi = rng
self._y_manual = (lo, hi)
self._y_mode = "manual"
self.yManualEngaged.emit()
else:
lo, hi = self._y_manual
self._y_drag = {"kind": kind, "y0": y, "lo": lo, "hi": hi,
"h": area.height(), "ref": ref}
if not self._drag_cursor:
QApplication.setOverrideCursor(QCursor(Qt.SizeVerCursor))
self._drag_cursor = True
self.update()
return True
def update_y_drag(self, y):
"""拖拽中: 手把直接跟鼠标(参考域映射+钳位), 中带按参考跨度平移"""
d = self._y_drag
if d is None:
return False
area, _rng, _ref_unused = self._cached_layout()
ref = d["ref"]
ref_span = max(ref[1] - ref[0], 1e-9)
h = max(d["h"], 1e-6)
lo, hi = d["lo"], d["hi"]
if d["kind"] == "max":
hi = max(self._rail_value(area, ref, y), lo + MIN_Y_SPAN)
elif d["kind"] == "min":
lo = min(self._rail_value(area, ref, y), hi - MIN_Y_SPAN)
else:
delta = -(y - d["y0"]) / h * ref_span
delta = min(max(delta, ref[0] - lo), ref[1] - hi)
lo, hi = lo + delta, hi + delta
self._y_manual = (lo, hi)
self.update()
return True
def end_y_drag(self):
"""松手: 量化 0.1℃ 并通知(防御四舍五入挤掉间距)"""
if self._y_drag is None:
return False
lo, hi = self._y_manual
lo, hi = round(lo, 1), round(hi + 1e-9, 1)
if not lo < hi:
hi = lo + 0.1
self._y_manual = (lo, hi)
self._y_drag = None
self.update()
self.yRangeChanged.emit(lo, hi)
return True
def mousePressEvent(self, event):
pos = event.position()
kind = self._gutter_hit(pos.x(), pos.y())
if kind is not None:
self.start_y_drag(kind, pos.y())
event.accept()
else:
event.ignore()
def mouseMoveEvent(self, event):
pos = event.position()
if self._y_drag is not None:
self.update_y_drag(pos.y())
self._set_y_hover(True)
return
hit = self._gutter_hit(pos.x(), pos.y()) is not None
self._set_y_hover(hit) # 值标签仅悬停/拖拽时显示
if hit:
self.setCursor(Qt.SizeVerCursor)
else:
self.unsetCursor()
def mouseReleaseEvent(self, event):
pos = event.position()
if self._y_drag is not None:
self.end_y_drag()
if self._drag_cursor:
QApplication.restoreOverrideCursor()
self._drag_cursor = False
self.update_y_drag_hover(pos)
event.ignore()
def update_y_drag_hover(self, pos):
hit = self._gutter_hit(pos.x(), pos.y()) is not None
self._set_y_hover(hit)
if hit:
self.setCursor(Qt.SizeVerCursor)
else:
self.unsetCursor()
def _set_y_hover(self, on):
"""滑条悬停态翻转才重绘(悬停/拖拽期间才画把手值标签)"""
if self._y_hover == on:
return
self._y_hover = on
self.update()
def leaveEvent(self, _event):
if self._y_drag is None:
self.unsetCursor()
self._set_y_hover(False) # 悬停态途中拖拽不退出(松手后按位置重算)
def _draw_y_gutter(self, p, area, ymin, ymax):
"""右缘滑条栏: 轨道 + 手动量程带 + 双把手。值标签仅鼠标悬停/拖拽时显示
(常驻会遮挡曲线, 用户标注); auto 模式把手半透明——直接按住即可切手动。"""
c = self._colors
accent = QColor(c["measured"])
manual = self._y_mode == "manual"
show_pills = self._y_hover or self._y_drag is not None
ref = self._y_ref()
x0, x1 = self._gutter()
x_c = (x0 + x1) / 2
y_hi = self._rail_px(area, ref, ymax)
y_lo = self._rail_px(area, ref, ymin)
p.setPen(QPen(QColor(c["grid"]), 1))
p.drawLine(QPointF(x_c, area.top()), QPointF(x_c, area.bottom()))
if manual and y_hi + 2 < y_lo:
band = QColor(accent)
band.setAlpha(38)
p.setPen(Qt.NoPen)
p.setBrush(band)
p.drawRect(QRectF(x_c - 2, y_hi, 4, y_lo - y_hi))
for y_px, label in ((y_hi, f"{ymax:.1f}"), (y_lo, f"{ymin:.1f}")):
bar = QColor(accent)
if not manual:
bar.setAlpha(110)
p.setPen(Qt.NoPen)
p.setBrush(bar)
p.drawRoundedRect(QRectF(x_c - 7, y_px - HANDLE_H / 2, 14, HANDLE_H), 3, 3)
p.setPen(QPen(QColor(c["bg"]), 1))
for dy in (-1.5, 1.5):
p.drawLine(QPointF(x_c - 4, y_px + dy), QPointF(x_c + 4, y_px + dy))
if not show_pills:
continue
pill = QRectF(area.right() - 41, y_px - 8, 38, 16)
p.setBrush(QColor(c["bg"]))
p.setPen(QPen(QColor(c["grid"]), 1))
p.drawRoundedRect(pill, 4, 4)
font = self.font()
font.setPointSizeF(7.5)
p.setFont(font)
p.setPen(QColor(c["measured"]))
p.drawText(pill, Qt.AlignCenter, label)
@staticmethod
def _y_range(vis_v, tgt_pts):
vals = [v for v in vis_v if v == v] + [v for _t, v in tgt_pts]
if not vals:
return -5.0, 55.0
lo, hi = min(vals), max(vals)
if hi - lo < 2.0:
mid = (hi + lo) / 2
lo, hi = mid - 1.0, mid + 1.0
pad = (hi - lo) * 0.1
return lo - pad, hi + pad
@staticmethod
def _nice_step(span, ticks):
if span <= 0:
return 1.0
raw = span / max(ticks, 1)
mag = 10 ** math.floor(math.log10(raw)) if raw > 0 else 1.0
for m in (1, 2, 2.5, 5, 10):
if raw <= m * mag:
return m * mag
return 10 * mag
@staticmethod
def _scale_y(area, ymin, ymax):
h = area.height()
def sy(v):
return area.bottom() - (v - ymin) / (ymax - ymin) * h
return sy
@staticmethod
def _scale_x(area, t0, t1):
w = area.width()
def sx(t):
return area.left() + (t - t0) / max(t1 - t0, 1e-6) * w
return sx
def _draw_grid(self, p, area, t0, t1, ymin, ymax, sx, sy):
c = self._colors
grid_pen = QPen(QColor(c["grid"]), 1)
text_pen = QColor(c["text"])
font = self.font()
font.setPointSizeF(8.0)
p.setFont(font)
ystep = self._nice_step(ymax - ymin, 5)
v = math.ceil(ymin / ystep) * ystep
while v <= ymax:
y = sy(v)
p.setPen(grid_pen)
p.drawLine(QPointF(area.left(), y), QPointF(area.right(), y))
p.setPen(text_pen)
p.drawText(QRectF(0, y - 8, area.left() - 6, 16),
Qt.AlignRight | Qt.AlignVCenter, f"{v:g}")
v += ystep
span = t1 - t0
xstep = next((s for s in X_TICK_STEPS if span / s <= 8), 3600)
t = math.ceil((t0 + 0.001) / xstep) * xstep
while t <= t1:
x = sx(t)
p.setPen(grid_pen)
p.drawLine(QPointF(x, area.top()), QPointF(x, area.bottom()))
p.setPen(text_pen)
p.drawText(QRectF(x - 46, area.bottom() + 4, 92, 16),
Qt.AlignHCenter | Qt.AlignVCenter,
time.strftime("%H:%M:%S", time.localtime(t)))
t += xstep
def _draw_target(self, p, tgt, sx, sy):
if not tgt:
return
p.setPen(QPen(QColor(self._colors["target"]), 1.4, Qt.DashLine))
path = QPainterPath()
for i, (t, v) in enumerate(tgt):
if i == 0:
path.moveTo(sx(t), sy(v))
else:
path.lineTo(sx(t), sy(v))
p.drawPath(path)
def _draw_measured(self, p, ts, vs, sx, sy):
if not ts:
return
p.setRenderHint(QPainter.Antialiasing, True)
pen = QPen(QColor(self._colors["measured"]), 1.6)
pen.setCapStyle(Qt.FlatCap)
p.setPen(pen)
path = QPainterPath()
drawing = False
for t, v in zip(ts, vs):
if v != v: # NaN: NTC 故障断口
drawing = False
continue
x, y = sx(t), sy(v)
if not drawing:
path.moveTo(x, y)
drawing = True
else:
path.lineTo(x, y)
p.drawPath(path)
def _draw_marker(self, p, area, ts, vs, sx, sy):
"""当前值标记: 视窗内最后一个有效样本。超出当前 Y 范围(手动模式)不画。"""
last_t = last_v = None
for t, v in zip(reversed(ts), reversed(vs)):
if v == v:
last_t, last_v = t, v
break
if last_t is None:
return
x, y = sx(last_t), sy(last_v)
if not area.top() <= y <= area.bottom():
return
p.setPen(Qt.NoPen)
p.setBrush(QColor(self._colors["measured"]))
p.drawEllipse(QPointF(x, y), 3.2, 3.2)
p.setPen(QColor(self._colors["measured"]))
font = self.font()
font.setPointSizeF(8.5)
font.setBold(True)
p.setFont(font)
# 文字通常画在点上方; 贴顶边放不下时换到点下方, 再钳回绘图区内
ty = y - 6 if y - 14 >= area.top() else y + 12
ty = min(max(ty, area.top() + 9), area.bottom() - 2)
p.drawText(QPointF(min(x + 6, area.right() - 44), ty),
f"{last_v:.1f}℃")
def _draw_faults(self, p, area, t0, t1):
flt = [(t, s) for t, s in self._faults if t0 <= t <= t1]
if not flt:
return
span = max(t1 - t0, 1e-6)
p.setPen(QColor(self._colors["fault"]))
font = self.font()
font.setPointSizeF(7.5)
p.setFont(font)
labeled = 0
last_x = -1e9
for t, s in flt:
x = area.left() + (t - t0) / span * area.width()
p.drawLine(QPointF(x, area.bottom() - 8), QPointF(x, area.bottom()))
if labeled < 12 and x - last_x > 52:
p.drawText(QPointF(x + 3, area.bottom() - 10), s)
last_x = x
labeled += 1
def _draw_legend(self, p, area):
c = self._colors
font = self.font()
font.setPointSizeF(8.5)
p.setFont(font)
# 图例放绘图区左上: 右侧被滑条把手/值标签占用(用户标注)
x = area.left() + 6
y = 16
for color, label in ((c["measured"], "实测"), (c["target"], "目标")):
p.setPen(Qt.NoPen)
p.setBrush(QColor(color))
p.drawRect(QRectF(x, y - 7, 14, 3))
p.setPen(QColor(c["text"]))
p.drawText(QPointF(x + 18, y), label)
x += 66
class App(QMainWindow):
def __init__(self, auto_close=0):
super().__init__()
app = QApplication.instance()
self._system_palette = app.palette() if app else QPalette()
self._system_style = app.styleSheet() if app else ""
self.setWindowTitle(f"{APP_NAME} v{APP_VERSION} — 热压模具温控监视台")
icon = _icon_path() # 标题栏图标(run_gui 另设 app 级 + 任务栏标识)
if os.path.exists(icon):
self.setWindowIcon(QIcon(icon))
self.resize(1440, 920)
self.setMinimumSize(1150, 720)
config.load_settings()
# ---- 监视数据(主线程持有; 串口线程只经队列投递) ----
self.samples_t = [] # time.time() 墙钟, 串口线程读取点打戳
self.samples_v = [] # ℃, NaN = NTC 故障断口
self.target_hist = [] # [(t, 目标℃)]
self.faults = [] # [(t, 'OPEN'|'SHRT')]
self.sensor_state = None # None / 'OK' / 'OPEN' / 'SHRT'
self.heater_state = None # None / 'ON' / 'OFF'
self.report_state = None # None / ('ON', ms) / ('OFF', None)
self.firmware = None
# 应答 FIFO: 半双工上固件严格按收到顺序应答, 连接序列/连发场景下
# 多个应答可能同时在途, 单 pending 槽会被乱序击穿(实测踩坑)
self._pending = deque() # [{'cmd','kind','t','apply','then'}]
self._frozen_at = None
self._want = False # 连接按钮意图(点击即翻转, 不等线程实际开/关)
self._want_since = None # 意图=连接但实际未建立的起始时刻(丢失判定)
self._last_set = None # 最近一条设定命令(t/cmd/apply): 晚到 ok 补偿
self._resume_report_ms = None # M4 查询前暂停的上报间隔(查询完自动恢复)
self._seq = [] # 连接同步序列剩余步骤
self._last_chart_repaint = 0.0
self._ports = []
self._rx_q = queue.Queue()
# ---- CSV 记录 ----
self._csv = None
self._csv_path = ""
self._csv_t0 = 0.0
self._csv_count = 0
self.link = ff.FF485(self._on_event, self._on_status, on_frame=self._on_frame)
self._connected = False
self._build()
self._apply_settings()
self._timer = QTimer(self)
self._timer.timeout.connect(self._tick)
self._timer.start(100)
if ff.HAS_SERIAL:
self._refresh_ports(silent=True)
else:
self.lbl_st.setText("未安装 pyserial: py -m pip install pyserial")
if auto_close > 0:
QTimer.singleShot(auto_close * 1000, self.close)
# ================= 界面构建 =================
def _button(self, text, callback, primary=False):
button = PrimaryPushButton(text) if primary else PushButton(text)
button.clicked.connect(callback)
return button
def _card(self, title, trailing=None):
card = CardWidget(self)
layout = QVBoxLayout(card)
layout.setContentsMargins(14, 12, 14, 14)
layout.setSpacing(8)
row = QHBoxLayout()
row.addWidget(StrongBodyLabel(title, card))
row.addStretch(1)
if trailing is not None:
row.addWidget(trailing)
layout.addLayout(row)
return card, layout
def _chip(self, text):
lbl = QLabel(text, self)
lbl.setMinimumHeight(30)
lbl.setAlignment(Qt.AlignCenter | Qt.AlignVCenter)
lbl.setStyleSheet(f"background: {CHIP_IDLE}; color: white; "
"border-radius: 6px; padding: 4px 8px;")
return lbl
@staticmethod
def _set_chip(lbl, text, color):
lbl.setText(text)
lbl.setStyleSheet(f"background: {color}; color: white; "
"border-radius: 6px; padding: 4px 8px;")
def _build(self):
root = QWidget(self)
root.setObjectName("root")
outer = QVBoxLayout(root)
outer.setContentsMargins(18, 14, 18, 18)
outer.setSpacing(12)
header = QHBoxLayout()
title = StrongBodyLabel("FilmForge", root)
# 标题字号走 QFont: FluentLabelBase 切主题时会重写自定义 qss 槽(DLPilot §3.12)
tf = title.font()
tf.setPixelSize(24)
tf.setBold(True)
title.setFont(tf)
header.addWidget(title)
header.addWidget(BodyLabel("热压模具温控 · RS485 闭环监视", root))
header.addStretch(1)
self.cbo_theme = ComboBox(root)
self.cbo_theme.addItems(["Dark", "Light", "System"])
self.cbo_theme.currentTextChanged.connect(self._set_theme)
header.addWidget(self.cbo_theme)
header.addWidget(self._button("关于", self._about))
outer.addLayout(header)
# 双列放进同一滚动容器: 两列高度锁定一致——不足一屏时各自拉伸到底边,
# 超出一屏时整页一起滚动; 各列独立滚动会让底边在窗口化下错位(实测)
split = QSplitter(Qt.Horizontal)
split.setChildrenCollapsible(False)
split.addWidget(self._column_host(self._build_left))
split.addWidget(self._column_host(self._build_right))
split.setStretchFactor(0, 3)
split.setStretchFactor(1, 2)
page = QWidget()
page_lay = QHBoxLayout(page)
page_lay.setContentsMargins(0, 0, 0, 0)
page_lay.addWidget(split)
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setFrameShape(QScrollArea.NoFrame)
scroll.setWidget(page)
outer.addWidget(scroll, 1)
self.setCentralWidget(root)
def _column_host(self, builder):
"""构建一列的宿主 QWidget(不再各自包 QScrollArea, 整页统一滚动)"""
host = QWidget()
layout = QVBoxLayout(host)
layout.setContentsMargins(2, 2, 8, 2)
layout.setSpacing(10)
builder(layout)
layout.addStretch(1)
return host
# ---------------- 左列: 串口 + 监视 ----------------
def _build_serial_card(self, layout):
"""串口连接卡片(置于左列顶部): 端口/波特率选择 + 连接开关 + 状态行"""
if ff.HAS_SERIAL:
card, body = self._card("串口 · RS485 → FilmForge 温控板")
row = QHBoxLayout()
self.cbo_port = ComboBox(card)
row.addWidget(self.cbo_port, 1)
row.addWidget(self._button("刷新", self._refresh_ports))
self.btn_conn = self._button("连接", self._toggle_connect, True)
row.addWidget(self.btn_conn)
body.addLayout(row)
params = QHBoxLayout()
self.cbo_baud = ComboBox(card)
self.cbo_baud.addItems(["115200", "57600", "38400", "9600"])
self.cbo_baud.setCurrentText(str(config.SERIAL_CFG["baud"]))
params.addWidget(BodyLabel("波特率", card))
params.addWidget(self.cbo_baud)
self.lbl_st = BodyLabel("未连接", card)
self.lbl_st.setWordWrap(True) # 长错误串不得撑大参数行(DLPilot 教训)
params.addWidget(self.lbl_st, 1)
body.addLayout(params)
layout.addWidget(card)
else:
self.cbo_port = None
self.btn_conn = None
self.lbl_st = BodyLabel("未安装 pyserial, 串口功能不可用。", self)
self.lbl_st.setWordWrap(True)
card, body = self._card("串口不可用")
body.addWidget(self.lbl_st)
layout.addWidget(card)
def _build_left(self, layout):
self._build_serial_card(layout)
win_combo = ComboBox(self)
for name, _sec in config.WINDOW_OPTIONS:
win_combo.addItem(name)
win_combo.setCurrentIndex(1) # 先设默认再连信号, 避免触发时 chart 未建
win_combo.currentIndexChanged.connect(self._on_window)
self.cbo_window = win_combo
card = CardWidget(self)
body = QVBoxLayout(card)
body.setContentsMargins(14, 12, 14, 14)
body.setSpacing(8)
# ---- 标题行 = [标题][暂停][清空] ... [Y轴手动][窗口] 一行放齐:
# 暂停/清空并入标题行、Y 轴按钮不再单独换行(用户标注: 两个按钮组都
# 悬在半空布局脱节)。不走 self._card 的通用头, 需要多 trailing 控件。
header = QHBoxLayout()
header.addWidget(StrongBodyLabel("实时温度曲线", card))
header.addSpacing(8)
self.btn_pause = self._button("暂停", self._toggle_pause)
header.addWidget(self.btn_pause)
self.btn_clear = self._button("清空", self._clear_plot)
header.addWidget(self.btn_clear)
header.addStretch(1)
# ---- Y 轴(温度轴)量程: 自动适配 / 图表右缘滑条手动拖拽 ----
self.btn_ymode = PushButton("Y轴自动", card)
self.btn_ymode.setCheckable(True)
self.btn_ymode.setChecked(True) # 默认自动; 先就位默认态再连信号
self.btn_ymode.setToolTip("勾选=随曲线数据自适应量程; 取消勾选或直接拖动"
"图表右缘滑条=固定量程, 再勾选恢复自动")
self.btn_ymode.toggled.connect(self._on_ymode)
header.addWidget(self.btn_ymode)
header.addWidget(win_combo)
body.addLayout(header)
self.chart = TemperatureChart(card)
self.chart.set_source(self.samples_t, self.samples_v, self.target_hist, self.faults)
self.chart.yManualEngaged.connect(self._on_y_drag_engaged)
self.chart.yRangeChanged.connect(self._on_y_range_changed)
body.addWidget(self.chart, 1)
layout.addWidget(card, 2)
card, body = self._card("实时读数")
big_row = QHBoxLayout()
self.lbl_temp = QLabel("—.— ℃", self)
bf = self.lbl_temp.font()
bf.setPixelSize(34)
bf.setBold(True)
self.lbl_temp.setFont(bf)
big_row.addWidget(self.lbl_temp)
big_row.addStretch(1)
chip_grid = QVBoxLayout()
top = QHBoxLayout()
bottom = QHBoxLayout()
self.chip_sensor = self._chip("NTC: —")
self.chip_heater = self._chip("加热: —")
self.chip_report = self._chip("上报: —")
self.chip_target = self._chip("目标: —")
top.addWidget(self.chip_sensor)
top.addWidget(self.chip_heater, 1)
bottom.addWidget(self.chip_report)
bottom.addWidget(self.chip_target, 1)
chip_grid.addLayout(top)
chip_grid.addLayout(bottom)
big_row.addLayout(chip_grid)
body.addLayout(big_row)
self.lbl_stats = BodyLabel("窗口统计: —", self)
body.addWidget(self.lbl_stats)
layout.addWidget(card)
card, body = self._card("数据记录 · CSV")
row = QHBoxLayout()
self.btn_csv = self._button("开始记录", self._toggle_csv)
self.btn_opendir = self._button("打开目录", self._open_log_dir)
row.addWidget(self.btn_csv)
row.addWidget(self.btn_opendir)
row.addStretch(1)
body.addLayout(row)
self.lbl_csv = BodyLabel("未记录。开始后逐行写入 logs/ 下 CSV(时间/温度/目标/加热/传感器)。", self)
self.lbl_csv.setWordWrap(True)
body.addWidget(self.lbl_csv)
layout.addWidget(card)
# ---------------- 右列: 控制 ----------------
def _build_right(self, layout):
card, body = self._card("温度控制 · 目标 / 加热（PID 闭环）")
row = QHBoxLayout()
self.ent_target = LineEdit(card)
self.ent_target.setPlaceholderText("0~200 整数℃")
self.ent_target.setFixedWidth(110)
self.sld_target = Slider(Qt.Horizontal, card)
self.sld_target.setRange(0, 120) # 工况温度 <=120℃; 协议上限 200 仍可手输
self.sld_target.setValue(120)
self.sld_target.valueChanged.connect(lambda v: self.ent_target.setText(str(v)))
row.addWidget(BodyLabel("目标温度", card))
row.addWidget(self.ent_target)
row.addWidget(self._button("查询", self._query_target))
row.addWidget(self._button("设定", self._send_target, True))
body.addLayout(row)
body.addWidget(self.sld_target)
row = QHBoxLayout()
row.addWidget(self._button("查加热 M3", lambda: self._query("M3", "heater")))
row.addWidget(self._button("开启加热", self._heater_on))
row.addWidget(self._button("关闭加热",
lambda: self._send_set("M3S0", "heater", "OFF")))
row.addStretch(1)
body.addLayout(row)
body.addWidget(self._note(card, "开启加热后固件按目标温度 PID 闭环真实控温(500ms 周期,"
"慢 PWM 执行); NTC 故障或 ≥150℃ 过温时固件自动断电保护,"
"上电默认关闭。当前占空比无查询命令, 可从温度曲线间接观察。"))
layout.addWidget(card)
card, body = self._card("温度监控 · 周期上报 M2S<ms>")
row = QHBoxLayout()
self.ent_report = LineEdit(card)
self.ent_report.setPlaceholderText("100~60000, 10ms 步进")
self.ent_report.setFixedWidth(150)
self.ent_report.setText(str(config.REPORT_MS))
row.addWidget(BodyLabel("间隔 ms", card))
row.addWidget(self.ent_report)
row.addWidget(self._button("开启上报", self._report_on, True))
row.addWidget(self._button("停止",
lambda: self._send_set("M2S0", "report", ("OFF", None))))
row.addWidget(self._button("查状态", lambda: self._query("M2", "report")))
body.addLayout(row)
body.addWidget(self._note(card, "温度监控统一走 M2: 开启上报后固件按设定周期主动推送"
"温度帧入曲线; 停止后曲线暂停, 需要单点时用命令区"
" 查温度 M1(两条路读同一全局最新温度)。"))
layout.addWidget(card)
card, body = self._card("命令 · 预设 / 自定义")
row = QHBoxLayout()
row.addWidget(self._button("查温度 M1", lambda: self._query("M1", "temp")))
row.addWidget(self._button("查版本", lambda: self._query("M999", "version")))
row.addStretch(1)
self.lbl_ver = BodyLabel("固件: —", card)
row.addWidget(self.lbl_ver)
body.addLayout(row)
row = QHBoxLayout()
self.cbo_preset = ComboBox(card)
self.cbo_preset.addItems([f"{n} {c}" for n, c, _ in ff.PRESETS])
row.addWidget(BodyLabel("预设:", card))
row.addWidget(self.cbo_preset, 1)
row.addWidget(self._button("发送", self._send_preset, True))
body.addLayout(row)
row = QHBoxLayout()
self.ent_custom = LineEdit(card)
self.ent_custom.setPlaceholderText("自定义命令, 如 M2S1000")
self.ent_custom.returnPressed.connect(self._send_custom)
row.addWidget(BodyLabel("自定义:", card))
row.addWidget(self.ent_custom, 1)
row.addWidget(self._button("发送", self._send_custom))
body.addLayout(row)
layout.addWidget(card)
card, body = self._card("串口数据 · 收发日志（译码 + 原始 hex）",
trailing=self._button("清空", self._clear_log))
self.txt_log = QPlainTextEdit(self)
self.txt_log.setReadOnly(True)
self.txt_log.setFont(MONO)
self.txt_log.setLineWrapMode(QPlainTextEdit.NoWrap)
self.txt_log.setMinimumHeight(200)
# 裁剪走原生环形上限: 手工 cursor 裁剪在 PySide6 6.11 上抛 AttributeError
# (实例级枚举访问已移除), 曾致每 tick 异常中止 + 日志无限增长 + 显示滞后
self.txt_log.setMaximumBlockCount(500)
body.addWidget(self.txt_log)
self._log(f"帧格式: D5 01 LEN <命令> CRC / B5 TYPE LEN <payload> CRC; "
f"响应超时建议 {ff.REPLY_TIMEOUT * 1000:.0f}ms 重发")
layout.addWidget(card, 1) # 弹性填充: 右列底边与左列对齐
card, body = self._card("危险操作 · 会使设备复位")
row = QHBoxLayout()
self.btn_m888 = self._button("软件复位 M888", self._danger_m888)
self.btn_ota = self._button("进 OTA M9999", self._danger_ota)
setCustomStyleSheet(self.btn_m888, DANGER_QSS, DANGER_QSS)
setCustomStyleSheet(self.btn_ota, DANGER_QSS, DANGER_QSS)
row.addWidget(self.btn_m888)
row.addWidget(self.btn_ota)
row.addStretch(1)
body.addLayout(row)
body.addWidget(self._note(card, "复位后固件回到上电默认(加热 OFF / 目标 0.0℃); "
"2 秒后自动重查设备状态。"))
layout.addWidget(card)
@staticmethod
def _note(parent, text):
lbl = BodyLabel(text, parent)
lbl.setWordWrap(True)
return lbl
def _apply_settings(self):
if self.cbo_port is not None and config.SERIAL_CFG["port"]:
self._pick_port(config.SERIAL_CFG["port"])
self.ent_report.editingFinished.connect(self._save_report_pref)
widx = next((i for i, (_n, s) in enumerate(config.WINDOW_OPTIONS)
if s == config.WINDOW_SEC), 1)
self.cbo_window.setCurrentIndex(widx)
self.chart.set_window(config.WINDOW_SEC)
self.cbo_theme.setCurrentText(config.THEME)
# ---- Y 轴量程记忆恢复: 存有合法手动值且上次为手动时才复用, 否则回自动 ----
stored = (config.Y_MIN is not None and config.Y_MAX is not None
and config.Y_MIN < config.Y_MAX)
manual = stored and config.Y_MODE == "manual"
self.btn_ymode.blockSignals(True)
self.btn_ymode.setChecked(not manual)
self.btn_ymode.blockSignals(False)
self.btn_ymode.setText("Y轴手动" if manual else "Y轴自动")
if manual:
self.chart.set_y_manual(config.Y_MIN, config.Y_MAX)
self.chart.set_y_mode("manual")
# ================= 主题 =================
def _set_theme(self, value):
config.THEME = value
app = QApplication.instance()
setTheme({"Dark": Theme.DARK, "Light": Theme.LIGHT, "System": Theme.AUTO}[value])
if value == "System":
app.setPalette(self._system_palette)
app.setStyleSheet(self._system_style)
is_dark = app.palette().color(QPalette.Window).value() < 128
else:
is_dark = value == "Dark"
self._apply_palette(is_dark)
self._apply_native_titlebar(is_dark)
self.chart.set_colors(CHART_COLORS_DARK if is_dark else CHART_COLORS_LIGHT)
def _apply_palette(self, is_dark):
"""原生 Qt 控件与 Fluent 卡片表面色对齐(沿用 DLPilot 配色)"""
app = QApplication.instance()
palette = QPalette()
if is_dark:
window, base, text, border = "#202020", "#2b2b2b", "#f5f5f5", "#3a3a3a"
accent = "#00c8d7"
else:
window, base, text, border = "#f4f6f8", "#ffffff", "#1f2328", "#d8dee4"
accent = "#0f6cbd"
for role in (QPalette.Window, QPalette.Base, QPalette.AlternateBase):
palette.setColor(role, QColor(window if role == QPalette.Window else base))
palette.setColor(QPalette.Text, QColor(text))
palette.setColor(QPalette.WindowText, QColor(text))
palette.setColor(QPalette.Button, QColor(base))
palette.setColor(QPalette.ButtonText, QColor(text))
palette.setColor(QPalette.PlaceholderText, QColor("#8b949e" if not is_dark else "#8a8a8a"))
palette.setColor(QPalette.Highlight, QColor(accent))
palette.setColor(QPalette.HighlightedText, QColor("#ffffff"))
palette.setColor(QPalette.ToolTipBase, QColor(base))
palette.setColor(QPalette.ToolTipText, QColor(text))
app.setPalette(palette)
app.setStyleSheet(f"""
QMainWindow, QWidget#root {{ background-color: {window}; color: {text}; }}
QScrollArea, QScrollArea > QWidget > QWidget {{ background: transparent; border: 0; }}
CardWidget {{ background-color: {base}; border: 1px solid {border}; border-radius: 8px; }}
QPlainTextEdit, QTextEdit, QLineEdit, QComboBox {{
background-color: {base}; color: {text};
border: 1px solid {border}; border-radius: 5px;
selection-background-color: {accent};
}}
""")
def _apply_native_titlebar(self, is_dark):
if os.name != "nt" or not hasattr(ctypes, "windll"):
return
try:
hwnd = int(self.winId())
dwm = ctypes.windll.dwmapi
dark = ctypes.c_int(1 if is_dark else 0)
dwm.DwmSetWindowAttribute(hwnd, 20, ctypes.byref(dark), ctypes.sizeof(dark))
color = QColor("#202020" if is_dark else "#f4f6f8")
colorref = ctypes.c_int(color.red() | (color.green() << 8) | (color.blue() << 16))
dwm.DwmSetWindowAttribute(hwnd, 35, ctypes.byref(colorref), ctypes.sizeof(colorref))
except Exception:
pass
# ================= 串口线程回调(后台线程, 只入队) =================
def _on_event(self, line):
self._rx_q.put(("event", line))
def _on_status(self, msg):
self._rx_q.put(("status", msg))
def _on_frame(self, frame, t_rx):
self._rx_q.put(("frame", frame, t_rx))
# ================= 主循环 tick =================
def _tick(self):
now = time.time()
try:
while True:
item = self._rx_q.get_nowait()
if item[0] == "status":
self._on_status_main(item[1])
elif item[0] == "event":
self._log(item[1])
else:
self._handle_frame(item[1], item[2])
except queue.Empty:
pass
connected = self.link.is_connected()
if connected != self._connected:
self._connected = connected
if connected and self._want:
self._connect_seq()
elif not connected:
self._pending.clear()
self._last_set = None
self._resume_report_ms = None
# 意图=连接但实际未建立(正在连接)或中途丢失(串口拔出/线程侧异常):
# 2s 宽限后判丢失, 按钮回"连接", 避免界面卡在"断开"与实际状态脱节
if self._want and not connected:
if self._want_since is None:
self._want_since = now
elif now - self._want_since > 2.0:
self._want = False
self._want_since = None
self.btn_conn.setText("连接")
self._pending.clear()
self._last_set = None
self._resume_report_ms = None
self.lbl_st.setText("连接未建立或已丢失, 请重试")
else:
self._want_since = None
# 应答超时: 按发送顺序逐个清理队首过期条目(监视台不自动重发,
# 命令全幂等, 交给用户重试; 设定命令的晚到 ok 走补偿, 见 _handle_frame)
while self._pending and now - self._pending[0]["t"] > PENDING_TIMEOUT:
entry = self._pending.popleft()
self._log(f'⏱ {entry["cmd"]} 无响应(>{PENDING_TIMEOUT:.1f}s)')
if entry["kind"] == "target":
self._resume_report() # M4 查询失败也要恢复上报
if entry.get("then"):
entry["then"]() # 链式流程(暂停上报→查目标)继续走
# 曲线时间轴推进 + 窗口统计(节流 0.5s)
if now - self._last_chart_repaint >= 0.5:
self._last_chart_repaint = now
self.chart.update()
self._update_stats()
def _on_status_main(self, msg):
self.lbl_st.setText(msg)
# 连接/断开/丢失全部入日志: 状态标签只留最后一条, 日志才可回溯排查
self._log(f"── {msg} ──")
# ================= 帧派发 =================
def _handle_frame(self, fr, t_rx):
if fr.type_code != 0xF0:
if self._pending: # 错误应答对应发送序队首
entry = self._pending.popleft()
self.lbl_st.setText(f'{entry["cmd"]} → {ff.describe(fr)}')
return
val, fault = ff.parse_temperature(fr.payload)
if fault: # NTC 开路/短路: 记故障, 不画点
if self._pending and self._pending[0]["kind"] == "temp":
self._pending.popleft()
self._record_fault(t_rx, fault)
return
if val is not None:
# 温度帧二义性(M1/M4/上报帧同为 ASCII 温度): 仅当队首在等目标温度
# 时拆出, 其余一律入实测曲线(M1 与上报读同一全局温度, 值恒相同)。
# M4 应答路由窗口 = 条目生命周期(适配器换向偶发延迟实测可 >0.5s);
# 上报开启时周期帧密集, 收紧到新鲜窗口, 视窗外温度帧按上报处理
head = self._pending[0] if self._pending else None
if head and head["kind"] == "target":
reporting = bool(self.report_state and self.report_state[0] == "ON")
window = RESPONSE_FRESH if reporting else PENDING_TIMEOUT
if t_rx - head["t"] <= window:
self._pending.popleft()
self._set_target(t_rx, val)
self._resume_report() # 暂停上报查目标的收尾
return
elif head and head["kind"] == "temp":
self._pending.popleft()
self._add_sample(t_rx, val)
return
# 非数值成功载荷: ok / ON / OFF / ON <ms> / v*
# 应答形态必须与队首期望匹配才消费, 不匹配记杂散(条目超时自清)
head = self._pending[0] if self._pending else None
pl = fr.payload.strip()
# 晚到 ok 补偿: 设定命令的 ok 延迟超过 1.2s 时已按超时清队(适配器换向
# 实测可 >4s), 迟到的 ok 在此按发送时记录的计划补偿应用, 避免状态
# chip 与设备脱节(实测: M2S100 ok 迟到, 上报 chip 停在 OFF 而设备在报)
if pl == "ok" and not (head and (head.get("apply") or head["kind"] == "ack")):
late = self._last_set
if late and t_rx - late["t"] <= 8.0:
self._last_set = None
self._log(f'⏱→ {late["cmd"]} ok 晚到, 补偿应用')
self._apply_ack({"cmd": late["cmd"], "kind": "ack",
"apply": late["apply"], "t": late["t"]}, t_rx)
elif head is not None:
self._log(f'杂散应答 ok(队首在等 {head["cmd"]})')
return
if head is None:
return
kind = head["kind"]
if pl == "ok" and (head.get("apply") or kind == "ack"):
self._pending.popleft()
self._apply_ack(head, t_rx)
if self._last_set and self._last_set.get("cmd") == head["cmd"]:
self._last_set = None
elif pl.startswith("v") and kind == "version" and len(pl) <= 16:
self._pending.popleft()
self.firmware = pl
self.lbl_ver.setText(f"固件: {pl}")
self.lbl_st.setText(f"固件版本 {pl}")
elif pl == "OFF" and kind == "report" or pl.startswith("ON ") and kind == "report":
st = ff.parse_report_state(pl)
if st[0]:
self._pending.popleft()
self._set_report_state(st[0], st[1])
elif pl in ("ON", "OFF") and kind == "heater":
self._pending.popleft()
self._set_heater(pl)
else:
self._log(f'杂散应答 {pl!r}(队首在等 {head["cmd"]})')
def _apply_ack(self, pending, t_rx):
"""M2S/M3S/M4S 设定命令的 ok 应答 -> 按 apply 计划更新本地状态"""
apply_plan = pending.get("apply")
if apply_plan:
what, value = apply_plan
if what == "target":
self._set_target(t_rx, float(value))
self.lbl_st.setText(f"目标温度已设定 {value} ℃")
elif what == "heater":
self._set_heater(value)
self.lbl_st.setText(f"加热已{'开启(PID 闭环)' if value == 'ON' else '关闭'}")
elif what == "report":
state, ms = value
self._set_report_state(state, ms)
self.lbl_st.setText("周期上报已开启" if state == "ON" else "周期上报已停止")
else:
self.lbl_st.setText("设备应答 ok")
then = pending.get("then")
if then:
then()
# ================= 数据通路 =================
def _add_sample(self, t, v):
self.sensor_state = "OK"
self._set_chip(self.chip_sensor, "NTC: 正常", CHIP_OK)
self.samples_t.append(t)
self.samples_v.append(v)
if len(self.samples_t) > config.MAX_SAMPLES:
excess = len(self.samples_t) - config.MAX_SAMPLES
del self.samples_t[:excess]
del self.samples_v[:excess]
self.lbl_temp.setText(f"{v:.1f} ℃")
self._csv_write(t, f"{v:.1f}", ok=True)
self.chart.update()
def _record_fault(self, t, code):
self.sensor_state = code
name = "开路" if code == ff.SENSOR_OPEN else "短路"
self._set_chip(self.chip_sensor, f"NTC: {name}({code})", CHIP_ERR)
self.samples_t.append(t)
self.samples_v.append(float("nan"))
self.faults.append((t, code))
if len(self.faults) > MAX_FAULTS: # 封顶: 抖动场景不无限增长(_draw_faults 按 paint 全量扫描)
del self.faults[:len(self.faults) - MAX_FAULTS]
self.lbl_temp.setText("—.— ℃")
self._csv_write(t, code, ok=False)
self.chart.update()
def _set_target(self, t, v):
self.target_hist.append((t, v))
if len(self.target_hist) > 5000:
del self.target_hist[:len(self.target_hist) - 5000]
self._set_chip(self.chip_target, f"目标: {v:g} ℃", CHIP_WARN)
self.chart.update()
def _set_heater(self, state):
self.heater_state = state
self._set_chip(self.chip_heater, f"加热: {state}",
CHIP_ERR if state == "ON" else CHIP_IDLE)
def _set_report_state(self, state, ms):
self.report_state = (state, ms)
text = f"上报: ON {ms}ms" if state == "ON" else "上报: OFF"
self._set_chip(self.chip_report, text, CHIP_OK if state == "ON" else CHIP_IDLE)
def _update_stats(self):
t0, t1 = self.chart._view()
vals = [v for t, v in zip(self.samples_t, self.samples_v)
if t0 <= t <= t1 and v == v]
if not vals:
self.lbl_stats.setText("窗口统计: —")
return
mean = sum(vals) / len(vals)
self.lbl_stats.setText(f"窗口统计: 最小 {min(vals):.1f} / 最大 {max(vals):.1f} / "
f"平均 {mean:.1f} ℃ · {len(vals)} 点")
# ================= 发送路径 =================
def _send_cmd(self, cmd, kind, apply_plan=None, then=None):
"""带应答跟踪发送: kind 决定应答派发, apply_plan 为 ok 后的本地状态计划,
then 为 ok 消费后的链式回调(暂停上报→查目标用)。
条目入 FIFO(半双工应答按发送序返回); 队列超长只留最新 8 条防失控"""
if not self.link.is_connected():
self.lbl_st.setText("未连接串口")
return False
# 幂等去重: 同一命令已有未决应答时忽略本次发送(连点"停止/开启"不再堆积,
# 真机实测连点 12 次 M2S0 挤爆 8 条 pending 队列; 协议全幂等, 同命令
# 一条在途足够, 超时自清后自然允许重发)
if any(e["cmd"] == cmd for e in self._pending):
self.lbl_st.setText(f"{cmd} 应答中, 已忽略重复发送")
return True
self._pending.append({"cmd": cmd, "kind": kind, "t": time.time(),
"apply": apply_plan, "then": then})
if apply_plan:
self._last_set = {"cmd": cmd, "t": time.time(), "apply": apply_plan}
while len(self._pending) > 8:
dropped = self._pending.popleft()
self._log(f'丢弃积压命令: {dropped["cmd"]}')
self.link.send(cmd)
return True
def _query(self, cmd, kind):
self._send_cmd(cmd, kind)
def _send_set(self, cmd, kind, value):
self._send_cmd(cmd, kind, apply_plan=(kind, value))
def _send_target(self):
try:
t = ff.validate_target_temp(self.ent_target.text())
except ValueError as e:
self.lbl_st.setText(f"目标温度非法: {e}")
return
self.ent_target.setText(str(t))
self.sld_target.setValue(t)
self._send_set(f"M4S{t}", "target", t)
def _heater_on(self):
if not self._confirm("开启加热",
"加热已实装: 开启后固件将按目标温度 PID 闭环真实加热模具,\n"
"表面可达 0~200℃(NTC 故障/过温时固件自动断电保护)。\n"
"确认开启?"):
return
self._send_set("M3S1", "heater", "ON")
def _report_on(self):
try:
ms = ff.validate_report_interval(self.ent_report.text())
except ValueError as e:
self.lbl_st.setText(f"上报间隔非法: {e}")
return
self.ent_report.setText(str(ms))
self._send_set(f"M2S{ms}", "report", ("ON", ms))
def _send_preset(self):
label = self.cbo_preset.currentText()
for name, cmd, _note in ff.PRESETS:
if not label.startswith(name + " "):
continue
if cmd in ff.RESET_COMMANDS:
self._danger_reset(cmd)
elif cmd == "M3S1":
self._heater_on()
elif cmd == "M2S500":
self._send_set(cmd, "report", ("ON", 500))
elif cmd == "M4S120":
self._send_set(cmd, "target", 120)
elif cmd == "M2S0":
self._send_set(cmd, "report", ("OFF", None))
elif cmd == "M3S0":
self._send_set(cmd, "heater", "OFF")
elif cmd == "M4":
self._query_target() # 上报开启时需暂停/恢复, 不能直发
else:
kind = {"M1": "temp", "M2": "report", "M3": "heater",
"M999": "version"}.get(cmd, "ack")
self._send_cmd(cmd, kind)
return
def _send_custom(self):
cmd = self.ent_custom.text().strip()
if not cmd:
return
if cmd in ff.RESET_COMMANDS:
self._danger_reset(cmd)
return
self._send_cmd(cmd, "ack")
self.ent_custom.clear()
def _danger_reset(self, cmd):
ota = cmd == "M9999"
if not self._confirm("软件复位" if not ota else "进入 OTA 模式",
"固件将复位" + ("进入 Bootloader(未部署前等同复位)。"
if ota else "。")
+ "复位后设备回到上电默认状态, 确定继续?"):
return
if self._send_cmd(cmd, "ack"):
self._log(f"⚠ {cmd} 已下发, 2 秒后自动重查设备状态")
QTimer.singleShot(2000, self._connect_seq_if_connected)
def _danger_m888(self):
self._danger_reset("M888")
def _danger_ota(self):
self._danger_reset("M9999")
def _confirm(self, title, text):
box = QMessageBox(self)
box.setWindowTitle(title)
box.setText(text)
box.setIcon(QMessageBox.Warning)
box.setStandardButtons(QMessageBox.Yes | QMessageBox.No)
box.setDefaultButton(QMessageBox.No)
return box.exec() == QMessageBox.Yes
# ================= 连接 =================
def _refresh_ports(self, _checked=False, silent=False):
self._ports = ff.FF485.list_ports_info()
self.cbo_port.clear()
self.cbo_port.addItems([label for _d, label in self._ports])
if self._ports:
self._pick_port(config.SERIAL_CFG["port"] or self._ports[0][0])
elif not silent:
self.lbl_st.setText("未发现串口设备")
def _pick_port(self, device):
for dev, label in self._ports:
if dev == device:
self.cbo_port.setCurrentText(label)
return
def _selected_port(self):
label = self.cbo_port.currentText()
for device, shown in self._ports:
if shown == label:
return device
return label.strip()
def _toggle_connect(self):
# 按钮按"意图"立即翻转, 实际开/关由串口线程异步完成并经状态日志反馈;
# 不再用 is_connected() 判断该走哪条分支(线程侧状态滞后会点不动)
if self._want:
self._want = False
self._want_since = None
self.btn_conn.setText("连接")
self.lbl_st.setText("正在断开…")
self._pending.clear()
self._last_set = None
self._resume_report_ms = None
self.link.close()
return
port = self._selected_port()
if not port:
self.lbl_st.setText("未选择串口, 请先选择或刷新")
return
try:
baud = int(self.cbo_baud.currentText())
except ValueError:
self.lbl_st.setText("波特率非法")
return
config.SERIAL_CFG.update(port=port, baud=baud)
config.save_settings()
self._want = True
self._want_since = time.time()
self.btn_conn.setText("断开")
self.lbl_st.setText("正在连接…")
self.link.open(port, baud)
def _connect_seq_if_connected(self):
if self.link.is_connected():
self._connect_seq()
def _connect_seq(self):
"""连接/复位后状态同步: 版本→上报→加热→(暂停上报→)目标。
每步发送前确认在线(线程异步打开期间 singleShot 早发会被"未连接串口"丢弃),
150ms 间隔规避半双工碰撞; 目标查询放最后并等 M2 应答消费完(见 _seq_target)"""
self._pending.clear()
self._last_set = None
self._seq = [("M999", "version"), ("M2", "report"), ("M3", "heater")]
QTimer.singleShot(0, self._seq_next)
def _seq_next(self):
if not self._seq:
QTimer.singleShot(0, self._seq_target)
return
if not self.link.is_connected():
return # 连接已断: 序列作废(重连会重跑)
cmd, kind = self._seq.pop(0)
self._query(cmd, kind)
QTimer.singleShot(150, self._seq_next)
def _seq_target(self):
if not self.link.is_connected():
return
if any(e["kind"] == "report" for e in self._pending):
QTimer.singleShot(100, self._seq_target) # 等 M2 应答落地再定分支
return
self._query_target()
def _query_target(self):
"""查询目标温度(M4)。上报开启时 M4 应答与周期温度帧同为 ASCII 浮点,
FIFO 窗口法在 100ms 上报下必现误占(实测目标 chip 显示成实测值)——
改为应答驱动的暂停/恢复: M2S0(ok)→M4→自动 M2S<ms> 恢复。"""
if not (self.report_state and self.report_state[0] == "ON"):
self._send_cmd("M4", "target")
return
self._resume_report_ms = self.report_state[1]
self.lbl_st.setText("查询目标: 先暂停上报…")
self._send_cmd("M2S0", "report", ("OFF", None), then=self._send_m4_after_pause)
def _send_m4_after_pause(self):
if self.link.is_connected():
QTimer.singleShot(120, lambda: self._send_cmd("M4", "target"))
def _resume_report(self):
ms, self._resume_report_ms = self._resume_report_ms, None
if ms and self.link.is_connected():
self._send_cmd(f"M2S{ms}", "report", ("ON", ms))
# ================= 曲线交互 =================
def _on_window(self, idx):
sec = config.WINDOW_OPTIONS[idx][1]
config.WINDOW_SEC = sec
config.save_settings()
self.chart.set_window(sec)
# ---- Y 轴(温度轴)量程 ----
def _on_ymode(self, checked):
"""勾选=随数据自适应量程; 取消勾选=固定量程(右缘滑条拖拽)"""
manual = not checked
self.chart.set_y_mode("manual" if manual else "auto")
self.btn_ymode.setText("Y轴手动" if manual else "Y轴自动")
config.Y_MODE = "manual" if manual else "auto"
config.save_settings()
def _on_y_drag_engaged(self):
"""滑条首次拖拽: 图表内部已切手动, 这里同步按钮态并落设置"""
self.btn_ymode.blockSignals(True)
self.btn_ymode.setChecked(False)
self.btn_ymode.blockSignals(False)
self.btn_ymode.setText("Y轴手动")
config.Y_MODE = "manual"
config.save_settings()
def _on_y_range_changed(self, lo, hi):
"""滑条松手: 图表已量化 0.1℃, 直接范围记忆"""
config.Y_MIN, config.Y_MAX = lo, hi
config.save_settings()
def _toggle_pause(self):
if self._frozen_at is None:
self._frozen_at = time.time()
self.btn_pause.setText("恢复")
else:
self._frozen_at = None
self.btn_pause.setText("暂停")
self.chart.set_frozen(self._frozen_at)
def _clear_plot(self):
del self.samples_t[:]
del self.samples_v[:]
del self.faults[:]
self.chart.update()
self._update_stats()
# ================= CSV 记录 =================
def _toggle_csv(self):
if self._csv is not None:
self._csv.close()
self._csv = None
self.btn_csv.setText("开始记录")
self.lbl_csv.setText(f"已停止, 共 {self._csv_count} 行 → "
f"{os.path.basename(self._csv_path)}")
return
os.makedirs(config.LOG_DIR, exist_ok=True)
path = os.path.join(config.LOG_DIR, time.strftime("temp_%Y%m%d_%H%M%S.csv"))
try:
f = open(path, "w", encoding="utf-8-sig", newline="")
except OSError as e:
self.lbl_csv.setText(f"创建 CSV 失败: {e}")
return
self._csv = f
self._csv_path = path
self._csv_t0 = time.time()
self._csv_count = 0
f.write("时间,相对秒,温度C,目标C,加热,传感器\n")
self.btn_csv.setText("停止记录")
self.lbl_csv.setText(f"记录中 → {path}")
def _csv_write(self, t, temp_text, ok):
if self._csv is None:
return
target = f"{self.target_hist[-1][1]:g}" if self.target_hist else ""
heater = self.heater_state or ""
sensor = self.sensor_state or ("OK" if ok else "")
try:
self._csv.write(f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(t))},"
f"{t - self._csv_t0:.3f},{temp_text},{target},{heater},{sensor}\n")
self._csv.flush()
except OSError as e:
# 写失败(磁盘满/记录盘被拔): 自动停记录。本函数在 _tick 帧处理路径上,
# 未捕获异常会中止整段消费(与 v0.2.8 日志裁剪同款失效管道, 勿回退)
self._csv.close()
self._csv = None
self.btn_csv.setText("开始记录")
self.lbl_csv.setText(f"记录失败已停止: {e}")
self._log(f"⚠ CSV 记录失败已停止: {e}")
return
self._csv_count += 1
self.lbl_csv.setText(f"记录中({self._csv_count} 行) → "
f"{os.path.basename(self._csv_path)}")
def _open_log_dir(self):
os.makedirs(config.LOG_DIR, exist_ok=True)
os.startfile(config.LOG_DIR)
# ================= 日志 =================
def _log(self, line):
sb = self.txt_log.verticalScrollBar()
at_bottom = sb.value() >= sb.maximum() - 4
self.txt_log.appendPlainText(time.strftime("%H:%M:%S ") + line)
if at_bottom: # 仅原本贴底时跟随, 不抢用户上翻看历史
sb.setValue(sb.maximum())
def _clear_log(self):
self.txt_log.clear()
def _about(self):
QMessageBox.about(self, f"关于 {APP_NAME}",
f"{APP_NAME} v{APP_VERSION}\n热压模具温控 RS485 上位机\n"
f"协议: protocols.md v1.0 (M1~M4 / M888 / M999 / M9999)\n\n{REPO_URL}")
# ================= 设置持久化小项 =================
def _save_report_pref(self):
try:
config.REPORT_MS = int(self.ent_report.text().strip())
except ValueError:
return
config.save_settings()
# ================= 生命周期 =================
def closeEvent(self, event):
config.THEME = self.cbo_theme.currentText()
config.save_settings()
if self._csv is not None:
self._csv.close()
self._csv = None
try:
self.link.close()
except Exception:
pass
super().closeEvent(event)
def _single_instance():
if os.name != "nt":
return True
try:
ctypes.windll.kernel32.CreateMutexW(None, False, "hulk.FilmForgeGUI.single")
return ctypes.windll.kernel32.GetLastError() != 183
except Exception:
return True
def _icon_path():
"""窗口图标路径: 打包版取 _MEIPASS(build_exe.py --add-data 内嵌), 源码版取 logos/。"""
bundled = os.path.join(getattr(sys, "_MEIPASS", ""), "filmforge.ico")
if os.path.exists(bundled):
return bundled
return os.path.join(os.path.dirname(os.path.abspath(__file__)),
"logos", "filmforge.ico")
def run_gui(auto_close=0):
if not _single_instance():
app = QApplication.instance() or QApplication(sys.argv)
QMessageBox.information(None, "FilmForge 已在运行", "检测到另一个 FilmForge 实例正在运行。")
return
app = QApplication.instance() or QApplication(sys.argv)
app.setApplicationName(APP_NAME)
if os.name == "nt":
try: # 任务栏独立标识: 缺省时任务栏可能沿用宿主默认图标并错误分组
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(
"hulk.FilmForge.GUI")
except Exception:
pass
icon = _icon_path()
if os.path.exists(icon):
app.setWindowIcon(QIcon(icon)) # 主窗/弹窗/任务栏统一图标
setTheme(Theme.AUTO)
window = App(auto_close)
window.show()
app.exec()
Reference in New Issue View Git Blame Copy Permalink