Heater: setpoint ramping to suppress thermal-lag overshoot

This commit is contained in:
2026-10-10 13:59:43 +08:00
parent 5b24b956d2
commit 12ba9de05f
11 changed files with 353 additions and 214 deletions
+37 -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,37 @@
安全保护齐全;实测 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:排版调整——串口连接卡片从右列移到左列顶部,左列图表卡/右列日志卡
弹性拉伸填充,两列底边对齐。
- selftest 全过(累计新增 14 项回归用例);exe 已本地打包冒烟,Gitea 发布
待用户确认后执行。
遗留见其 README"已知事项":应用图标、USB-RS485 适配器更换。
- [ ] **LED 引脚确认**:FilmForge 自制板 LED 引脚未定(当前沿用模板默认 PA15,
仅为构建占位),按原理图修改。
- [ ] **delay_10us 校准**:M4@168MHz 系数为估算值(/300000),需实测修正。
+28 -12
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,28 @@ 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,真机实测确认)。
- Fluent 控件自定义样式必须走 `setCustomStyleSheet`(覆盖 hover/pressed/disabled),
直接 `setStyleSheet` 会在切主题时被库重刷(DLPilot 同款坑);标题字号走 QFont。
- 图表刷新双节流:数据到达即重绘(≤10Hz)+ 0.5s 时间轴推进;“全部”窗口超 4000 点
@@ -67,8 +82,9 @@ python publish_gitea.py release/FilmForge-GUI-v<版本>-win64.zip --replace --in
## 已知事项
- 当前 USB-RS485 适配器自动换向偶发丢响应(固件侧已立“适配器更换”TODO):
表现为个别查询 1.2s 超时,队列自动清理后轮询自然重试,曲线无碍。
- 加热(M3)/目标温度(M4)为固件预留:仅软件状态(ack+可回读),硬件执行体接入后
协议不变,本工具无需改动即可联调真实控温闭环。
- 当前 USB-RS485 适配器自动换向偶发丢响应/大延迟(固件侧已立”适配器更换”TODO):
查询 1.2s 超时提示后重发即可;设定命令的晚到 ok 已自动补偿应用;曲线走上报帧
(设备单向推送),不受影响。
- 加热(M3S1)为真实 PID 闭环:开启后模具真实升温(NTC 故障/过温 150℃ 固件自动
断电保护,上电默认关闭);当前占空比无查询命令,从温度曲线间接观察。
- 打包暂无应用图标(`logos/filmforge.ico` 缺位,build_exe.py 自动跳过)。
+188 -93
View File
@@ -5,7 +5,7 @@
协议与串口模块(filmforge_rs485)保持不变, 本文件只做呈现层:
PySide6-Fluent-Widgets 卡片式双栏仪表盘 + 自绘滚动温度曲线(无第三方图表依赖)。
闭环监视 = 实测温度曲线 + 目标温度阶梯线同图 + 加热/上报/传感器状态联动;
设备未开启 M2 上报时可由上位机周期轮询 M1 补齐曲线。
温度监控走固件 M2 周期上报帧(M2S<ms> 开启), M1 仅作手动单点查询。
"""
import ctypes
@@ -60,7 +60,7 @@ import config
import filmforge_rs485 as ff
APP_NAME = "FilmForge"
APP_VERSION = "0.1.0"
APP_VERSION = "0.2.5"
REPO_URL = "https://gitea.hulk.wang/hulk/FilmForge"
MONO = QFont("Consolas", 10)
CHIP_IDLE = "#666666"
@@ -153,7 +153,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 +164,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 +198,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):
@@ -373,9 +380,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()
@@ -481,8 +492,40 @@ class App(QMainWindow):
scroll.setWidget(host)
return scroll
# ---------------- 左列: 监视 ----------------
# ---------------- 左列: 串口 + 监视 ----------------
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 +587,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 +609,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 +627,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("命令 · 预设 / 自定义")
@@ -659,7 +667,7 @@ class App(QMainWindow):
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 +692,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 +789,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,8 +830,8 @@ class App(QMainWindow):
def _on_status_main(self, msg):
self.lbl_st.setText(msg)
if msg.startswith("已连接"):
self._log(f"── {msg} ──")
# 连接/断开/丢失全部入日志: 状态标签只留最后一条, 日志才可回溯排查
self._log(f"── {msg} ──")
# ================= 帧派发 =================
def _handle_frame(self, fr, t_rx):
@@ -843,6 +859,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 +869,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 +917,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):
@@ -947,14 +982,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 +1023,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 +1055,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 +1123,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 +1146,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 +1157,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):
@@ -1189,10 +1288,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())
+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,
}
+10 -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'),
@@ -348,7 +349,10 @@ class FF485:
cmd, frame = item
self.ser.write(frame)
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):
+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) 会复位设备, 界面内已做二次确认。
"""
+54
View File
@@ -150,6 +150,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(),
+3 -2
View File
@@ -152,9 +152,10 @@ void usart1_irq_handler(void);
#define HEATER_PWM_PERIOD_MS 1000U /* PWM 载波周期(1s,恒 1Hz 开关频率) */
#define HEATER_PWM_STEP_PCT 10U /* 占空比步进(每档脉宽 = 100ms) */
#define HEATER_PID_PERIOD_MS 500U /* PID 计算周期 */
#define HEATER_PID_KP 8.0f /* PID 参数:初值,需上板实测整定 */
#define HEATER_PID_KP 4.0f /* PID 参数:小 Kp 换取低过冲(接近目标时减速) */
#define HEATER_PID_KI 0.05f
#define HEATER_PID_KD 0.0f /* NTC 噪声对微分放大,先禁用 */
#define HEATER_PID_KD 30.0f /* 微分阻尼:按误差变化率提前刹车(抑制爬升惯性) */
#define HEATER_RAMP_CPS 0.8f /* 目标斜坡速率(℃/s):内部设定值缓升,抑制热滞后过冲 */
#define HEATER_OVERTEMP_C 150.0f /* 过温强制关断阈值(滞回保护) */
#define HEATER_OVERTEMP_HYST 5.0f /* 过温恢复滞回 */
+24 -3
View File
@@ -19,7 +19,8 @@
/* 内部状态 */
static bool s_enable = false; /* 加热使能(M3) */
static float s_target = 0.0f; /* 目标温度(M4) */
static float s_target = 0.0f; /* 目标温度(M4,外部设定值) */
static float s_ramp_target = 0.0f; /* 内部斜坡目标(以 HEATER_RAMP_CPS 滑向 s_target) */
static uint8_t s_duty10 = 0; /* 量化占空档位 0~10(x10%) */
static bool s_overtemp = false; /* 过温锁存(滞回恢复) */
static float s_integral = 0.0f; /* PID 积分项 */
@@ -91,9 +92,22 @@ static void pid_compute(void) {
return;
}
/* 目标斜坡:内部设定值以受限速率滑向 s_target。
满功率爬升段 ramp 超前于实际温升(不受影响);接近目标后 ramp 先到站停住,
误差随实际温度逼近而平滑收缩 -> 功率缓退,抑制热传导滞后造成的过冲。 */
float diff = s_target - s_ramp_target;
float step = HEATER_RAMP_CPS * (HEATER_PID_PERIOD_MS / 1000.0f);
if (diff > step) {
s_ramp_target += step;
} else if (diff < -step) {
s_ramp_target -= step;
} else {
s_ramp_target = s_target;
}
/* 位置式 PID(带抗积分饱和):输出饱和时积分仅沿退饱和方向更新,
防止升温段积分顶满后长时间维持 100% 输出(此前实测目标 50℃ 冲到 59℃) */
float err = s_target - t;
float err = s_ramp_target - t;
float u_noi = HEATER_PID_KP * err
+ HEATER_PID_KD * (err - s_prev_err); /* 比例 + 微分 */
float integ_new = s_integral + err;
@@ -150,7 +164,14 @@ void heat_task(void) {
void heat_set_enable(bool on) {
if (s_enable == on) return; /* 幂等 */
s_enable = on;
if (!on) {
if (on) {
/* 从当前实际温度起坡,避免内部目标从 0 跳变造成功率冲击 */
s_ramp_target = ntc_get_temperature();
if (s_ramp_target < 0.0f) {
s_ramp_target = 0.0f; /* 传感器故障标记值非法,退化为 0 */
}
} else {
s_ramp_target = 0.0f;
s_duty10 = 0;
pid_reset();
heat_out(false);
+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')