Tools: PID debug toolchain from V0.1.x bring-up

test_run.py: judged closed-loop test over COM5 (fixed sequence, J1~J4
This commit is contained in:
2026-10-15 20:55:29 +08:00
parent 8900240c65
commit 2f43efb94b
7 changed files with 618 additions and 0 deletions
+18
View File
@@ -0,0 +1,18 @@
# -*- coding: utf-8 -*-
"""Parse live temperature frames from test_events.log tail (test writes it live)."""
import re
import sys
n = int(sys.argv[1]) if len(sys.argv) > 1 else 60
rows = []
for ln in open("test_events.log", encoding="utf-8", errors="replace"):
m = re.match(r"^([\d.]+) . ([\d.]+) \[B5", ln)
if m:
rows.append((float(m.group(1)), float(m.group(2))))
print(f"{len(rows)} temp frames total")
if len(rows) > 10:
t1, v1 = rows[8]
t2, v2 = rows[-1]
print(f"{v1:.1f}C@{t1:.0f}s -> {v2:.1f}C@{t2:.0f}s slope={(v2-v1)/(t2-t1):.3f} C/s")
for t, v in rows[-8:]:
print(f" t={t:7.1f}s {v:6.1f}C")
+51
View File
@@ -0,0 +1,51 @@
# -*- coding: utf-8 -*-
"""Decode obs_samples.txt (openocd mrw words, V0.1.2 layout) into quantities."""
import struct
import sys
def f(w):
return struct.unpack("<f", struct.pack("<i", int(w)))[0]
def u8(word, byte):
return (int(word) >> (8 * byte)) & 0xFF
path = sys.argv[1] if len(sys.argv) > 1 else "obs_samples.txt"
rows = []
for ln in open(path, encoding="utf-8"):
parts = ln.split()
if len(parts) != 11 or not parts[0].isdigit():
continue
v = parts
rows.append({
"t": int(v[0]) / 1000.0,
"stash_T": f(v[1]), "stash_I": f(v[2]), "fault": u8(v[3], 0),
"prev": f(v[4]), "integ": f(v[5]), "duty": u8(v[6], 1),
"ramp": f(v[7]), "tgt": f(v[8]), "en": u8(v[9], 0), "temp": f(v[10]),
})
if not rows:
print("no rows")
sys.exit(1)
t0 = rows[0]["t"]
print(f"{'t':>7} {'temp':>7} {'tgt':>6} {'ramp':>7} {'duty%':>5} {'integ':>8} {'prev':>7} {'fl':>3} {'en'}")
for r in rows:
print(f"{r['t']-t0:7.1f} {r['temp']:7.2f} {r['tgt']:6.1f} {r['ramp']:7.2f} "
f"{r['duty']*5:5d} {r['integ']:8.2f} {r['prev']:7.2f} {r['fault']:3d} {r['en']}")
print("\n-- events (duty/fault/integ-zero 变化) --")
last = None
for r in rows:
key = (r["fault"], r["duty"], r["integ"] == 0.0)
if last is not None and key != last[0]:
print(f"t={r['t']-t0:7.1f}s fault={r['fault']} duty={r['duty']*5}% integ={r['integ']:.2f} "
f"temp={r['temp']:.2f}")
last = (key, r)
n_fault = sum(1 for r in rows if r["fault"] > 0)
temps = [r["temp"] for r in rows]
ints = [r["integ"] for r in rows if r["en"]]
print(f"\nsamples={len(rows)} fault_beats={n_fault} temp[{min(temps):.2f},{max(temps):.2f}] "
f"integ[{min(ints) if ints else 0:.1f},{max(ints) if ints else 0:.1f}] (en 段)")
+29
View File
@@ -0,0 +1,29 @@
# -*- coding: utf-8 -*-
"""Decode the LAST N raw rows of obs_samples.txt (guaranteed current writer)."""
import struct
import sys
def f(w):
w = int(w) & 0xFFFFFFFF
return struct.unpack("<f", struct.pack("<I", w))[0]
def u8(w, b):
return (int(w) >> (8 * b)) & 0xFF
n = int(sys.argv[1]) if len(sys.argv) > 1 else 400
stride = int(sys.argv[2]) if len(sys.argv) > 2 else 10
lines = open("obs_samples.txt", encoding="utf-8").read().splitlines()
rows = [l.split() for l in lines if len(l.split()) == 11 and l.split()[0].isdigit()]
rows = rows[-n:]
t0 = int(rows[0][0])
print(f"{len(rows)} rows (last), stride {stride}:")
for r in rows[::stride]:
print(f"t={(int(r[0])-t0)/1000:7.1f} temp={f(r[10]):6.2f} duty={u8(r[6],1)*5:3d}% "
f"integ={f(r[5]):6.2f} fault={u8(r[3],0)} en={u8(r[9],0)}")
temps = [f(r[10]) for r in rows]
ints = [f(r[5]) for r in rows]
print(f"window: temp[{min(temps):.2f},{max(temps):.2f}] p2p={max(temps)-min(temps):.2f} "
f"integ[{min(ints):.1f},{max(ints):.1f}]")
+28
View File
@@ -0,0 +1,28 @@
# DAPLink observer V0.1.12: writes PID state to obs_samples.txt every 500ms
# W1@0x20000460 fault_stash_T(f) W2@0x20000464 fault_stash_I(f) W3@0x20000468 b0=fault_debounce
# W4@0x20000470 prev_temp(f) W5@0x20000474 integral(f) W6@0x20000478 b1=duty
# W7@0x2000047c ramp_target(f) W8@0x20000480 target(f) W9@0x20000484 b0=enable
# W10@0x200001d0 ntc_last_temp(f)
set LOGF "E:/Hulk-Coding/FilmForge/obs_samples.txt"
set g_f [open $LOGF a]
proc logline {s} { global g_f; puts $g_f $s; flush $g_f }
logline "OBS start [clock milliseconds]"
if {[catch {init} err]} { logline "ERR init: $err" }
while {1} {
if {[catch {
set w1 [mrw 0x20000460]
set w2 [mrw 0x20000464]
set w3 [mrw 0x20000468]
set w4 [mrw 0x20000470]
set w5 [mrw 0x20000474]
set w6 [mrw 0x20000478]
set w7 [mrw 0x2000047c]
set w8 [mrw 0x20000480]
set w9 [mrw 0x20000484]
set w10 [mrw 0x200001d0]
logline "[clock milliseconds] $w1 $w2 $w3 $w4 $w5 $w6 $w7 $w8 $w9 $w10"
} err]} { logline "ERR loop: $err" }
sleep 500
}
+26
View File
@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""Segment climb-rate analysis from test_events.log (2C buckets)."""
import re
rows = []
for ln in open("test_events.log", encoding="utf-8", errors="replace"):
m = re.match(r"^([\d.]+) . ([\d.]+) \[B5", ln)
if m:
rows.append((float(m.group(1)), float(m.group(2))))
if not rows:
print("no frames")
raise SystemExit
t0 = rows[0][0]
print(f"{len(rows)} frames, span {rows[-1][0]-t0:.0f}s")
segs = {}
for t, v in rows:
segs.setdefault(int(v // 2) * 2, []).append((t, v))
for k in sorted(segs):
seg = segs[k]
dt = seg[-1][0] - seg[0][0]
dv = seg[-1][1] - seg[0][1]
rate = dv / dt if dt > 0 else 0
print(f" [{k:3d}-{k+2:3d}C] {dt:6.1f}s rate={rate:+.3f} C/s ({seg[0][1]:.1f}->{seg[-1][1]:.1f})")
+355
View File
@@ -0,0 +1,355 @@
# -*- coding: utf-8 -*-
"""V1(已烧录 0.1.0)vs V2(连续 PI 候选)离线对比仿真。
V1 问题:err<0 硬切 0% + err>=0 只有 Kp*err,稳态维持功率全靠 Ki=0.05 慢慢填
(填到 ~10% 要几分钟)→ 目标附近等效"P+上侧死区"继电器 → ±1℃/15~20s 极限环
(真机 2026-10-15 截图实测形态)。
V2:连续 PID 公式覆盖 err<0(输出钳 [0,100] 天然处理"不能制冷"),
Ki 0.05→0.5、I_MAX 2000→200(Ki*I_MAX=100% 不变)。
热模型按本次实测标定:50℃ 处冷却 ~0.1 C/s、~12% 占空净升 ~0.12 C/s
→ a≈1.9 C/s@100%,b≈0.0036/s(50℃ 维持功率 ≈5%)。
"""
KP, KI, KD = 10.0, 0.05, 30.0 # V1
KP2, KI2, KD2 = 10.0, 0.5, 30.0 # V2 候选
RAMP_CPS = 0.8
SLOWBAND_C, SLOWBAND_DUTY = 7.0, 25.0
I_MAX, I_MAX2 = 2000.0, 200.0
PWM_PERIOD_MS, PWM_STEP, PID_PERIOD_MS = 2000, 5, 500
A, B, TA, TAU_S = 1.9, 0.0036, 22.0, 4.0
def plant_step(Tc, Ts, power, dt):
Tc += (A * power - B * (Tc - TA)) * dt
Ts += (Tc - Ts) / TAU_S * dt
return Tc, Ts
class Base:
_last_power = 0.0 # 上一拍 PWM 输出(传感器加热偏置建模用)
def __init__(self, init_temp=None):
self.enable = False
self.target = 0.0
self.duty = 0
self.integral = 0.0
self.phase_ms = 0
self.divider = 0
self.Tc = TA if init_temp is None else init_temp
self.Ts = self.Tc
def tick(self):
prev_duty = self.duty
self.phase_ms = (self.phase_ms + 10) % PWM_PERIOD_MS
self.divider += 10
if self.divider >= PID_PERIOD_MS:
self.divider = 0
self.pid_compute()
if self.duty > prev_duty:
self.phase_ms = 0
on_ms = self.duty * (PWM_PERIOD_MS // (100 // PWM_STEP))
power = 0.0 if self.duty == 0 else (
1.0 if self.duty >= 100 // PWM_STEP else (1.0 if self.phase_ms < on_ms else 0.0))
self.Tc, self.Ts = plant_step(self.Tc, self.Ts, power, 0.01)
self._last_power = power
return power
def quant(self, u):
u = max(0.0, min(100.0, u))
return int((u + PWM_STEP / 2.0) / PWM_STEP)
def ramp_to(self):
diff = self.target - self.ramp_target
step = RAMP_CPS * (PID_PERIOD_MS / 1000.0)
if diff > step:
self.ramp_target += step
elif diff < -step:
self.ramp_target -= step
else:
self.ramp_target = self.target
def set_enable(self, on):
if self.enable == on:
return
self.enable = on
if on:
self.ramp_target = max(0.0, self.Ts)
self.prev_temp = self.ramp_target
else:
self.ramp_target = self.duty = 0
self.integral = 0.0
class V1(Base):
"""已烧录 0.1.0:err<0 硬切 + 慢积分。"""
def pid_compute(self):
t = self.Ts
if not self.enable:
self.duty, self.integral, self.prev_temp = 0, 0.0, t
return
self.ramp_to()
err = self.ramp_target - t
d_t = t - self.prev_temp
if err < 0:
self.integral = max(0.0, self.integral + err)
self.prev_temp = t
self.duty = 0
return
u_pd = KP * err - KD * d_t
if err < SLOWBAND_C:
u_pd = min(SLOWBAND_DUTY, u_pd)
u_pd = max(0.0, u_pd)
u = u_pd + KI * self.integral
if u < 100.0:
self.integral = min(I_MAX, self.integral + err)
self.prev_temp = t
self.duty = self.quant(u)
class V2(Base):
"""候选:连续 PID(无 err<0 硬切)+ 快积分 + 执行器受限不吸收。
吸收条件:err>0 时仅当 P/D 未被钳(0 < u_pd_raw < 当前上限,即执行器
真实响应误差)才吸收——爬升/温和带钳位段不积分,防瞬态 windup;
err<0 时只要 u>0 就回退(维持功率受 P 负项与 D 共同压制)。
"""
def pid_compute(self):
t = self.Ts
if not self.enable:
self.duty, self.integral, self.prev_temp = 0, 0.0, t
return
self.ramp_to()
err = self.ramp_target - t
d_t = t - self.prev_temp
u_pd_raw = KP2 * err - KD2 * d_t
cap = SLOWBAND_DUTY if err < SLOWBAND_C else 100.0
u_pd = max(0.0, min(cap, u_pd_raw))
u = u_pd + KI2 * self.integral
if (err > 0 and 0.0 < u_pd_raw < cap) or (err < 0 and u > 0.0):
self.integral = max(0.0, min(I_MAX2, self.integral + err))
self.prev_temp = t
self.duty = self.quant(u)
class V3(Base):
"""V3:连续 PID + 积分吸收限定在目标近旁窄带(|err|<ZONE)。
远处/爬升段不积分(slowband 钳位 + P 已经在工作),只有贴近目标才允许
积分建立稳态功率——杜绝瞬态 windup;err<0 且 u>0 时照常回退。
参数从 V3_KP/KI/KD 全局取,便于扫参。"""
def pid_compute(self):
t = self.Ts
if not self.enable:
self.duty, self.integral, self.prev_temp = 0, 0.0, t
return
self.ramp_to()
err = self.ramp_target - t
d_t = t - self.prev_temp
u_pd_raw = V3_KP * err - V3_KD * d_t
cap = SLOWBAND_DUTY if err < SLOWBAND_C else 100.0
u_pd = max(0.0, min(cap, u_pd_raw))
u = u_pd + V3_KI * self.integral
if (0.0 < err < V3_ZONE and 0.0 < u_pd_raw < cap) or (err < 0 and u > 0.0):
self.integral = max(0.0, min(V3_IMAX, self.integral + err))
self.prev_temp = t
self.duty = self.quant(u)
V3_KP, V3_KI, V3_KD, V3_ZONE, V3_IMAX = 10.0, 0.25, 30.0, 2.0, 400.0
class V4(Base):
"""V4(final) = V3 + ① err<0 时 P 负项直接抵扣积分(总量钳 [0,100])
② 积分窄带 2.0->0.6(单次爬升最多喂 ~I=12 ≈ 真机维持功率,防膨胀)
③ 故障无扰恢复(<3s 且 |ΔT|<2℃ 还原暂存积分)
真机根因(2026-10-15 DAPLink 抓证 t=397~486s):积分爬升段膨胀到 65
(维持仅需 ~12)→ 过线后 P 钳 0、Ki·I 顶 10~20% 慢泄 → +2℃ 飙升 →
泄光下潜 → 再爬再喂,自持极限环(无毛刺也发生)。"""
def __init__(self, *a, **kw):
super().__init__(*a, **kw)
self.fault_beats = 0
self.stash_I = 0.0
self.stash_T = 0.0
self.glitch_now = False
self.noise = __import__("random").Random(20261015)
def sensor(self):
t = -273.15 if self.glitch_now else self.Ts
if t > -100.0:
t += self.noise.uniform(-0.05, 0.05) # 读数噪声
t -= 0.15 if self._last_power > 0.5 else 0.0 # 加热时鳍片梯度偏低读
return t
def pid_compute(self):
t = self.sensor()
if t <= -273.0:
if self.fault_beats < 255:
self.fault_beats += 1
if self.fault_beats == 3:
self.stash_I = self.integral
self.stash_T = self.prev_temp
if self.fault_beats >= 3:
self.duty = 0
self.integral = 0.0
self.prev_temp = t
return
if self.fault_beats >= 3:
dur = self.fault_beats * (PID_PERIOD_MS / 1000.0)
if dur <= 3.0 and abs(t - self.stash_T) < 2.0:
self.integral = min(V4_IMAX, self.stash_I)
self.fault_beats = 0
if not self.enable:
self.duty, self.integral, self.prev_temp = 0, 0.0, t
return
self.ramp_to()
err = self.ramp_target - t
d_t = t - self.prev_temp
u_pd_raw = V3_KP * err - V3_KD * d_t
cap = SLOWBAND_DUTY if err < SLOWBAND_C else 100.0
if err >= 0.0:
u_pd = max(0.0, min(cap, u_pd_raw))
u = u_pd + V3_KI * self.integral
else:
u = u_pd_raw + V3_KI * self.integral # P 负项直接抵扣积分
if (0.0 < err < V4_ZONE and 0.0 < u_pd_raw < cap) or (err < 0 and u > 0.0):
self.integral = max(0.0, min(V4_IMAX, self.integral + err))
u = max(0.0, min(100.0, u))
self.prev_temp = t
self.duty = self.quant(u)
V4_ZONE, V4_IMAX = 0.6, 160.0
class V4_OLD(Base):
"""V3 原样(err<0 积分独撑 + 窄带 2.0 + 故障清零重建),毛刺/噪声注入对照。"""
def __init__(self, *a, **kw):
super().__init__(*a, **kw)
self.fault_beats = 0
self.glitch_now = False
self.noise = __import__("random").Random(20261015)
def sensor(self):
t = -273.15 if self.glitch_now else self.Ts
if t > -100.0:
t += self.noise.uniform(-0.05, 0.05)
t -= 0.15 if self._last_power > 0.5 else 0.0
return t
def pid_compute(self):
t = self.sensor()
if t <= -273.0:
if self.fault_beats < 3:
self.fault_beats += 1
if self.fault_beats >= 3:
self.duty = 0
self.integral = 0.0
self.prev_temp = t
return
self.fault_beats = 0
if not self.enable:
self.duty, self.integral, self.prev_temp = 0, 0.0, t
return
self.ramp_to()
err = self.ramp_target - t
d_t = t - self.prev_temp
u_pd_raw = V3_KP * err - V3_KD * d_t
cap = SLOWBAND_DUTY if err < SLOWBAND_C else 100.0
u_pd = max(0.0, min(cap, u_pd_raw))
u = u_pd + V3_KI * self.integral # err<0:P 钳 0,积分独撑
if (0.0 < err < V3_ZONE and 0.0 < u_pd_raw < cap) or (err < 0 and u > 0.0):
self.integral = max(0.0, min(V3_IMAX, self.integral + err))
u = max(0.0, min(100.0, u))
self.prev_temp = t
self.duty = self.quant(u)
def run(cls, label, events, init_temp=None, end=420, plant=None, glitches=()):
global A, B, TAU_S
if plant:
A, B, TAU_S = plant
p = cls(init_temp)
samples = []
idx, t_sim = 0, 0.0
events = sorted(events, key=lambda e: e[0])
while t_sim < end:
if idx < len(events) and t_sim >= events[idx][0] - 1e-9:
_, act = events[idx]
if act == "enable":
p.set_enable(True)
elif act == "disable":
p.set_enable(False)
else:
p.target = act
idx += 1
if hasattr(p, "glitch_now"):
p.glitch_now = any(gs <= t_sim < gs + gd for gs, gd in glitches)
p.tick()
samples.append(p.Ts)
t_sim += 0.01
def stats(t0, t1, tgt):
seg = samples[int(t0 / 0.01):int(t1 / 0.01)]
if not seg:
return
reach = next((i * 0.01 for i, v in enumerate(seg) if abs(v - tgt) <= 0.3), None)
if reach is None:
print(f" [{t0:4.0f}-{t1:4.0f}s tgt={tgt:g}] 未进 ±0.3 带(段内 "
f"{min(seg):.2f}~{max(seg):.2f})")
return
r = int(reach / 0.01)
post = seg[r:]
tail = seg[max(r, len(seg) - 6000):]
print(f" [{t0:4.0f}-{t1:4.0f}s tgt={tgt:g}] 进入±0.3带 {reach:5.1f}s | 到达后 "
f"[{min(post):.2f},{max(post):.2f}] | 末60s [{min(tail):.2f},{max(tail):.2f}] "
f"±{(max(tail)-min(tail))/2:.2f}")
print(f"\n== {label} ==")
for i, (t0, act) in enumerate(events):
if isinstance(act, float) or isinstance(act, int):
t1 = events[i + 1][0] if i + 1 < len(events) else end
stats(t0 + 2, min(t1, end), act)
for gs, gd in glitches:
seg = samples[int(gs / 0.01):int(min(gs + 120, end) / 0.01)]
if seg:
print(f" [glitch @{gs:g}s +{gd:g}s] 120s 窗内 [{min(seg):.2f},{max(seg):.2f}] "
f"max|Δ|={max(abs(v - p.target) for v in seg):.2f}")
return samples
if __name__ == "__main__":
# 基准工况(按实测斜率标定):a=1.9 b=0.0036 tau=4
BASE = (1.9, 0.0036, 4.0)
# 恶劣工况:维持功率 ~11%(b=0.0075)+ 传感器滞后 8s
HARSH = (1.9, 0.0075, 8.0)
import sys
if len(sys.argv) > 1 and sys.argv[1] == "sweep":
# 扫参:V3 参数网格,两工况各跑 50C 阶跃 + 从上方使能
for kp, ki, kd in [(10, 0.5, 30), (10, 0.5, 10), (10, 0.5, 0),
(15, 0.5, 30), (10, 0.25, 30), (10, 1.0, 30),
(10, 0.5, 50), (15, 0.5, 10)]:
V3_KP, V3_KI, V3_KD = float(kp), float(ki), float(kd)
print(f"\n### V3 Kp={kp} Ki={ki} Kd={kd}")
run(V3, f"HARSH 50C 阶跃", [(0, "enable"), (1, 50.0)], end=420, plant=HARSH)
run(V3, f"HARSH 从上方使能", [(0, "enable"), (1, 50.0)], init_temp=51.0, end=420, plant=HARSH)
run(V3, f"BASE 50C 阶跃", [(0, "enable"), (1, 50.0)], end=420, plant=BASE)
sys.exit(0)
# C:镜像本次实测——从 51C(高于目标)使能,目标 50
GL = [(200, 1.4), (320, 1.4)] # 毛刺注入:1.4s NTC 开路 ×2(真机 DAPLink 抓证形态)
EV = [(0, "enable"), (1, 50.0)]
run(V4_OLD, "V3(旧) 毛刺注入 tgt=50(恶劣)", EV, init_temp=51.0, end=460, plant=HARSH, glitches=GL)
run(V4, "V4 毛刺注入 tgt=50(恶劣)", EV, init_temp=51.0, end=460, plant=HARSH, glitches=GL)
run(V4, "V4 毛刺注入 tgt=50(基准)", EV, init_temp=51.0, end=460, plant=BASE, glitches=GL)
# 回归:无毛刺主场景
run(V4, "V4 50C 阶跃(恶劣)", EV, end=420, plant=HARSH)
run(V4, "V4 30C 阶跃(基准)", [(0, "enable"), (1, 30.0)], plant=BASE)
run(V4, "V4 50->40 下调(恶劣)", [(0, "enable"), (1, 50.0), (400, 40.0)], end=700, plant=HARSH)
+111
View File
@@ -0,0 +1,111 @@
# -*- coding: utf-8 -*-
"""FilmForge 温控闭环自动测试(COM5,RS485)。
固定测试序列:M999(版本) -> M2S500(0.5s 上报) -> M4S50(目标) -> M3S1(加热)
-> 采样 DURATION 秒 -> M3S0/M2S0 收尾。
不设温度上限(用户明确:真机靠物理温度开关保护,测试环境上限由用户控制)。
判定标准(脚本自动输出,不掺人工解释):
J1 上报流:>2 帧/s 连续; J2 稳态窗(最后 300s):|均值-目标|<=0.5C
J3 稳态窗极值:[目标-0.5, 目标+0.5](用户规格 ±0.5C)
J4 无 OPEN/SHRT 故障值
输出:test_data.csv(t,temp 全序列)+ test_events.log(全部收发)+ 终端摘要。
"""
import sys
import time
sys.path.insert(0, "GUI")
import filmforge_rs485 as ff
PORT = "COM5"
TARGET = 50.0
DURATION = float(sys.argv[1]) if len(sys.argv) > 1 else 600.0
rows = [] # (t, temp or None)
events = []
faults = []
version = None
t0 = time.time()
def on_event(line):
events.append((time.time() - t0, line))
with open("test_events.log", "a", encoding="utf-8") as f:
f.write(f"{time.time():.3f} {line}\n")
def on_status(msg):
on_event(f"[STATUS] {msg}")
def on_frame(frame, t_rx):
payload = frame.payload if isinstance(frame.payload, str) else \
frame.payload.decode("ascii", errors="replace")
try:
val = float(payload)
except ValueError:
val = None
if val is not None and val > -100.0: # 正常温度
rows.append((time.time() - t0, val))
elif val is not None: # OPEN=-273.15 / SHORT=-274.15
faults.append((time.time() - t0, val))
on_event(f"[FAULT] {payload}")
else:
if payload.startswith("v"):
global version
version = payload
on_event(f"[FRAME] {payload!r}")
link = ff.FF485(on_event=on_event, on_status=on_status, on_frame=on_frame)
link.open(PORT)
time.sleep(0.5)
if not link.is_connected():
print("OPEN COM5 FAILED")
sys.exit(1)
# ---- 序列 ----
link.send("M999")
time.sleep(1.0)
link.send("M2S500")
time.sleep(1.0)
link.send("M4S50")
time.sleep(1.0)
link.send("M3S1")
on_event("[SEQ] heat on, target 50")
t_end = time.time() + DURATION
while time.time() < t_end:
time.sleep(1.0)
link.send("M3S0")
time.sleep(0.5)
link.send("M2S0")
time.sleep(1.0)
link.close()
# ---- 判定输出 ----
with open("test_data.csv", "w", encoding="utf-8") as f:
f.write("t,temp\n")
for t, v in rows:
f.write(f"{t:.2f},{v:.2f}\n")
print(f"version={version} frames={len(rows)} faults={len(faults)}")
if rows:
temps = [v for _, v in rows]
print(f"info global_max={max(temps):.2f} global_min={min(temps):.2f}")
settle = [v for t, v in rows if t >= rows[-1][0] - 300.0] or temps
mean = sum(settle) / len(settle)
print(f"J2 settle_mean={mean:.2f} (|{-TARGET + mean:.2f}|<=0.5? {'PASS' if abs(mean - TARGET) <= 0.5 else 'FAIL'})")
ok3 = max(settle) <= TARGET + 0.5 and min(settle) >= TARGET - 0.5
print(f"J3 settle_min={min(settle):.2f} settle_max={max(settle):.2f} "
f"p2p={max(settle) - min(settle):.2f} ([{TARGET - 0.5:.1f},{TARGET + 0.5:.1f}]? {'PASS' if ok3 else 'FAIL'})")
cross = next((t for t, v in rows if abs(v - TARGET) <= 0.3), None)
if cross:
post = [(t, v) for t, v in rows if t >= cross]
print(f"reach={cross:.0f}s post_max={max(v for _, v in post):.2f} "
f"(overshoot {max(v for _, v in post) - TARGET:+.2f})")
print(f"J4 faults={'PASS' if not faults else 'FAIL(' + str(len(faults)) + ')'}")
print(f"J1 report_rate={'PASS' if len(rows) >= DURATION * 1.6 else 'FAIL'}")