diff --git a/AGENTS.md b/AGENTS.md index a20da09..087beb1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,7 +28,9 @@ - `LD/` —— 链接脚本（默认 xE 512K）； - `Inc/`+`Src/` —— 应用层，板级引脚/开关集中在 `Inc/board_config.h`； - `OpenOCD/` —— 工程内会话配置（GD-LINK/CMSIS-DAP + stm32f4x）； - - `doc/` —— 芯片文档（数据手册/用户手册/固件库指南/AN，已齐备）。 + - `doc/` —— 芯片文档（数据手册/用户手册/固件库指南/AN，已齐备）； + - `GUI/` —— 温控上位机（PySide6-Fluent，详见其 `README.md`，运行态 + `filmforge_gui.json`/`logs/`/`build/`/`release/` 由 `GUI/.gitignore` 挡）。 - **编译方法**： ```bash cmake --preset Debug && cmake --build Build/Debug @@ -70,7 +72,27 @@ - M3/M4 为**预留**：仅维护软件状态（ack+可回读），真实加热硬件执行体待接入 （`command.c` 的 `heater_apply()` 桩）；GUI 可先行全量联调协议 - 协议要点：全幂等（ack 丢可重发）、响应超时建议 300ms×3 次、温度一律 ASCII - - 上位机工具：用户另行开发 FilmForge 专用 GUI（协议按 protocols.md） +- **专用 GUI 上位机 v0.1.0 完成**（`GUI/`，参照 DLPilot 分层：协议模块纯标准库 + + PySide6-Fluent 界面 + 自检 + PyInstaller）： + - 文件角色：`filmforge_rs485.py`（帧/解析/串口线程 FF485）、`app_fluent.py` + （双栏仪表盘 + **自绘温度曲线**，未引第三方图表库）、`config.py`（设置记忆 + filmforge_gui.json）、`filmforge_gui.py`（入口+CLI `--port/--cmd/--seconds-report`）、 + `selftest_rs485.py`/`selftest_gui.py`（协议向量 / 离屏渲染+派发回归）、 + `build_exe.py` → `release/FilmForge.exe`+发布 zip、`publish_gitea.py` + （tag `FilmForge-GUI-vX.Y.Z`） + - 功能：实测温度曲线+目标温度阶梯线同图（闭环监视）、M2 上报帧直采 / 上报关闭时 + 上位机自动轮询 M1 补点、NTC OPEN/SHRT 断口+故障标记、时间窗/暂停/窗口统计、 + 目标(M4S 0~200)/加热(M3S1 确认)/上报(M2S 100~60000)控制、连接自动同步 + 版本/上报/加热/目标、M888/M9999 二次确认+复位后重查、收发日志、CSV 记录 + - **COM5 真机全链路验证通过**：M999=v41.0.0.1、M1/M2/M3/M4 查询、M2S500 上报 + 3s 收 6 帧、M2S0 停止；GUI 端到端（串口线程→队列→派发→曲线/状态芯片）实测通过 + - 关键教训（勿重蹈，详见 `GUI/README.md`）：①应答派发必须是 **FIFO 队列**（半双工 + 应答按发送序返回；单 pending 槽在连接序列 150ms 连发下被乱序击穿，实测 M2 的 + OFF 被 M3 槽吃掉、M4 目标值混入曲线）；②时间戳在串口线程读取点打（SerialMate + 教训）；③M4 目标应答路由窗口取条目生命周期 1.2s（适配器换向延迟实测可 >0.5s， + 上报开启时才收紧 0.5s）；④Fluent 控件样式走 setCustomStyleSheet（DLPilot 同款坑） + - 发布策略：GUI 产物（release/）暂不入库，本地 `python build_exe.py` 生成；发布走 + `publish_gitea.py` 上传 zip（tag `FilmForge-GUI-vX.Y.Z`，与固件 tag 区分） - **SerialMate 串口工具问题已定位**（见 `E:\Hulk-Coding\HW-Debug-Tools\SerialMate\ISSUE_rx_timestamp_and_lag.md`）： 时间戳在 GUI 消费时打导致失真 + 接收日志无限增长致卡顿；待该工具修复后 @@ -87,7 +109,9 @@ - [ ] **适配器更换**：换可靠的 USB-RS485 适配器（当前自动方向切换间歇丢响应）。 - [ ] **加热驱动**：加热功率器件驱动与控制（MOS/PWM/继电器方案待定）， 接入 `command.c` 的 `heater_apply()` 桩；随后控温闭环（PID）。 -- [ ] **专用 GUI 上位机**：用户开发中，协议按 `protocols.md`。 +- [x] **专用 GUI 上位机 v0.1.0**：`GUI/` 已完成并通过 COM5 真机全链路验证 + （协议按 `protocols.md`；遗留见其 README"已知事项"：应用图标、加热硬件 + 接入后的闭环联调、固件 M3/M4 硬件执行体实现）。 - [ ] **LED 引脚确认**：FilmForge 自制板 LED 引脚未定（当前沿用模板默认 PA15， 仅为构建占位），按原理图修改。 - [ ] **delay_10us 校准**：M4@168MHz 系数为估算值（/300000），需实测修正。 diff --git a/CMakeLists.txt b/CMakeLists.txt index e3bcfc1..7a2072d 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -39,6 +39,7 @@ set(TARGET_SRC Src/rs485.c Src/ntc.c Src/command.c + Src/pwr_hold.c # TODO: 待从 GD32E23x 工程移植后启用 # Src/i2c.c diff --git a/GUI/.gitignore b/GUI/.gitignore new file mode 100644 index 0000000..7a30795 --- /dev/null +++ b/GUI/.gitignore @@ -0,0 +1,6 @@ +# 生成物与运行态(不入库) +__pycache__/ +build/ +release/ +logs/ +filmforge_gui.json diff --git a/GUI/README.md b/GUI/README.md new file mode 100644 index 0000000..0db2983 --- /dev/null +++ b/GUI/README.md @@ -0,0 +1,74 @@ +# FilmForge GUI — 热压模具温控闭环监视台 + +FilmForge 热压模具温控板（GD32F405, RS485 数据口）的桌面上位机：实时温度曲线、 +目标温度/加热/周期上报控制、收发日志与 CSV 数据记录。协议权威定义见固件仓根目录 +[`../protocols.md`](../protocols.md)（v1.0，帧 hex 由脚本生成）。 + +## 功能 + +- **实时温度曲线**（自绘 QPainter 控件，无第三方图表依赖）：实测温度折线 + + 目标温度橙色阶梯线同图对照（闭环监视），NTC 开路/短路处曲线断开并打故障标记； + 时间窗 2/5/15/60 分钟/全部，暂停/恢复，窗口统计（最小/最大/平均）。 +- **温度数据双通道**：固件 M2 周期上报帧直接入曲线；未开启上报时可由上位机按 + 500ms/1s/2s 自动轮询 M1 补点（两条路读同一全局最新温度）。 +- **控制面板**：目标温度 M4S（0~200 整数℃，滑条+输入框）、加热 M3S1/M3S0 + （预留功能，开启需确认）、周期上报 M2S（100~60000，10ms 步进）、预设/自定义命令。 +- **状态联动**：连接后自动查询版本(M999)/上报(M2)/加热(M3)/目标(M4)并同步到状态芯片； + 软件复位(M888)/进 OTA(M9999) 二次确认，复位后 2 秒自动重查状态。 +- **收发日志**：译码 + 原始 hex，500 行上限防卡顿。 +- **CSV 记录**：逐行写入 `logs/temp_*.csv`（时间/相对秒/温度/目标/加热/传感器）。 + +## 快速开始 + +源码运行（Python 3.12+，Windows）： + +```powershell +py -m pip install PySide6 PySide6-Fluent-Widgets pyserial +python filmforge_gui.py +``` + +选择串口（板上 RS485 数据口，本机通常 COM5）→ 连接，115200-8-N-1。 + +命令行模式（自动化/无 GUI 联调）： + +```powershell +python filmforge_gui.py --port COM5 --cmd M999 --cmd M1 # 逐条命令并打印译码响应 +python filmforge_gui.py --port COM5 --cmd M2S500 --seconds-report 3 # 收 3s 上报帧 +python filmforge_gui.py --seconds 5 # GUI 冒烟(自动关闭) +``` + +## 打包 / 自检 / 发布 + +```powershell +python selftest_rs485.py # 协议回归(对照 protocols.md 帧向量, 无硬件) +python selftest_gui.py # GUI 离屏回归(曲线渲染/帧派发/主题/CSV) +python build_exe.py # → release/FilmForge.exe + 发布 zip +python publish_gitea.py release/FilmForge-GUI-v<版本>-win64.zip --replace --insecure +``` + +## 实现要点（改代码前必读） + +- 分层：`filmforge_rs485.py`（协议+串口线程，纯标准库+可选 pyserial）/ + `app_fluent.py`（PySide6-Fluent 呈现层）/ `config.py`（运行目录+设置记忆）/ + `filmforge_gui.py`（入口+CLI）。协议模块不依赖 GUI。 +- **应答派发是 FIFO 队列**（`app_fluent._pending`）：半双工上固件严格按收到顺序应答， + 连接序列(150ms 间隔连发 4 条)与自动轮询并存，单 pending 槽会被乱序击穿——实测踩坑： + M2 的 OFF 被 M3 槽吃掉、M4 的目标值混进温度曲线。应答形态与队首期望匹配才消费， + 不匹配记杂散，1.2s 超时自清。温度帧二义性（M1/M4/上报同为 ASCII 温度）靠 + 队首期望 + 生命周期窗口拆分（上报开启时收紧到 0.5s）。 +- **时间戳在串口线程读取点打**（`FF485.on_frame(fr, t_rx)`），不晚于 GUI 轮询消费—— + 避免重蹈 SerialMate 的时间戳失真问题（见 `E:\Hulk-Coding\HW-Debug-Tools\SerialMate`）。 +- M4 目标应答的路由窗口 = 条目生命周期（1.2s）：USB-RS485 适配器换向偶发延迟 + 实测可超 0.5s；上报开启时周期帧密集，才收紧到 0.5s 新鲜窗口。 +- Fluent 控件自定义样式必须走 `setCustomStyleSheet`（覆盖 hover/pressed/disabled）， + 直接 `setStyleSheet` 会在切主题时被库重刷（DLPilot 同款坑）；标题字号走 QFont。 +- 图表刷新双节流：数据到达即重绘（≤10Hz）+ 0.5s 时间轴推进；“全部”窗口超 4000 点 + 按步进抽稀（NaN 断口保留）。 + +## 已知事项 + +- 当前 USB-RS485 适配器自动换向偶发丢响应（固件侧已立“适配器更换”TODO）： + 表现为个别查询 1.2s 超时，队列自动清理后轮询自然重试，曲线无碍。 +- 加热(M3)/目标温度(M4)为固件预留：仅软件状态（ack+可回读），硬件执行体接入后 + 协议不变，本工具无需改动即可联调真实控温闭环。 +- 打包暂无应用图标（`logos/filmforge.ico` 缺位，build_exe.py 自动跳过）。 diff --git a/GUI/app_fluent.py b/GUI/app_fluent.py new file mode 100644 index 0000000..e164b0b --- /dev/null +++ b/GUI/app_fluent.py @@ -0,0 +1,1237 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""FilmForge Fluent UI — 热压模具温控闭环监视台。 + +协议与串口模块(filmforge_rs485)保持不变, 本文件只做呈现层: +PySide6-Fluent-Widgets 卡片式双栏仪表盘 + 自绘滚动温度曲线(无第三方图表依赖)。 +闭环监视 = 实测温度曲线 + 目标温度阶梯线同图 + 加热/上报/传感器状态联动; +设备未开启 M2 上报时可由上位机周期轮询 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 +from PySide6.QtGui import QColor, QFont, 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.1.0" +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 # 应答新鲜度: 该窗口内的温度帧优先按“查询应答”派发 + +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) +DANGER_QSS = ("PushButton, PushButton:hover, PushButton:pressed, PushButton:disabled " + "{ background-color: #D9534F; color: white; border: none; }") + + +class TemperatureChart(QWidget): + """自绘滚动温度曲线(QWidget + QPainter, 不引第三方图表库)。 + + 数据由 App 持有, 经 set_source() 传入引用; 视窗/冻结/配色独立设置。 + 曲线语义: 实测=折线(NTC 故障处 NaN 断开+底部红标), 目标=阶梯横线(橙虚线)。 + """ + + 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._has_data = False + + # ---- 数据/状态注入 ---- + def set_source(self, samples_t, samples_v, target_hist, faults): + self._samples_t, self._samples_v = samples_t, samples_v + self._target_hist, self._faults = target_hist, faults + self._has_data = bool(samples_t) or bool(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 _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) + t_out, v_out = ts[lo:], vs[lo:] + 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)...] 持有至下一变更时刻, 展开成阶梯点并延伸到右缘""" + 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 + b = min(tn, t1) + if not pts: + pts.append((max(ti, t0), vi)) + pts.append((b, vi)) + if pts and pts[-1][0] < t1: + pts.append((t1, pts[-1][1])) + return pts + + # ---- 绘制 ---- + def paintEvent(self, _event): + p = QPainter(self) + c = self._colors + p.fillRect(self.rect(), QColor(c["bg"])) + area = QRectF(self.rect()).adjusted(52, 10, -12, -26) + t0, t1 = self._view() + vis_t, vis_v = self._visible(t0, t1) + tgt = self._target_points(t0, t1) + ymin, ymax = self._y_range(vis_v, tgt) + 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) + self._draw_target(p, tgt, sx, sy) + self._draw_measured(p, vis_t, vis_v, sx, sy) + self._draw_faults(p, area, t0, t1) + self._draw_legend(p) + if not self._has_data: + p.setPen(QColor(c["text"])) + p.drawText(self.rect(), Qt.AlignCenter, + "暂无数据 — 连接设备后查询温度(M1)或开启上报(M2S)") + + @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) + + 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 not None: + x, y = sx(last_t), sy(last_v) + 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) + p.drawText(QPointF(min(x + 6, self.width() - 54), max(y - 6, 14)), + 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): + c = self._colors + font = self.font() + font.setPointSizeF(8.5) + p.setFont(font) + x = self.width() - 150 + 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} — 热压模具温控监视台") + 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'}] + self._last_poll = 0.0 + self._frozen_at = None + 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, root) + split.setChildrenCollapsible(False) + split.addWidget(self._scroll_column(self._build_left)) + split.addWidget(self._scroll_column(self._build_right)) + split.setStretchFactor(0, 3) + split.setStretchFactor(1, 2) + outer.addWidget(split, 1) + self.setCentralWidget(root) + + def _scroll_column(self, builder): + host = QWidget() + layout = QVBoxLayout(host) + layout.setContentsMargins(2, 2, 8, 2) + layout.setSpacing(10) + builder(layout) + layout.addStretch(1) + scroll = QScrollArea() + scroll.setWidgetResizable(True) + scroll.setFrameShape(QScrollArea.NoFrame) + scroll.setWidget(host) + return scroll + + # ---------------- 左列: 监视 ---------------- + def _build_left(self, 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 + row = QHBoxLayout() + self.btn_pause = self._button("暂停", self._toggle_pause) + self.btn_clear = self._button("清空", self._clear_plot) + row.addWidget(self.btn_pause) + row.addWidget(self.btn_clear) + card, body = self._card("实时温度曲线", trailing=win_combo) + body.addLayout(row) + self.chart = TemperatureChart(card) + self.chart.set_source(self.samples_t, self.samples_v, self.target_hist, self.faults) + 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): + 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) + + card, body = self._card("温度控制 · 目标 / 加热（固件预留）") + 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, 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("查询", lambda: self._query("M4", "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, "当前固件加热/目标仅维护软件状态(ack+可回读), 硬件执行体" + "待接入; 接入后协议不变, 本页即可直接联调真实控温。")) + layout.addWidget(card) + + card, body = self._card("周期上报 · M2S") + 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) + row = QHBoxLayout() + row.addWidget(BodyLabel("自动轮询 M1", card)) + self.cbo_poll = ComboBox(card) + for name, _ms in config.POLL_OPTIONS: + self.cbo_poll.addItem(name) + row.addWidget(self.cbo_poll) + row.addStretch(1) + body.addLayout(row) + body.addWidget(self._note(card, "上报开启时曲线走固件周期帧; 未开启时可由上位机按所选" + "周期自动查询 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) + body.addWidget(self.txt_log) + self._log(f"帧格式: D5 01 LEN <命令> CRC / B5 TYPE LEN CRC; " + f"响应超时建议 {ff.REPLY_TIMEOUT * 1000:.0f}ms 重发") + layout.addWidget(card) + + 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"]) + idx = next((i for i, (_n, ms) in enumerate(config.POLL_OPTIONS) + if ms == config.POLL_MS), 2) + self.cbo_poll.setCurrentIndex(idx) # 先设初值再连信号 + self.cbo_poll.currentIndexChanged.connect(self._on_poll_changed) + 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) + + # ================= 主题 ================= + 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 self.btn_conn is not None: + self.btn_conn.setText("断开" if connected else "连接") + if connected: + self._connect_seq() + else: + self._pending.clear() + + # 应答超时: 按发送顺序逐个清理队首过期条目(监视台不自动重发, + # 命令全幂等, 交给用户/自动轮询自然重试) + while self._pending and now - self._pending[0]["t"] > PENDING_TIMEOUT: + entry = self._pending.popleft() + self._log(f'⏱ {entry["cmd"]} 无响应(>{PENDING_TIMEOUT:.1f}s)') + + # 自动轮询 M1: 仅在连接、未开启上报、无未决命令时补点 + if (connected and config.POLL_MS > 0 + and not (self.report_state and self.report_state[0] == "ON") + and not self._pending + and now - self._last_poll >= config.POLL_MS / 1000.0): + self._last_poll = now + self._send_cmd("M1", "temp") + + # 曲线时间轴推进 + 窗口统计(节流 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) + if msg.startswith("已连接"): + 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) + return + elif head and head["kind"] == "temp": + self._pending.popleft() + self._add_sample(t_rx, val) + return + + # 非数值成功载荷: ok / ON / OFF / ON / v* + # 应答形态必须与队首期望匹配才消费, 不匹配记杂散(条目超时自清) + head = self._pending[0] if self._pending else None + if head is None: + return + pl = fr.payload.strip() + kind = head["kind"] + if pl == "ok" and (head.get("apply") or kind == "ack"): + self._pending.popleft() + self._apply_ack(head, t_rx) + 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"加热已{'开启' 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") + + # ================= 数据通路 ================= + 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)) + 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): + """带应答跟踪发送: kind 决定应答派发, apply_plan 为 ok 后的本地状态计划。 + 条目入 FIFO(半双工应答按发送序返回); 队列超长只留最新 8 条防失控""" + if not self.link.is_connected(): + self.lbl_st.setText("未连接串口") + return False + self._pending.append({"cmd": cmd, "kind": kind, "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("开启加热", + "当前固件加热为预留功能(仅软件状态, 硬件执行体待接入)。\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") + else: + kind = {"M1": "temp", "M2": "report", "M3": "heater", + "M4": "target", "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): + if self.link.is_connected(): + 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.link.open(port, baud) + + def _connect_seq_if_connected(self): + if self.link.is_connected(): + self._connect_seq() + + def _connect_seq(self): + """连接/复位后状态同步: 版本→上报→加热→目标, 150ms 间隔规避半双工碰撞""" + self._pending.clear() + self._last_poll = time.time() # 序列期间抑制自动轮询, 防止抢占 M999 位 + seq = (("M999", "version"), ("M2", "report"), ("M3", "heater"), ("M4", "target")) + for i, (cmd, kind) in enumerate(seq): + QTimer.singleShot(i * 150, lambda c=cmd, k=kind: self._query(c, k)) + + # ================= 曲线交互 ================= + def _on_window(self, idx): + sec = config.WINDOW_OPTIONS[idx][1] + config.WINDOW_SEC = sec + config.save_settings() + self.chart.set_window(sec) + + 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 "") + 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() + 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): + self.txt_log.appendPlainText(time.strftime("%H:%M:%S ") + line) + if self.txt_log.blockCount() > 500: + cursor = self.txt_log.textCursor() + cursor.movePosition(cursor.Start) + cursor.select(cursor.BlockUnderCursor) + cursor.removeSelectedText() + cursor.deleteChar() + sb = self.txt_log.verticalScrollBar() + 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 _on_poll_changed(self, idx): + config.POLL_MS = config.POLL_OPTIONS[idx][1] + config.save_settings() + + 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 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) + setTheme(Theme.AUTO) + window = App(auto_close) + window.show() + app.exec() diff --git a/GUI/build_exe.py b/GUI/build_exe.py new file mode 100644 index 0000000..a81a8cf --- /dev/null +++ b/GUI/build_exe.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +把 filmforge_gui.py 打包成单文件 exe(工具名: FilmForge) +================================ +使用 PyInstaller --onefile --noconsole: Python 解释器、PySide6 和 +PySide6-Fluent-Widgets 全部打进一个 exe, 目标机器无需安装依赖。 + +用法: + python build_exe.py 自动安装 PyInstaller(如缺)并打包 + 生成发布 zip + +产物: + release/FilmForge.exe 完整可直接运行的发布文件夹成员 + release/FilmForge-GUI-v<版本>-win64.zip exe + README 的发布包(供 publish_gitea.py 上传) +首次运行 onefile exe 会先解压到临时目录, 启动比脚本慢一两秒, 属正常现象。 +""" + +import os +import re +import subprocess +import sys +import zipfile + +# 本脚本位于固件仓 GUI/, 打包产物输出到 GUI/release/ +SRC = os.path.dirname(os.path.abspath(__file__)) +DIST = os.path.join(SRC, 'release') +WORK = os.path.join(SRC, 'build') +ENTRY = os.path.join(SRC, 'filmforge_gui.py') +README = os.path.join(SRC, 'README.md') +ICO = os.path.join(SRC, 'logos', 'filmforge.ico') + + +def app_version(): + """从 app_fluent.py 源码读 APP_VERSION(避免构建机导入 PySide6)""" + with open(os.path.join(SRC, 'app_fluent.py'), encoding='utf-8') as f: + m = re.search(r'APP_VERSION\s*=\s*"([^"]+)"', f.read()) + return m.group(1) if m else '0.0.0' + + +def ensure_pyinstaller(): + try: + import PyInstaller # noqa: F401 + return + except ImportError: + pass + print('PyInstaller 未安装, 正在安装...') + subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'pyinstaller']) + + +def extra_collect_args(): + """Fluent Widgets 的主题资源必须收集; PyInstaller 自带 Qt hooks 会处理 PySide6。""" + try: + import qfluentwidgets # noqa: F401 + except ImportError: + print('警告: 未安装 PySide6-Fluent-Widgets, 请先执行 pip install PySide6-Fluent-Widgets') + return [] + return ['--collect-all', 'qfluentwidgets'] + + +def qt_icu_args(): + """Qt6Core imports the unversioned Windows ICU API. PyInstaller can + otherwise pick Poppler's ICU 78 from PATH, whose suffixed exports make + QtCore.pyd fail with WinError 127 on startup.""" + if os.name == 'nt' and os.path.exists(r'C:\Windows\System32\icuuc.dll'): + return ['--add-binary', r'C:\Windows\System32\icuuc.dll;.'] + return [] + + +def remove_stale(exe_path): + """删除旧 exe; 若被占用(上次验证的进程还在), 先结束同名进程再重试""" + if not os.path.exists(exe_path): + return + try: + os.remove(exe_path) + return + except PermissionError: + pass + import time + print(f'{exe_path} 被占用, 结束残留进程后重试...') + subprocess.run(['taskkill', '/F', '/IM', os.path.basename(exe_path)], + capture_output=True) + time.sleep(1.0) + os.remove(exe_path) + + +def build(): + remove_stale(os.path.join(DIST, 'FilmForge.exe')) + cmd = [sys.executable, '-m', 'PyInstaller', + '--onefile', '--noconsole', '--clean', '--name', 'FilmForge', + '--distpath', DIST, '--workpath', WORK, '--specpath', WORK, + ] + extra_collect_args() + qt_icu_args() + [ENTRY] + if os.path.exists(ICO): + # exe 文件图标 + 运行时窗口图标资源(从 _MEIPASS 读取) + cmd += ['--icon', ICO, '--add-data', f'{ICO};.'] + else: + print('提示: 未找到 logos/filmforge.ico, 本次打包不带图标') + print('\n>>', ' '.join(cmd)) + subprocess.check_call(cmd) + out = os.path.join(DIST, 'FilmForge.exe') + print(f'完成: {out} ({os.path.getsize(out) / 1024 / 1024:.1f} MB)') + return out + + +def make_zip(exe_path): + """exe + README 打成发布 zip(命名供 publish_gitea.py 从文件名推 tag)""" + version = app_version() + zip_path = os.path.join(DIST, f'FilmForge-GUI-v{version}-win64.zip') + with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as z: + z.write(exe_path, 'FilmForge.exe') + if os.path.exists(README): + z.write(README, 'README.md') + print(f'发布包: {zip_path} ({os.path.getsize(zip_path) / 1024 / 1024:.1f} MB)') + return zip_path + + +def main(): + ensure_pyinstaller() + make_zip(build()) + + +if __name__ == '__main__': + main() diff --git a/GUI/config.py b/GUI/config.py new file mode 100644 index 0000000..39e700e --- /dev/null +++ b/GUI/config.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +工具运行配置: 运行目录定位 / filmforge_gui.json(串口记忆 + 监视设置) +================================================================ +PyInstaller onefile 打包后 __file__ 指向临时解压目录, +配置必须落在 exe 旁边, 故 frozen 时改用 sys.executable 定位。 +""" + +import json +import os +import sys + +import filmforge_rs485 + +if getattr(sys, 'frozen', False): + BASE_DIR = os.path.dirname(os.path.abspath(sys.executable)) +else: + BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +SETTINGS_FILE = os.path.join(BASE_DIR, 'filmforge_gui.json') +LOG_DIR = os.path.join(BASE_DIR, 'logs') + +# 曲线时间窗候选(秒): 下拉框文案 -> 秒 +WINDOW_OPTIONS = (('2 分钟', 120), ('5 分钟', 300), ('15 分钟', 900), + ('60 分钟', 3600), ('全部', 0)) +DEFAULT_WINDOW_SEC = 300 +MAX_SAMPLES = 200000 # 曲线内存上限(点), 超出丢最旧; 10Hz 下约 5.5 小时 + +# 自动轮询 M1 候选(毫秒): 仅在设备未开启 M2 上报时生效 +POLL_OPTIONS = (('关闭', 0), ('500 ms', 500), ('1 s', 1000), ('2 s', 2000)) +DEFAULT_POLL_MS = 1000 + +# 串口记忆(协议固定 115200-8-N-1, 仍允许改以便调试) +SERIAL_CFG = {'port': '', 'baud': filmforge_rs485.DEFAULT_BAUD, + 'bytesize': 8, 'parity': 'N', 'stopbits': 1} +REPORT_MS = 500 # 上报间隔输入框默认值 +POLL_MS = DEFAULT_POLL_MS # 自动轮询 M1 周期(0=关) +POLL_ENABLED = True +WINDOW_SEC = DEFAULT_WINDOW_SEC +THEME = 'System' + + +def load_settings(): + """读 filmforge_gui.json(缺失/损坏按默认值), 更新模块级默认字典""" + try: + with open(SETTINGS_FILE, encoding='utf-8') as f: + data = json.load(f) + except Exception: + return + if not isinstance(data, dict): + return + ser = data.get('serial') + if isinstance(ser, dict): + SERIAL_CFG['port'] = str(ser.get('port') or '') + try: + SERIAL_CFG['baud'] = int(ser.get('baud') or filmforge_rs485.DEFAULT_BAUD) + except (TypeError, ValueError): + pass + try: + SERIAL_CFG['bytesize'] = int(ser.get('bytesize') or 8) + except (TypeError, ValueError): + pass + SERIAL_CFG['parity'] = str(ser.get('parity') or 'N').upper()[:1] + try: + SERIAL_CFG['stopbits'] = int(ser.get('stopbits') or 1) + except (TypeError, ValueError): + pass + for key, target in (('report_ms', 'REPORT_MS'), ('poll_ms', 'POLL_MS'), + ('window_sec', 'WINDOW_SEC')): + try: + globals()[target] = int(data.get(key)) + except (TypeError, ValueError): + pass + global POLL_ENABLED, THEME + POLL_ENABLED = bool(data.get('poll_enabled', POLL_ENABLED)) + theme = str(data.get('theme') or '') + if theme in ('Dark', 'Light', 'System'): + THEME = theme + + +def save_settings(): + """写出全部设置(始终写全键)""" + data = { + 'serial': dict(SERIAL_CFG), + 'report_ms': REPORT_MS, + 'poll_ms': POLL_MS, + 'poll_enabled': POLL_ENABLED, + 'window_sec': WINDOW_SEC, + 'theme': THEME, + } + try: + with open(SETTINGS_FILE, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + except OSError: + pass # 只读目录(exe 旁)下静默放弃, 不影响运行 diff --git a/GUI/filmforge_gui.py b/GUI/filmforge_gui.py new file mode 100644 index 0000000..36d42d8 --- /dev/null +++ b/GUI/filmforge_gui.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +FilmForge 温控闭环监视台 — 入口与命令行模式 +================================================ +配合 FilmForge 热压模具温控板(RS485 数据口, 协议见 ../protocols.md)使用。 + +模块划分: + app_fluent.py PySide6-Fluent-Widgets 主界面 + 自绘温度曲线 + filmforge_rs485.py RS485 协议与串口链路(纯标准库 + 可选 pyserial) + config.py 运行目录与 filmforge_gui.json 设置 + +依赖: GUI 需 PySide6 + PySide6-Fluent-Widgets; 串口需 pyserial。CLI 仅标准库+pyserial。 + +命令行模式(自动化/联调用): + python filmforge_gui.py 启动 GUI + python filmforge_gui.py --seconds 5 GUI 5 秒自动关闭(冒烟测试) + python filmforge_gui.py --port COM5 --cmd M999 --cmd M1 + RS485 逐条发命令并打印译码响应 + python filmforge_gui.py --port COM5 --cmd M2S500 --seconds-report 5 + 开 500ms 上报并连续收 5 秒打印 +""" + +import argparse +import sys +import time + +import filmforge_rs485 as ff + + +def run_gui(auto_close=0): + """启动 PySide6-Fluent-Widgets 主界面(实现在 app_fluent.py)""" + import app_fluent + app_fluent.run_gui(auto_close) + + +def run_cli(port, baud, cmds, seconds_report=0): + """RS485 直连模式: 逐条发送命令并打印译码响应(自动化脚本用)。 + + 响应超时 2s(协议建议 300ms 重发, 此处不重发直接判失败)。 + 返回码: 任一命令非法/超时/响应非 F0 -> 1, 全部成功 -> 0。 + """ + if not ff.HAS_SERIAL: + print('错误: 未安装 pyserial, 请先执行 py -m pip install pyserial') + return 2 + try: + ser = ff.serial.Serial(port, baud, timeout=0.1) + except Exception as e: + print(f'打开串口失败: {port} @ {baud}: {e}') + return 1 + print(f'已连接 {port} @ {baud} (RS485 半双工, 单发单收)') + parser = ff.FrameParser() + rc = 0 + + def wait_reply(deadline_s): + got = None + deadline = time.time() + deadline_s + while got is None and time.time() < deadline: + data = ser.read(256) + if data: + got = next(iter(parser.feed(data)), None) + return got + + for cmd in cmds: + try: + frame = ff.build_frame(cmd) + except ValueError as e: + print(f'→ {cmd} 命令非法: {e}') + rc = 1 + continue + ser.write(frame) + print(f'→ {cmd} [{frame.hex(" ").upper()}]') + got = wait_reply(2.0) + if got is None: + print('← (2s 超时无响应)') + rc = 1 + else: + print(f'← {ff.describe(got)} [{got.raw.hex(" ").upper()}]') + if got.type_code != 0xF0: + rc = 1 + time.sleep(0.05) # 协议: 命令间隔 >=50ms + + if seconds_report > 0: # 连续收帧(验证 M2 周期上报) + print(f'── 连续接收 {seconds_report}s ──') + deadline = time.time() + seconds_report + n = 0 + while time.time() < deadline: + data = ser.read(256) + for fr in parser.feed(data): + n += 1 + print(f'← {ff.describe(fr)} [{fr.raw.hex(" ").upper()}]') + print(f'── 共 {n} 帧 ──') + ser.close() + return rc + + +def main(): + ap = argparse.ArgumentParser(description='FilmForge 热压模具温控闭环监视台') + ap.add_argument('--seconds', type=int, default=0, + help='GUI 自动退出秒数(冒烟测试)') + ap.add_argument('--port', default=None, help='RS485 串口(如 COM5), 配合 --cmd') + ap.add_argument('--baud', type=int, default=ff.DEFAULT_BAUD, + help='波特率(默认 115200)') + ap.add_argument('--cmd', action='append', default=None, dest='cmds', + metavar='CMD', help='RS485 发送命令(可多次), 如 --cmd M999 --cmd M1') + ap.add_argument('--seconds-report', type=int, default=0, metavar='SEC', + help='发送完命令后连续接收 SEC 秒(打印周期上报帧)') + args = ap.parse_args() + + if args.cmds or args.seconds_report: + if not args.port: + print('错误: --cmd/--seconds-report 需要配合 --port 指定串口(如 --port COM5)') + sys.exit(2) + sys.exit(run_cli(args.port, args.baud, args.cmds or [], args.seconds_report)) + run_gui(args.seconds) + + +if __name__ == '__main__': + main() diff --git a/GUI/filmforge_rs485.py b/GUI/filmforge_rs485.py new file mode 100644 index 0000000..2bfbb15 --- /dev/null +++ b/GUI/filmforge_rs485.py @@ -0,0 +1,369 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +FilmForge 热压模具 RS485 温控链路 +================================================================ +协议以 ../protocols.md 为准(固件仓根目录, 帧示例 hex 全部脚本生成): + 物理层 RS485 半双工 115200 8N1 无流控(PC 经 USB-RS485 适配器接板) + 命令帧 D5 01 LEN CRC 01=板类型, LEN=命令字节数 + 响应帧 B5 TYPE LEN CRC payload 一律 ASCII(ok / err / 24.5 / ON 500 ...) + CRC 帧内第 2 字节(板类型)起累加到最后一个 payload 字节, 取低 8 位 + 上报帧 M2S 开启后固件按周期主动发 B5 F0 <温度ASCII> CRC, 格式与 M1 应答相同 + +命令语义速记(全表见 protocols.md): + M1 查温度(OPEN=NTC 开路/SHRT=短路) M2 查上报状态 M2S 设间隔并开启(100~60000, + 10ms 步进) M2S0 停止 M3/M3S1/M3S0 加热(预留, 当前仅软件状态) M4/M4S 目标温度 + (0~200 整数) M888 软复位 M999 查版本 M9999 进 OTA(均会复位, GUI 需二次确认) +通信要点: 全部命令幂等(ack 丢了直接重发); 命令间隔 >=50ms; 响应超时 300ms 重发 x3。 + +仅标准库 + 可选 pyserial(缺失只影响串口功能, 其余照常)。 +""" + +import collections +import queue +import re +import threading +import time + +try: + import serial + import serial.tools.list_ports + HAS_SERIAL = True +except ImportError: + HAS_SERIAL = False + +HEADER_TX = 0xD5 # 命令帧头(Host -> Device) +HEADER_RX = 0xB5 # 响应帧头(Device -> Host) +BOARD_TYPE = 0x01 # 板类型字节 +FRAME_MIN, FRAME_MAX = 5, 32 # 整帧长度上下限(含帧头/CRC) +DEFAULT_BAUD = 115200 + +# --- 参数范围(protocols.md 命令表) --- +TEMP_MIN, TEMP_MAX = 0, 200 # 目标温度整数摄氏度 +REPORT_MIN, REPORT_MAX, REPORT_STEP = 100, 60000, 10 # 上报间隔 ms +REPLY_TIMEOUT = 0.3 # 响应超时建议值(秒), 供 GUI/CLI 参考 + +# --- 响应 TYPE 码语义(protocols.md "响应 TYPE 码"表) --- +TYPE_NAMES = { + 0xF0: '成功', + 0xF1: 'CRC 校验错', + 0xF3: '板类型错', + 0xF4: '命令格式错', + 0xFD: '参数越界', + 0xFE: '未知命令', +} + +SENSOR_OPEN = 'OPEN' # NTC 开路 +SENSOR_SHORT = 'SHRT' # NTC 短路 + +# GUI 预设命令表: (显示名, 命令, 说明), 顺序即下拉顺序 +PRESETS = ( + ('查温度', 'M1', '单次测温; OPEN/SHRT 为 NTC 故障态'), + ('查上报状态', 'M2', 'ON / OFF'), + ('开上报500ms', 'M2S500', '设上报间隔并开启(100~60000, 10ms 步进)'), + ('停上报', 'M2S0', '停止周期上报'), + ('查加热状态', 'M3', 'ON/OFF(预留, 当前仅软件状态)'), + ('开加热', 'M3S1', '预留: 硬件执行体待接入, 仅维护软件状态'), + ('关加热', 'M3S0', '预留'), + ('查目标温度', 'M4', 'ASCII 摄氏度(如 120.0)'), + ('设目标120C', 'M4S120', '0~200 整数摄氏度, 越界回 FD'), + ('查固件版本', 'M999', '如 v41.0.0.1'), + ('软件复位', 'M888', '回 ok 后复位(慎发)'), + ('进OTA', 'M9999', '复位进 Bootloader(慎发)'), +) + +# 会复位设备的命令: 任何入口发送前都须二次确认(protocols.md 通信注意事项 3) +RESET_COMMANDS = ('M888', 'M9999') + +Frame = collections.namedtuple('Frame', 'type_code payload raw') + + +def sum_crc(data): + """帧内第 2 字节(板类型)起累加到末 payload 字节, 取低 8 位(不含包头与 CRC)""" + return sum(data) & 0xFF + + +def build_frame(cmd): + """ASCII 命令字符串 -> 命令帧 bytes; 空命令/超长/非 ASCII 抛 ValueError""" + payload = cmd.strip().encode('ascii') + if not payload: + raise ValueError('空命令') + if len(payload) > FRAME_MAX - 4: + raise ValueError(f'命令过长(>{FRAME_MAX - 4} 字节, 帧上限 {FRAME_MAX})') + frame = bytes((HEADER_TX, BOARD_TYPE, len(payload))) + payload + return frame + bytes((sum_crc(frame[1:]),)) + + +class FrameParser: + """增量字节流 -> 完整响应帧: 帧头+LEN 定界, CRC 校验, 失配丢当前帧头重新同步。 + + 命令/响应/上报帧共用同一字节流; payload 理论上可能含 0xB5 造成同步偏移 + (伪帧头 CRC 必错会被跳过), 下一帧自动恢复, 对监视工具可接受。 + """ + + def __init__(self): + self._buf = bytearray() + + def feed(self, data): + """喂入新收到的字节, 返回本次解析出的完整 Frame 列表""" + self._buf += data + frames = [] + while True: + i = self._buf.find(HEADER_RX) + if i < 0: + self._buf.clear() + return frames + if i: # 丢弃帧头前的噪声 + del self._buf[:i] + if len(self._buf) < 3: # 帧头/TYPE/LEN 尚不完整 + return frames + length = self._buf[2] + if not 1 <= length <= FRAME_MAX - 4: + del self._buf[:1] # LEN 非法 -> 伪帧头, 前进 1 字节重同步 + continue + total = 4 + length + if len(self._buf) < total: # 整帧未收齐 + return frames + raw = bytes(self._buf[:total]) + if raw[-1] == sum_crc(raw[1:-1]): + frames.append(Frame(self._buf[1], + raw[3:-1].decode('ascii', 'replace').strip(), raw)) + del self._buf[:total] + else: # CRC 失配 -> 丢帧头重同步 + del self._buf[:1] + + +def _clean(s): + """控制字符 -> '·': 异常载荷按 ASCII 解出的 NUL 等会污染日志行, + 且 Windows 剪贴板遇 NUL 直接截断复制内容""" + return ''.join(c if ord(c) >= 0x20 and ord(c) != 0x7F else '·' for c in s) + + +def describe(fr): + """响应帧 -> 人读文本: TYPE 译码 + payload""" + pl = _clean(fr.payload) + if fr.type_code == 0xF0: + return pl or 'ok' + text = TYPE_NAMES.get(fr.type_code, f'TYPE=0x{fr.type_code:02X}') + if pl: + text += f' {pl}' + return text + + +# ------------------------------------------------------------ +# payload 译码助手(全部只认成功帧 F0 的 ASCII 载荷) +# ------------------------------------------------------------ +_FLOAT_RE = re.compile(r'^[+-]?\d+(?:\.\d+)?$') + + +def parse_temperature(payload): + """M1 应答/上报帧 payload -> (值, 故态)。解析成功返回 (float, None); + OPEN/SHRT 返回 (None, 'OPEN'|'SHRT'); 其余返回 (None, None) 表示非温度载荷。""" + p = payload.strip() + if p in (SENSOR_OPEN, SENSOR_SHORT): + return None, p + if _FLOAT_RE.match(p): + try: + v = float(p) + except ValueError: # pragma: no cover - 正则已挡 + return None, None + if v == v and abs(v) != float('inf'): # 拒绝 nan/inf + return v, None + return None, None + + +def parse_report_state(payload): + """M2 应答 'ON '/'OFF' -> ('ON', ms) / ('OFF', None) / (None, None)""" + p = payload.strip() + if p == 'OFF': + return 'OFF', None + m = re.match(r'^ON\s+(\d+)$', p) + if m: + return 'ON', int(m.group(1)) + return None, None + + +def parse_onoff(payload): + """M3 应答 'ON'/'OFF' -> 'ON'/'OFF'/None""" + p = payload.strip() + return p if p in ('ON', 'OFF') else None + + +def validate_report_interval(text): + """上报间隔输入 -> 合法 int(100~60000, 10ms 步进), 否则抛 ValueError""" + ms = int(str(text).strip()) + if not REPORT_MIN <= ms <= REPORT_MAX: + raise ValueError(f'上报间隔须在 {REPORT_MIN}~{REPORT_MAX} ms') + if ms % REPORT_STEP: + raise ValueError(f'上报间隔须为 {REPORT_STEP}ms 的整数倍') + return ms + + +def validate_target_temp(text): + """目标温度输入 -> 合法 int(0~200 整数摄氏度), 否则抛 ValueError""" + t = float(str(text).strip()) + if t != int(t): + raise ValueError('目标温度只支持整数摄氏度') + t = int(t) + if not TEMP_MIN <= t <= TEMP_MAX: + raise ValueError(f'目标温度须在 {TEMP_MIN}~{TEMP_MAX} ℃') + return t + + +class FF485: + """串口工作线程: 端口/波特率热切换 + 命令队列 + 回调(收发全在后台线程)。 + + on_event(str) 收发日志行(已译码, 附原始 hex) + on_status(str) 连接状态变化(已连接/断开/失败/设备拔出) + on_frame(frame, t_rx) 结构化帧, t_rx 为串口线程收到该帧时刻的 + time.time()(在读取点打戳, 不受 GUI 轮询延迟影响) + 回调在后台线程触发, GUI 侧需自行投递回主线程(queue + Qt 定时器轮询)。 + """ + + def __init__(self, on_event, on_status, on_frame=None): + self.on_event, self.on_status, self.on_frame = on_event, on_status, on_frame + self.q = queue.Queue() + self._target = None # 想态 (port, baud, bytesize, parity, stopbits) + self._open_key = None # 当前已打开串口的参数元组 + self.ser = None + self.parser = FrameParser() + self._lock = threading.Lock() + threading.Thread(target=self._run, daemon=True).start() + + @staticmethod + def list_ports_info(): + """枚举串口, 返回 [(设备名, 显示标签), ...]。 + + 标签 = "COMx 设备描述"(描述里的 "(COMx)" 尾巴去掉, COM 号已在最前), + 多口描述完全相同(同型号多口)时追加物理位置/序列号消歧。 + """ + if not HAS_SERIAL: + return [] + items = [] + for p in serial.tools.list_ports.comports(): + desc = re.sub(r'\s*\((COM\d+)\)\s*$', '', (p.description or '').strip(), + flags=re.I) + if desc.upper() == p.device.upper(): + desc = '' + label = f'{p.device} {desc}'.rstrip() + tail = p.location or p.serial_number or '' + items.append((p.device, label, tail)) + counts = collections.Counter(label for _d, label, _t in items) + out = [] + for dev, label, tail in items: + if counts[label] > 1 and tail: # 同标签多口: 附物理位置消歧 + label = f'{label} [{tail}]' + out.append((dev, label)) + return out + + def open(self, port, baud=DEFAULT_BAUD, bytesize=8, parity='N', stopbits=1): + if not HAS_SERIAL: + self.on_status('未安装 pyserial: py -m pip install pyserial') + return + with self._lock: + self._target = (port, int(baud), int(bytesize), + str(parity).upper(), int(stopbits)) + + def close(self): + with self._lock: + self._target = None + + def is_connected(self): + with self._lock: + return self.ser is not None + + def send(self, cmd): + """发送一条 ASCII 命令(自动组帧); 未连接/命令非法走 on_status 提示""" + with self._lock: + ser = self.ser + if ser is None: + self.on_status('未连接串口, 命令被丢弃') + return + try: + frame = build_frame(cmd) + except ValueError as e: + self.on_status(f'命令非法: {e}') + return + self.q.put((cmd, frame)) + + def _run(self): + # 串口参数映射(pyserial 常量): 数据位/校验/停止位 + bytesizes = {8: serial.EIGHTBITS, 7: serial.SEVENBITS, + 6: serial.SIXBITS, 5: serial.FIVEBITS} + parities = {'N': serial.PARITY_NONE, 'E': serial.PARITY_EVEN, + 'O': serial.PARITY_ODD} + stopbits = {1: serial.STOPBITS_ONE, 2: serial.STOPBITS_TWO} + while True: + with self._lock: + target = self._target + cur = self._open_key + if target != cur: + if self.ser is not None: + try: + self.ser.close() + except Exception: + pass + self.ser = None + self._open_key = None + self.on_status('已断开') + if target: + port, baud, dbsize, parity, sbits = target + try: + self.ser = serial.Serial( + port, baud, bytesize=bytesizes.get(dbsize, serial.EIGHTBITS), + parity=parities.get(parity, serial.PARITY_NONE), + stopbits=stopbits.get(sbits, serial.STOPBITS_ONE), + timeout=0.15, + write_timeout=1.0) # 写超时: 转换器停滞时 write 会 + # 永久阻塞并卡死线程, 必须设上限 + self.parser = FrameParser() + self._open_key = target + self.on_status(f'已连接 {port} @ {baud} ' + f'{dbsize}{parity}{sbits}') + except Exception as e: + self.ser = None + with self._lock: + self._target = None + blob = str(e) + repr(e) + if 'PermissionError' in blob or '拒绝访问' in blob \ + or 'Access is denied' in blob: + hint = '串口被占用（串口助手/另一实例?）' + elif 'FileNotFoundError' in blob or '找不到指定的' in blob \ + or 'No such file' in blob: + hint = '串口不存在，请刷新后重选' + else: + hint = '无法打开，详情见日志' + # 状态标签只放短提示(长异常串会撑大参数行把窗口拉长), + # 完整异常走 on_event 进收发日志 + self.on_status(f'连接失败: {hint}') + self.on_event(f'连接失败: {e!r}') + if self.ser is not None: + try: + item = None + try: + item = self.q.get_nowait() + except queue.Empty: + pass + if item is not None: + cmd, frame = item + self.ser.write(frame) + self.on_event(f'→ {cmd} [{frame.hex(" ").upper()}]') + data = self.ser.read(256) + if data: + t_rx = time.time() # 在读取点打时间戳(曲线/CSV 用) + for fr in self.parser.feed(data): + if self.on_frame is not None: + self.on_frame(fr, t_rx) + self.on_event(f'← {describe(fr)} ' + f'[{fr.raw.hex(" ").upper()}]') + except OSError: + try: + self.ser.close() + except Exception: + pass + self.ser = None + self._open_key = None # 必须同步清除, 否则同参数重连被当作已打开而跳过 + with self._lock: + self._target = None + self.on_status('串口丢失(设备拔出?), 请重新连接') + time.sleep(0.02) diff --git a/GUI/publish_gitea.py b/GUI/publish_gitea.py new file mode 100644 index 0000000..f4aff69 --- /dev/null +++ b/GUI/publish_gitea.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +创建/更新 Gitea Release 并上传发布 zip(纯标准库, 无需安装依赖) +================================================================ +用法(在固件仓任意目录执行, 自动从 origin 解析仓库): + python publish_gitea.py release/FilmForge-GUI-v0.1.0-win64.zip + python publish_gitea.py --notes-file release/notes-v0.1.0.md + python publish_gitea.py --tag FilmForge-GUI-v0.1.0 --replace + +行为: + - tag 与版本默认取 zip 文件名(FilmForge-GUI-vX.Y.Z-win64.zip → FilmForge-GUI-vX.Y.Z) + - release 不存在则创建(tag 不存在时 Gitea 在默认分支头自动建 tag); + 已存在则复用, 仅上传/更新资产 + - release 上已有同名资产时: 默认跳过, --replace 删旧再传 + - API token: --token 参数或环境变量 GITEA_TOKEN(需该仓库写权限) + - 本实例 python urllib 间歇性 SSL UNEXPECTED_EOF, 已内置连接级自动重试; + 证书校验失败时加 --insecure +""" + +import argparse +import json +import os +import re +import ssl +import subprocess +import sys +import time +import urllib.error +import urllib.request +import uuid + +DEFAULT_NOTES = """## FilmForge GUI v{version} + +FilmForge 热压模具温控板(RS485)上位机: 实时温度曲线(自绘, 无第三方图表依赖)、 +目标温度/加热/周期上报控制、收发日志、CSV 数据记录。协议见固件仓 `protocols.md`。 + +### 安装 +下载 `FilmForge-GUI-v{version}-win64.zip`, 解压后双击 `FilmForge.exe`(Windows 10/11 x64, 免安装)。 + +### 包内容 +- `FilmForge.exe` — 主程序 +- `README.md` — 完整使用说明 + +### 注意 +- 加热(M3)/目标温度(M4)为固件预留功能: 当前仅软件状态(ack+可回读), 硬件执行体接入后协议不变。 +- 软件复位(M888)/进 OTA(M9999) 会复位设备, 界面内已做二次确认。 +""" + + +def git_remote_info(): + """从 origin 远程地址解析 API base 与 owner/repo(ssh 端口不影响 https API)""" + url = subprocess.check_output(['git', 'remote', 'get-url', 'origin']).decode().strip() + m = re.match(r'(?:ssh://git@|https?://)(?:[^/@]+@)?([\w.-]+)(?::\d+)?[:/]([\w.-]+)/([\w.-]+?)(?:\.git)?$', url) + if not m: + sys.exit(f'无法从 origin 解析 Gitea 地址: {url}') + host, owner, repo = m.groups() + return f'https://{host}', owner, repo + + +class Api: + def __init__(self, base, owner, repo, token, insecure): + self.api = f'{base}/api/v1/repos/{owner}/{repo}' + self.headers = {'Authorization': f'token {token}'} + self.ctx = ssl._create_unverified_context() if insecure else None + + def request(self, method, path, data=None, headers=None, retries=4): + """发请求; 实例 TLS 间歇性抽风(SSL UNEXPECTED_EOF), 连接级错误自动重试""" + last = None + for attempt in range(retries): + req = urllib.request.Request(self.api + path, data=data, method=method, + headers={**self.headers, **(headers or {})}) + try: + with urllib.request.urlopen(req, context=self.ctx) as resp: + body = resp.read() + return resp.status, json.loads(body) if body else None + except urllib.error.HTTPError as e: + return e.code, json.loads(e.read() or b'{}') + except (urllib.error.URLError, ssl.SSLError, OSError) as e: + last = e + time.sleep(1.0 + attempt) + raise RuntimeError(f'Gitea 请求连续 {retries} 次失败: {last}') + + def get_json(self, path): + status, data = self.request('GET', path) + return data if status == 200 else None + + def upload_asset(self, release_id, filepath, replace): + name = os.path.basename(filepath) + status, rel = self.request('GET', f'/releases/{release_id}') + if status != 200: + sys.exit(f'读取 release {release_id} 失败: {status} {rel}') + # Gitea 版本差异: assets 可能是 {items:[...]} 或纯数组 + assets = rel.get('assets') or [] + if isinstance(assets, dict): + assets = assets.get('items', []) + for asset in assets: + if asset['name'] == name: + if not replace: + print(f'资产 {name} 已存在, 跳过(覆盖请加 --replace)') + return + self.request('DELETE', f"/releases/{release_id}/assets/{asset['id']}") + print(f'已删除旧资产 {name}') + boundary = uuid.uuid4().hex + payload = open(filepath, 'rb').read() + body = (f'--{boundary}\r\n' + f'Content-Disposition: form-data; name="attachment"; filename="{name}"\r\n' + f'Content-Type: application/octet-stream\r\n\r\n').encode() + payload + \ + f'\r\n--{boundary}--\r\n'.encode() + status, data = self.request( + 'POST', f'/releases/{release_id}/assets?name={name}', body, + {'Content-Type': f'multipart/form-data; boundary={boundary}'}) + if status != 201: + sys.exit(f'上传资产失败: {status} {data}') + print(f'已上传资产: {name} ({len(payload) / 1024 / 1024:.1f} MB)') + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument('zipfile_path', help='发布 zip(release/ 下)') + ap.add_argument('--tag', help='release 标签, 默认取 zip 文件名去掉 -win64.zip') + ap.add_argument('--notes-file', help='release 说明 markdown 文件, 默认用内置模板') + ap.add_argument('--token', default=os.environ.get('GITEA_TOKEN'), help='API token') + ap.add_argument('--replace', action='store_true', help='覆盖已存在的同名资产') + ap.add_argument('--insecure', action='store_true', help='跳过 TLS 证书校验(自签证书)') + args = ap.parse_args() + + if not os.path.exists(args.zipfile_path): + sys.exit(f'找不到 {args.zipfile_path}') + if not args.token: + sys.exit('缺少 API token: 传 --token 或设环境变量 GITEA_TOKEN') + + tag = args.tag or re.sub(r'-win64\.zip$', '', os.path.basename(args.zipfile_path)) + version = tag.removeprefix('FilmForge-GUI-v') + notes = open(args.notes_file, encoding='utf-8').read() if args.notes_file \ + else DEFAULT_NOTES.format(version=version) + + base, owner, repo = git_remote_info() + api = Api(base, owner, repo, args.token, args.insecure) + print(f'目标: {base}/{owner}/{repo} tag={tag}') + + rel = api.get_json(f'/releases/tags/{tag}') + if rel: + print(f'release 已存在(id={rel["id"]}), 复用') + else: + status, rel = api.request('POST', '/releases', json.dumps({ + 'tag_name': tag, + 'name': f'FilmForge GUI v{version}', + 'body': notes, + }).encode(), {'Content-Type': 'application/json'}) + if status != 201: + sys.exit(f'创建 release 失败: {status} {rel}') + print(f'release 已创建(id={rel["id"]})') + api.upload_asset(rel['id'], args.zipfile_path, args.replace) + print(f'完成: {rel.get("html_url", "") or f"{base}/{owner}/{repo}/releases/tag/{tag}"}') + + +if __name__ == '__main__': + main() diff --git a/GUI/selftest_gui.py b/GUI/selftest_gui.py new file mode 100644 index 0000000..ed4e6b3 --- /dev/null +++ b/GUI/selftest_gui.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""app_fluent 离屏回归(无硬件): 曲线渲染像素 + 帧派发状态机 + 主题往返 + CSV + +用法: python selftest_gui.py (全部通过退出码 0; 需 PySide6-Fluent-Widgets) +""" +import math +import os +from collections import deque +import sys +import time + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import filmforge_rs485 as ff +from PySide6.QtGui import QImage, QPainter +from PySide6.QtWidgets import QApplication + +fails = [] + + +def check(name, got, want): + ok = got == want + print(('PASS' if ok else 'FAIL'), name, '->', got, '' if ok else f'(期望 {want})') + if not ok: + fails.append(name) + + +def mk_frame(payload, type_code=0xF0): + body = bytes([type_code, len(payload)]) + payload.encode('ascii') + return ff.FrameParser().feed(bytes([0xB5]) + body + bytes([ff.sum_crc(body)]))[0] + + +app = QApplication([]) +import app_fluent + +window = app_fluent.App(0) +window.show() +app.processEvents() + +# 1) 空数据渲染不抛异常(占位文案) +img = QImage(window.chart.size(), QImage.Format_ARGB32) +window.chart.render(img) +check('空数据渲染', not img.isNull(), True) + +# 2) 注入合成数据后渲染: 实测曲线/目标线/故障标记像素均存在 +t0 = time.time() - 120 +for i in range(600): + window.samples_t.append(t0 + i * 0.2) + window.samples_v.append(24 + 5 * math.sin(i / 40) + i * 0.01) +window.target_hist.append((t0, 30.0)) +window.target_hist.append((t0 + 60, 35.0)) +window.faults.append((t0 + 30, ff.SENSOR_OPEN)) +window._set_theme('Dark') # 像素比对按 Dark 配色 +window.chart.update() +img = QImage(window.chart.size(), QImage.Format_ARGB32) +window.chart.render(img) + + +def count_pixels(img, rgb, tol=30): + n = 0 + for y in range(0, img.height(), 2): + for x in range(0, img.width(), 2): + c = img.pixelColor(x, y) + if (abs(c.red() - rgb[0]) < tol and abs(c.green() - rgb[1]) < tol + and abs(c.blue() - rgb[2]) < tol): + n += 1 + return n + + +measured = (0, 200, 215) # Dark 主题实测曲线色 #00c8d7 +target = (232, 131, 58) # 目标线 #E8833A +fault = (217, 83, 79) # 故障 #D9534F +n_meas, n_tgt, n_flt = count_pixels(img, measured), count_pixels(img, target), count_pixels(img, fault) +print(f'曲线像素={n_meas} 目标线像素={n_tgt} 故障像素={n_flt}') +check('实测曲线已绘制', n_meas > 50, True) +check('目标阶梯线已绘制', n_tgt > 5, True) +check('NTC 故障标记已绘制', n_flt > 0, True) + +# 3) 帧派发状态机(经 _handle_frame, 等价真实串口路径) +n0 = len(window.samples_t) +window._handle_frame(mk_frame('24.5'), time.time()) +check('上报帧入曲线', len(window.samples_t), n0 + 1) +check('大字读数更新', window.lbl_temp.text(), '24.5 ℃') + +window._pending = deque([{'cmd': 'M4', 'kind': 'target', 't': time.time(), 'apply': None}]) +window._handle_frame(mk_frame('120.0'), time.time()) +check('M4 应答拆到目标', window.target_hist[-1][1], 120.0) +check('目标 chip 更新', window.chip_target.text(), '目标: 120 ℃') +check('pending 已消费', len(window._pending), 0) + +window._pending = deque([{'cmd': 'M4', 'kind': 'target', 't': time.time() - 2, 'apply': None}]) +n1 = len(window.samples_t) +window._handle_frame(mk_frame('33.3'), time.time()) +check('过期 pending 不拦截温度帧', len(window.samples_t), n1 + 1) +check('目标不被污染', window.target_hist[-1][1], 120.0) +window._pending.clear() + +window._pending = deque([{'cmd': 'M2', 'kind': 'report', 't': time.time(), 'apply': None}]) +window._handle_frame(mk_frame('ON 500'), time.time()) +check('M2 状态解析', window.report_state, ('ON', 500)) +check('上报 chip 更新', window.chip_report.text(), '上报: ON 500ms') + +window._pending = deque([{'cmd': 'M3', 'kind': 'heater', 't': time.time(), 'apply': None}]) +window._handle_frame(mk_frame('OFF'), time.time()) +check('M3 状态解析', window.heater_state, 'OFF') + +window._pending = deque([{'cmd': 'M4S77', 'kind': 'ack', 't': time.time(), + 'apply': ('target', 77)}]) +window._handle_frame(mk_frame('ok'), time.time()) +check('M4S ack 落目标', window.target_hist[-1][1], 77.0) + +nf = len(window.faults) +window._handle_frame(mk_frame('OPEN'), time.time()) +check('OPEN 记故障', len(window.faults), nf + 1) +check('故障断口 NaN', math.isnan(window.samples_v[-1]), True) +check('传感器 chip 故障态', window.chip_sensor.text(), 'NTC: 开路(OPEN)') + +window._pending = deque([{'cmd': 'M999', 'kind': 'version', 't': time.time(), 'apply': None}]) +window._handle_frame(mk_frame('v41.0.0.1'), time.time()) +check('版本解析', window.firmware, 'v41.0.0.1') + +window._pending = deque([{'cmd': 'M4S300', 'kind': 'ack', 't': time.time(), + 'apply': ('target', 300)}]) +window._handle_frame(mk_frame('err', 0xFD), time.time()) +check('FD 错误消费 pending', len(window._pending), 0) +check('目标不被错误帧污染', window.target_hist[-1][1], 77.0) + +# 3b) 连接序列 FIFO 回归: M999/M2/M3/M4 同时在途, 应答按序到达 +window._pending.clear() +for cmd, kind in (('M999', 'version'), ('M2', 'report'), ('M3', 'heater'), ('M4', 'target')): + window._pending.append({'cmd': cmd, 'kind': kind, 't': time.time(), 'apply': None}) +window._handle_frame(mk_frame('v41.0.0.1'), time.time()) +window._handle_frame(mk_frame('OFF'), time.time()) # M2 应答, 不得被 heater 吃掉 +window._handle_frame(mk_frame('OFF'), time.time()) # M3 应答 +window._handle_frame(mk_frame('0.0'), time.time()) # M4 应答, 不得混入曲线 +check('FIFO 版本', window.firmware, 'v41.0.0.1') +check('FIFO 上报', window.report_state, ('OFF', None)) +check('FIFO 加热', window.heater_state, 'OFF') +check('FIFO 目标', window.target_hist[-1][1], 0.0) +check('FIFO 队列清空', len(window._pending), 0) + +# 3c) 杂散应答不消费队首: 队首等版本, 先来 OFF(杂散)再来 v41 +window._pending.clear() +window._pending.append({'cmd': 'M999', 'kind': 'version', 't': time.time(), 'apply': None}) +window._handle_frame(mk_frame('OFF'), time.time()) +check('杂散不消费队首', len(window._pending), 1) +window._handle_frame(mk_frame('v41.0.0.1'), time.time()) +check('杂散后正常消费', len(window._pending), 0) + +# 4) 主题往返后曲线配色随动 + 渲染不崩 +window._set_theme('Light') +check('Light 曲线配色', window.chart._colors['measured'].lower(), + app_fluent.CHART_COLORS_LIGHT['measured']) +window._set_theme('Dark') +check('Dark 曲线配色', window.chart._colors['measured'].lower(), + app_fluent.CHART_COLORS_DARK['measured']) +window._set_theme('Light') +window._set_theme('System') + +# 5) CSV 记录往返 +window._toggle_csv() +window._add_sample(time.time(), 25.1) +window._record_fault(time.time(), ff.SENSOR_SHORT) +window._toggle_csv() +with open(window._csv_path, encoding='utf-8-sig') as f: + lines = f.read().strip().splitlines() +check('CSV 表头+2 行', len(lines), 3) +check('CSV 故障行', lines[2].split(',')[2], 'SHRT') + +print('RESULT:', 'FAIL:' + ','.join(fails) if fails else 'ALL PASS') +sys.exit(1 if fails else 0) diff --git a/GUI/selftest_rs485.py b/GUI/selftest_rs485.py new file mode 100644 index 0000000..80c6140 --- /dev/null +++ b/GUI/selftest_rs485.py @@ -0,0 +1,128 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""filmforge_rs485 协议自检(无硬件): 对照 ../protocols.md 的权威帧向量做回归 + +用法: python selftest_rs485.py (全部通过退出码 0) +""" +import sys + +import filmforge_rs485 as ff + +fails = [] + + +def check(name, got, want): + ok = got == want + print(('PASS' if ok else 'FAIL'), name, '->', got, '' if ok else f'(期望 {want})') + if not ok: + fails.append(name) + + +# 1) 帧构造: protocols.md 命令表全部 12 条权威 hex 向量 +for cmd, hex_want in ( + ('M1', 'D5 01 02 4D 31 81'), + ('M2', 'D5 01 02 4D 32 82'), + ('M2S500', 'D5 01 06 4D 32 53 35 30 30 6E'), + ('M2S0', 'D5 01 04 4D 32 53 30 07'), + ('M3', 'D5 01 02 4D 33 83'), + ('M3S1', 'D5 01 04 4D 33 53 31 09'), + ('M3S0', 'D5 01 04 4D 33 53 30 08'), + ('M4', 'D5 01 02 4D 34 84'), + ('M4S120', 'D5 01 06 4D 34 53 31 32 30 6E'), + ('M888', 'D5 01 04 4D 38 38 38 FA'), + ('M999', 'D5 01 04 4D 39 39 39 FD'), + ('M9999', 'D5 01 05 4D 39 39 39 39 37'), +): + check(f'build_frame({cmd})', ff.build_frame(cmd).hex(' ').upper(), hex_want) + +# 2) 增量解析: ok 应答跨两个半包喂入 +p = ff.FrameParser() +r1 = p.feed(bytes.fromhex('B5 F0')) +r2 = p.feed(bytes.fromhex('02 6F 6B CC')) +check('split-feed frames', len(r1) + len(r2), 1) +fr = (r1 + r2)[0] +check('type_code', fr.type_code, 0xF0) +check('payload', fr.payload, 'ok') +check('describe(ok)', ff.describe(fr), 'ok') + +# 3) 温度上报帧(protocols.md 实例: 24.5C)与前导噪声 +body = bytes.fromhex('F0 04 32 34 2E 35') # F0 + "24.5" +good = bytes([0xB5]) + body + bytes([ff.sum_crc(body)]) +frames = ff.FrameParser().feed(b'\x01\x02' + good) +check('noise+report count', len(frames), 1) +check('report payload', frames[0].payload, '24.5') +check('parse_temperature', ff.parse_temperature(frames[0].payload), (24.5, None)) +check('describe(成功)', ff.describe(frames[0]), '24.5') + +# 4) 传感器故障态 OPEN/SHRT 译码 +for s in (ff.SENSOR_OPEN, ff.SENSOR_SHORT): + b = bytes.fromhex('F0 04') + s.encode() + f = ff.FrameParser().feed(bytes([0xB5]) + b + bytes([ff.sum_crc(b)])) + check(f'parse_temperature({s})', ff.parse_temperature(f[0].payload), (None, s)) + +# 5) 错误 TYPE 译码(FD 参数越界 / FE 未知命令) +for tc, name in ((0xFD, '参数越界'), (0xFE, '未知命令'), (0xF1, 'CRC 校验错'), + (0xF3, '板类型错'), (0xF4, '命令格式错')): + b = bytes([tc, 0x03]) + b'err' + f = ff.FrameParser().feed(bytes([0xB5]) + b + bytes([ff.sum_crc(b)])) + check(f'describe(TYPE={tc:02X})', ff.describe(f[0]), f'{name} err') + +# 6) CRC 坏帧被丢弃后继续同步到下一帧 +bad = bytearray(good) +bad[-1] ^= 0xFF +frames = ff.FrameParser().feed(bytes(bad) + good) +check('bad-crc then good', len(frames), 1) +check('recovered payload', frames[0].payload, '24.5') + +# 7) M2 状态与 M3 开关译码 +check("parse_report_state('ON 500')", ff.parse_report_state('ON 500'), ('ON', 500)) +check("parse_report_state('ON 60000')", ff.parse_report_state('ON 60000'), ('ON', 60000)) +check("parse_report_state('OFF')", ff.parse_report_state('OFF'), ('OFF', None)) +check("parse_report_state('ok')", ff.parse_report_state('ok'), (None, None)) +check("parse_onoff('ON')", ff.parse_onoff('ON'), 'ON') +check("parse_onoff('OFF')", ff.parse_onoff('OFF'), 'OFF') +check("parse_onoff('24.5')", ff.parse_onoff('24.5'), None) +check("parse_temperature('120.0')", ff.parse_temperature('120.0'), (120.0, None)) +check("parse_temperature('ok')", ff.parse_temperature('ok'), (None, None)) + +# 8) 参数校验: 上报间隔(100~60000, 10ms 步进)与目标温度(0~200 整数) +check('interval 500 ok', ff.validate_report_interval('500'), 500) +check('interval 60000 ok', ff.validate_report_interval('60000'), 60000) +for bad_v in ('99', '60010', '505', 'abc', '0'): + try: + ff.validate_report_interval(bad_v) + fails.append(f'interval {bad_v} not rejected') + except ValueError: + pass +check('temp 120 ok', ff.validate_target_temp('120'), 120) +check('temp 0 ok', ff.validate_target_temp('0'), 0) +check('temp 200.0 ok', ff.validate_target_temp('200.0'), 200) +for bad_v in ('-1', '201', '120.5', 'abc'): + try: + ff.validate_target_temp(bad_v) + fails.append(f'temp {bad_v} not rejected') + except ValueError: + pass +print('参数越界/非法输入拒绝: PASS') + +# 9) 非法输入拒绝 +for bad_cmd in ('', ' '): + try: + ff.build_frame(bad_cmd) + fails.append('empty-cmd-not-rejected') + except ValueError: + pass +try: + ff.build_frame('M' * 40) + fails.append('long-cmd-not-rejected') +except ValueError: + pass +print('空/超长命令拒绝: PASS') + +# 10) 预设命令全部可组帧 +for _name, cmd, _note in ff.PRESETS: + ff.build_frame(cmd) +print(f'预设 {len(ff.PRESETS)} 条全部可组帧: PASS') + +print('RESULT:', 'FAIL:' + ','.join(fails) if fails else 'ALL PASS') +sys.exit(1 if fails else 0) diff --git a/Inc/board_config.h b/Inc/board_config.h index 3b60a54..7f1d6a1 100644 --- a/Inc/board_config.h +++ b/Inc/board_config.h @@ -136,6 +136,15 @@ void usart1_irq_handler(void); /******************************************************************************/ +/* >>>>>>>>>>>>>>>>>>>>[POWER HOLD（PMOS 自锁保持脚）]<<<<<<<<<<<<<<<<<<<< */ +/* PA11：上电保持脚——main 入口即初始化为推挽输出高（先于一切外设初始化）， + 维持 PMOS 导通；复位期间由外部电路（按键/上拉）保持。 */ +#define PWR_HOLD_RCU RCU_GPIOA +#define PWR_HOLD_PORT GPIOA +#define PWR_HOLD_PIN GPIO_PIN_11 + +/******************************************************************************/ + #define DEBUG_UART USART1 #endif //BOARD_CONFIG_H diff --git a/Inc/pwr_hold.h b/Inc/pwr_hold.h new file mode 100644 index 0000000..2c05bb4 --- /dev/null +++ b/Inc/pwr_hold.h @@ -0,0 +1,26 @@ +#ifndef PWR_HOLD_H +#define PWR_HOLD_H + +#include "gd32f4xx.h" +#include "board_config.h" + +/** + * @brief 初始化电源保持脚（PA11，PMOS 自锁保持）。 + * @details 在 main 入口、一切外设初始化之前调用：推挽输出高，立即维持 + * PMOS 导通（手动上电后由 MCU 接替按键/外部电路保持供电）。 + * 引脚定义见 board_config.h 的 PWR_HOLD_*。 + */ +void pwr_hold_init(void); + +/** + * @brief 保持脚置高（PMOS 导通，维持供电）。 + */ +void pwr_hold_on(void); + +/** + * @brief 保持脚置低（预留软关机：PMOS 断开，设备自行掉电）。 + * @warning 调用后本系统失电，函数不会返回——仅用于软关机流程收尾。 + */ +void pwr_hold_off(void); + +#endif // PWR_HOLD_H diff --git a/Src/main.c b/Src/main.c index a55591c..7ab53a3 100644 --- a/Src/main.c +++ b/Src/main.c @@ -14,6 +14,7 @@ #include "command.h" #include "ntc.h" #include "led.h" +#include "pwr_hold.h" #include #include "board_config.h" @@ -25,6 +26,9 @@ */ int main(void) { + /* 电源保持脚最先初始化：上电立即拉高 PA11 维持 PMOS 导通（先于一切外设） */ + pwr_hold_init(); + led_init(); systick_config(); diff --git a/Src/pwr_hold.c b/Src/pwr_hold.c new file mode 100644 index 0000000..7c4f690 --- /dev/null +++ b/Src/pwr_hold.c @@ -0,0 +1,23 @@ +#include "pwr_hold.h" +#include "gd32f4xx_rcu.h" +#include "gd32f4xx_gpio.h" +#include "board_config.h" + +void pwr_hold_init(void) { + /* 使能 GPIO 时钟，推挽输出高（PMOS 导通，维持供电） */ + rcu_periph_clock_enable(PWR_HOLD_RCU); + gpio_mode_set(PWR_HOLD_PORT, GPIO_MODE_OUTPUT, GPIO_PUPD_NONE, PWR_HOLD_PIN); + gpio_output_options_set(PWR_HOLD_PORT, GPIO_OTYPE_PP, GPIO_OSPEED_50MHZ, PWR_HOLD_PIN); + gpio_bit_set(PWR_HOLD_PORT, PWR_HOLD_PIN); +} + +void pwr_hold_on(void) { + gpio_bit_set(PWR_HOLD_PORT, PWR_HOLD_PIN); +} + +void pwr_hold_off(void) { + /* 软关机：置低后系统失电 */ + gpio_bit_reset(PWR_HOLD_PORT, PWR_HOLD_PIN); + while (1) { + } +}