227 lines
9.7 KiB
Python
227 lines
9.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:
|
||
"""串口工作线程。回调(后台线程触发, 第二参数 ts=事件发生的 epoch 秒):
|
||
on_status(text, ts) 连接状态变化(已连接/断开/失败/设备拔出)
|
||
on_frame(fr, ts) 解析模式: 完整 Frame(protocol.Frame); ts=本批字节到达时刻
|
||
on_raw(data, ts) 透传模式: 原始接收 bytes; ts=到达时刻
|
||
on_tx(label, data, ts) 发送回显: label=payload 字符串(自动包装)/None(HEX直发);
|
||
ts=写入串口时刻
|
||
时间戳必须由本模块在读线程打(字节到达时刻), GUI 消费侧再打会随 UI 卡顿
|
||
成批挤压失真——日志时间就不再是到达时间。
|
||
"""
|
||
|
||
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', time.time())
|
||
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('未连接串口, 命令被丢弃', time.time())
|
||
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('已断开', time.time())
|
||
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 会
|
||
# 永久阻塞并卡死线程, 必须设上限
|
||
inter_byte_timeout=0.02) # 字节间超时(Win32
|
||
# ReadIntervalTimeout): timeout=0.15 是"总超时"
|
||
# (150ms 内必返回, 不会无限攒批), 周期上报帧下 read
|
||
# 会 ~150ms 攒 1~2 帧才返回一批; 设 20ms 后按帧粒度
|
||
# 返回, 帧到达时间戳精确到单帧。20ms 仍大于 1200 波特
|
||
# 单字节时间(~8.3ms)×2, 不误切帧
|
||
with self._lock:
|
||
self._open_key = target
|
||
self.parser = None # 换口重建解析器(下次接收时惰性建)
|
||
self.on_status(f'已连接 {port} @ {baud} {dbsize}{parity}{sbits}',
|
||
time.time())
|
||
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}', time.time())
|
||
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, time.time())
|
||
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.time())
|
||
time.sleep(0.02)
|
||
|
||
def _dispatch_rx(self, data):
|
||
ts = time.time() # 字节到达时刻: 后台线程打, 随帧携带
|
||
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, ts)
|
||
elif self.on_raw is not None:
|
||
self.on_raw(data, ts)
|