18 changed files with 1068 additions and 276 deletions
+76 -3
View File
@@ -79,6 +79,12 @@
- 实测 30℃ 闭环(NTC 贴 PTC 鳍片):30s 爬升、超调 1.4℃(7%)、60s 收敛中
- M3/M4 已从软件桩切到 heat 模块(协议帧不变)
- 协议要点:全幂等(ack 丢可重发)、响应超时建议 300ms×3 次、温度一律 ASCII
- **PID 抗积分饱和**(5b24b95):条件积分(back-calculation lite)抑制输出限幅
期间的积分累积。
- **50℃→40℃ 阶跃验证脚本**(`temp_test_com.py`,工作区修改未提交):M2S500
全曲线采样,输出超调/settle 时间/稳态误差/降温段统计(结果以实跑为准)。
- **电源保持 PA11 实装**(`pwr_hold.h/pwr_hold.c`):上电最先拉高 PA11 维持
PMOS 自锁供电(先于一切外设初始化);`pwr_hold_off()` 预留软关机(置低即失电)。
- **专用 GUI 上位机 v0.1.0 完成**(`GUI/`,参照 DLPilot 分层:协议模块纯标准库 +
PySide6-Fluent 界面 + 自检 + PyInstaller):
- 文件角色:`filmforge_rs485.py`(帧/解析/串口线程 FF485)、`app_fluent.py`
@@ -118,9 +124,76 @@
安全保护齐全;实测 30℃ 闭环通过:30s 爬升、超调 1.4℃、60s 收敛中)。
- [ ] **PID 参数整定**:初值 Kp=8/Ki=0.05/Kd=0(board_config.h 宏),按真实
工况温度(80~120℃)与负载重新整定,抑制 ~7% 超调。
- [x] **专用 GUI 上位机 v0.1.0**:`GUI/` 已完成并通过 COM5 真机全链路验证
(协议按 `protocols.md`;遗留见其 README"已知事项":应用图标、加热硬件
接入后的闭环联调、固件 M3/M4 硬件执行体实现)。
- [x] **专用 GUI 上位机 v0.2.0**:v0.1.0 已完成 COM5 真机全链路验证;v0.2.0 与
实装固件对齐——去除"固件预留/仅软件状态"旧表述(协议模块预设说明、温控
卡片标题/备注、开启加热确认弹窗改为真实控温警示、ack 提示),GUI README
与 publish_gitea 发布说明模板同步,protocols.md M3 表项去"预留";
监控逻辑简化:温度监控统一走 M2 周期上报(移除"M1 自动轮询"双通道及
poll 配置项,M1 仅手动单点查询),目标温度滑条范围 0~200→0~120(协议
上限 200 仍可手输)。**v0.2.1 真机回归修复**(COM5 实测暴露):①M4 查询
应答与周期上报帧同为 ASCII 浮点被误占(目标 chip 显示成实测值)→ 查目标
改应答驱动"暂停上报 M2S0→M4→自动恢复 M2S<ms>"链;②设定 ok 延迟 >1.2s
超时清队后状态 chip 与设备脱节(实测 M2S100 ok 迟到、chip 停 OFF)→ 8s
内晚到 ok 按发送时计划补偿应用;③断开按钮偶发不切换、断开后日志仍刷 →
连接按钮改意图驱动(点击即翻转、2s 未建立判丢失回退),已连接/已断开/
丢失全部入日志,连接同步序列每步发送前确认在线(修复序列整组被
"未连接串口"丢弃)。selftest 全过(新增 8 项回归用例);已打包冒烟并发布
Gitea Release `FilmForge-GUI-v0.2.1`(无应用图标——`logos/filmforge.ico`
仍缺位)。
- **GUI v0.2.2~v0.2.4(已修复待真机确认,未发布)**:
- v0.2.2:串口线程 `read(256)` 在持续帧流下攒满 256 字节才返回(win32 字节
间隔超时因帧间隔更短永不触发,100ms 上报 ≈2.5s 一批),曲线/命令/响应全部
3 秒一卡、一股脑一大坨——真机日志确认每簇 ≈25 帧、间隔恒 3s。已改
`read(in_waiting or 1)`,真机复测曲线实时、命令即发即回。
- v0.2.3:发送入口幂等去重——同命令未决时忽略重复发送(真机连点"停止"12 次
挤爆 8 条 pending 队列、"丢弃积压命令"刷屏),状态栏提示"应答中"。
- v0.2.4:目标阶梯线渲染 bug(v0.1.0 起潜伏)——`_target_points` 对最后一次
目标变更只画终点,从旧值斜拉直线到图表右缘(实测 57→41 被画成 2 分钟斜坡,
截图确认);改为纯阶梯(变更时刻垂直落变+水平延伸),数据本身无误。
- v0.2.5:排版调整——串口连接卡片从右列移到左列顶部,左列图表卡/右列日志卡
弹性拉伸填充,两列底边对齐。
- v0.2.6:对齐修复——双列改放同一滚动容器(整页统一滚动):原先两列各自包
QScrollArea 独立滚动,窗口化下底边错位(全屏正常);现两列高度锁定一致,
不足一屏各自拉伸到底、超出一屏整页滚动。
- v0.2.7:图标接入(logos/ 下 ico/png/svg 已就位)——exe 嵌入图标
(build_exe.py --icon,此前缺文件自动跳过)、窗口标题栏/任务栏/弹窗图标
(app+window 两级 setWindowIcon,_icon_path() 兼容源码态与 _MEIPASS 打包态)、
Windows 任务栏独立标识 AppUserModelID "hulk.FilmForgeGUI"。README"无应用
图标"遗留项关闭。
- v0.2.8:**慢性卡顿修复**——`_log()` 手工 cursor 裁剪(`cursor.Start`/
`BlockUnderCursor` 实例级枚举访问)在 PySide6 6.11 上必然抛 AttributeError,
日志超 500 行(100ms 上报 ≈50s)后每个 100ms tick 异常中止 `_tick` 后半段:
应答超时清理失效(pending 永不清 → "停止"点击被幂等去重永久拦截,实拍状态栏
"M2S0 应答中,已忽略重复发送")、RX 队列积压 ratchet(帧处理滞后 54ms→均值
1037ms/峰值 2701ms,日志/chip/曲线显示滞后数秒,"停止 5s 后才见命令发出"实为
日志显示积压,TX 本身 5~166ms 即出)、日志无限增长、CPU +30%。改原生
`setMaximumBlockCount(500)` 环形裁剪;selftest 新增"500 行封顶/封顶后可追加/
最新行在末尾"3 条回归。离屏 A/B 实测(GUI/_diag_latency_buggy.py,假固件
100ms 上报+定时点击):buggy 组 5.5 分钟异常 2241 次、修复组全程漂移 ≤15ms/
零异常/零积压。v0.2.7 exe(2026-10-10 打包)内嵌同版本 PySide6 6.11.2,
同带此 bug,需重新打包生效。
- v0.2.9:**整体审查修复 5 项**——①CSV 写失败(磁盘满/记录盘被拔)自动停记录
并提示,不再沿 _tick 帧处理路径抛 OSError 中止主循环(与 v0.2.8 日志裁剪同款
失效管道);②故障标记 faults 封顶 1000 条(曾是唯一无上限缓冲:NTC 抖动时
10 条/s 无限增长,且 _draw_faults 每 paint 全量扫描);③串口写超时单独捕获
SerialTimeoutException——只丢该条命令、链路保留,不再按 OSError 当"设备拔出"
关端口断链(适配器停滞曾迫使手动重连);④日志自动滚动仅在贴底时跟随(上报
运行时可上翻历史,不再被拽回底部);⑤冻结视图切片右缘同样二分(暂停时长不再
增加 render 成本)。selftest 新增 7 项回归(封顶/滚动跟随/CSV 失败自停/冻结
切片)共 59 项全过。整体审查报告(2026-10-12)其余 P3 遗留:selftest 未覆盖
_tick 主循环(超时清理/丢失判定)、"全部"窗口全量切片有 MAX_SAMPLES 兜底、
_last_set 单槽位罕见误补偿(8s 自愈)、build_exe remove_stale 无提示强杀
运行实例、M888 后固定 2s 重查可能早于固件就绪、_danger_reset 把去重忽略记成
"已下发"。协议建议的 300ms×3 自动重发维持不做(设计决策,用户拍板)。
- selftest 全过(累计新增 15 项回归用例,含窗口图标非空检查);exe 已本地
打包冒烟,Gitea 发布待用户确认后执行。
遗留见其 README"已知事项":USB-RS485 适配器更换。
- [x] **GUI 慢性卡顿修复(根因已定位并修复 v0.2.8,待真机确认)**:
根因/机理/实测数据见上方 v0.2.8 条目(`cursor.Start` 实例级枚举访问在
PySide6 6.11 必抛 → tick 异常中止 → 应答超时清理失效 + RX 积压 ratchet +
日志无限增长)。复现/验证脚本:`GUI/_diag_latency.py`(修复基线)/
`GUI/_diag_latency_buggy.py`(带 bug 对照,假固件 100ms 上报+定时点击)。
遗留:待真机长时间运行确认 + 重新打包 exe。
- [ ] **LED 引脚确认**:FilmForge 自制板 LED 引脚未定(当前沿用模板默认 PA15,
仅为构建占位),按原理图修改。
- [ ] **delay_10us 校准**:M4@168MHz 系数为估算值(/300000),需实测修正。
+16 -2
View File
@@ -88,11 +88,25 @@ target_link_options(${PROJECT_NAME} PRIVATE
# 容量宏(代码中如需按容量分支使用 GD32_FLASH_SIZE_KB)
target_compile_definitions(${PROJECT_NAME} PRIVATE GD32_FLASH_SIZE_KB=${GD32_FLASH_SIZE_KB})
# 生成 map 文件等
# 生成 map 文件等。`--no-warn-rwx-segments` 仅用于抑制新版本
# binutils 的 RWX 段警告,旧版(例如 xPack GCC 11.3 搭配 binutils 2.38)
# 不认识该选项,因此先探测链接器能力,避免因工具链版本差异导致链接失败。
target_link_options(${PROJECT_NAME} PRIVATE
-Wl,-Map=${CMAKE_RUNTIME_OUTPUT_DIRECTORY}/Application.map
-Wl,--no-warn-rwx-segments
)
execute_process(
COMMAND "${CMAKE_LINKER}" --help
OUTPUT_VARIABLE FILMFORGE_LINKER_HELP
ERROR_VARIABLE FILMFORGE_LINKER_HELP_ERROR
RESULT_VARIABLE FILMFORGE_LINKER_HELP_RESULT
OUTPUT_STRIP_TRAILING_WHITESPACE
)
if(FILMFORGE_LINKER_HELP_RESULT EQUAL 0 AND
FILMFORGE_LINKER_HELP MATCHES "no-warn-rwx-segments")
target_link_options(${PROJECT_NAME} PRIVATE -Wl,--no-warn-rwx-segments)
else()
message(STATUS "Linker does not support --no-warn-rwx-segments; skipping it")
endif()
# 链接SDK库
target_link_libraries(${PROJECT_NAME} PRIVATE CMSIS)
+41 -13
View File
@@ -9,10 +9,11 @@ FilmForge 热压模具温控板(GD32F405, RS485 数据口)的桌面上位机
- **实时温度曲线**(自绘 QPainter 控件,无第三方图表依赖):实测温度折线 +
目标温度橙色阶梯线同图对照(闭环监视),NTC 开路/短路处曲线断开并打故障标记;
时间窗 2/5/15/60 分钟/全部,暂停/恢复,窗口统计(最小/最大/平均)。
- **温度数据双通道**:固件 M2 周期上报帧直接入曲线;未开启上报时可由上位机按
500ms/1s/2s 自动轮询 M1 补点(两条路读同一全局最新温度)。
- **温度数据**:温度监控统一走固件 M2 周期上报帧(M2S<ms> 开启后主动推送入曲线);
M1 仅作手动单点查询(命令区按钮/预设)。
- **控制面板**:目标温度 M4S<n>(0~200 整数℃,滑条+输入框)、加热 M3S1/M3S0
(预留功能,开启需确认)、周期上报 M2S<ms>(100~60000,10ms 步进)、预设/自定义命令。
(真实 PID 闭环控温,开启需确认)、周期上报 M2S<ms>(100~60000,10ms 步进)、
预设/自定义命令。
- **状态联动**:连接后自动查询版本(M999)/上报(M2)/加热(M3)/目标(M4)并同步到状态芯片;
软件复位(M888)/进 OTA(M9999) 二次确认,复位后 2 秒自动重查状态。
- **收发日志**:译码 + 原始 hex,500 行上限防卡顿。
@@ -52,14 +53,41 @@ python publish_gitea.py release/FilmForge-GUI-v<版本>-win64.zip --replace --in
`app_fluent.py`(PySide6-Fluent 呈现层)/ `config.py`(运行目录+设置记忆)/
`filmforge_gui.py`(入口+CLI)。协议模块不依赖 GUI。
- **应答派发是 FIFO 队列**(`app_fluent._pending`):半双工上固件严格按收到顺序应答,
连接序列(150ms 间隔连发 4 条)与自动轮询并存,单 pending 槽会被乱序击穿——实测踩坑:
连接序列(150ms 间隔连发)场景下单 pending 槽会被乱序击穿——实测踩坑:
M2 的 OFF 被 M3 槽吃掉、M4 的目标值混进温度曲线。应答形态与队首期望匹配才消费,
不匹配记杂散,1.2s 超时自清。温度帧二义性(M1/M4/上报同为 ASCII 温度)靠
队首期望 + 生命周期窗口拆分(上报开启时收紧到 0.5s)。
不匹配记杂散,1.2s 超时自清。
- **M4 查询走"暂停上报→查询→恢复"应答驱动链**:M4 应答与周期上报帧同为 ASCII
浮点,仅靠时间窗拆分在 100ms 上报下实测必现误占(目标 chip 显示成实测温度)——
上报开启时查目标自动执行 M2S0(ok)→M4→M2S<ms> 恢复;连接同步序列同样走此路径。
- **晚到 ok 补偿**:USB-RS485 适配器换向延迟实测可 >4s,设定命令的 ok 迟到超过
1.2s 已按超时清队,8s 内的迟到 ok 按发送时记录的计划补偿应用(否则状态 chip
与设备脱节,实测:M2S100 ok 迟到,chip 停在 OFF 而设备在报)。
- **连接按钮按"意图"翻转**:点击立即切换文本,实际开/关由串口线程异步完成,
已连接/已断开/丢失全部写入日志;期望连接但 2s 未建立按丢失处理回退按钮。
连接同步序列每步发送前确认在线(线程异步打开期间早发会被"未连接串口"丢弃)。
- **时间戳在串口线程读取点打**(`FF485.on_frame(fr, t_rx)`),不晚于 GUI 轮询消费——
避免重蹈 SerialMate 的时间戳失真问题(见 `E:\Hulk-Coding\HW-Debug-Tools\SerialMate`)。
- M4 目标应答的路由窗口 = 条目生命周期(1.2s):USB-RS485 适配器换向偶发延迟
实测可超 0.5s;上报开启时周期帧密集,才收紧到 0.5s 新鲜窗口。
- **发送入口幂等去重**:同一命令已有未决应答时忽略重复发送(状态栏提示
"应答中, 已忽略重复发送")——真机实测连点"停止"12 次会把 8 条 pending 队列
挤爆("丢弃积压命令"刷屏);协议全幂等,同命令一条在途足够,超时自清后
自然允许重发。
- **串口读取按到达量收,不攒大块**:固定 `read(256)` 在持续帧流下要攒满 256 字节
才返回(Windows 串口"字节间隔超时"150ms 因帧间隔更短永不触发,100ms 上报时
≈2.5s 一批),表现为曲线/命令全部 3 秒一卡、一股脑一大坨。已改为
`read(in_waiting or 1)`,延迟回到 ≤150ms 量级(v0.2.2,真机实测确认)。
- **日志裁剪走原生 `setMaximumBlockCount(500)`**:手工 `textCursor` 裁剪用了
`cursor.Start`/`cursor.BlockUnderCursor` 实例级枚举访问,PySide6 6.11 已移除
(类上仍在),必然抛 AttributeError——日志超 500 行(100ms 上报 ≈50s)后每个
100ms tick 在 `_log` 处异常中止:`_tick` 后半段(应答超时清理/连接状态机/统计
节流)永不执行(pending 永不清 → "停止"点击被幂等去重永久拦截)、RX 队列积压
ratchet(帧处理滞后涨到 1~2.7s,日志/chip/曲线显示全面滞后)、日志无限增长。
v0.2.8 改原生环形裁剪修复(离屏 A/B 实测:修复后 6 分钟漂移 ≤15ms、零异常、
零积压;selftest 新增 3 条回归用例)。
- **日志自动滚动仅在贴底时跟随**:上报运行时可上翻查看历史(每行强制滚底会抢
滚动条);CSV 写失败(磁盘满/记录盘被拔)自动停止记录并提示——`_csv_write` 在
`_tick` 帧处理路径上,未捕获异常会中止主循环(v0.2.8 日志裁剪同款教训);故障
标记封顶 1000 条;冻结视图切片右缘二分(暂停时长不增加渲染成本);串口写超时
只丢该条命令、链路保留(不按设备拔出断链)。均 v0.2.9。
- Fluent 控件自定义样式必须走 `setCustomStyleSheet`(覆盖 hover/pressed/disabled),
直接 `setStyleSheet` 会在切主题时被库重刷(DLPilot 同款坑);标题字号走 QFont。
- 图表刷新双节流:数据到达即重绘(≤10Hz)+ 0.5s 时间轴推进;“全部”窗口超 4000 点
@@ -67,8 +95,8 @@ python publish_gitea.py release/FilmForge-GUI-v<版本>-win64.zip --replace --in
## 已知事项
- 当前 USB-RS485 适配器自动换向偶发丢响应(固件侧已立“适配器更换”TODO):
表现为个别查询 1.2s 超时,队列自动清理后轮询自然重试,曲线无碍。
- 加热(M3)/目标温度(M4)为固件预留:仅软件状态(ack+可回读),硬件执行体接入后
协议不变,本工具无需改动即可联调真实控温闭环。
- 打包暂无应用图标(`logos/filmforge.ico` 缺位,build_exe.py 自动跳过)。
- 当前 USB-RS485 适配器自动换向偶发丢响应/大延迟(固件侧已立”适配器更换”TODO):
查询 1.2s 超时提示后重发即可;设定命令的晚到 ok 已自动补偿应用;曲线走上报帧
(设备单向推送),不受影响。
- 加热(M3S1)为真实 PID 闭环:开启后模具真实升温(NTC 故障/过温 150℃ 固件自动
断电保护,上电默认关闭);当前占空比无查询命令,从温度曲线间接观察。
+232
View File
@@ -0,0 +1,232 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""诊断 v2: 干净基线 — 旁路 app_fluent._log 的坏裁剪(cursor.Start 在 PySide6
6.11.2 上抛 AttributeError), 用 Qt 原生 setMaximumBlockCount 替代, 其余不动源码。
目的: 验证 100ms 持续上报数分钟下, 修复裁剪后 点击->TX 是否仍瞬时、事件循环
是否仍流畅(对照 buggy 版的 tick 中止 + 队列积压)。"""
import ctypes
import gc
import os
import sys
import threading
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
class FakeSerial:
"""模拟 CH34x + 固件: M2S<ms> 控制周期温度帧, M2S0 回 ok 并停流。"""
def __init__(self, *a, **kw):
self._buf = bytearray()
self._lock = threading.Lock()
self._alive = True
self.write_log = [] # [(time, cmd_bytes)]
self.report_ms = 0
self._next_tx = 0.0
self._th = threading.Thread(target=self._fw, daemon=True)
self._th.start()
FakeSerial.last = self
def _fw(self):
while self._alive:
now = time.time()
if self.report_ms and now >= self._next_tx:
self._push(self._frame("34.6"))
self._next_tx = now + self.report_ms / 1000.0
time.sleep(0.002)
@staticmethod
def _frame(payload, type_code=0xF0):
body = bytes([type_code, len(payload)]) + payload.encode("ascii")
return bytes([0xB5]) + body + bytes([ff.sum_crc(body)])
def _push(self, b):
with self._lock:
self._buf += b
@property
def in_waiting(self):
with self._lock:
return len(self._buf)
def read(self, n):
deadline = time.time() + 0.15
while True:
with self._lock:
if self._buf:
out = bytes(self._buf[:n])
del self._buf[:n]
return out
if time.time() >= deadline:
return b""
time.sleep(0.004)
def write(self, data):
self.write_log.append((time.time(), bytes(data)))
cmd = bytes(data[3:-1]).decode("ascii", "replace").strip()
threading.Timer(0.05, lambda: self._on_cmd(cmd)).start()
return len(data)
def _on_cmd(self, cmd):
if cmd.startswith("M2S"):
try:
ms = int(cmd[3:])
except ValueError:
ms = 0
self.report_ms = ms
self._next_tx = time.time() + ms / 1000.0
self._push(self._frame("ok"))
elif cmd in ("M1", "M2", "M3", "M4", "M999"):
self._push(self._frame({"M1": "34.6", "M2": "OFF", "M3": "OFF",
"M4": "35.0", "M999": "v41.0.0.1"}[cmd]))
else:
self._push(self._frame("ok"))
def close(self):
self._alive = False
try:
self._th.join(timeout=0.5)
except Exception:
pass
ff.serial.Serial = FakeSerial
from PySide6.QtCore import QTimer
from PySide6.QtWidgets import QApplication
app = QApplication([])
import app_fluent
# ---- 测试侧修复: 旁路坏裁剪(原生上限 500 行, 语义等同原意图) ----
window = app_fluent.App(0)
window.txt_log.setMaximumBlockCount(500) # blockCount 永不 >500, 坏分支不触发
window.show()
fake = None
exception_count = [0]
# ---- 指标 ----
stats = {"paint": [0, 0.0, 0.0], "tick": [0, 0.0, 0.0],
"gc": [0, 0.0, 0.0], "drift": [], "qmax": [0], "click": []}
_orig_paint = app_fluent.TemperatureChart.paintEvent
def timed_paint(self, ev):
t0 = time.perf_counter()
_orig_paint(self, ev)
dt = time.perf_counter() - t0
s = stats["paint"]; s[0] += 1; s[1] += dt; s[2] = max(s[2], dt)
app_fluent.TemperatureChart.paintEvent = timed_paint
_orig_tick = app_fluent.App._tick
def timed_tick(self):
t0 = time.perf_counter()
try:
_orig_tick(self)
except Exception as e:
exception_count[0] += 1
if exception_count[0] <= 3:
print("TICK EXCEPTION:", e, flush=True)
dt = time.perf_counter() - t0
s = stats["tick"]; s[0] += 1; s[1] += dt; s[2] = max(s[2], dt)
app_fluent.App._tick = timed_tick
_gc_t = [0.0]
def gc_cb(phase, info):
if phase == "start":
_gc_t[0] = time.perf_counter()
else:
dt = time.perf_counter() - _gc_t[0]
s = stats["gc"]; s[0] += 1; s[1] += dt; s[2] = max(s[2], dt)
gc.callbacks.append(gc_cb)
probe = {"expect": None}
def probe_fire():
now = time.time()
if probe["expect"] is None:
probe["expect"] = now + 0.02
return
stats["drift"].append((now, now - probe["expect"]))
probe["expect"] += 0.02
if now - probe["expect"] > 0.5:
probe["expect"] = now + 0.02
pt = QTimer(); pt.timeout.connect(probe_fire); pt.start(20)
CLICK_PERIOD = 30
click = {"expect": None, "n": 0}
def do_click_stop():
now = time.time()
drift_ms = (now - click["expect"]) * 1000 if click["expect"] else 0.0
q0 = window._rx_q.qsize()
window._send_set("M2S0", "report", ("OFF", None)) # "停止"按钮真实路径
click["n"] += 1
stats["click"].append(dict(n=click["n"], t=now, drift_ms=drift_ms,
qsize=q0, pending=len(window._pending),
samples=len(window.samples_t),
logblk=window.txt_log.blockCount()))
QTimer.singleShot(3000, do_click_restart)
def do_click_restart():
window._send_set("M2S100", "report", ("ON", 100))
click["expect"] = time.time() + CLICK_PERIOD
QTimer.singleShot(CLICK_PERIOD * 1000, do_click_stop)
def start():
global fake
window.link.open("COMFAKE", 115200)
def wait_conn():
if window.link.is_connected():
fake = FakeSerial.last
click["expect"] = time.time() + 15
QTimer.singleShot(15000, do_click_stop)
print("[t0] connected; stop-clicks every %ds from +15s" % CLICK_PERIOD,
flush=True)
else:
QTimer.singleShot(200, wait_conn)
QTimer.singleShot(200, wait_conn)
QTimer.singleShot(100, start)
t_start = time.time()
DURATION = float(sys.argv[1]) if len(sys.argv) > 1 else 320.0
INTERVAL = float(sys.argv[2]) if len(sys.argv) > 2 else 40.0
def monitor():
while time.time() - t_start < DURATION:
time.sleep(INTERVAL)
el = time.time() - t_start
recent = [d for t, d in stats["drift"] if t >= time.time() - INTERVAL]
rmax = max(recent) * 1000 if recent else 0
stats["qmax"][0] = max(stats["qmax"][0], window._rx_q.qsize())
print(f"[{el:6.1f}s] cpu={time.process_time():7.2f}s "
f"samples={len(window.samples_t):6d} q={window._rx_q.qsize():4d} "
f"(max {stats['qmax'][0]:4d}) logblk={window.txt_log.blockCount():4d} "
f"paint(n={stats['paint'][0]:5d} max={stats['paint'][2]*1000:6.1f}ms) "
f"tick_max={stats['tick'][2]*1000:6.1f}ms "
f"gc_max={stats['gc'][2]*1000:5.1f}ms drift_max={rmax:7.1f}ms "
f"exc={exception_count[0]}", flush=True)
print("==== CLICK 明细 ====", flush=True)
for c in stats["click"]:
tx = next((t for t, d in FakeSerial.last.write_log
if d[3:-1].decode("ascii", "replace") == "M2S0"
and c["t"] <= t <= c["t"] + 2.0), None)
total = (tx - c["t"]) * 1000 if tx else -1
print(f"click#{c['n']} t=+{c['t']-t_start:6.1f}s 调度漂移={c['drift_ms']:8.1f}ms "
f"click->TX={total:7.1f}ms q={c['qsize']:4d} pending={c['pending']} "
f"samples={c['samples']:6d} logblk={c['logblk']}", flush=True)
ds = sorted(d for _t, d in stats["drift"])
if ds:
print(f"drift: n={len(ds)} p50={ds[len(ds)//2]*1000:.1f}ms "
f"p99={ds[int(len(ds)*0.99)]*1000:.1f}ms max={ds[-1]*1000:.1f}ms "
f">500ms次数={sum(1 for d in ds if d>0.5)}", flush=True)
print(f"paint total={stats['paint'][1]:.2f}s max={stats['paint'][2]*1000:.1f}ms | "
f"tick total={stats['tick'][1]:.2f}s max={stats['tick'][2]*1000:.1f}ms | "
f"gc total={stats['gc'][1]:.2f}s max={stats['gc'][2]*1000:.1f}ms "
f"n={stats['gc'][0]} | exceptions={exception_count[0]}", flush=True)
os._exit(0)
threading.Thread(target=monitor, daemon=True).start()
app.exec()
+225
View File
@@ -0,0 +1,225 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""诊断 v2b: BUGGY 对照组 — 不加裁剪修复, 复现 PySide6 6.11.2 上
_log 裁剪 AttributeError 触发后的行为: tick 中止 / RX 处理滞后 ratchet /
事件循环漂移 / 点击->TX。指标同 v2, 另增帧处理年龄(frame_age)。"""
import gc
import os
import sys
import threading
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
class FakeSerial:
def __init__(self, *a, **kw):
self._buf = bytearray()
self._lock = threading.Lock()
self._alive = True
self.write_log = []
self.report_ms = 0
self._next_tx = 0.0
self._th = threading.Thread(target=self._fw, daemon=True)
self._th.start()
FakeSerial.last = self
def _fw(self):
while self._alive:
now = time.time()
if self.report_ms and now >= self._next_tx:
self._push(self._frame("34.6"))
self._next_tx = now + self.report_ms / 1000.0
time.sleep(0.002)
@staticmethod
def _frame(payload, type_code=0xF0):
body = bytes([type_code, len(payload)]) + payload.encode("ascii")
return bytes([0xB5]) + body + bytes([ff.sum_crc(body)])
def _push(self, b):
with self._lock:
self._buf += b
@property
def in_waiting(self):
with self._lock:
return len(self._buf)
def read(self, n):
deadline = time.time() + 0.15
while True:
with self._lock:
if self._buf:
out = bytes(self._buf[:n])
del self._buf[:n]
return out
if time.time() >= deadline:
return b""
time.sleep(0.004)
def write(self, data):
self.write_log.append((time.time(), bytes(data)))
cmd = bytes(data[3:-1]).decode("ascii", "replace").strip()
threading.Timer(0.05, lambda: self._on_cmd(cmd)).start()
return len(data)
def _on_cmd(self, cmd):
if cmd.startswith("M2S"):
try:
ms = int(cmd[3:])
except ValueError:
ms = 0
self.report_ms = ms
self._next_tx = time.time() + ms / 1000.0
self._push(self._frame("ok"))
elif cmd in ("M1", "M2", "M3", "M4", "M999"):
self._push(self._frame({"M1": "34.6", "M2": "OFF", "M3": "OFF",
"M4": "35.0", "M999": "v41.0.0.1"}[cmd]))
else:
self._push(self._frame("ok"))
def close(self):
self._alive = False
try:
self._th.join(timeout=0.5)
except Exception:
pass
ff.serial.Serial = FakeSerial
from PySide6.QtCore import QTimer
from PySide6.QtWidgets import QApplication
app = QApplication([])
import app_fluent
# 注意: 不打 setMaximumBlockCount 补丁 —— 原样复现 bug
window = app_fluent.App(0)
window.show()
fake = None
exception_count = [0]
stats = {"tick": [0, 0.0, 0.0], "gc": [0, 0.0, 0.0], "drift": [],
"qmax": [0], "click": [], "age": [0, 0.0, 0.0]} # age: n,total,max
_orig_tick = app_fluent.App._tick
def timed_tick(self):
t0 = time.perf_counter()
try:
_orig_tick(self)
except Exception as e:
exception_count[0] += 1
if exception_count[0] <= 3:
print("TICK EXCEPTION:", type(e).__name__, e, flush=True)
dt = time.perf_counter() - t0
s = stats["tick"]; s[0] += 1; s[1] += dt; s[2] = max(s[2], dt)
app_fluent.App._tick = timed_tick
_orig_handle = app_fluent.App._handle_frame
def timed_handle(self, fr, t_rx):
_orig_handle(self, fr, t_rx)
age = time.time() - t_rx # 帧从收到到被主线程处理的滞后
s = stats["age"]; s[0] += 1; s[1] += age; s[2] = max(s[2], age)
app_fluent.App._handle_frame = timed_handle
_gc_t = [0.0]
def gc_cb(phase, info):
if phase == "start":
_gc_t[0] = time.perf_counter()
else:
dt = time.perf_counter() - _gc_t[0]
s = stats["gc"]; s[0] += 1; s[1] += dt; s[2] = max(s[2], dt)
gc.callbacks.append(gc_cb)
probe = {"expect": None}
def probe_fire():
now = time.time()
if probe["expect"] is None:
probe["expect"] = now + 0.02
return
stats["drift"].append((now, now - probe["expect"]))
probe["expect"] += 0.02
if now - probe["expect"] > 0.5:
probe["expect"] = now + 0.02
pt = QTimer(); pt.timeout.connect(probe_fire); pt.start(20)
CLICK_PERIOD = 30
click = {"expect": None, "n": 0}
def do_click_stop():
now = time.time()
drift_ms = (now - click["expect"]) * 1000 if click["expect"] else 0.0
q0 = window._rx_q.qsize()
window._send_set("M2S0", "report", ("OFF", None))
click["n"] += 1
stats["click"].append(dict(n=click["n"], t=now, drift_ms=drift_ms,
qsize=q0, pending=len(window._pending),
samples=len(window.samples_t),
logblk=window.txt_log.blockCount()))
QTimer.singleShot(3000, do_click_restart)
def do_click_restart():
window._send_set("M2S100", "report", ("ON", 100))
click["expect"] = time.time() + CLICK_PERIOD
QTimer.singleShot(CLICK_PERIOD * 1000, do_click_stop)
def start():
global fake
window.link.open("COMFAKE", 115200)
def wait_conn():
if window.link.is_connected():
fake = FakeSerial.last
click["expect"] = time.time() + 15
QTimer.singleShot(15000, do_click_stop)
print("[t0] connected; stop-clicks every %ds from +15s (BUGGY mode)"
% CLICK_PERIOD, flush=True)
else:
QTimer.singleShot(200, wait_conn)
QTimer.singleShot(200, wait_conn)
QTimer.singleShot(100, start)
t_start = time.time()
DURATION = float(sys.argv[1]) if len(sys.argv) > 1 else 320.0
INTERVAL = float(sys.argv[2]) if len(sys.argv) > 2 else 40.0
def monitor():
while time.time() - t_start < DURATION:
time.sleep(INTERVAL)
el = time.time() - t_start
recent = [d for t, d in stats["drift"] if t >= time.time() - INTERVAL]
rmax = max(recent) * 1000 if recent else 0
stats["qmax"][0] = max(stats["qmax"][0], window._rx_q.qsize())
n, tot, mx = stats["age"]
print(f"[{el:6.1f}s] cpu={time.process_time():7.2f}s "
f"samples={len(window.samples_t):6d} q={window._rx_q.qsize():5d} "
f"(max {stats['qmax'][0]:5d}) logblk={window.txt_log.blockCount():5d} "
f"frame_age avg={tot/max(n,1)*1000:7.1f}ms max={mx*1000:8.1f}ms "
f"tick_max={stats['tick'][2]*1000:6.1f}ms drift_max={rmax:8.1f}ms "
f"exc={exception_count[0]}", flush=True)
print("==== CLICK 明细 ====", flush=True)
for c in stats["click"]:
tx = next((t for t, d in FakeSerial.last.write_log
if d[3:-1].decode("ascii", "replace") == "M2S0"
and c["t"] <= t <= c["t"] + 2.0), None)
total = (tx - c["t"]) * 1000 if tx else -1
print(f"click#{c['n']} t=+{c['t']-t_start:6.1f}s 调度漂移={c['drift_ms']:8.1f}ms "
f"click->TX={total:7.1f}ms q={c['qsize']:5d} pending={c['pending']} "
f"samples={c['samples']:6d} logblk={c['logblk']}", flush=True)
ds = sorted(d for _t, d in stats["drift"])
if ds:
print(f"drift: n={len(ds)} p50={ds[len(ds)//2]*1000:.1f}ms "
f"p99={ds[int(len(ds)*0.99)]*1000:.1f}ms max={ds[-1]*1000:.1f}ms "
f">500ms次数={sum(1 for d in ds if d>0.5)}", flush=True)
n, tot, mx = stats["age"]
print(f"frame_age avg={tot/max(n,1)*1000:.1f}ms max={mx*1000:.1f}ms | "
f"exceptions={exception_count[0]} | tick n={stats['tick'][0]} "
f"max={stats['tick'][2]*1000:.1f}ms", flush=True)
os._exit(0)
threading.Thread(target=monitor, daemon=True).start()
app.exec()
+254 -112
View File
@@ -5,7 +5,7 @@
协议与串口模块(filmforge_rs485)保持不变, 本文件只做呈现层:
PySide6-Fluent-Widgets 卡片式双栏仪表盘 + 自绘滚动温度曲线(无第三方图表依赖)。
闭环监视 = 实测温度曲线 + 目标温度阶梯线同图 + 加热/上报/传感器状态联动;
设备未开启 M2 上报时可由上位机周期轮询 M1 补齐曲线。
温度监控走固件 M2 周期上报帧(M2S<ms> 开启), M1 仅作手动单点查询。
"""
import ctypes
@@ -46,7 +46,7 @@ if getattr(sys, "frozen", False) and os.name == "nt":
pass
from PySide6.QtCore import QPointF, QRectF, Qt, QTimer
from PySide6.QtGui import QColor, QFont, QPainter, QPainterPath, QPen, QPalette
from PySide6.QtGui import QColor, QFont, QIcon, QPainter, QPainterPath, QPen, QPalette
from PySide6.QtWidgets import (
QApplication, QHBoxLayout, QLabel, QMainWindow, QMessageBox, QPlainTextEdit,
QScrollArea, QSplitter, QVBoxLayout, QWidget,
@@ -60,7 +60,7 @@ import config
import filmforge_rs485 as ff
APP_NAME = "FilmForge"
APP_VERSION = "0.1.0"
APP_VERSION = "0.2.9"
REPO_URL = "https://gitea.hulk.wang/hulk/FilmForge"
MONO = QFont("Consolas", 10)
CHIP_IDLE = "#666666"
@@ -69,6 +69,7 @@ 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)
@@ -133,7 +134,7 @@ class TemperatureChart(QWidget):
ts, vs = self._samples_t, self._samples_v
if not ts:
return [], []
lo, hi = 0, len(ts) # 二分找左边界(留一点前文保证连线)
lo, hi = 0, len(ts) # 左边界二分(留一点前文保证连线)
while lo < hi:
m = (lo + hi) // 2
if ts[m] < t0:
@@ -141,7 +142,14 @@ class TemperatureChart(QWidget):
else:
hi = m
lo = max(lo - 1, 0)
t_out, v_out = ts[lo:], vs[lo:]
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 = [], []
@@ -153,7 +161,9 @@ class TemperatureChart(QWidget):
return t_out, v_out
def _target_points(self, t0, t1):
"""目标阶梯线: [(t,v)...] 持有至下一变更时刻, 展开成阶梯点并延伸到右缘"""
"""目标阶梯线: [(t,v)...] 每段从自身时刻起垂直落变并水平延伸至下一变更,
最后一段延伸到右缘。旧实现对"最后一次变更"只画终点, 会从旧值斜拉一条
直线到右缘(实测 57→41 被画成 2 分钟斜坡), 已改为纯阶梯。"""
hist = self._target_hist
if not hist:
return []
@@ -162,10 +172,15 @@ class TemperatureChart(QWidget):
tn = hist[i + 1][0] if i + 1 < len(hist) else t1
if tn < t0 or ti > t1:
continue
b = min(tn, t1)
start, end = max(ti, t0), min(tn, t1)
if not pts:
pts.append((max(ti, t0), vi))
pts.append((b, vi))
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
@@ -191,7 +206,7 @@ class TemperatureChart(QWidget):
if not self._has_data:
p.setPen(QColor(c["text"]))
p.drawText(self.rect(), Qt.AlignCenter,
"暂无数据 — 连接设备后查询温度(M1)或开启上报(M2S<ms>)")
"暂无数据 — 连接设备后开启周期上报(M2S<ms>)即可实时推送; 单点查询用 M1")
@staticmethod
def _y_range(vis_v, tgt_pts):
@@ -357,6 +372,9 @@ class App(QMainWindow):
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)
@@ -373,9 +391,13 @@ class App(QMainWindow):
self.firmware = None
# 应答 FIFO: 半双工上固件严格按收到顺序应答, 连接序列/连发场景下
# 多个应答可能同时在途, 单 pending 槽会被乱序击穿(实测踩坑)
self._pending = deque() # [{'cmd','kind','t','apply'}]
self._last_poll = 0.0
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()
@@ -459,30 +481,69 @@ class App(QMainWindow):
header.addWidget(self._button("关于", self._about))
outer.addLayout(header)
split = QSplitter(Qt.Horizontal, root)
# 双列放进同一滚动容器: 两列高度锁定一致——不足一屏时各自拉伸到底边,
# 超出一屏时整页一起滚动; 各列独立滚动会让底边在窗口化下错位(实测)
split = QSplitter(Qt.Horizontal)
split.setChildrenCollapsible(False)
split.addWidget(self._scroll_column(self._build_left))
split.addWidget(self._scroll_column(self._build_right))
split.addWidget(self._column_host(self._build_left))
split.addWidget(self._column_host(self._build_right))
split.setStretchFactor(0, 3)
split.setStretchFactor(1, 2)
outer.addWidget(split, 1)
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 _scroll_column(self, builder):
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)
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setFrameShape(QScrollArea.NoFrame)
scroll.setWidget(host)
return scroll
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)
@@ -544,47 +605,18 @@ class App(QMainWindow):
# ---------------- 右列: 控制 ----------------
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("温度控制 · 目标 / 加热(固件预留)")
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, 200)
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("查询", lambda: self._query("M4", "target")))
row.addWidget(self._button("查询", self._query_target))
row.addWidget(self._button("设定", self._send_target, True))
body.addLayout(row)
body.addWidget(self.sld_target)
@@ -595,11 +627,12 @@ class App(QMainWindow):
lambda: self._send_set("M3S0", "heater", "OFF")))
row.addStretch(1)
body.addLayout(row)
body.addWidget(self._note(card, "当前固件加热/目标仅维护软件状态(ack+可回读), 硬件执行体"
"待接入; 接入后协议不变, 本页即可直接联调真实控温。"))
body.addWidget(self._note(card, "开启加热后固件按目标温度 PID 闭环真实控温(500ms 周期,"
"慢 PWM 执行); NTC 故障或 ≥150℃ 过温时固件自动断电保护,"
"上电默认关闭。当前占空比无查询命令, 可从温度曲线间接观察。"))
layout.addWidget(card)
card, body = self._card("周期上报 · M2S<ms>")
card, body = self._card("温度监控 · 周期上报 M2S<ms>")
row = QHBoxLayout()
self.ent_report = LineEdit(card)
self.ent_report.setPlaceholderText("100~60000, 10ms 步进")
@@ -612,16 +645,9 @@ class App(QMainWindow):
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 补点(两条路读同一全局最新温度)。"))
body.addWidget(self._note(card, "温度监控统一走 M2: 开启上报后固件按设定周期主动推送"
"温度帧入曲线; 停止后曲线暂停, 需要单点时用命令区"
" 查温度 M1(两条路读同一全局最新温度)。"))
layout.addWidget(card)
card, body = self._card("命令 · 预设 / 自定义")
@@ -656,10 +682,13 @@ class App(QMainWindow):
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)
layout.addWidget(card, 1) # 弹性填充: 右列底边与左列对齐
card, body = self._card("危险操作 · 会使设备复位")
row = QHBoxLayout()
@@ -684,10 +713,6 @@ class App(QMainWindow):
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)
@@ -785,26 +810,38 @@ class App(QMainWindow):
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:
if connected and self._want:
self._connect_seq()
else:
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)')
# 自动轮询 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")
if entry["kind"] == "target":
self._resume_report() # M4 查询失败也要恢复上报
if entry.get("then"):
entry["then"]() # 链式流程(暂停上报→查目标)继续走
# 曲线时间轴推进 + 窗口统计(节流 0.5s)
if now - self._last_chart_repaint >= 0.5:
@@ -814,7 +851,7 @@ class App(QMainWindow):
def _on_status_main(self, msg):
self.lbl_st.setText(msg)
if msg.startswith("已连接"):
# 连接/断开/丢失全部入日志: 状态标签只留最后一条, 日志才可回溯排查
self._log(f"── {msg} ──")
# ================= 帧派发 =================
@@ -843,6 +880,7 @@ class App(QMainWindow):
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()
@@ -852,13 +890,28 @@ class App(QMainWindow):
# 非数值成功载荷: 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
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)
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
@@ -885,13 +938,16 @@ class App(QMainWindow):
self.lbl_st.setText(f"目标温度已设定 {value} ℃")
elif what == "heater":
self._set_heater(value)
self.lbl_st.setText(f"加热已{'开启' if value == 'ON' else '关闭'}(软件状态)")
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):
@@ -914,6 +970,8 @@ class App(QMainWindow):
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()
@@ -947,14 +1005,23 @@ class App(QMainWindow):
f"平均 {mean:.1f} ℃ · {len(vals)} 点")
# ================= 发送路径 =================
def _send_cmd(self, cmd, kind, apply_plan=None):
"""带应答跟踪发送: kind 决定应答派发, apply_plan 为 ok 后的本地状态计划。
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})
"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"]}')
@@ -979,8 +1046,9 @@ class App(QMainWindow):
def _heater_on(self):
if not self._confirm("开启加热",
"当前固件加热为预留功能(仅软件状态, 硬件执行体待接入)。\n"
"接入硬件后此操作将真实加热模具, 确认开启?"):
"加热已实装: 开启后固件将按目标温度 PID 闭环真实加热模具,\n"
"表面可达 0~200℃(NTC 故障/过温时固件自动断电保护)。\n"
"确认开启?"):
return
self._send_set("M3S1", "heater", "ON")
@@ -1010,9 +1078,11 @@ class App(QMainWindow):
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",
"M4": "target", "M999": "version"}.get(cmd, "ack")
"M999": "version"}.get(cmd, "ack")
self._send_cmd(cmd, kind)
return
@@ -1076,7 +1146,16 @@ class App(QMainWindow):
return label.strip()
def _toggle_connect(self):
if self.link.is_connected():
# 按钮按"意图"立即翻转, 实际开/关由串口线程异步完成并经状态日志反馈;
# 不再用 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()
@@ -1090,6 +1169,10 @@ class App(QMainWindow):
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):
@@ -1097,12 +1180,51 @@ class App(QMainWindow):
self._connect_seq()
def _connect_seq(self):
"""连接/复位后状态同步: 版本→上报→加热→目标, 150ms 间隔规避半双工碰撞"""
"""连接/复位后状态同步: 版本→上报→加热→(暂停上报→)目标。
每步发送前确认在线(线程异步打开期间 singleShot 早发会被"未连接串口"丢弃),
150ms 间隔规避半双工碰撞; 目标查询放最后并等 M2 应答消费完(见 _seq_target)"""
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))
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):
@@ -1157,9 +1279,19 @@ class App(QMainWindow):
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)}")
@@ -1170,14 +1302,10 @@ class App(QMainWindow):
# ================= 日志 =================
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()
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):
@@ -1189,10 +1317,6 @@ class App(QMainWindow):
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())
@@ -1224,6 +1348,15 @@ def _single_instance():
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)
@@ -1231,6 +1364,15 @@ def run_gui(auto_close=0):
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()
+2 -12
View File
@@ -26,16 +26,10 @@ WINDOW_OPTIONS = (('2 分钟', 120), ('5 分钟', 300), ('15 分钟', 900),
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'
@@ -65,14 +59,12 @@ def load_settings():
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')):
for key, target in (('report_ms', 'REPORT_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))
global THEME
theme = str(data.get('theme') or '')
if theme in ('Dark', 'Light', 'System'):
THEME = theme
@@ -83,8 +75,6 @@ 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,
}
+19 -6
View File
@@ -12,8 +12,9 @@ FilmForge 热压模具 RS485 温控链路
命令语义速记(全表见 protocols.md):
M1 查温度(OPEN=NTC 开路/SHRT=短路) M2 查上报状态 M2S<ms> 设间隔并开启(100~60000,
10ms 步进) M2S0 停止 M3/M3S1/M3S0 加热(预留, 当前仅软件状态) M4/M4S<n> 目标温度
(0~200 整数) M888 软复位 M999 查版本 M9999 进 OTA(均会复位, GUI 需二次确认)
10ms 步进) M2S0 停止 M3/M3S1/M3S0 加热使能(实装: PB12 慢 PWM + PID 按目标温度
闭环, 开启后真实加热) M4/M4S<n> 目标温度(0~200 整数) M888 软复位 M999 查版本
M9999 进 OTA(均会复位, GUI 需二次确认)
通信要点: 全部命令幂等(ack 丢了直接重发); 命令间隔 >=50ms; 响应超时 300ms 重发 x3。
仅标准库 + 可选 pyserial(缺失只影响串口功能, 其余照常)。
@@ -62,9 +63,9 @@ PRESETS = (
('查上报状态', 'M2', 'ON <ms> / OFF'),
('开上报500ms', 'M2S500', '设上报间隔并开启(100~60000, 10ms 步进)'),
('停上报', 'M2S0', '停止周期上报'),
('查加热状态', 'M3', 'ON/OFF(预留, 当前仅软件状态)'),
('开加热', 'M3S1', '预留: 硬件执行体待接入, 仅维护软件状态'),
('关加热', 'M3S0', '预留'),
('查加热状态', 'M3', 'ON/OFF(加热使能状态)'),
('开加热', 'M3S1', '开启 PID 闭环控温, 模具将真实加热(需确认)'),
('关加热', 'M3S0', '关闭加热输出'),
('查目标温度', 'M4', 'ASCII 摄氏度(如 120.0)'),
('设目标120C', 'M4S120', '0~200 整数摄氏度, 越界回 FD'),
('查固件版本', 'M999', '如 v41.0.0.1'),
@@ -346,9 +347,21 @@ class FF485:
pass
if item is not None:
cmd, frame = item
sent = False
try:
self.ser.write(frame)
sent = True
except serial.SerialTimeoutException:
# 写超时(适配器停滞): 只丢这条命令, 链路保留。
# 勿并入下方 OSError——瞬时停滞曾被当成"设备拔出"
# 关端口断链, 迫使用户手动重连(适配器换向本就不稳)
self.on_event(f'✕ {cmd} 写超时未发出, 链路保留(可重发)')
if sent:
self.on_event(f'→ {cmd} [{frame.hex(" ").upper()}]')
data = self.ser.read(256)
# 按到达量读取, 不攒大块: 固定 read(256) 在持续帧流下要攒满
# 256 字节才返回(win32 字节间隔超时 150ms 因帧间隔更短永不
# 触发, 100ms 上报 ≈ 2.5s 一批), 表现为曲线/命令全按 3s 一卡
data = self.ser.read(self.ser.in_waiting or 1)
if data:
t_rx = time.time() # 在读取点打时间戳(曲线/CSV 用)
for fr in self.parser.feed(data):
Binary file not shown.

After

Width:  |  Height:  |  Size: 23 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 10 KiB

+6
View File
@@ -0,0 +1,6 @@
<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 512 512">
<rect width="512" height="512" fill="#000000"/>
<rect x="38" y="64" width="436" height="186" rx="34" fill="#00C8D7"/>
<text x="256" y="210" font-family="Arial, 'Microsoft YaHei', sans-serif" font-size="146" font-weight="900" fill="#000000" text-anchor="middle">FILM</text>
<text x="256" y="421" font-family="Arial, 'Microsoft YaHei', sans-serif" font-size="146" font-weight="900" fill="#FFFFFF" text-anchor="middle">Forge</text>
</svg>

After

Width:  |  Height:  |  Size: 507 B

+4 -4
View File
@@ -4,9 +4,9 @@
创建/更新 Gitea Release 并上传发布 zip(纯标准库, 无需安装依赖)
================================================================
用法(在固件仓任意目录执行, 自动从 origin 解析仓库):
python publish_gitea.py release/FilmForge-GUI-v0.1.0-win64.zip
python publish_gitea.py <zip> --notes-file release/notes-v0.1.0.md
python publish_gitea.py <zip> --tag FilmForge-GUI-v0.1.0 --replace
python publish_gitea.py release/FilmForge-GUI-v0.2.0-win64.zip
python publish_gitea.py <zip> --notes-file release/notes-v0.2.0.md
python publish_gitea.py <zip> --tag FilmForge-GUI-v0.2.0 --replace
行为:
- tag 与版本默认取 zip 文件名(FilmForge-GUI-vX.Y.Z-win64.zip → FilmForge-GUI-vX.Y.Z)
@@ -43,7 +43,7 @@ FilmForge 热压模具温控板(RS485)上位机: 实时温度曲线(自绘, 无
- `README.md` — 完整使用说明
### 注意
- 加热(M3)/目标温度(M4)为固件预留功能: 当前仅软件状态(ack+可回读), 硬件执行体接入后协议不变。
- 加热(M3S1)为真实 PID 闭环控温: 开启后模具将真实加热(NTC 故障/过温 150℃ 固件自动断电保护)。
- 软件复位(M888)/进 OTA(M9999) 会复位设备, 界面内已做二次确认。
"""
+109
View File
@@ -40,6 +40,9 @@ window = app_fluent.App(0)
window.show()
app.processEvents()
# 0) 图标链路: logos/filmforge.ico 存在时窗口图标必须已加载
check('窗口图标已设置', not window.windowIcon().isNull(), True)
# 1) 空数据渲染不抛异常(占位文案)
img = QImage(window.chart.size(), QImage.Format_ARGB32)
window.chart.render(img)
@@ -150,6 +153,60 @@ check('杂散不消费队首', len(window._pending), 1)
window._handle_frame(mk_frame('v41.0.0.1'), time.time())
check('杂散后正常消费', len(window._pending), 0)
# 3d) 晚到 ok 补偿: 设定命令超时清队后, 迟到的 ok 仍按发送时计划应用
window._pending.clear()
window._last_set = {'cmd': 'M2S500', 't': time.time(), 'apply': ('report', ('ON', 500))}
window._handle_frame(mk_frame('ok'), time.time())
check('晚到 ok 补偿上报', window.report_state, ('ON', 500))
check('晚到 ok 已清记录', window._last_set, None)
# 3e) M4 暂停-恢复链: 上报开启时查目标 → M2S0(ok)→M4→应答落地→自动恢复上报
window._pending.clear()
window.link.is_connected = lambda: True # 仅测试: 跳过串口在线检查
window._set_report_state('ON', 500)
window._query_target()
check('查目标先停上报', window._pending[0]['cmd'], 'M2S0')
check('挂起恢复标记', window._resume_report_ms, 500)
window._handle_frame(mk_frame('ok'), time.time()) # M2S0 ok → then 定时发 M4
window._pending.clear()
window._pending.append({'cmd': 'M4', 'kind': 'target', 't': time.time(), 'apply': None})
window._handle_frame(mk_frame('57.0'), time.time()) # M4 应答(替换定时器路径)
check('M4 应答落目标', window.target_hist[-1][1], 57.0)
check('恢复标记已清', window._resume_report_ms, None)
check('恢复上报已入队', window._pending[0]['cmd'], 'M2S500')
window.link.is_connected = lambda: False # 还原
window._pending.clear()
# 3f) 上报开启时, 上报帧不得被当成 M4 目标应答误占(队首非 target 时入曲线)
window._set_report_state('ON', 100)
n2 = len(window.samples_t)
window._handle_frame(mk_frame('60.0'), time.time())
check('无 pending 上报帧入曲线', len(window.samples_t), n2 + 1)
window._set_report_state('OFF', None)
# 3g) 幂等去重: 同命令未决时重复发送被忽略, 队列不再堆积(真机连点 M2S0 回归)
window._pending.clear()
window.link.is_connected = lambda: True
window._send_set('M2S0', 'report', ('OFF', None))
window._send_set('M2S0', 'report', ('OFF', None))
window._send_set('M2S0', 'report', ('OFF', None))
check('重复命令去重', len(window._pending), 1)
check('不同命令不去重', 'M4' if window._send_cmd('M4', 'target') else '', 'M4')
check('去重后队列 2 条', len(window._pending), 2)
window.link.is_connected = lambda: False
window._pending.clear()
# 3h) 目标阶梯线: 目标变更须在变更时刻垂直落变, 不得从旧值斜拉到右缘
t_a, t_b, t_now = time.time() - 120, time.time() - 60, time.time()
window.chart._target_hist = [(t_a, 57.0), (t_b, 41.0)]
pts = window.chart._target_points(t_a, t_now)
check('目标线终点延到右缘', abs(pts[-1][0] - t_now) < 1e-6 and pts[-1][1], 41.0)
diag = [1 for (xa, ya), (xb, yb) in zip(pts, pts[1:])
if abs(xb - xa) > 1e-9 and abs(yb - ya) > 1e-9]
check('目标线无斜段(纯阶梯)', diag, [])
check('变更时刻有垂直落变', (t_b, 57.0) in pts and (t_b, 41.0) in pts, True)
window.chart._target_hist = [] # 还原
# 4) 主题往返后曲线配色随动 + 渲染不崩
window._set_theme('Light')
check('Light 曲线配色', window.chart._colors['measured'].lower(),
@@ -170,5 +227,57 @@ with open(window._csv_path, encoding='utf-8-sig') as f:
check('CSV 表头+2 行', len(lines), 3)
check('CSV 故障行', lines[2].split(',')[2], 'SHRT')
# 6) 日志环形裁剪回归: 写满 500 行后封顶, 继续追加不抛异常、不无限增长。
# (手工 cursor 裁剪用 cursor.Start/BlockUnderCursor 实例级枚举访问,
# PySide6 6.11 上必然抛 AttributeError → 每 tick 异常中止, v0.2.8 改原生裁剪)
for i in range(600):
window._log(f'日志压力行 {i}')
check('日志 500 行封顶', window.txt_log.blockCount() <= 500, True)
window._log('封顶后追加行')
check('封顶后可继续追加', window.txt_log.blockCount() <= 500, True)
check('封顶后最新行在末尾',
window.txt_log.toPlainText().splitlines()[-1].endswith('封顶后追加行'), True)
# 7) v0.2.9 回归: 故障标记封顶 / 滚动条跟随 / CSV 写失败自停 / 冻结切片右边界
t_end = time.time()
for i in range(app_fluent.MAX_FAULTS + 50):
window._record_fault(t_end + i * 0.01, ff.SENSOR_OPEN)
check('故障标记封顶', len(window.faults), app_fluent.MAX_FAULTS)
sb = window.txt_log.verticalScrollBar()
sb.setValue(0)
window._log('上翻后新行')
check('上翻时不抢滚动条', sb.value() <= 1, True)
sb.setValue(sb.maximum())
window._log('贴底跟随行')
check('贴底时跟随滚动', abs(sb.value() - sb.maximum()) <= 4, True)
class _BoomFile:
"""模拟写坏盘: write/flush 必抛 OSError(磁盘满/记录盘被拔)"""
def write(self, *_a):
raise OSError(28, '模拟磁盘满')
def flush(self):
raise OSError(28, '模拟磁盘满')
def close(self):
pass
window._csv = _BoomFile()
window._csv_path = 'dummy.csv'
window._add_sample(time.time(), 25.0)
check('CSV 写失败自动停止记录', window._csv is None, True)
check('CSV 失败后按钮复位', window.btn_csv.text(), '开始记录')
c = window.chart
c._samples_t, c._samples_v = [1.0, 2.0, 3.0, 4.0], [1, 2, 3, 4]
c._frozen_at = 2.5
ts, _vs = c._visible(0, 2.5)
check('冻结切片不含冻结尾后样本', ts, [1.0, 2.0])
c._frozen_at = None
ts, _vs = c._visible(0, 99)
check('实时切片完整', ts, [1.0, 2.0, 3.0, 4.0])
c._samples_t, c._samples_v = window.samples_t, window.samples_v
print('RESULT:', 'FAIL:' + ','.join(fails) if fails else 'ALL PASS')
sys.exit(1 if fails else 0)
+7 -4
View File
@@ -149,12 +149,15 @@ void usart1_irq_handler(void);
#define HEATER_EN_RCU RCU_GPIOB
#define HEATER_EN_PORT GPIOB
#define HEATER_EN_PIN GPIO_PIN_12
#define HEATER_PWM_PERIOD_MS 1000U /* PWM 载波周期(1s,恒 1Hz 开关频率) */
#define HEATER_PWM_STEP_PCT 10U /* 占空比步进(每档脉宽 = 100ms) */
#define HEATER_PWM_PERIOD_MS 2000U /* PWM 载波周期(2s,载波 0.5Hz) */
#define HEATER_PWM_STEP_PCT 5U /* 占空比步进(每档脉宽 = 100ms,维持最小开关频率约束) */
#define HEATER_PID_PERIOD_MS 500U /* PID 计算周期 */
#define HEATER_PID_KP 8.0f /* PID 参数:初值,需上板实测整定 */
#define HEATER_PID_KP 10.0f /* PID 参数:超温前退功率更快(配合超温强制关断) */
#define HEATER_PID_KI 0.05f
#define HEATER_PID_KD 0.0f /* NTC 噪声对微分放大,先禁用 */
#define HEATER_PID_KD 30.0f /* 微分阻尼:按误差变化率提前刹车(抑制爬升惯性) */
#define HEATER_SLOWBAND_C 7.0f /* 温和带宽度:距目标小于此值时占空上限受限
(低速爬完最后一段,减少蓄热惯性过冲) */
#define HEATER_SLOWBAND_DUTY 25U /* 温和带内占空上限(%) */
#define HEATER_OVERTEMP_C 150.0f /* 过温强制关断阈值(滞回保护) */
#define HEATER_OVERTEMP_HYST 5.0f /* 过温恢复滞回 */
+1 -1
View File
@@ -40,7 +40,7 @@ void heat_set_target(float temp_c);
float heat_get_target(void);
/**
* @brief 查询当前量化占空比档位(0~10,对应 0~100%,10% 步进)。
* @brief 查询当前量化占空比档位(0~20,对应 0~100%,5% 步进)。
*/
uint8_t heat_get_duty(void);
+66 -33
View File
@@ -1,10 +1,14 @@
/**
* @file heat.c
* @brief 加热控制:慢 PWM 驱动(PB12)+ PID 控温闭环 + 安全保护。
* @brief 加热控制:慢 PWM 驱动(PB12)+ 分段 PID 控温闭环 + 安全保护。
* @details 硬件:PB12=HEAT_EN,高 = 加热开(2N7002K 电平转换 → VBE2412 PMOS → 24V)。
* 开关频率约束(<=10Hz)由"1s 载波周期 + 10% 占空步进"保证——每档脉宽
* 恒为 100ms 的整数倍。PID 以 500ms 周期计算(NTC 后台采样 100ms),
* 输出量化到 0~10 档占空。
* 开关频率约束(<=10Hz)由"2s 载波周期 + 5% 占空步进"保证——每档脉宽
* 恒为 100ms 的整数倍(0.5Hz 载波、20 档占空分辨率)。
* 控制为分段式:
* err >= 温和带 全量 PID(满功率爬升);
* 0 <= err < 温和带 正常 PID 但占空上限压到温和档(小功率缓升,
* 减少进带蓄热,降低越过目标后的惯性过冲);
* err < 0(超温) 强制关断 + 积分按负误差回退(执行器只能加热不能制冷)。
* @note 安全保护:上电默认关闭;NTC 传感器故障(OPEN/SHRT)强制关断;
* 过温(>= HEATER_OVERTEMP_C)滞回保护;未使能强制关断。
*/
@@ -20,18 +24,19 @@
static bool s_enable = false; /* 加热使能(M3) */
static float s_target = 0.0f; /* 目标温度(M4) */
static uint8_t s_duty10 = 0; /* 量化占空档位 0~10(x10%) */
static uint8_t s_duty = 0; /* 量化占空档位 0~HEATER_PWM_STEPS(x STEP%) */
static bool s_overtemp = false; /* 过温锁存(滞回恢复) */
static float s_integral = 0.0f; /* PID 积分项 */
static float s_prev_err = 0.0f; /* PID 上次误差 */
static uint16_t s_pwm_phase_ms = 0; /* PWM 载波相位 0..PERIOD-1 */
static uint16_t s_pid_divider = 0; /* PID 周期分频(10ms 圈) */
/* 档位总数 = 100% / 步进% */
#define HEATER_PWM_STEPS (100U / HEATER_PWM_STEP_PCT)
/******************************************************************************/
/* 底层输出 */
static void heat_out(bool on);
static void heat_out(bool on) {
if (on) {
gpio_bit_set(HEATER_EN_PORT, HEATER_EN_PIN);
@@ -60,21 +65,28 @@ static void pid_reset(void) {
}
/**
* @brief PID 计算(500ms 周期):误差 -> 占空档位 s_duty10(0~10)。
* @brief 积分试算值(err 一步积分;供抗饱和逻辑条件采纳)。
*/
static float integ_step(float err) {
return s_integral + err;
}
/**
* @brief 分段 PID 计算(500ms 周期):误差 -> 占空档位 s_duty。
*/
static void pid_compute(void) {
float t = ntc_get_temperature();
/* 传感器故障(开路/短路):安全第一,强制关断并复位 PID */
if (t <= -273.0f) {
s_duty10 = 0;
s_duty = 0;
pid_reset();
return;
}
/* 未使能:强制关断 */
if (!s_enable) {
s_duty10 = 0;
s_duty = 0;
pid_reset();
return;
}
@@ -86,39 +98,60 @@ static void pid_compute(void) {
s_overtemp = false;
}
if (s_overtemp) {
s_duty10 = 0;
s_duty = 0;
pid_reset();
return;
}
/* 位置式 PID(带抗积分饱和):输出饱和时积分仅沿退饱和方向更新,
防止升温段积分顶满后长时间维持 100% 输出(此前实测目标 50℃ 冲到 59℃) */
float err = s_target - t;
float u_noi = HEATER_PID_KP * err
+ HEATER_PID_KD * (err - s_prev_err); /* 比例 + 微分 */
float integ_new = s_integral + err;
float u = u_noi + HEATER_PID_KI * integ_new;
/* 超温区间:执行器只能加热不能制冷,输出必须为 0;
积分按负误差回退,为温度回落后的重新加热保留正确初始态 */
if (err < 0.0f) {
s_integral = integ_step(err);
s_prev_err = err;
s_duty = 0;
return;
}
/* 位置式 PID(带抗积分饱和) */
float u = HEATER_PID_KP * err
+ HEATER_PID_KD * (err - s_prev_err)
+ HEATER_PID_KI * integ_step(err);
if (u > 100.0f) {
u = 100.0f;
if (err < 0.0f) s_integral = integ_new; /* 超上限:仅误差为负(降温方向)才吸收 */
} else if (u < 0.0f) {
u = 0.0f;
if (err > 0.0f) s_integral = integ_new; /* 超下限:仅误差为正(升温方向)才吸收 */
/* 饱和:err>=0 分支积分不吸收(防 windup) */
} else {
s_integral = integ_new; /* 未饱和:正常积分 */
s_integral = integ_step(err);
}
s_prev_err = err;
/* 输出量化到 10% 档位(四舍五入) */
s_duty10 = (uint8_t)((u + (float)HEATER_PWM_STEP_PCT / 2.0f) / (float)HEATER_PWM_STEP_PCT);
/* 温和带:距目标不足 HEATER_SLOWBAND_C 时,比例/微分即时响应受限
(小功率缓升,减少进带蓄热、降低惯性过冲);积分项不受限——
稳态维持功率由积分缓慢顶到实际需要值 */
if (err < HEATER_SLOWBAND_C) {
float u_pd = HEATER_PID_KP * err + HEATER_PID_KD * (err - s_prev_err);
if (u_pd > (float)HEATER_SLOWBAND_DUTY) {
u_pd = (float)HEATER_SLOWBAND_DUTY;
}
if (u_pd < 0.0f) {
u_pd = 0.0f;
}
u = u_pd + HEATER_PID_KI * s_integral;
if (u > 100.0f) {
u = 100.0f;
}
}
/* 输出量化到档位(四舍五入) */
s_duty = (uint8_t)((u + (float)HEATER_PWM_STEP_PCT / 2.0f) / (float)HEATER_PWM_STEP_PCT);
}
/******************************************************************************/
/* 公共接口 */
void heat_task(void) {
uint8_t prev_duty = s_duty10;
uint8_t prev_duty = s_duty;
/* PWM 载波相位推进(10ms tick) */
s_pwm_phase_ms += 10U;
@@ -131,16 +164,16 @@ void heat_task(void) {
if (s_pid_divider >= HEATER_PID_PERIOD_MS) {
s_pid_divider = 0U;
pid_compute();
if (s_duty10 != prev_duty) {
if (s_duty != prev_duty) {
s_pwm_phase_ms = 0U; /* 占空变更:从新周期起点开始,响应即时 */
}
}
/* 按占空档位输出:0 档常关、10 档常开、其余按相位斩波 */
uint16_t on_ms = (uint16_t)s_duty10 * (HEATER_PWM_PERIOD_MS / 10U);
if (s_duty10 == 0U) {
/* 按占空档位输出:0 档常关、满档常开、其余按相位斩波 */
uint16_t on_ms = (uint16_t)s_duty * (HEATER_PWM_PERIOD_MS / HEATER_PWM_STEPS);
if (s_duty == 0U) {
heat_out(false);
} else if (s_duty10 >= 10U) {
} else if (s_duty >= HEATER_PWM_STEPS) {
heat_out(true);
} else {
heat_out(s_pwm_phase_ms < on_ms);
@@ -151,7 +184,7 @@ void heat_set_enable(bool on) {
if (s_enable == on) return; /* 幂等 */
s_enable = on;
if (!on) {
s_duty10 = 0;
s_duty = 0;
pid_reset();
heat_out(false);
}
@@ -170,5 +203,5 @@ float heat_get_target(void) {
}
uint8_t heat_get_duty(void) {
return s_duty10;
return s_duty;
}
+3 -3
View File
@@ -39,9 +39,9 @@
| `M2` | 查询上报状态 | `D5 01 02 4D 32 82` | `ON <ms>` 或 `OFF` |
| `M2S500` | 设上报间隔并开启 | `D5 01 06 4D 32 53 35 30 30 6E` | `ok`;范围 100~60000ms,10ms 步进 |
| `M2S0` | 停止上报 | `D5 01 04 4D 32 53 30 07` | `ok` |
| `M3` | 查询加热状态 | `D5 01 02 4D 33 83` | `ON` / `OFF`(预留,当前仅软件状态) |
| `M3S1` | 开启加热 | `D5 01 04 4D 33 53 31 09` | `ok`(预留) |
| `M3S0` | 关闭加热 | `D5 01 04 4D 33 53 30 08` | `ok`(预留) |
| `M3` | 查询加热状态 | `D5 01 02 4D 33 83` | `ON` / `OFF`(加热使能状态) |
| `M3S1` | 开启加热 | `D5 01 04 4D 33 53 31 09` | `ok`(PID 闭环真实控温) |
| `M3S0` | 关闭加热 | `D5 01 04 4D 33 53 30 08` | `ok` |
| `M4` | 查询目标温度 | `D5 01 02 4D 34 84` | ASCII 摄氏度(如 `120.0`) |
| `M4S120` | 设定目标温度 | `D5 01 06 4D 34 53 31 32 30 6E` | `ok`;整数摄氏度 0~200,越界回 PARAM_ERR |
| `M888` | 软件复位 | `D5 01 04 4D 38 38 38 FA` | `ok` 后复位(慎发) |
-76
View File
@@ -1,76 +0,0 @@
# -*- coding: utf-8 -*-
"""FilmForge PID anti-windup verification: target 50C, track steady-state."""
import serial
import time
import threading
def sum8(seq):
return sum(seq) & 0xFF
def frame(cmd):
p = cmd.encode()
body = bytes([0xD5, 0x01, len(p)]) + p
return body + bytes([sum8(body[1:])])
rx = bytearray()
stop = threading.Event()
def reader(buf, handle):
while not stop.is_set():
data = handle.read(256)
if data:
buf.extend(data)
handle.close()
def frames(buf):
out, i, b = [], 0, bytes(buf)
while i < len(b):
if b[i] == 0xB5 and i + 3 < len(b):
total = 3 + b[i + 2] + 1
if i + total <= len(b):
out.append(b[i:i + total]); i += total; continue
i += 1
return out
s5 = serial.Serial('COM5', 115200, timeout=0.5)
t = threading.Thread(target=reader, args=(rx, s5))
t.start()
time.sleep(0.5)
def cmd(c, wait=0.8):
rx.clear()
s5.write(frame(c))
time.sleep(wait)
fs = frames(rx)
payload = [f[3:3+f[2]].decode('ascii','replace') for f in fs if len(f) >= 4 and f[1] == 0xF0]
print('[%s] resp: %s' % (c, payload if payload else '<none>'))
cmd('M4S50')
cmd('M3S1')
cmd('M2S500')
print('--- heating to 50C, 150s sampling ---')
t0 = time.time()
last_print = -10
peak = 0
while time.time() - t0 < 150:
time.sleep(2.0)
fs = frames(rx)
vals = []
for f in fs:
if len(f) >= 4 and f[1] == 0xF0 and f[2] >= 2:
try:
vals.append(float(f[3:3+f[2]]))
except ValueError:
pass
if vals:
peak = max(peak, max(vals))
dt = time.time() - t0
if dt >= last_print + 10 and vals:
print(' t+%03ds %.1f C (peak %.1f)' % (dt, vals[-1], peak))
last_print = dt - (dt % 10)
cmd('M3S0', wait=1.0)
cmd('M2S0')
stop.set(); t.join()
print('PEAK=%0.1f (target 50, overshoot %+.1f)' % (peak, peak - 50))
print('DONE')