#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ DLPC3421 光机 RS485 控制链路(板上 GD32E230 作 RS485→I2C 网关) ================================================================ 协议以 ../LT8619C_DLPC3421_HDMI/protocols.md 为准(上游固件仓, 含逐条命令的 响应语义/错误码/就绪门禁与实测帧示例): 物理层 RS485 半双工 115200 8N1(PC 经 USB-RS485 转换器接板, DE 由 MCU 硬件换向) 命令帧 D5 01 LEN CRC 01=板类型, LEN=命令字节数 响应帧 B5 TYPE LEN CRC payload 为 ASCII(ok / err:OL / okTL:246 ...) CRC 帧内第 2 字节(板类型)起累加到最后一个 payload 字节, 取低 8 位 帧长 最小 6 字节, 最大 32 字节 就绪门禁: DLPC3421 启动完成前 M730/M731/M732 被拒(回 0xFF), 属预期。 常用命令见 PRESETS(全表见 protocols.md); M7S0 是"开启"TEC 自动控温, 方向别记反。 仅标准库 + 可选 pyserial(缺失只影响串口功能, 同 hdmi_gui.py 的 Pillow 模式)。 """ import collections import queue import re import threading import time try: import serial import serial.tools.list_ports HAS_SERIAL = True except ImportError: HAS_SERIAL = False HEADER_TX = 0xD5 # 命令帧头(Host -> Device) HEADER_RX = 0xB5 # 响应帧头(Device -> Host) BOARD_TYPE = 0x01 # 板类型字节 FRAME_MIN, FRAME_MAX = 6, 32 # 整帧长度上下限(含帧头/CRC) DEFAULT_BAUD = 115200 # --- 响应 TYPE 码语义(protocols.md "响应 TYPE 码"表) --- TYPE_NAMES = { 0xF0: '成功', 0xF1: 'CRC 校验错', 0xF2: '包头错', 0xF3: '板类型错', 0xF4: '格式/缺参数', 0xFD: '参数越界/读回复核不一致', 0xFE: '未知命令', 0xFF: '通用错误(含 DLPC 就绪门禁拒绝)', 0xFA: '开光机通信失败', 0xFB: '关光机通信失败', 0xFC: '读回复核通信失败', } # --- payload 中 err: 的 DLPC 侧错误码 --- DLPC_ERR_NAMES = { 0: 'DLPC 成功', 1: 'DLPC 通用错误', 2: 'DLPC 通信超时', 3: 'I2C NACK(器件无应答)', 4: 'DLPC 参数无效', 5: 'DLPC 忙', } # GUI 预设命令表: (显示名, 命令, 说明), 顺序即下拉顺序 PRESETS = ( ('开光机', 'M730S0', '开启 RGB 投光(需 DLPC 就绪)'), ('关光机', 'M730S1', '关闭投光并停定时器'), ('设光强', 'M731S100', 'RGB 三路同值, 范围 54-800, 写后读回复核'), ('查电流', 'M732', 'okRGB, 三路电流原始值 0-65535'), ('查DMD温度', 'M201', 'okTL:246 = 24.6°C(有符号, 0.1°C)'), ('查LAMP温度', 'M202', '同 M201'), ('查投光结束', 'M737', '定时投光中不立即回包, 到期自动关灯后补发 ok'), ('查版本', 'M999', 'okv20.0.0.1 形式'), ('握手', 'M1', '协议握手, 回 ok'), ('TEC自动控温开', 'M7S0', 'S0=开启 PID 自动控温(方向别记反)'), ('TEC自动控温关', 'M7S1', 'S1=关闭'), ('凝露点33.0°C', 'M8S330', 'TEC 凝露点, 0.1°C 单位, 范围 200-350'), ('软件重启', 'M888', '回 ok 后 100ms 复位'), ) Frame = collections.namedtuple('Frame', 'type_code payload raw') def sum_crc(data): """帧内第 2 字节(板类型)起累加到末 payload 字节, 取低 8 位(不含包头与 CRC)""" return sum(data) & 0xFF def build_frame(cmd): """ASCII 命令字符串 -> 命令帧 bytes; 空命令/超长/非 ASCII 抛 ValueError""" payload = cmd.strip().encode('ascii') if not payload: raise ValueError('空命令') if len(payload) > FRAME_MAX - 4: raise ValueError(f'命令过长(>{FRAME_MAX - 4} 字节, 帧上限 {FRAME_MAX})') frame = bytes((HEADER_TX, BOARD_TYPE, len(payload))) + payload return frame + bytes((sum_crc(frame[1:]),)) class FrameParser: """增量字节流 -> 完整响应帧: 帧头+LEN 定界, CRC 校验, 失配丢当前帧头重新同步。 payload 理论上可能含 0xB5 造成同步偏移(伪帧头 CRC 必错会被跳过), 下一帧自动恢复, 对调试工具可接受。 """ def __init__(self): self._buf = bytearray() def feed(self, data): """喂入新收到的字节, 返回本次解析出的完整 Frame 列表""" self._buf += data frames = [] while True: i = self._buf.find(HEADER_RX) if i < 0: self._buf.clear() return frames if i: # 丢弃帧头前的噪声 del self._buf[:i] if len(self._buf) < 3: # 帧头/TYPE/LEN 尚不完整 return frames length = self._buf[2] if not FRAME_MIN - 4 <= length <= FRAME_MAX - 4: del self._buf[:1] # LEN 非法 -> 伪帧头, 前进 1 字节重同步 continue total = 4 + length if len(self._buf) < total: # 整帧未收齐 return frames raw = bytes(self._buf[:total]) if raw[-1] == sum_crc(raw[1:-1]): frames.append(Frame(self._buf[1], raw[3:-1].decode('ascii', 'replace').strip(), raw)) del self._buf[:total] else: # CRC 失配 -> 丢帧头重同步 del self._buf[:1] def describe(fr): """响应帧 -> 人读文本: TYPE 译码 + payload, err:N 追加 DLPC 错误释义""" if fr.type_code == 0xF0: text = fr.payload or 'ok' else: text = TYPE_NAMES.get(fr.type_code, f'TYPE=0x{fr.type_code:02X}') if fr.payload: text += f' {fr.payload}' m = re.search(r'err:(\d+)', fr.payload) if m: text += f'（{DLPC_ERR_NAMES.get(int(m.group(1)), "未知错误码")}）' return text class DLPC485: """串口工作线程: 端口/波特率热切换 + 命令队列 + 回调(收发全在后台线程)。 on_event(str) 收发日志行(已译码, 附原始 hex) on_status(str) 连接状态变化(已连接/断开/失败/设备拔出) 回调在后台线程触发, GUI 侧需自行投递回主线程(见 hdmi_gui 的轮询队列)。 """ def __init__(self, on_event, on_status): self.on_event, self.on_status = on_event, on_status self.q = queue.Queue() self._target = None # 想态 (port, baud); None=断开 self.ser = None self.parser = FrameParser() self._lock = threading.Lock() threading.Thread(target=self._run, daemon=True).start() @staticmethod def list_ports_info(): """枚举串口, 返回 [(设备名, 显示标签), ...]。 标签 = "COMx 设备描述"(描述里的 "(COMx)" 尾巴去掉, COM 号已在最前), CH344 这类多路适配器的每一路描述不同(如 serial-A/B/C/D), 依此区分; 多口描述完全相同(同型号多口)时追加物理位置/序列号消歧。 """ if not HAS_SERIAL: return [] items = [] for p in serial.tools.list_ports.comports(): desc = re.sub(r'\s*\((COM\d+)\)\s*$', '', (p.description or '').strip(), flags=re.I) if desc.upper() == p.device.upper(): desc = '' label = f'{p.device} {desc}'.rstrip() tail = p.location or p.serial_number or '' items.append((p.device, label, tail)) counts = collections.Counter(label for _d, label, _t in items) out = [] for dev, label, tail in items: if counts[label] > 1 and tail: # 同标签多口: 附物理位置消歧 label = f'{label} [{tail}]' out.append((dev, label)) return out def open(self, port, baud=DEFAULT_BAUD): if not HAS_SERIAL: self.on_status('未安装 pyserial: py -m pip install pyserial') return with self._lock: self._target = (port, int(baud)) def close(self): with self._lock: self._target = None def is_connected(self): with self._lock: return self.ser is not None def send(self, cmd): """发送一条 ASCII 命令(自动组帧); 未连接/命令非法走 on_status 提示""" with self._lock: ser = self.ser if ser is None: self.on_status('未连接串口, 命令被丢弃') return try: frame = build_frame(cmd) except ValueError as e: self.on_status(f'命令非法: {e}') return self.q.put((cmd, frame)) def _run(self): while True: with self._lock: target = self._target cur = (self.ser.port, self.ser.baudrate) if self.ser else None if target != cur: # 端口/波特率变化或断开: 统一在此切换 if self.ser is not None: try: self.ser.close() except Exception: pass self.ser = None self.on_status('已断开') if target: try: self.ser = serial.Serial(target[0], target[1], timeout=0.15) self.parser = FrameParser() self.on_status(f'已连接 {target[0]} @ {target[1]}') except Exception as e: self.ser = None with self._lock: self._target = None self.on_status(f'连接失败: {e}') if self.ser is not None: try: item = None try: item = self.q.get_nowait() except queue.Empty: pass if item is not None: cmd, frame = item self.ser.write(frame) self.on_event(f'→ {cmd} [{frame.hex(" ").upper()}]') data = self.ser.read(256) if data: for fr in self.parser.feed(data): self.on_event(f'← {describe(fr)} ' f'[{fr.raw.hex(" ").upper()}]') except OSError: self.ser = None with self._lock: self._target = None self.on_status('串口丢失(设备拔出?), 请重新连接') time.sleep(0.02)