hulk/HW-Debug-Tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 8 Wiki Activity

feat(DLPilot): 电源自检补全 FPDO 挡位判定; 功率检测实现

Browse Source
This commit is contained in:
hulk
2026-09-18 00:58:42 +08:00
parent 8e6107b490
commit 9585fae947
4 changed files with 141 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DLPilot/src/app_gui.py
+119 -57
View File
@@ -95,12 +95,17 @@ class App(ctk.CTk):
self._orig_modes = {} # device -> 切改前的原始 DEVMODEW
# 光机控制链路: 回调发生在串口线程, 经队列投递回 GUI 主线程轮询
self._dlpc_log_q = queue.Queue()
self.dlpc = dlpc_rs485.DLPC485(self._dlpc_on_event, self._dlpc_on_status)
self.dlpc = dlpc_rs485.DLPC485(self._dlpc_on_event, self._dlpc_on_status,
on_frame=self._dlpc_on_frame)
self._dlpc_connected = False
self._poll_job = None
self._tl_order = [] # 温度查询应答顺序: 'DMD'/'LAMP' 先到先得
self._pd_seq = [] # 电源自检待执行步骤: [(命令, kind), ...]
self._pd_values = {} # 自检采集值: {'ma': 挡位电流, 'mv': 实测电压}
self._pd_timer = None # 自检单步超时定时器
self._pwr_seq = [] # 功率检测待执行步骤
self._pwr_values = {} # 功率检测采集值: {'mv', 'ma'}
self._pwr_timer = None
self._build()
self._load_displays()
# 主窗口按 ESC 也能关闭所有投屏(投屏窗口失焦时的兜底出口)
@@ -340,8 +345,10 @@ class App(ctk.CTk):
text_color=CHIP_IDLE_TEXT)
self.chip_pd.pack(side='left')
self.var_fpdo = tk.StringVar(value='FPDO: —')
ctk.CTkLabel(rowp1, textvariable=self.var_fpdo, anchor='w',
text_color='#3B8ED0').pack(side='left', padx=(12, 0))
self.chip_fpdo = ctk.CTkLabel(rowp1, textvariable=self.var_fpdo, width=150,
height=26, corner_radius=6, fg_color=CHIP_IDLE,
text_color=CHIP_IDLE_TEXT, anchor='w', padx=8)
self.chip_fpdo.pack(side='left', padx=(12, 0))
rowp2 = ctk.CTkFrame(c, fg_color='transparent')
rowp2.pack(fill='x', pady=(6, 0))
ctk.CTkButton(rowp2, text='功率检测', width=90,
@@ -349,8 +356,7 @@ class App(ctk.CTk):
self.var_pwr = tk.StringVar(value='电压: — 电流: — 功率: —')
ctk.CTkLabel(rowp2, textvariable=self.var_pwr, height=28,
corner_radius=6, fg_color=('#F2F2F2', '#181818'),
anchor='w').pack(side='left', fill='x', expand=True,
padx=(8, 0))
anchor='w', padx=8).pack(side='left', padx=(8, 0))
# -- 光机命令 --
# 标题行右侧: 电流实时读数 + 查电流按钮(okR..G..B.. 应答解析)
@@ -713,6 +719,13 @@ class App(ctk.CTk):
# ============================================================
# 光机控制(RS485)
# ============================================================
def _dlpc_on_frame(self, fr):
"""结构化应答帧(串口线程): 电源自检/功率检测序列进行中时路由给状态机"""
if self._pd_seq:
self._dlpc_log_q.put(('pdframe', fr))
elif self._pwr_seq:
self._dlpc_log_q.put(('pwrframe', fr))
def _dlpc_on_event(self, line):
# 温度查询应答: okTL:<0.1°C>, 依发送顺序归入 DMD/LAMP 实时显示框
m = re.search(r'← okTL:(-?\d+)', line)
@@ -765,10 +778,12 @@ class App(ctk.CTk):
self.var_lamp.set(text)
elif kind == 'current':
self.var_current.set(text)
self._dlpc_log(text)
if (self._pd_seq and kind == 'event'
and text.startswith('← ')):
self._pd_on_reply(text) # 电源自检: 应答路由到状态机
elif kind == 'pdframe':
self._pd_step_done(payload.type_code == 0xF0, payload)
elif kind == 'pwrframe':
self._pwr_step_done(payload)
else:
self._dlpc_log(text)
except queue.Empty:
pass
connected = self.dlpc.is_connected()
@@ -853,14 +868,15 @@ class App(ctk.CTk):
# -- 电源检查: 自检序列 + 功率读数 --
def _pd_selftest(self):
"""电源自检: 按序自动执行 M301/M302(电压检测命令待固件定义后在序列追加),
每步应答刷新状态位: 绿=正常回包, 红=错误/超时, 灰=未检测"""
"""电源自检: 依序 M301(协议) → M302(挡位最大电流) → M303(实测母线电压),
应答帧驱动序列前进(成功=TYPE 0xF0), 结束后按 _pd_fpdo_verdict 判定 FPDO"""
if not self.dlpc.is_connected():
self.lbl_dlpc_st.configure(text='未连接串口', text_color='#E07A3F')
return
if self._pd_seq:
return # 进行中, 忽略重复触发
self._pd_seq = [('M301', 'pd'), ('M302', 'fpdo')]
self._pd_seq = [('M301', 'pd'), ('M302', 'fpdo_ma'), ('M303', 'meas_mv')]
self._pd_values = {}
self._pd_chip_reset()
self._dlpc_log('电源自检: 开始')
self._pd_step_next()
@@ -868,72 +884,116 @@ class App(ctk.CTk):
def _pd_chip_reset(self):
self.chip_pd.configure(fg_color=CHIP_IDLE,
text_color=CHIP_IDLE_TEXT, text='PD')
self.chip_fpdo.configure(fg_color=CHIP_IDLE, text_color=CHIP_IDLE_TEXT)
self.var_fpdo.set('FPDO: —')
def _pd_step_next(self):
if not self._pd_seq:
self._pd_fpdo_verdict()
self._dlpc_log('电源自检: 完成')
return
self.dlpc.send(self._pd_seq[0][0])
self._pd_timer = self.after(3000, lambda: self._pd_step_done(False, '超时'))
self._pd_timer = self.after(3000, lambda: self._pd_step_done(False, None))
def _pd_on_reply(self, line):
self._pd_step_done(self._pd_reply_ok(line), line)
@staticmethod
def _pd_reply_ok(line):
"""应答行 -> 是否成功帧(TYPE=0xF0): 成功行正文是 payload 原文(如 'PD'/'okTL:246',
不必有 ok 前缀), 失败行正文是 TYPE 译码词或带 err:, 据此区分"""
body = line[2:].split(' [')[0].strip()
if 'err:' in body:
return False
return not any(n in body for n in dlpc_rs485.TYPE_NAMES.values()
if n != '成功') # '成功' 不会出现在日志正文, 仅防 payload 撞词
def _pd_step_done(self, ok, line=''):
def _pd_step_done(self, ok, fr):
if not self._pd_seq:
return
if self._pd_timer:
self.after_cancel(self._pd_timer)
self._pd_timer = None
_cmd, kind = self._pd_seq.pop(0)
payload = ''
if ok:
body = line[2:].split(' [')[0].strip() # 去 '← ' 与尾部 hex
payload = '' if body.lower() == 'ok' else \
(body[2:].strip() if body[:2].lower() == 'ok' else body)
if kind == 'pd':
self.chip_pd.configure(fg_color=CHIP_OK if ok else CHIP_ERR,
text_color='#FFFFFF',
text=payload if ok and payload else 'PD')
elif kind == 'fpdo':
mv = re.search(r'(\d+(?:\.\d+)?)\s*V', payload, re.I)
ma = re.search(r'(\d+(?:\.\d+)?)\s*A', payload, re.I)
if not ok:
self.var_fpdo.set('FPDO: 应答错误')
elif mv and ma: # 电压 + 电源宣称最大电流 -> FPDO 功率
v, a = float(mv.group(1)), float(ma.group(1))
self.var_fpdo.set(f'FPDO: {v:g}V / {a:g}A ({v * a:.0f}W)')
elif payload:
self.var_fpdo.set(f'FPDO: {payload}')
cmd, kind = self._pd_seq.pop(0)
raw = dlpc_rs485.payload_bytes(fr) if fr is not None else b''
if ok and kind == 'pd':
proto = raw.decode('ascii', 'replace').strip()
good = proto in ('PD', 'EPR')
self.chip_pd.configure(fg_color=CHIP_OK if good else CHIP_ERR,
text_color='#FFFFFF', text=proto or 'PD')
elif ok and kind == 'fpdo_ma' and len(raw) == 2:
self._pd_values['ma'] = dlpc_rs485.payload_u16_be(raw)
elif ok and kind == 'meas_mv' and len(raw) == 2:
self._pd_values['mv'] = dlpc_rs485.payload_u16_be(raw)
if ok:
self._pd_step_next()
else: # 出错即中止, 后续状态位保持未检测灰
else: # 出错即中止
self._pd_seq.clear()
why = '超时无应答' if not line.startswith('← ') else '应答错误'
self._dlpc_log(f'电源自检: {why}, 序列中止')
if kind == 'pd':
self.chip_pd.configure(fg_color=CHIP_ERR, text_color='#FFFFFF',
text='ERR')
elif kind == 'fpdo_ma':
self.chip_fpdo.configure(fg_color=CHIP_ERR, text_color='#FFFFFF')
self.var_fpdo.set('FPDO: ERR')
self._dlpc_log('电源自检: 应答错误/超时, 序列中止')
def _pd_fpdo_verdict(self):
"""FPDO 判定: 电压挡位由 M303 实测母线电压就近判定(标准 PD 固定挡
5/9/12/15/20/28V, 容差 ±5%; CH224Q 0x0A 只写不可读, 挡位只能实测判定),
电流为 M302 挡位最大电流。默认挡 12V 且 ≥3A -> 绿底;
挡位不符(如 9V 挡) -> 红底; 实测电压不落在任何标准挡 -> ERR"""
mv = self._pd_values.get('mv')
ma = self._pd_values.get('ma')
if mv is None or ma is None:
self.chip_fpdo.configure(fg_color=CHIP_ERR, text_color='#FFFFFF')
self.var_fpdo.set('FPDO: ERR')
return
gear = next((g for g in (5, 9, 12, 15, 20, 28)
if abs(mv - g * 1000) <= g * 50), None) # ±5% = g*50 mV
a_txt = f'{ma / 1000:g}'
if gear is None:
self.chip_fpdo.configure(fg_color=CHIP_ERR, text_color='#FFFFFF')
self.var_fpdo.set('FPDO: ERR')
elif gear == 12 and ma >= 3000:
self.chip_fpdo.configure(fg_color=CHIP_OK, text_color='#FFFFFF')
self.var_fpdo.set(f'FPDO: 12V@{a_txt}A')
else:
self.chip_fpdo.configure(fg_color=CHIP_ERR, text_color='#FFFFFF')
self.var_fpdo.set(f'FPDO: {gear}V@{a_txt}A')
def _pwr_query(self):
"""功率检测: 命令待固件实现, 先占位(定义后在此发送并接 _pwr_update)"""
"""功率检测: M303 实测电压 + M304 实测电流(按有符号) -> 换算功率"""
if not self.dlpc.is_connected():
self.lbl_dlpc_st.configure(text='未连接串口', text_color='#E07A3F')
return
self._dlpc_log('功率检测: 命令待固件实现')
if self._pwr_seq:
return
self._pwr_seq = [('M303', 'mv'), ('M304', 'ma')]
self._pwr_values = {}
self._dlpc_log('功率检测: 开始')
self._pwr_step_next()
def _pwr_update(self, volts, amps):
"""功率读数框更新入口: 电压/电流到手后换算功率并刷新"""
self.var_pwr.set(f'电压: {volts:.1f} V 电流: {amps:.2f} A '
f'功率: {volts * amps:.1f} W')
def _pwr_step_next(self):
if not self._pwr_seq:
v_mv = self._pwr_values.get('mv', 0)
ma = self._pwr_values.get('ma', 0)
self._pwr_update(v_mv, ma)
self._dlpc_log('功率检测: 完成')
return
self.dlpc.send(self._pwr_seq[0][0])
self._pwr_timer = self.after(3000, lambda: self._pwr_step_done(None))
def _pwr_step_done(self, fr):
if not self._pwr_seq:
return
if self._pwr_timer:
self.after_cancel(self._pwr_timer)
self._pwr_timer = None
cmd, kind = self._pwr_seq.pop(0)
raw = dlpc_rs485.payload_bytes(fr) if fr is not None else b''
if fr is None or fr.type_code != 0xF0 or len(raw) != 2:
self._pwr_seq.clear()
self.var_pwr.set('电压: ERR 电流: — 功率: —')
self._dlpc_log('功率检测: 应答错误/超时, 中止')
return
if kind == 'mv':
self._pwr_values['mv'] = dlpc_rs485.payload_u16_be(raw)
elif kind == 'ma':
self._pwr_values['ma'] = dlpc_rs485.payload_s16_be(raw)
self._pwr_step_next()
def _pwr_update(self, v_mv, ma):
"""功率读数框更新: 电压 V / 电流 mA / 功率 W"""
v = v_mv / 1000
self.var_pwr.set(f'电压: {v:.2f} V 电流: {ma} mA '
f'功率: {v * ma / 1000:.2f} W')
def _light_entry_apply(self, _evt=None):
"""滑条数值输入框回车: 校验并限幅 54-800, 同步回滑条"""
@@ -1000,11 +1060,13 @@ class App(ctk.CTk):
self._dlpc_send(f'M8S{v}')
def destroy(self):
try: # 取消待触发的轮询/自检超时回调, 避免销毁后 Tcl 报错
try: # 取消待触发的轮询/自检/功率超时回调, 避免销毁后 Tcl 报错
if self._poll_job:
self.after_cancel(self._poll_job)
if self._pd_timer:
self.after_cancel(self._pd_timer)
if self._pwr_timer:
self.after_cancel(self._pwr_timer)
except Exception:
pass
try: # 两种退出路径(关窗/--seconds 自动关)都收掉串口线程
DLPilot/src/dlpc_rs485.py
+4 -2
View File
@@ -196,8 +196,8 @@ class DLPC485:
回调在后台线程触发, GUI 侧需自行投递回主线程(见 hdmi_gui 的轮询队列)。
"""
def __init__(self, on_event, on_status):
self.on_event, self.on_status = on_event, on_status
def __init__(self, on_event, on_status, on_frame=None):
self.on_event, self.on_status, self.on_frame = on_event, on_status, on_frame
self.q = queue.Queue()
self._target = None # 想态 (port, baud, bytesize, parity, stopbits)
self._open_key = None # 当前已打开串口的参数元组
@@ -314,6 +314,8 @@ class DLPC485:
data = self.ser.read(256)
if data:
for fr in self.parser.feed(data):
if self.on_frame is not None:
self.on_frame(fr) # 结构化帧(二进制载荷场景用)
self.on_event(f'← {describe(fr)} '
f'[{fr.raw.hex(" ").upper()}]')
except OSError: