216 lines
8.7 KiB
Python
216 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
串口链路线程: 端口/波特率热切换 + 发送队列 + 接收分发
|
|
======================================================
|
|
与 DLPilot 的 DLPC485 同一套路(D I2C-Inject Injector 同源), 差异点:
|
|
- 帧格式参数(ProtoConfig)可运行中热替换, 解析器配置随之切换并清残余半帧;
|
|
- 支持协议解析(Frame 回调)与透传(原始 bytes 回调)两种接收模式, 运行中切换;
|
|
- 发送一律传完整帧 bytes(组帧由 GUI 侧按 ProtoConfig 完成, 本模块不关心格式)。
|
|
|
|
回调全在后台线程触发, GUI 侧必须自行投递回主线程(见 main.py 的轮询队列)。
|
|
pyserial 为本模块独占依赖(缺失只影响串口功能)。
|
|
"""
|
|
|
|
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
|
|
|
|
DEFAULT_BAUD = 115200
|
|
DEFAULTS = dict(bytesize=8, parity='N', stopbits=1)
|
|
|
|
|
|
class SerialLink:
|
|
"""串口工作线程。回调(后台线程触发):
|
|
on_status(str) 连接状态变化(已连接/断开/失败/设备拔出)
|
|
on_frame(fr) 解析模式: 完整 Frame(protocol.Frame)
|
|
on_raw(data) 透传模式: 原始接收 bytes
|
|
on_tx(label, data) 发送回显: label=payload 字符串(自动包装)/None(HEX直发)
|
|
"""
|
|
|
|
def __init__(self, cfg, on_status):
|
|
self.on_status = on_status
|
|
self.on_frame = None
|
|
self.on_raw = None
|
|
self.on_tx = None
|
|
self.q = queue.Queue()
|
|
self._cfg = cfg
|
|
self._parse = True
|
|
self._target = None # 意图 (port, baud, bytesize, parity, stopbits)
|
|
self._open_key = None # 当前已打开串口的参数元组
|
|
self.ser = None
|
|
self.parser = None # 惰性建: 协议模块无 pyserial 依赖, 此处才用到
|
|
self._lock = threading.Lock()
|
|
threading.Thread(target=self._run, daemon=True).start()
|
|
|
|
# --- 配置与模式(线程安全) ---
|
|
|
|
def set_config(self, cfg):
|
|
"""热替换帧格式; 解析器半帧缓冲随之清空, 避免新旧格式混接"""
|
|
with self._lock:
|
|
self._cfg = cfg
|
|
if self.parser is not None:
|
|
self.parser.cfg = cfg
|
|
self.parser.clear()
|
|
|
|
def set_parse(self, enabled):
|
|
"""切换 解析协议帧 / 原始透传; 切换时清解析缓冲"""
|
|
with self._lock:
|
|
self._parse = bool(enabled)
|
|
if self.parser is not None:
|
|
self.parser.clear()
|
|
|
|
# --- 串口管理 ---
|
|
|
|
@staticmethod
|
|
def list_ports_info():
|
|
"""枚举串口, 返回 [(设备名, 显示标签), ...]。
|
|
|
|
标签 = "COMx 设备描述"(描述里的 "(COMx)" 尾巴去掉), 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, **kw):
|
|
if not HAS_SERIAL:
|
|
self.on_status('未安装 pyserial: py -m pip install pyserial')
|
|
return
|
|
params = {**DEFAULTS, **kw}
|
|
with self._lock:
|
|
self._target = (port, int(baud), int(params['bytesize']),
|
|
str(params['parity']).upper(), int(params['stopbits']))
|
|
|
|
def close(self):
|
|
with self._lock:
|
|
self._target = None
|
|
|
|
def is_connected(self):
|
|
with self._lock:
|
|
return self.ser is not None
|
|
|
|
# --- 发送(队列化, 后台线程写串口) ---
|
|
|
|
def send_frame(self, frame, label=''):
|
|
"""发送一个完整帧 bytes; label 用于回显(自动包装时为 payload 文本)。
|
|
未连接时直接丢弃并提示(防积压到重连后一起发出, DLPilot 同策略)"""
|
|
with self._lock:
|
|
ser = self.ser
|
|
if ser is None:
|
|
self.on_status('未连接串口, 命令被丢弃')
|
|
return False
|
|
self.q.put((label, bytes(frame)))
|
|
return True
|
|
|
|
# --- 工作线程 ---
|
|
|
|
def _run(self):
|
|
bytesizes = {8: serial.EIGHTBITS, 7: serial.SEVENBITS,
|
|
6: serial.SIXBITS, 5: serial.FIVEBITS}
|
|
parities = {'N': serial.PARITY_NONE, 'E': serial.PARITY_EVEN,
|
|
'O': serial.PARITY_ODD}
|
|
stopbits = {1: serial.STOPBITS_ONE, 2: serial.STOPBITS_TWO}
|
|
while True:
|
|
with self._lock:
|
|
target = self._target
|
|
cur = self._open_key
|
|
if target != cur:
|
|
if self.ser is not None:
|
|
try:
|
|
self.ser.close()
|
|
except Exception:
|
|
pass
|
|
self.ser = None
|
|
self._open_key = None
|
|
self.on_status('已断开')
|
|
if target:
|
|
port, baud, dbsize, parity, sbits = target
|
|
try:
|
|
maker = (serial.serial_for_url if '://' in port
|
|
else serial.Serial) # '://' 支持 loop:// 等虚拟端口(自检用)
|
|
self.ser = maker(
|
|
port, baud, bytesize=bytesizes.get(dbsize, serial.EIGHTBITS),
|
|
parity=parities.get(parity, serial.PARITY_NONE),
|
|
stopbits=stopbits.get(sbits, serial.STOPBITS_ONE),
|
|
timeout=0.15,
|
|
write_timeout=1.0) # 写超时: 转换器停滞时 write 会
|
|
# 永久阻塞并卡死线程, 必须设上限
|
|
with self._lock:
|
|
self._open_key = target
|
|
self.parser = None # 换口重建解析器(下次接收时惰性建)
|
|
self.on_status(f'已连接 {port} @ {baud} {dbsize}{parity}{sbits}')
|
|
except Exception as e:
|
|
self.ser = None
|
|
with self._lock:
|
|
self._target = None
|
|
et = str(e)
|
|
if 'PermissionError' in repr(e) or '拒绝访问' in et \
|
|
or 'Access is denied' in et:
|
|
et += '（串口被占用: 检查串口助手/另一个本工具实例）'
|
|
self.on_status(f'连接失败: {et}')
|
|
if self.ser is not None:
|
|
try:
|
|
item = None
|
|
try:
|
|
item = self.q.get_nowait()
|
|
except queue.Empty:
|
|
pass
|
|
if item is not None:
|
|
label, frame = item
|
|
self.ser.write(frame)
|
|
if self.on_tx is not None:
|
|
self.on_tx(label, frame)
|
|
data = self.ser.read(4096)
|
|
if data:
|
|
self._dispatch_rx(data)
|
|
except OSError:
|
|
try:
|
|
self.ser.close()
|
|
except Exception:
|
|
pass
|
|
self.ser = None
|
|
self._open_key = None # 必须同步清除, 否则同参数重连被当作已打开而跳过
|
|
with self._lock:
|
|
self._target = None
|
|
self.on_status('串口丢失(设备拔出?), 请重新连接')
|
|
time.sleep(0.02)
|
|
|
|
def _dispatch_rx(self, data):
|
|
with self._lock:
|
|
parse = self._parse
|
|
if parse and self.parser is None: # 惰性建, 带当前 ProtoConfig
|
|
from protocol import FrameParser
|
|
self.parser = FrameParser(self._cfg)
|
|
if parse:
|
|
for fr in self.parser.feed(data):
|
|
if self.on_frame is not None:
|
|
self.on_frame(fr)
|
|
elif self.on_raw is not None:
|
|
self.on_raw(data)
|