feat(HDMI-Tool): v0.2.0 集成光机控制(RS485→GD32网关→DLPC3421) — 本地提交, 待验证后再推送发布

2026-09-11 00:25:28 +08:00
parent efaf4d658f
commit 2459b33162
8 changed files with 732 additions and 82 deletions
@@ -6,15 +6,18 @@ HDMI EDID 查看器 + 测试图案投屏工具
 配合 LT8619C(HDMI 转 RGB888)→ DLPC3421 转接板调试使用。

 功能:
-  1. 枚举各显示输出口接的设备, 从注册表读取其 EDID 原始数据,
+  1. 枚举各显示输出口, 从注册表读取其 EDID 原始数据,
     按 EDID 1.4 / CEA-861 标准解算: 厂商/产品/名称/尺寸/Gamma/输入类型、
     支持的全部模式(详细时序 DTD / 已建立时序 / 标准时序 / CEA VIC)、
     首选时序完整参数(像素时钟/消隐/同步极性)、CEA 扩展能力、校验和。
  2. 向所选屏幕全屏投测试图案(黑底 1:1 居中, 测试卡区域可选
     640x360 / 800x600 / full 铺满): 纯红 / 纯绿 / 纯蓝 / 黑白棋盘格 /
     网格线 + 3 个自定义图片栏位, 点击或 ESC 退出。
+  3. 光机控制: RS485 直连板上 GD32E230 网关(→DLPC3421), 开关光机/设光强/
+     查温度电流/定时投光, 协议见 dlpc_rs485.py 头注释(上游仓 protocols.md)。

-仅用标准库(ctypes / winreg / tkinter), 无需 pip 安装任何依赖。
+仅用标准库(ctypes / winreg / tkinter), 无需 pip 安装任何依赖;
+可选 pyserial(光机控制)与 Pillow(自定义图片 JPG/BMP/WebP), 缺了不影响其余功能。

 命令行模式(自动化测试用):
  python hdmi_gui.py                        启动 GUI
@@ -24,6 +27,8 @@ HDMI EDID 查看器 + 测试图案投屏工具
                                            直接出图案, 3 秒自动退出
  python hdmi_gui.py --test-pattern Blue --area 800x600
                                            指定测试卡区域(full=铺满所选屏)
+  python hdmi_gui.py --port COM9 --dlpc-cmd M730S0 --dlpc-cmd M999
+                                            RS485 逐条发光机命令并打印译码响应
  python hdmi_gui.py --seconds 5            GUI 5 秒自动关闭(冒烟测试)
 """

@@ -33,11 +38,15 @@ import ctypes.wintypes as wt
 import json
 import math
 import os
+import queue
 import shutil
 import struct
 import sys
+import time
 import winreg

+import dlpc_rs485
+
 # 进程级 DPI 感知: 保证图案窗口按物理像素精确定位/绘制
 try:
    ctypes.windll.shcore.SetProcessDpiAwareness(2)
@@ -65,11 +74,15 @@ SLOT_COUNT = 3
 SLOT_CONFIG = os.path.join(BASE_DIR, 'pattern_slots.json')
 IMG_EXTS = ['.png', '.gif'] + (['.jpg', '.jpeg', '.bmp', '.webp'] if HAS_PIL else [])

+# 光机控制串口记忆(与栏位配置同存 pattern_slots.json 的 "serial" 键)
+SERIAL_CFG = {'port': '', 'baud': dlpc_rs485.DEFAULT_BAUD}
+

 def load_slots():
    """读 pattern_slots.json, 返回 (slots, area)。

-    新格式为 {'area': 'WxH'|'full', 'slots': [...]}; 兼容旧版纯数组格式(无 area)。
+    新格式为 {'area': 'WxH'|'full', 'slots': [...], 'serial': {...}};
+    兼容旧版纯数组格式(无 area)与无 serial 键的中间格式。
    """
    slots = [{'name': f'自定义{i + 1}', 'file': None} for i in range(SLOT_COUNT)]
    try:
@@ -80,6 +93,13 @@ def load_slots():
    if isinstance(data, dict):
        area = str(data.get('area') or DEFAULT_AREA)
        items = data.get('slots') or []
+        ser = data.get('serial')
+        if isinstance(ser, dict):
+            SERIAL_CFG['port'] = str(ser.get('port') or '')
+            try:
+                SERIAL_CFG['baud'] = int(ser.get('baud') or dlpc_rs485.DEFAULT_BAUD)
+            except (TypeError, ValueError):
+                pass
    else:
        area, items = DEFAULT_AREA, data
    for i, item in enumerate(items[:SLOT_COUNT]):
@@ -91,7 +111,8 @@ def load_slots():

 def save_slots(slots, area):
    with open(SLOT_CONFIG, 'w', encoding='utf-8') as f:
-        json.dump({'area': area, 'slots': slots}, f, ensure_ascii=False, indent=2)
+        json.dump({'area': area, 'slots': slots, 'serial': SERIAL_CFG},
+                  f, ensure_ascii=False, indent=2)


 def copy_image_into_script_dir(src_path):
@@ -1008,8 +1029,8 @@ def run_gui(auto_close=0):
    class App(tk.Tk):
        def __init__(self):
            super().__init__()
-            self.title('HDMI-Tool — EDID 查看器 + 测试图案  (LT8619C 转接板调试)')
-            self.geometry('960x880')
+            self.title('HDMI-Tool — EDID 查看器 + 测试图案 + 光机控制  (LT8619C 转接板调试)')
+            self.geometry('960x1100')
            ic = get_icon_path()
            if ic:
                try:
@@ -1020,6 +1041,10 @@ def run_gui(auto_close=0):
            self.slots, self.area = load_slots()
            self._modes = {}          # 'WxH@Hz' -> DEVMODEW(当前所选屏)
            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_connected = False
            self._build()

        def _build(self):
@@ -1113,6 +1138,85 @@ def run_gui(auto_close=0):
            ttk.Label(row2, text='自定义: 点击投图, ⚙ 配置名称与图片(图片保存在脚本目录)',
                      foreground='#666666').pack(side='left', padx=10)

+            # --- 光机控制: RS485 -> 板载 GD32E230 网关 -> DLPC3421 ---
+            grp2 = ttk.LabelFrame(
+                self, text='光机控制 — RS485 → 板载 GD32 网关 → DLPC3421 (115200 8N1)',
+                padding=(10, 6))
+            grp2.pack(fill='x', padx=8, pady=(6, 0))
+            if not dlpc_rs485.HAS_SERIAL:
+                ttk.Label(grp2, foreground='#B05000',
+                          text='未安装 pyserial, 光机控制不可用 — '
+                               '执行 py -m pip install pyserial 后重启本工具').pack(anchor='w')
+            else:
+                r1 = ttk.Frame(grp2)
+                r1.pack(fill='x')
+                ttk.Label(r1, text='串口:').pack(side='left')
+                self.cbo_dlpc = ttk.Combobox(r1, width=12, state='readonly',
+                                             values=dlpc_rs485.DLPC485.list_ports())
+                if SERIAL_CFG['port']:
+                    self.cbo_dlpc.set(SERIAL_CFG['port'])
+                self.cbo_dlpc.pack(side='left', padx=(4, 2))
+                ttk.Button(r1, text='刷新', width=6,
+                           command=self._dlpc_refresh).pack(side='left')
+                ttk.Label(r1, text='波特率:').pack(side='left', padx=(10, 0))
+                self.var_baud = tk.StringVar(value=str(SERIAL_CFG['baud']))
+                ttk.Combobox(r1, width=8, textvariable=self.var_baud,
+                             values=('115200', '57600', '38400', '9600')
+                             ).pack(side='left', padx=4)
+                self.btn_dlpc = ttk.Button(r1, text='连接', width=8,
+                                           command=self._dlpc_toggle)
+                self.btn_dlpc.pack(side='left', padx=(10, 0))
+                self.lbl_dlpc_st = ttk.Label(r1, text='未连接', foreground='#666666')
+                self.lbl_dlpc_st.pack(side='left', padx=10)
+
+                r2 = ttk.Frame(grp2)
+                r2.pack(fill='x', pady=(4, 0))
+                for name, cmd in (('开光机', 'M730S0'), ('关光机', 'M730S1'),
+                                  ('查电流', 'M732'), ('查版本', 'M999'),
+                                  ('软件重启', 'M888')):
+                    ttk.Button(r2, text=name, width=9,
+                               command=lambda c=cmd: self._dlpc_send(c)
+                               ).pack(side='left', padx=3)
+                ttk.Label(r2, text='光强(54-800):').pack(side='left', padx=(12, 2))
+                self.var_light = tk.StringVar(value='100')
+                ttk.Entry(r2, textvariable=self.var_light, width=6).pack(side='left')
+                ttk.Button(r2, text='设光强', width=7,
+                           command=self._dlpc_set_light).pack(side='left', padx=4)
+
+                r3 = ttk.Frame(grp2)
+                r3.pack(fill='x', pady=(4, 0))
+                ttk.Label(r3, text='命令:').pack(side='left')
+                self._dlpc_presets = [(n, c) for n, c, _ in dlpc_rs485.PRESETS]
+                self.cbo_dlpc_cmd = ttk.Combobox(
+                    r3, state='readonly', width=26,
+                    values=[f'{n}  {c}' for n, c in self._dlpc_presets])
+                self.cbo_dlpc_cmd.current(0)
+                self.cbo_dlpc_cmd.pack(side='left', padx=(4, 2))
+                ttk.Button(r3, text='发送', width=7,
+                           command=self._dlpc_send_preset).pack(side='left')
+                ttk.Label(r3, text='自定义:').pack(side='left', padx=(12, 2))
+                self.var_custom_cmd = tk.StringVar()
+                ent_cmd = ttk.Entry(r3, textvariable=self.var_custom_cmd, width=14)
+                ent_cmd.pack(side='left')
+                ent_cmd.bind('<Return>', lambda _e: self._dlpc_send_custom())
+                ttk.Button(r3, text='发送', width=7,
+                           command=self._dlpc_send_custom).pack(side='left', padx=4)
+                ttk.Label(r3, text='如 M730S0T2000(定时 2 秒投光)',
+                          foreground='#666666').pack(side='left', padx=6)
+
+                r4 = ttk.Frame(grp2)
+                r4.pack(fill='x', pady=(4, 0))
+                self.txt_dlpc = tk.Text(r4, height=4, font=('Consolas', 9),
+                                        state='disabled', wrap='none',
+                                        background='#101010', foreground='#C8C8C8')
+                sb4 = ttk.Scrollbar(r4, orient='vertical', command=self.txt_dlpc.yview)
+                self.txt_dlpc.configure(yscrollcommand=sb4.set)
+                self.txt_dlpc.pack(side='left', fill='both', expand=True)
+                sb4.pack(side='right', fill='y')
+                self._dlpc_log('帧格式: D5 01 LEN <命令> CRC / B5 TYPE LEN <payload> CRC, '
+                               '累加和校验; 全部命令见 README"光机控制"')
+                self.after(80, self._poll_dlpc)
+
            # --- 底部: EDID hex dump + 逐字节解析按钮 ---
            bot = ttk.LabelFrame(self, text='EDID 原始数据', padding=4)
            bot.pack(fill='both', expand=False, padx=8, pady=(6, 8))
@@ -1181,6 +1285,104 @@ def run_gui(auto_close=0):
            self.area = self.area_var.get()
            save_slots(self.slots, self.area)

+        # ---- 光机控制(RS485) ----
+        def _dlpc_on_event(self, line):
+            self._dlpc_log_q.put(('event', line))
+
+        def _dlpc_on_status(self, msg):
+            self._dlpc_log_q.put(('status', msg))
+
+        def _poll_dlpc(self):
+            """串口线程 -> GUI 主线程: 80ms 轮询日志队列并同步连接状态"""
+            try:
+                while True:
+                    kind, text = self._dlpc_log_q.get_nowait()
+                    if kind == 'status':
+                        if text.startswith('已连接'):
+                            color = '#2E7D32'
+                        elif ('失败' in text or '丢失' in text or '未' in text
+                                or '非法' in text or '丢弃' in text):
+                            color = '#B05000'
+                        else:
+                            color = '#666666'
+                        self.lbl_dlpc_st.config(text=text, foreground=color)
+                    self._dlpc_log(text)
+            except queue.Empty:
+                pass
+            connected = self.dlpc.is_connected()
+            if connected != self._dlpc_connected:
+                self._dlpc_connected = connected
+                self.btn_dlpc.config(text='断开' if connected else '连接')
+            self.after(80, self._poll_dlpc)
+
+        def _dlpc_log(self, line):
+            self.txt_dlpc.configure(state='normal')
+            self.txt_dlpc.insert('end', time.strftime('%H:%M:%S  ') + line + '\n')
+            if int(self.txt_dlpc.index('end-1c').split('.')[0]) > 300:
+                self.txt_dlpc.delete('1.0', '2.0')     # 只保留最近约 300 行
+            self.txt_dlpc.see('end')
+            self.txt_dlpc.configure(state='disabled')
+
+        def _dlpc_refresh(self):
+            ports = dlpc_rs485.DLPC485.list_ports()
+            self.cbo_dlpc['values'] = ports
+            if SERIAL_CFG['port'] in ports:
+                self.cbo_dlpc.set(SERIAL_CFG['port'])
+            elif ports:
+                self.cbo_dlpc.current(0)
+
+        def _dlpc_toggle(self):
+            if self.dlpc.is_connected():
+                self.dlpc.close()
+                return
+            port = self.cbo_dlpc.get().strip()
+            if not port:
+                self.lbl_dlpc_st.config(text='未选择串口', foreground='#B05000')
+                return
+            try:
+                baud = int(self.var_baud.get())
+            except ValueError:
+                self.lbl_dlpc_st.config(text='波特率非法', foreground='#B05000')
+                return
+            SERIAL_CFG['port'], SERIAL_CFG['baud'] = port, baud
+            save_slots(self.slots, self.area_var.get())     # 记住本次串口
+            self.dlpc.open(port, baud)
+
+        def _dlpc_send(self, cmd):
+            if not self.dlpc.is_connected():
+                self.lbl_dlpc_st.config(text='未连接串口', foreground='#B05000')
+                return
+            self.dlpc.send(cmd)
+
+        def _dlpc_send_preset(self):
+            i = self.cbo_dlpc_cmd.current()
+            if i >= 0:
+                self._dlpc_send(self._dlpc_presets[i][1])
+
+        def _dlpc_send_custom(self):
+            cmd = self.var_custom_cmd.get().strip()
+            if cmd:
+                self._dlpc_send(cmd)
+
+        def _dlpc_set_light(self):
+            from tkinter import messagebox
+            try:
+                v = int(self.var_light.get())
+            except ValueError:
+                messagebox.showinfo('光强非法', '光强需为 54-800 的整数')
+                return
+            if not 54 <= v <= 800:
+                messagebox.showinfo('光强越界', '光强范围 54-800(越界固件会回 err:OL)')
+                return
+            self._dlpc_send(f'M731S{v}')
+
+        def destroy(self):
+            try:                    # 两种退出路径(关窗/--seconds 自动关)都收掉串口线程
+                self.dlpc.close()
+            except Exception:
+                pass
+            super().destroy()
+
        def _show_edid_fields(self):
            """逐字节字段解析弹窗(当前所选显示器), 自动换行文本视图"""
            from tkinter import messagebox
@@ -1384,8 +1586,50 @@ def run_dump(edid_fields=False):
    return 0


+def run_dlpc(port, baud, cmds):
+    """RS485 直连模式: 逐条发送光机命令, 等响应打印译码结果(自动化脚本用)。
+
+    返回码: 任一命令非法/超时/响应非 0xF0 成功 -> 1, 全部成功 -> 0。
+    """
+    if not dlpc_rs485.HAS_SERIAL:
+        print('错误: 未安装 pyserial, 请先执行 py -m pip install pyserial')
+        return 2
+    try:
+        ser = dlpc_rs485.serial.Serial(port, baud, timeout=0.1)
+    except Exception as e:
+        print(f'打开串口失败: {port} @ {baud}: {e}')
+        return 1
+    print(f'已连接 {port} @ {baud} (RS485 半双工, 单发单收)')
+    parser = dlpc_rs485.FrameParser()
+    rc = 0
+    for cmd in cmds:
+        try:
+            frame = dlpc_rs485.build_frame(cmd)
+        except ValueError as e:
+            print(f'→ {cmd}   命令非法: {e}')
+            rc = 1
+            continue
+        ser.write(frame)
+        print(f'→ {cmd}   [{frame.hex(" ").upper()}]')
+        deadline = time.time() + 35.0     # M737 定时投光中延迟应答(定时上限 32.7s)
+        got = None
+        while got is None and time.time() < deadline:
+            data = ser.read(256)
+            if data:
+                got = next(iter(parser.feed(data)), None)
+        if got is None:
+            print('← (35s 超时无响应)')
+            rc = 1
+        else:
+            print(f'← {dlpc_rs485.describe(got)}   [{got.raw.hex(" ").upper()}]')
+            if got.type_code != 0xF0:
+                rc = 1
+    ser.close()
+    return rc
+
+
 def main():
-    ap = argparse.ArgumentParser(description='HDMI EDID 查看器 + 测试图案')
+    ap = argparse.ArgumentParser(description='HDMI EDID 查看器 + 测试图案 + 光机控制')
    ap.add_argument('--dump', action='store_true', help='控制台输出 EDID 解算结果, 不开 GUI')
    ap.add_argument('--edid-fields', action='store_true',
                    help='控制台输出 EDID 逐字节字段解析表(配合 --dump 风格, 不开 GUI); exe 版不支持')
@@ -1396,6 +1640,12 @@ def main():
                    help='测试卡区域: WxH 黑底居中 或 full 铺满所选显示器(默认 640x360)')
    ap.add_argument('--monitor', type=int, default=1, help='--test-pattern 用的显示器序号(默认 1)')
    ap.add_argument('--seconds', type=int, default=0, help='自动退出秒数(GUI/图案冒烟测试)')
+    ap.add_argument('--port', default=None, help='光机 RS485 串口(如 COM9), 配合 --dlpc-cmd')
+    ap.add_argument('--baud', type=int, default=dlpc_rs485.DEFAULT_BAUD,
+                    help='光机串口波特率(默认 115200)')
+    ap.add_argument('--dlpc-cmd', action='append', default=None, dest='dlpc_cmds',
+                    metavar='CMD',
+                    help='RS485 发送光机命令(可多次), 如 --dlpc-cmd M730S0 --dlpc-cmd M999')
    args = ap.parse_args()

    # LT8619C 板卡热插拔后可能停留在"仅电脑屏幕"拓扑, 先强制扩展,
@@ -1413,6 +1663,11 @@ def main():
                                'python hdmi_gui.py --edid-fields')
            sys.exit(1)
        sys.exit(run_dump(edid_fields=args.edid_fields))
+    if args.dlpc_cmds:
+        if not args.port:
+            print('错误: --dlpc-cmd 需要配合 --port 指定串口(如 --port COM9)')
+            sys.exit(2)
+        sys.exit(run_dlpc(args.port, args.baud, args.dlpc_cmds))
    if args.test_pattern:
        displays = get_displays()
        if not displays: