feat: 新增 SerialMate 串口调试助手 v0.1.0
- 协议帧构成令牌序列可定制(包头/板ID/LEN口径/5种CRC x 3种覆盖), 默认兼容 DLPilot 光机协议
This commit is contained in:
@@ -0,0 +1,108 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
把 src/main.py 打包成单文件 exe(工具名: SerialMate)
|
||||
================================================
|
||||
与 EDID-Viewer/src/build_exe.py 同一套路: onefile + --noconsole, Python 解释器、
|
||||
PySide6 和 PySide6-Fluent-Widgets 全部打进一个 exe, 目标机器无需安装依赖。
|
||||
|
||||
用法:
|
||||
python build_exe.py 自动安装 PyInstaller(如缺)并打包
|
||||
python build_exe.py --debug 额外打一个带控制台的变体
|
||||
|
||||
产物: release/SerialMate.exe
|
||||
spec 由本脚本生成到 src/build/(不入库), 调整打包逻辑直接改这里的模板。
|
||||
"""
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
# 本脚本位于 src/, 打包产物输出到工具根的 release/
|
||||
SRC = os.path.dirname(os.path.abspath(__file__))
|
||||
BASE = os.path.dirname(SRC)
|
||||
RELEASE = os.path.join(BASE, 'release')
|
||||
WORK = os.path.join(SRC, 'build')
|
||||
ICO = os.path.join(BASE, 'assets', 'serial-mate.ico')
|
||||
|
||||
SPEC_TEMPLATE = """\
|
||||
# -*- coding: utf-8 -*-
|
||||
# 由 build_exe.py 生成, 不要手改; 调整打包逻辑请改 src/build_exe.py。
|
||||
from PyInstaller.utils.hooks import collect_all
|
||||
|
||||
qfluent_datas, qfluent_binaries, qfluent_hidden = collect_all('qfluentwidgets')
|
||||
a = Analysis(
|
||||
[r'{entry}'], pathex=[r'{src}'], binaries=qfluent_binaries,
|
||||
datas=qfluent_datas + {icon_datas}, hiddenimports=qfluent_hidden,
|
||||
noarchive=False)
|
||||
# Qt 6.11 使用 Windows 非版本化 ICU 导出; PATH 上 Poppler/Conda 的版本化
|
||||
# ICU(icuuc.dll/icudt78.dll)会被 PyInstaller 误收, 令 QtCore.pyd 启动报
|
||||
# WinError 127 —— 打包时剔除, 运行时用系统自带 ICU。
|
||||
a.binaries = [b for b in a.binaries if b[0].lower() not in ('icuuc.dll', 'icudt78.dll')]
|
||||
pyz = PYZ(a.pure)
|
||||
exe = EXE(pyz, a.scripts, a.binaries, a.datas, [], name='{name}',
|
||||
debug=False, strip=False, upx=False, console={console}{icon_clause})
|
||||
"""
|
||||
|
||||
|
||||
def ensure_pyinstaller():
|
||||
try:
|
||||
import PyInstaller # noqa: F401
|
||||
return
|
||||
except ImportError:
|
||||
pass
|
||||
print('PyInstaller 未安装, 正在安装...')
|
||||
subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'pyinstaller'])
|
||||
|
||||
|
||||
def remove_stale(exe_path):
|
||||
"""删除旧 exe; 若被占用(上次验证的进程还在), 先结束同名进程再重试"""
|
||||
if not os.path.exists(exe_path):
|
||||
return
|
||||
try:
|
||||
os.remove(exe_path)
|
||||
return
|
||||
except PermissionError:
|
||||
pass
|
||||
import time
|
||||
print(f'{exe_path} 被占用, 结束残留进程后重试...')
|
||||
subprocess.run(['taskkill', '/F', '/IM', os.path.basename(exe_path)],
|
||||
capture_output=True)
|
||||
time.sleep(1.0)
|
||||
os.remove(exe_path)
|
||||
|
||||
|
||||
def build(name, console):
|
||||
remove_stale(os.path.join(RELEASE, name + '.exe'))
|
||||
if os.path.exists(ICO):
|
||||
icon_datas = f"[(r'{ICO}', 'assets')]"
|
||||
icon_clause = f", icon=r'{ICO}'"
|
||||
else:
|
||||
print('警告: 未找到 assets/serial-mate.ico, 本次打包不带图标')
|
||||
icon_datas, icon_clause = '[]', ''
|
||||
spec = SPEC_TEMPLATE.format(entry=os.path.join(SRC, 'main.py'), src=SRC,
|
||||
icon_datas=icon_datas, name=name,
|
||||
console=console, icon_clause=icon_clause)
|
||||
os.makedirs(WORK, exist_ok=True)
|
||||
spec_path = os.path.join(WORK, name + '.spec')
|
||||
with open(spec_path, 'w', encoding='utf-8') as f:
|
||||
f.write(spec)
|
||||
cmd = [sys.executable, '-m', 'PyInstaller', '--noconfirm', '--clean',
|
||||
'--distpath', RELEASE, '--workpath', WORK, spec_path]
|
||||
print('\n>>', ' '.join(cmd))
|
||||
subprocess.check_call(cmd)
|
||||
out = os.path.join(RELEASE, name + '.exe')
|
||||
print(f'完成: {out} ({os.path.getsize(out) / 1024 / 1024:.1f} MB)')
|
||||
|
||||
|
||||
def main():
|
||||
build_console = '--debug' in sys.argv
|
||||
ensure_pyinstaller()
|
||||
# 常规版: 双击即用, 无黑框
|
||||
build('SerialMate', 'False')
|
||||
if build_console:
|
||||
# 调试版: 带控制台, 异常栈可见
|
||||
build('SerialMate-console', 'True')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,147 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
运行目录定位与设置持久化(BASE_DIR 语义与 DLPilot config.py 一致)
|
||||
==================================================================
|
||||
frozen(PyInstaller onefile)时 BASE_DIR=exe 所在目录(sys.executable),
|
||||
源码运行时=config.py 所在目录(即 src/)。设置写/读自 BASE_DIR 平铺的
|
||||
serial_mate.json, 不建子目录。
|
||||
|
||||
读容错: 文件缺失/损坏/缺键一律回落默认值, 并保证返回的 dict 键全齐,
|
||||
GUI 无需判空。
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
|
||||
if getattr(sys, 'frozen', False):
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(sys.executable))
|
||||
else:
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
CONFIG_FILE = os.path.join(BASE_DIR, 'serial_mate.json')
|
||||
LEGACY_CONFIG = os.path.join(BASE_DIR, 'serial_assistant.json') # 更名前的配置
|
||||
PROFILES_FILE = os.path.join(BASE_DIR, 'board_profiles.json')
|
||||
|
||||
|
||||
def _migrate_legacy_config():
|
||||
"""SerialAssistant -> SerialMate 更名迁移: 旧配置搬为新名(一次性, 已存在则跳过)"""
|
||||
if os.path.exists(CONFIG_FILE) or not os.path.exists(LEGACY_CONFIG):
|
||||
return
|
||||
try:
|
||||
shutil.copy2(LEGACY_CONFIG, CONFIG_FILE)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
DEFAULTS = {
|
||||
'serial': {'port': '', 'baud': 115200, 'bytesize': 8, 'parity': 'N', 'stopbits': 1},
|
||||
'proto': {'tx_hdr': 'D5', 'rx_hdr': 'B5', 'board': '01',
|
||||
'crc_algo': 'SUM8', 'crc_scope': 'no_header', 'parse': True,
|
||||
'layout': 'HDR ID LEN PAYLOAD CRC', 'len_sem': 'payload'},
|
||||
'tx': {'mode': 'wrap', 'periodic': False, 'interval_ms': 1000, 'profile': ''},
|
||||
'rx': {'display': 'text', 'show_raw': True},
|
||||
'channels': [
|
||||
{'enabled': False, 'name': '通道1', 'mode': 'num', 'offset': 0,
|
||||
'fmt': 'u16be', 'index': 1, 'pattern': '(?<=:)\\d+', 'scale': 1.0},
|
||||
{'enabled': False, 'name': '通道2', 'mode': 'bin', 'offset': 0,
|
||||
'fmt': 's16be', 'index': 1, 'pattern': '(?<=:)\\d+', 'scale': 1.0},
|
||||
{'enabled': False, 'name': '通道3', 'mode': 'bin', 'offset': 0,
|
||||
'fmt': 'u16be', 'index': 1, 'pattern': '(?<=:)\\d+', 'scale': 1.0},
|
||||
{'enabled': False, 'name': '通道4', 'mode': 'bin', 'offset': 0,
|
||||
'fmt': 'u16be', 'index': 1, 'pattern': '(?<=:)\\d+', 'scale': 1.0},
|
||||
],
|
||||
}
|
||||
|
||||
# 板型档案的帧协议字段(不含 parse —— 解析开关是查看偏好, 不随板型走)
|
||||
_PROFILE_PROTO_KEYS = ('tx_hdr', 'rx_hdr', 'board', 'layout', 'len_sem',
|
||||
'crc_algo', 'crc_scope')
|
||||
|
||||
# 首个板型的命令预设(DLPilot 光机命令集, 全表见上游 protocols.md)
|
||||
DEFAULT_COMMANDS = (
|
||||
('握手', 'M1'), ('查版本', 'M999'),
|
||||
('开光机', 'M730S0'), ('关光机', 'M730S1'),
|
||||
('设光强', 'M731S100'), ('查电流', 'M732'),
|
||||
('查DMD温度', 'M201'), ('查LAMP温度', 'M202'),
|
||||
('TEC自动控温开', 'M7S0'), ('TEC自动控温关', 'M7S1'),
|
||||
('凝露点33.0°C', 'M8S330'), ('软件重启', 'M888'),
|
||||
)
|
||||
|
||||
|
||||
def default_profiles():
|
||||
"""首个板型档案: DLPilot 光机(协议字段取默认值)"""
|
||||
return [{'name': 'DLPilot光机',
|
||||
'proto': {k: DEFAULTS['proto'][k] for k in _PROFILE_PROTO_KEYS},
|
||||
'commands': [list(c) for c in DEFAULT_COMMANDS]}]
|
||||
|
||||
|
||||
def _merge(base, over):
|
||||
"""递归覆盖: over 缺键用 base 补齐, over 多余键保留(向前兼容旧配置)"""
|
||||
out = dict(base)
|
||||
for k, v in over.items():
|
||||
if k in base and isinstance(base[k], dict) and isinstance(v, dict):
|
||||
out[k] = _merge(base[k], v)
|
||||
else:
|
||||
out[k] = v
|
||||
return out
|
||||
|
||||
|
||||
def load():
|
||||
_migrate_legacy_config()
|
||||
"""读设置; 任何异常回落默认(键全齐), 不抛出"""
|
||||
try:
|
||||
with open(CONFIG_FILE, encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError('配置根不是对象')
|
||||
except Exception:
|
||||
return json.loads(json.dumps(DEFAULTS)) # 深拷贝默认值
|
||||
return _merge(DEFAULTS, data)
|
||||
|
||||
|
||||
def save(cfg):
|
||||
"""写设置; 失败静默(只读目录等场景不阻塞退出)"""
|
||||
try:
|
||||
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
|
||||
json.dump(cfg, f, ensure_ascii=False, indent=1)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def load_profiles():
|
||||
"""读板型档案(board_profiles.json); 缺失/损坏回落默认并落盘。
|
||||
档案结构: {'profiles': [{'name', 'proto'{7字段}, 'commands'[[名,命令]..]}..]}"""
|
||||
try:
|
||||
with open(PROFILES_FILE, encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
profiles = data.get('profiles') if isinstance(data, dict) else data
|
||||
if isinstance(profiles, list) and profiles:
|
||||
out = []
|
||||
for p in profiles:
|
||||
if not isinstance(p, dict) or not p.get('name'):
|
||||
continue
|
||||
out.append({'name': str(p['name']),
|
||||
'proto': {k: p.get('proto', {}).get(k,
|
||||
DEFAULTS['proto'][k])
|
||||
for k in _PROFILE_PROTO_KEYS},
|
||||
'commands': [[str(n), str(c)]
|
||||
for n, c in p.get('commands', [])
|
||||
if isinstance(n, str) and isinstance(c, str)]})
|
||||
if out:
|
||||
return out
|
||||
except Exception:
|
||||
pass
|
||||
profiles = default_profiles()
|
||||
save_profiles(profiles)
|
||||
return profiles
|
||||
|
||||
|
||||
def save_profiles(profiles):
|
||||
"""写板型档案; 失败静默并返回 False"""
|
||||
try:
|
||||
with open(PROFILES_FILE, 'w', encoding='utf-8') as f:
|
||||
json.dump({'profiles': profiles}, f, ensure_ascii=False, indent=1)
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
@@ -0,0 +1,215 @@
|
||||
#!/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)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,61 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
组装 SerialMate 发布包: exe + README → 一个 zip(解压即用)
|
||||
================================================
|
||||
把主 README 复制进 release/ 后, 与单文件 exe 一起压成 zip —— 两者内容一致。
|
||||
|
||||
zip 解压后根目录为 SerialMate/:
|
||||
SerialMate.exe 主程序(先运行 build_exe.py 打包)
|
||||
README.md 使用说明
|
||||
|
||||
用法:
|
||||
python package_release.py 版本号自动取最近 git tag(SerialMate-v*)
|
||||
python package_release.py 0.1.0 手动指定版本号
|
||||
产物: release/SerialMate-v<版本>-win64.zip(zip 不入库, 上传到 Gitea Release)
|
||||
"""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import zipfile
|
||||
|
||||
SRC = os.path.dirname(os.path.abspath(__file__))
|
||||
BASE = os.path.dirname(SRC)
|
||||
RELEASE = os.path.join(BASE, 'release')
|
||||
|
||||
|
||||
def detect_version():
|
||||
"""从最近的 SerialMate-v* tag 推断版本号, 无 tag 则报错提示手动指定"""
|
||||
try:
|
||||
out = subprocess.check_output(
|
||||
['git', 'describe', '--tags', '--abbrev=0', '--match', 'SerialMate-v*'],
|
||||
cwd=BASE, stderr=subprocess.DEVNULL)
|
||||
return out.decode().strip().removeprefix('SerialMate-v')
|
||||
except Exception:
|
||||
sys.exit('未能从 git tag 推断版本号, 请手动指定: python package_release.py <版本号>')
|
||||
|
||||
|
||||
def main():
|
||||
version = (sys.argv[1] if len(sys.argv) > 1 else detect_version()).lstrip('v')
|
||||
exe = os.path.join(RELEASE, 'SerialMate.exe')
|
||||
readme = os.path.join(BASE, 'README.md')
|
||||
if not os.path.exists(exe):
|
||||
sys.exit('缺少 release/SerialMate.exe, 先运行: python build_exe.py')
|
||||
if not os.path.exists(readme):
|
||||
sys.exit(f'缺少 {readme}')
|
||||
|
||||
shutil.copy2(readme, os.path.join(RELEASE, 'README.md'))
|
||||
zip_name = f'SerialMate-v{version}-win64.zip'
|
||||
zip_path = os.path.join(RELEASE, zip_name)
|
||||
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as z:
|
||||
z.write(exe, f'SerialMate/{os.path.basename(exe)}')
|
||||
z.write(os.path.join(RELEASE, 'README.md'), 'SerialMate/README.md')
|
||||
print(f'发布文件夹: {RELEASE} (exe + README)')
|
||||
print(f'发布包: {zip_path} ({os.path.getsize(zip_path) / 1024 / 1024:.1f} MB)')
|
||||
print('下一步上传: python publish_gitea.py', zip_name)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,405 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
轻量滚动曲线控件(QPainter 绘制, 不引入 matplotlib/pyqtgraph)
|
||||
================================================================
|
||||
N 通道时间序列: add(ch, value) 追加点(内部取相对时间), deque 限长滚动丢弃;
|
||||
y 轴自动缩放(全平数据回落 ±1)或 set_yrange 固定量程; 颜色/文字取自当前
|
||||
palette, 明暗主题自动跟随。每通道 >1000 点时抽稀绘制, 3000 点 × 4 通道
|
||||
重绘开销可控。
|
||||
"""
|
||||
|
||||
import bisect
|
||||
import time
|
||||
from collections import deque
|
||||
|
||||
from PySide6.QtCore import QPointF, QRect, Qt
|
||||
from PySide6.QtGui import QColor, QFont, QPainter, QPainterPath, QPen, QPalette
|
||||
from PySide6.QtWidgets import QSizePolicy, QWidget
|
||||
|
||||
|
||||
def fmt_val(v):
|
||||
"""数值可读显示: 普通小数带千分位(最多 4 位小数, 去尾零), 达 10 亿量级
|
||||
才退回科学计数 —— 调试现场 2.821e+07 这类写法不直观"""
|
||||
if v != v or v in (float('inf'), float('-inf')):
|
||||
return str(v)
|
||||
if abs(v) >= 1e9:
|
||||
return f'{v:.4g}'
|
||||
s = f'{v:,.4f}' if abs(v) < 1 else f'{v:,.3f}'
|
||||
return s.rstrip('0').rstrip('.') or '0'
|
||||
|
||||
|
||||
class PlotWidget(QWidget):
|
||||
|
||||
COLORS = ('#0f6cbd', '#d83b01', '#107c10', '#8764b8')
|
||||
MAXLEN = 3000 # 每通道缓冲点数(超限滚动丢弃)
|
||||
M_LEFT, M_TOP, M_RIGHT, M_BOTTOM = 52, 10, 12, 22 # 绘图区内边距
|
||||
|
||||
def __init__(self, channels=4, parent=None):
|
||||
super().__init__(parent)
|
||||
n = channels
|
||||
self._times = [deque(maxlen=self.MAXLEN) for _ in range(n)]
|
||||
self._vals = [deque(maxlen=self.MAXLEN) for _ in range(n)]
|
||||
self._names = [''] * n
|
||||
self._latest = [None] * n
|
||||
self._paused = False
|
||||
self._yrange = None # None=自动 或 (min, max)
|
||||
self._span = None # 可见时间窗宽(秒); None=跟随全部
|
||||
self._offset = 0.0 # 右端距最新数据的秒数(0=实时)
|
||||
self._frozen = False # 暂停渲染: 继续收数但不重绘
|
||||
self._drag_y = None # Y 轴拖动基准(仅手动量程)
|
||||
self.on_yrange_changed = None # 拖动改 Y 量程回调(主界面同步控件)
|
||||
self.on_span_changed = None # 滚轮改时间窗回调(主界面同步下拉)
|
||||
self._hover_pos = None # 悬停光标位置(None=不在绘图区)
|
||||
self._hover_lines = [] # 悬停读值(顶部小框内容)
|
||||
self._t0 = time.perf_counter()
|
||||
self.setMinimumHeight(150)
|
||||
self.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
|
||||
self.setMouseTracking(True) # 无按键也接收 mouseMove(悬停读值)
|
||||
|
||||
# --- 数据接口(主线程调用) ---
|
||||
|
||||
def set_channel_name(self, ch, name):
|
||||
self._names[ch] = name
|
||||
self._touch()
|
||||
|
||||
def set_yrange(self, yr):
|
||||
self._yrange = yr
|
||||
self._touch()
|
||||
|
||||
def set_view(self, span):
|
||||
"""设置可见时间窗宽(秒); None=跟随全部数据"""
|
||||
self._span = None if span is None else max(float(span), 1.0)
|
||||
self._touch()
|
||||
|
||||
def set_offset(self, offset):
|
||||
"""可见窗右端距最新数据的秒数(0=实时跟随最新)"""
|
||||
self._offset = max(float(offset), 0.0)
|
||||
self._touch()
|
||||
|
||||
def view_offset(self):
|
||||
return self._offset
|
||||
|
||||
def view_span(self):
|
||||
return self._span
|
||||
|
||||
def view_range(self):
|
||||
"""(全部数据 tmin, tmax, 可回看的最大滞后秒数) —— 供时间滑条换算"""
|
||||
t_all = [t for ts in self._times if ts for t in (ts[0], ts[-1])]
|
||||
tmin = min(t_all) if t_all else 0.0
|
||||
tmax = max(t_all) if t_all else tmin + 5.0
|
||||
span = self._span or max(tmax - tmin, 5.0)
|
||||
return tmin, tmax, max(tmax - tmin - span, 0.0)
|
||||
|
||||
def visible_domain(self, tmin, tmax):
|
||||
"""按时间窗/滞后换算可见横轴域(纯计算, 供单测)。
|
||||
不做数据边界收敛: 选 5min 窗而数据只有 23s 时, 显示完整的 5min 轴,
|
||||
数据贴右端随采集逐渐填满(示波器式); 滞后超出全部历史则为空窗"""
|
||||
if self._span is None:
|
||||
return tmin, tmax
|
||||
right = tmax - self._offset
|
||||
left = right - self._span
|
||||
if right <= left: # 滞后超出全部历史 -> 空窗保宽
|
||||
right = left + self._span
|
||||
return left, right
|
||||
|
||||
def auto_yrange(self):
|
||||
"""自动 Y 量程: 可见数据的 min/max 加 8% 边距(全平数据回落 ±1)"""
|
||||
visible = [v for ch in range(len(self._vals)) for v in self._vals[ch]]
|
||||
if not visible:
|
||||
return -1.0, 1.0
|
||||
ylo, yhi = min(visible), max(visible)
|
||||
if ylo == yhi:
|
||||
return ylo - 1.0, yhi + 1.0
|
||||
pad = (yhi - ylo) * 0.08
|
||||
return ylo - pad, yhi + pad
|
||||
|
||||
def set_paused(self, paused):
|
||||
self._paused = bool(paused)
|
||||
|
||||
def is_paused(self):
|
||||
return self._paused
|
||||
|
||||
def set_frozen(self, frozen):
|
||||
"""暂停渲染(弹窗用): 继续接收入库但不重绘; 解冻后一次性渲染全部积累。
|
||||
冻结是绝对状态: 覆写 update() 吞掉一切重绘请求, 任何代码路径都无法
|
||||
在冻结期打破(悬停/滑条等), 只有 set_frozen(False) 恢复"""
|
||||
self._frozen = bool(frozen)
|
||||
if not self._frozen:
|
||||
super().update()
|
||||
|
||||
def is_frozen(self):
|
||||
return self._frozen
|
||||
|
||||
def update(self):
|
||||
if self._frozen:
|
||||
return
|
||||
super().update()
|
||||
|
||||
def _touch(self):
|
||||
self.update()
|
||||
|
||||
def add(self, ch, value):
|
||||
if self._paused or value is None:
|
||||
return
|
||||
self._times[ch].append(time.perf_counter() - self._t0)
|
||||
self._vals[ch].append(float(value))
|
||||
self._latest[ch] = float(value)
|
||||
self._touch()
|
||||
|
||||
def clear(self):
|
||||
for t in self._times:
|
||||
t.clear()
|
||||
for v in self._vals:
|
||||
v.clear()
|
||||
self._latest = [None] * len(self._latest)
|
||||
self._touch()
|
||||
|
||||
# --- 悬停读值 ---
|
||||
|
||||
def _nearest(self, ch, t, tol):
|
||||
"""通道 ch 在时刻 t 附近(±tol 秒)最近的采样点 -> (名称, 值) 或 None"""
|
||||
ts, vs = self._times[ch], self._vals[ch]
|
||||
if not ts:
|
||||
return None
|
||||
lst = list(ts) # deque 无随机 bisect, 拷贝代价可接受
|
||||
i = bisect.bisect_left(lst, t)
|
||||
best = None
|
||||
for j in (i - 1, i):
|
||||
if 0 <= j < len(lst) and abs(lst[j] - t) <= tol:
|
||||
if best is None or abs(lst[j] - t) < abs(lst[best] - t):
|
||||
best = j
|
||||
if best is None:
|
||||
return None
|
||||
return self._names[ch] or f'CH{ch + 1}', vs[best]
|
||||
|
||||
def _plot_geometry(self):
|
||||
"""绘图区几何与可见横轴域: (left, top, pw, ph, tmin, tmax)"""
|
||||
left, top = self.M_LEFT, self.M_TOP
|
||||
pw = self.width() - left - self.M_RIGHT
|
||||
ph = self.height() - top - self.M_BOTTOM
|
||||
t_all = [t for ts in self._times if ts for t in (ts[0], ts[-1])]
|
||||
if t_all:
|
||||
tmin, tmax = min(t_all), max(t_all)
|
||||
else:
|
||||
tmin, tmax = 0.0, 5.0
|
||||
tmin, tmax = self.visible_domain(tmin, tmax)
|
||||
return left, top, pw, ph, tmin, tmax
|
||||
|
||||
def wheelEvent(self, ev):
|
||||
"""滚轮缩放时间窗: 上滚放大(窗宽变短), 下滚缩小; 滞后按比例缩放"""
|
||||
delta = ev.angleDelta().y()
|
||||
if delta:
|
||||
factor = 0.8 if delta > 0 else 1.25
|
||||
old = self._span
|
||||
if old is None: # 跟随模式 -> 以当前数据跨度为基准
|
||||
_tmin, tmax, _mp = self.view_range()
|
||||
old = max(tmax - _tmin, 5.0)
|
||||
new = max(min(old * factor, 900.0), 2.0)
|
||||
if self._offset and old:
|
||||
self._offset = self._offset * new / old
|
||||
self.set_view(new)
|
||||
if self.on_span_changed is not None:
|
||||
self.on_span_changed()
|
||||
super().wheelEvent(ev)
|
||||
|
||||
def mousePressEvent(self, ev):
|
||||
if ev.button() == Qt.LeftButton and self._yrange is not None:
|
||||
self._drag_y = ev.position().toPoint().y()
|
||||
super().mousePressEvent(ev)
|
||||
|
||||
def mouseReleaseEvent(self, ev):
|
||||
if ev.button() == Qt.LeftButton:
|
||||
self._drag_y = None
|
||||
super().mouseReleaseEvent(ev)
|
||||
|
||||
def mouseMoveEvent(self, ev):
|
||||
pos = ev.position().toPoint()
|
||||
if self._drag_y is not None and self._yrange is not None:
|
||||
_left, _top, _pw, ph, *_rest = self._plot_geometry()
|
||||
if ph > 40:
|
||||
ylo, yhi = self._yrange
|
||||
shift = (yhi - ylo) * (pos.y() - self._drag_y) / ph
|
||||
yr = (ylo - shift, yhi - shift) # 下拖数据下移, 上拖上移
|
||||
self.set_yrange(yr)
|
||||
if self.on_yrange_changed is not None:
|
||||
self.on_yrange_changed(yr)
|
||||
self._drag_y = pos.y()
|
||||
self._hover_pos = pos
|
||||
self.update()
|
||||
super().mouseMoveEvent(ev)
|
||||
|
||||
def leaveEvent(self, ev):
|
||||
self._hover_pos = None
|
||||
self.update()
|
||||
super().leaveEvent(ev)
|
||||
|
||||
def _hover_lines_at(self, pos):
|
||||
"""光标位置 -> 读值框内容; 不在绘图区/无命中返回空列表。
|
||||
由 paintEvent 每帧调用(而非仅在鼠标移动时), 鼠标停住不动,
|
||||
实时数据到来时读值依然持续刷新(用户实测反馈)"""
|
||||
left, top, pw, ph, tmin, tmax = self._plot_geometry()
|
||||
span = max(tmax - tmin, 0.001)
|
||||
if pw < 40 or ph < 40 or not (left <= pos.x() <= left + pw) \
|
||||
or not (top <= pos.y() <= top + ph):
|
||||
return []
|
||||
t = tmin + (pos.x() - left) / pw * span
|
||||
tol = span / max(pw, 1) * 10 # 光标 ±10px 内的采样算命中
|
||||
lines = [f'T = {t:.2f}s']
|
||||
for ch in range(len(self._vals)):
|
||||
got = self._nearest(ch, t, tol)
|
||||
if got:
|
||||
lines.append(f'{got[0]} = {fmt_val(got[1])}')
|
||||
return lines if len(lines) > 1 else []
|
||||
|
||||
# --- 绘制 ---
|
||||
|
||||
def paintEvent(self, _ev):
|
||||
p = QPainter(self)
|
||||
p.setRenderHint(QPainter.Antialiasing)
|
||||
rect = self.rect()
|
||||
pal = self.palette()
|
||||
p.fillRect(rect, pal.color(QPalette.Base))
|
||||
text = pal.color(QPalette.WindowText)
|
||||
dim = QColor(text)
|
||||
dim.setAlpha(110)
|
||||
grid = QColor(text)
|
||||
grid.setAlpha(38)
|
||||
|
||||
left, top = self.M_LEFT, self.M_TOP
|
||||
right, bottom = self.M_RIGHT, self.M_BOTTOM
|
||||
pw, ph = rect.width() - left - right, rect.height() - top - bottom
|
||||
if pw < 40 or ph < 40:
|
||||
return
|
||||
|
||||
p.setFont(QFont(self.font().family(), 8))
|
||||
p.setPen(dim)
|
||||
|
||||
# y 量程: 固定 或 自动(auto_yrange: 8% 边距, 全平数据回落 ±1)
|
||||
if self._yrange is not None:
|
||||
ylo, yhi = self._yrange
|
||||
if ylo >= yhi:
|
||||
yhi = ylo + 1.0
|
||||
else:
|
||||
ylo, yhi = self.auto_yrange()
|
||||
|
||||
def my(y): # 值 -> 像素 y
|
||||
return top + (yhi - y) / (yhi - ylo) * ph
|
||||
|
||||
# 横向参考线(4 等分)与每条线上的纵轴数值刻度
|
||||
p.setPen(QPen(grid, 1, Qt.DashLine))
|
||||
for i in range(1, 4):
|
||||
y = top + ph * i / 4
|
||||
p.drawLine(left, int(y), left + pw, int(y))
|
||||
for i in range(1, 6):
|
||||
x = left + pw * i / 6
|
||||
p.drawLine(int(x), top, int(x), top + ph)
|
||||
p.setPen(dim)
|
||||
for i in range(5):
|
||||
v = yhi - (yhi - ylo) * i / 4
|
||||
y = top + ph * i / 4
|
||||
p.drawText(QRect(2, int(y) - 9, left - 8, 18),
|
||||
Qt.AlignRight | Qt.AlignVCenter, fmt_val(v))
|
||||
|
||||
# x 范围: 时间窗(span)内、按滞后(offset)取可见域; 无窗宽则跟随全部数据
|
||||
t_all = [t for ch in range(len(self._times)) if self._times[ch]
|
||||
for t in (self._times[ch][0], self._times[ch][-1])]
|
||||
if t_all:
|
||||
tmin, tmax = min(t_all), max(t_all)
|
||||
else:
|
||||
tmin, tmax = 0.0, 5.0
|
||||
data_span = tmax - tmin
|
||||
tmin, tmax = self.visible_domain(tmin, tmax)
|
||||
span = max(tmax - tmin, 0.001)
|
||||
|
||||
def mx(t): # 时间 -> 像素 x
|
||||
return left + (t - tmin) / span * pw
|
||||
|
||||
empty = True
|
||||
for ch in range(len(self._vals)):
|
||||
ts, vs = self._times[ch], self._vals[ch]
|
||||
n = len(vs)
|
||||
if n == 0:
|
||||
continue
|
||||
empty = False
|
||||
color = QColor(self.COLORS[ch % len(self.COLORS)])
|
||||
p.setPen(QPen(color, 1.6))
|
||||
if n == 1:
|
||||
p.drawEllipse(QPointF(mx(ts[0]), my(vs[0])), 2.2, 2.2)
|
||||
continue
|
||||
step = max(1, n // 1000)
|
||||
path = QPainterPath(QPointF(mx(ts[0]), my(vs[0])))
|
||||
for i in range(step, n, step):
|
||||
path.lineTo(mx(ts[i]), my(vs[i]))
|
||||
path.lineTo(mx(ts[-1]), my(vs[-1]))
|
||||
p.drawPath(path)
|
||||
|
||||
p.setPen(dim)
|
||||
tag = '窗口' if self._span is not None else '最近'
|
||||
if self._span is not None and data_span < self._span - 0.5:
|
||||
label = f'已采集 {data_span:.0f}s / 窗口 {self._span:.0f}s'
|
||||
else:
|
||||
label = f'{tag} {span:.0f}s' if span >= 10 else f'{tag} {span:.1f}s'
|
||||
p.drawText(rect.adjusted(0, rect.height() - 18, -4, 0),
|
||||
Qt.AlignRight | Qt.AlignVCenter, label)
|
||||
if self._hover_pos is not None:
|
||||
self._hover_lines = self._hover_lines_at(self._hover_pos)
|
||||
if self._hover_lines:
|
||||
self._hover_box(p)
|
||||
if empty:
|
||||
p.drawText(rect, Qt.AlignCenter, '等待数据…(勾选通道并使能接收)')
|
||||
elif self._paused:
|
||||
p.drawText(rect.adjusted(0, top, 0, 0), Qt.AlignHCenter, '已暂停')
|
||||
elif self._frozen:
|
||||
p.drawText(rect.adjusted(0, top, 0, 0), Qt.AlignHCenter,
|
||||
'已冻结渲染(采集继续)')
|
||||
self._legend(p, rect, text)
|
||||
|
||||
def _hover_box(self, p):
|
||||
"""顶部居中的悬停读值小框(反色底), 内容在 mouseMove 里持续刷新"""
|
||||
p.setFont(QFont(self.font().family(), 9))
|
||||
fm = p.fontMetrics()
|
||||
w = max(fm.horizontalAdvance(s) for s in self._hover_lines) + 16
|
||||
h = fm.height() * len(self._hover_lines) + 10
|
||||
x = self.M_LEFT + (self.width() - self.M_LEFT - self.M_RIGHT - w) // 2
|
||||
y = self.M_TOP + 6
|
||||
bg = QColor(self.palette().color(QPalette.WindowText))
|
||||
bg.setAlpha(210)
|
||||
p.setPen(Qt.NoPen)
|
||||
p.setBrush(bg)
|
||||
p.drawRoundedRect(x, y, w, h, 4, 4)
|
||||
p.setPen(self.palette().color(QPalette.Base))
|
||||
for i, s in enumerate(self._hover_lines):
|
||||
p.drawText(x + 8, y + 5 + i * fm.height(), w - 16, fm.height(),
|
||||
Qt.AlignLeft | Qt.AlignVCenter, s)
|
||||
p.setBrush(Qt.NoBrush)
|
||||
|
||||
def _legend(self, p, rect, text):
|
||||
"""右上角图例: 色块 + 通道名 + 最新值"""
|
||||
lines = []
|
||||
for ch in range(len(self._vals)):
|
||||
if self._latest[ch] is None:
|
||||
continue
|
||||
name = self._names[ch] or f'CH{ch + 1}'
|
||||
lines.append((ch, f'{name} = {fmt_val(self._latest[ch])}'))
|
||||
if not lines:
|
||||
return
|
||||
p.setFont(QFont(self.font().family(), 8))
|
||||
fm = p.fontMetrics()
|
||||
w = max(fm.horizontalAdvance(s) for _c, s in lines) + 26
|
||||
h = fm.height() * len(lines) + 8
|
||||
x1, y1 = rect.width() - 16 - w, 8
|
||||
bg = QColor(text)
|
||||
bg.setAlpha(18)
|
||||
p.setPen(Qt.NoPen)
|
||||
p.setBrush(bg)
|
||||
p.drawRoundedRect(x1, y1, w, h, 4, 4)
|
||||
p.setBrush(Qt.NoBrush)
|
||||
for i, (ch, s) in enumerate(lines):
|
||||
y = y1 + 4 + i * fm.height()
|
||||
p.setPen(Qt.NoPen)
|
||||
p.setBrush(QColor(self.COLORS[ch % len(self.COLORS)]))
|
||||
p.drawRect(x1 + 6, y + fm.ascent() - 8, 8, 8)
|
||||
p.setPen(self.palette().color(QPalette.WindowText))
|
||||
p.drawText(x1 + 20, y, w - 22, fm.height(),
|
||||
Qt.AlignLeft | Qt.AlignVCenter, s)
|
||||
@@ -0,0 +1,450 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
SerialAssistant 串口协议模块(纯标准库, 帧构成可定制)
|
||||
======================================================
|
||||
帧由"构成令牌序列"定义(默认与 DLPilot 光机 RS485 协议完全一致):
|
||||
HDR ID LEN PAYLOAD CRC -> 包头 + 板ID + LEN + payload + CRC
|
||||
LEN = payload 字节数(口径可选, 见 LEN_SEMS), 默认即"包长对应 payload 长度"
|
||||
令牌: HDR=包头 ID=板ID LEN=长度 PAYLOAD=载荷 CRC=校验(必须最后),
|
||||
其余 1~2 位十六进制(如 AA 55)为固定字面量字节(双字节同步头)。
|
||||
约束: LEN/PAYLOAD/CRC 恰好各一次; LEN 前至少一个 HDR 或字面量(同步定界用)。
|
||||
同一布局收发共用, 仅包头值分方向(TX=tx_hdr / RX=rx_hdr); ID 在 RX 侧
|
||||
按上游语义多为响应 TYPE 码, 解析时不校验、原样透出(Frame.code, 可为 None)。
|
||||
|
||||
CRC 算法(2 字节算法在线上的字节序: MODBUS 低位在前, 其余高位在前):
|
||||
SUM8 累加和取低 8 位(1 字节)
|
||||
CRC8 poly 0x07 init 0x00(1 字节, 标准校验值 check("123456789")=0xF4)
|
||||
CRC16_MODBUS poly 0x8005 反射 init 0xFFFF(2 字节, check=0x4B37)
|
||||
CRC16_XMODEM poly 0x1021 init 0x0000(2 字节, check=0x31C3)
|
||||
CRC16_CCITT poly 0x1021 init 0xFFFF(2 字节, check=0x29B1)
|
||||
|
||||
CRC 覆盖范围(CRC_SCOPES, 按帧内位置定义, 任意布局通用):
|
||||
with_header 包头(第1字节)起至 CRC 前
|
||||
no_header 第 2 字节起至 CRC 前(DLPilot 默认)
|
||||
payload_only 仅 payload
|
||||
|
||||
仅标准库; pyserial 只允许出现在 link.py。
|
||||
"""
|
||||
|
||||
import collections
|
||||
import re
|
||||
import struct
|
||||
|
||||
LEN_MAX = 255 # LEN 为单字节, payload 上限 255
|
||||
|
||||
TYPE_NAMES = { # DLPilot/上游 protocols.md 的响应 TYPE 码语义, 用于 RX 译码显示
|
||||
0xF0: '成功',
|
||||
0xF1: 'CRC 校验错',
|
||||
0xF2: '包头错',
|
||||
0xF3: '板类型错',
|
||||
0xF4: '格式/缺参数',
|
||||
0xFD: '参数越界/读回复核不一致',
|
||||
0xFE: '未知命令',
|
||||
0xFF: '通用错误(含就绪门禁拒绝)',
|
||||
0xFA: '开光机通信失败',
|
||||
0xFB: '关光机通信失败',
|
||||
0xFC: '读回复核通信失败',
|
||||
}
|
||||
|
||||
|
||||
# --- CRC 实现(查表, 供高波特率下增量解析用) ---------------------------------
|
||||
|
||||
def _table_crc8():
|
||||
tbl = []
|
||||
for b in range(256):
|
||||
c = b
|
||||
for _ in range(8):
|
||||
c = ((c << 1) ^ 0x07) & 0xFF if c & 0x80 else (c << 1) & 0xFF
|
||||
tbl.append(c)
|
||||
return tuple(tbl)
|
||||
|
||||
|
||||
def _table_crc16_msb():
|
||||
"""MSB-first 表(poly 0x1021, 用于 XMODEM/CCITT)"""
|
||||
tbl = []
|
||||
for b in range(256):
|
||||
c = (b << 8) & 0xFFFF
|
||||
for _ in range(8):
|
||||
c = ((c << 1) ^ 0x1021) & 0xFFFF if c & 0x8000 else (c << 1) & 0xFFFF
|
||||
tbl.append(c)
|
||||
return tuple(tbl)
|
||||
|
||||
|
||||
def _table_crc16_reflected():
|
||||
"""反射表(poly 0x8005 → 0xA001, 用于 MODBUS)"""
|
||||
tbl = []
|
||||
for b in range(256):
|
||||
c = b
|
||||
for _ in range(8):
|
||||
c = (c >> 1) ^ 0xA001 if c & 1 else c >> 1
|
||||
tbl.append(c)
|
||||
return tuple(tbl)
|
||||
|
||||
|
||||
_TAB_CRC8 = _table_crc8()
|
||||
_TAB_CRC16_MSB = _table_crc16_msb()
|
||||
_TAB_CRC16_REF = _table_crc16_reflected()
|
||||
|
||||
|
||||
def _sum8(data):
|
||||
return sum(data) & 0xFF
|
||||
|
||||
|
||||
def _crc8(data, _t=_TAB_CRC8):
|
||||
c = 0
|
||||
for b in data:
|
||||
c = _t[c ^ b]
|
||||
return c
|
||||
|
||||
|
||||
def _crc16_msb(data, init, _t=_TAB_CRC16_MSB):
|
||||
c = init
|
||||
for b in data:
|
||||
c = ((c << 8) & 0xFFFF) ^ _t[((c >> 8) ^ b) & 0xFF]
|
||||
return c
|
||||
|
||||
|
||||
def _crc16_xmodem(data):
|
||||
return _crc16_msb(data, 0x0000)
|
||||
|
||||
|
||||
def _crc16_ccitt(data):
|
||||
return _crc16_msb(data, 0xFFFF)
|
||||
|
||||
|
||||
def _crc16_modbus(data, _t=_TAB_CRC16_REF):
|
||||
c = 0xFFFF
|
||||
for b in data:
|
||||
c = (c >> 8) ^ _t[(c ^ b) & 0xFF]
|
||||
return c
|
||||
|
||||
|
||||
# name -> (CRC 字节数, 计算函数, 线上字节序: 'big'=高字节在前 / 'little'=低字节在前)
|
||||
CRC_ALGOS = {
|
||||
'SUM8': (1, _sum8, 'big'),
|
||||
'CRC8': (1, _crc8, 'big'),
|
||||
'CRC16_MODBUS': (2, _crc16_modbus, 'little'),
|
||||
'CRC16_XMODEM': (2, _crc16_xmodem, 'big'),
|
||||
'CRC16_CCITT': (2, _crc16_ccitt, 'big'),
|
||||
}
|
||||
CRC_ALGO_LABELS = { # 下拉框显示文本
|
||||
'SUM8': 'SUM8 累加和',
|
||||
'CRC8': 'CRC8 (poly 07)',
|
||||
'CRC16_MODBUS': 'CRC16 MODBUS',
|
||||
'CRC16_XMODEM': 'CRC16 XMODEM',
|
||||
'CRC16_CCITT': 'CRC16 CCITT-FALSE',
|
||||
}
|
||||
CRC_SCOPES = ('with_header', 'no_header', 'payload_only')
|
||||
CRC_SCOPE_LABELS = {
|
||||
'with_header': '含包头(整帧不含CRC)',
|
||||
'no_header': '不含包头(跳过第1字节, DLPilot)',
|
||||
'payload_only': '仅 payload',
|
||||
}
|
||||
LEN_SEMS = ('payload', 'body', 'total')
|
||||
LEN_SEM_LABELS = { # LEN 字节的计数口径
|
||||
'payload': 'payload 长度(默认)',
|
||||
'body': '到帧尾(payload+CRC)',
|
||||
'total': '整帧(含包头)',
|
||||
}
|
||||
|
||||
DEFAULT_LAYOUT = 'HDR ID LEN PAYLOAD CRC'
|
||||
_FRAME_TOKEN_RE = re.compile(r'^(?:0x)?([0-9A-Fa-f]{1,2})$')
|
||||
_KEYWORDS = {'HDR': 'hdr', 'ID': 'id', 'BOARD': 'id', 'LEN': 'len',
|
||||
'PAYLOAD': 'payload', 'DATA': 'payload', 'CRC': 'crc'}
|
||||
|
||||
|
||||
class FrameLayout:
|
||||
"""帧构成: 令牌序列(parse 文本而来), 供组帧与解析共用。
|
||||
|
||||
tokens 元素: 'hdr'/'id'/'len'/'payload'/'crc' 或 ('lit', 字节值)。
|
||||
约束(parse 内校验): LEN/PAYLOAD/CRC 恰好各一次; CRC 必须最后;
|
||||
LEN 之前至少一个 HDR 或字面量(否则无同步定界); ID 最多一次。
|
||||
payload 之前的令牌都是 1 字节(CRC 在最后), 故 payload 字节偏移=payload_pos。
|
||||
"""
|
||||
|
||||
def __init__(self, tokens):
|
||||
self.tokens = tuple(tokens)
|
||||
counts = collections.Counter(
|
||||
t if isinstance(t, str) else 'lit' for t in self.tokens)
|
||||
for key, cn in (('len', 'LEN'), ('payload', 'PAYLOAD'), ('crc', 'CRC')):
|
||||
if counts[key] != 1:
|
||||
raise ValueError(f'{cn} 必须恰好出现一次(当前 {counts[key]} 次)')
|
||||
if self.tokens[-1] != 'crc':
|
||||
raise ValueError('CRC 必须是最后一个令牌')
|
||||
if counts['id'] > 1:
|
||||
raise ValueError('ID 最多出现一次')
|
||||
self.len_pos = self.tokens.index('len')
|
||||
self.payload_pos = self.tokens.index('payload')
|
||||
id_positions = [i for i, t in enumerate(self.tokens) if t == 'id']
|
||||
self.id_pos = id_positions[0] if id_positions else None
|
||||
if not any(isinstance(t, tuple) or t == 'hdr'
|
||||
for t in self.tokens[:self.len_pos]):
|
||||
raise ValueError('LEN 之前需要至少一个 HDR 或十六进制字面量作同步头')
|
||||
|
||||
@classmethod
|
||||
def parse(cls, text):
|
||||
"""'AA 55 ID LEN PAYLOAD CRC' 之类 -> FrameLayout; 非法抛 ValueError"""
|
||||
parts = [p for p in re.split(r'[\s,，;；]+', (text or '').strip()) if p]
|
||||
if not parts:
|
||||
raise ValueError('帧构成为空')
|
||||
tokens = []
|
||||
for p in parts:
|
||||
kw = _KEYWORDS.get(p.upper())
|
||||
if kw:
|
||||
tokens.append(kw)
|
||||
continue
|
||||
m = _FRAME_TOKEN_RE.match(p)
|
||||
if not m:
|
||||
raise ValueError(f'未知令牌 "{p}"(可用: HDR ID LEN PAYLOAD CRC '
|
||||
f'或 0~FF 十六进制字面量)')
|
||||
tokens.append(('lit', int(m.group(1), 16)))
|
||||
return cls(tokens)
|
||||
|
||||
# --- 尺寸(依赖 CRC 字节数, 由 ProtoConfig 传参) ---
|
||||
|
||||
def rest_size(self, crc_size):
|
||||
"""payload 之后的字节数(ID/HDR/字面量 + CRC)"""
|
||||
return crc_size + len(self.tokens) - self.payload_pos - 2
|
||||
|
||||
def total_size(self, payload_len, crc_size):
|
||||
# payload 之前的令牌都是 1 字节, payload 字节偏移即 payload_pos
|
||||
return self.payload_pos + payload_len + self.rest_size(crc_size)
|
||||
|
||||
def len_extra(self, len_sem, crc_size):
|
||||
"""LEN 字节值 = payload_len + 该口径下的附加字节数"""
|
||||
if len_sem == 'body':
|
||||
return self.rest_size(crc_size)
|
||||
if len_sem == 'total':
|
||||
return self.payload_pos + self.rest_size(crc_size)
|
||||
return 0
|
||||
|
||||
|
||||
Frame = collections.namedtuple('Frame', 'code payload raw')
|
||||
"""code=ID 位置字节(RX 侧多为响应 TYPE 码; 布局无 ID 时为 None),
|
||||
payload=原始字节, raw=整帧含 CRC"""
|
||||
|
||||
|
||||
class ProtoConfig:
|
||||
"""协议配置: 帧构成(FrameLayout) + 包头(TX/RX)/板ID/LEN 口径/CRC 算法与覆盖"""
|
||||
|
||||
def __init__(self, tx_hdr=0xD5, rx_hdr=0xB5, board=0x01,
|
||||
crc_algo='SUM8', crc_scope='no_header',
|
||||
layout=None, len_sem='payload'):
|
||||
for name, v in (('tx_hdr', tx_hdr), ('rx_hdr', rx_hdr), ('board', board)):
|
||||
if not 0 <= int(v) <= 0xFF:
|
||||
raise ValueError(f'{name} 超出 0~255: {v}')
|
||||
if crc_algo not in CRC_ALGOS:
|
||||
raise ValueError(f'未知 CRC 算法: {crc_algo}')
|
||||
if crc_scope not in CRC_SCOPES:
|
||||
raise ValueError(f'未知 CRC 覆盖范围: {crc_scope}')
|
||||
if len_sem not in LEN_SEMS:
|
||||
raise ValueError(f'未知 LEN 口径: {len_sem}')
|
||||
self.tx_hdr, self.rx_hdr, self.board = int(tx_hdr), int(rx_hdr), int(board)
|
||||
self.crc_algo, self.crc_scope = crc_algo, crc_scope
|
||||
self.len_sem = len_sem
|
||||
self.layout = layout if layout is not None else FrameLayout.parse(DEFAULT_LAYOUT)
|
||||
|
||||
@property
|
||||
def crc_size(self):
|
||||
return CRC_ALGOS[self.crc_algo][0]
|
||||
|
||||
# --- CRC ---
|
||||
|
||||
def compute_crc(self, nocrc_frame, payload_len):
|
||||
"""对不含 CRC 的整帧按覆盖范围切片算 CRC 值(切片按位置定义, 任意布局通用)"""
|
||||
_size, fn, _order = CRC_ALGOS[self.crc_algo]
|
||||
if self.crc_scope == 'no_header':
|
||||
region = nocrc_frame[1:]
|
||||
elif self.crc_scope == 'payload_only':
|
||||
po = self.layout.payload_pos
|
||||
region = nocrc_frame[po:po + payload_len]
|
||||
else:
|
||||
region = nocrc_frame
|
||||
return fn(region)
|
||||
|
||||
def encode_crc(self, value):
|
||||
"""CRC 值 -> 线上字节(2 字节算法按算法约定字节序)"""
|
||||
size, _fn, order = CRC_ALGOS[self.crc_algo]
|
||||
if size == 1:
|
||||
return bytes((value & 0xFF,))
|
||||
hi, lo = (value >> 8) & 0xFF, value & 0xFF
|
||||
return bytes((hi, lo)) if order == 'big' else bytes((lo, hi))
|
||||
|
||||
# --- 组帧 ---
|
||||
|
||||
def build_frame(self, payload):
|
||||
"""payload bytes -> 完整帧 bytes; 空载荷允许(LEN=0), 超限抛 ValueError"""
|
||||
payload = bytes(payload)
|
||||
if len(payload) > LEN_MAX:
|
||||
raise ValueError(f'payload 超长(>{LEN_MAX} 字节)')
|
||||
cs = self.crc_size
|
||||
len_val = len(payload) + self.layout.len_extra(self.len_sem, cs)
|
||||
if len_val > LEN_MAX:
|
||||
raise ValueError(f'LEN 字节溢出(={len_val}>255), 检查帧构成与 LEN 口径')
|
||||
parts = []
|
||||
for tok in self.layout.tokens:
|
||||
if tok == 'crc':
|
||||
continue # CRC 最后, 跳过待拼
|
||||
if tok == 'hdr':
|
||||
parts.append(bytes((self.tx_hdr,)))
|
||||
elif tok == 'id':
|
||||
parts.append(bytes((self.board,)))
|
||||
elif tok == 'len':
|
||||
parts.append(bytes((len_val,)))
|
||||
elif tok == 'payload':
|
||||
parts.append(payload)
|
||||
else:
|
||||
parts.append(bytes((tok[1],)))
|
||||
nocrc = b''.join(parts)
|
||||
return nocrc + self.encode_crc(self.compute_crc(nocrc, len(payload)))
|
||||
|
||||
# --- 校验与拆帧 ---
|
||||
|
||||
def _plen_of(self, raw):
|
||||
"""从整帧 LEN 字节解出 payload 长度; 非法(负值)返回 None"""
|
||||
plen = raw[self.layout.len_pos] - self.layout.len_extra(self.len_sem, self.crc_size)
|
||||
return plen if plen >= 0 else None
|
||||
|
||||
def verify_frame(self, raw):
|
||||
"""整帧(含 CRC)校验: 长度与 LEN 口径自洽 + CRC 匹配"""
|
||||
cs = self.crc_size
|
||||
if len(raw) < self.layout.len_pos + 1 + cs:
|
||||
return False
|
||||
plen = self._plen_of(raw)
|
||||
if plen is None or len(raw) != self.layout.total_size(plen, cs):
|
||||
return False
|
||||
return self.encode_crc(self.compute_crc(raw[:-cs], plen)) == raw[-cs:]
|
||||
|
||||
def parse_frame(self, raw):
|
||||
"""整帧 -> Frame(不校验 CRC, 调用方先用 verify_frame); 长度不符抛 ValueError"""
|
||||
cs = self.crc_size
|
||||
plen = self._plen_of(raw)
|
||||
if plen is None or len(raw) != self.layout.total_size(plen, cs):
|
||||
raise ValueError('帧长与 LEN 口径不符')
|
||||
po = self.layout.payload_pos
|
||||
code = raw[self.layout.id_pos] if self.layout.id_pos is not None else None
|
||||
return Frame(code, raw[po:po + plen], bytes(raw))
|
||||
|
||||
|
||||
class FrameParser:
|
||||
"""增量字节流 -> 完整帧: 前缀令牌逐位对齐(帧头/字面量, ID 为通配) + LEN 定长
|
||||
+ CRC 校验; 任一处失配丢 1 字节重新同步。LEN 值经口径换算后即 payload 长度,
|
||||
负值视为伪帧头逐出 —— 伪帧头最多滞留一帧即被逐出重同步, 对调试工具可接受。
|
||||
"""
|
||||
|
||||
def __init__(self, cfg):
|
||||
self.cfg = cfg
|
||||
lay = cfg.layout
|
||||
cs = cfg.crc_size
|
||||
# LEN 之前的前缀令牌: (帧内位置, 期望字节|None=通配); hdr 按 RX 包头匹配
|
||||
self._prefix = []
|
||||
for pos, tok in enumerate(lay.tokens[:lay.len_pos]):
|
||||
exp = None
|
||||
if tok == 'hdr':
|
||||
exp = cfg.rx_hdr
|
||||
elif isinstance(tok, tuple):
|
||||
exp = tok[1]
|
||||
self._prefix.append((pos, exp))
|
||||
self._len_pos = lay.len_pos
|
||||
self._extra = lay.len_extra(cfg.len_sem, cs)
|
||||
self._buf = bytearray()
|
||||
|
||||
def clear(self):
|
||||
self._buf.clear()
|
||||
|
||||
def _aligned(self):
|
||||
"""把缓冲对齐到前缀匹配; 字节不足且已匹配部分则保持等待(返回 True),
|
||||
失配丢 1 字节重试, 缓冲耗尽返回 False"""
|
||||
buf = self._buf
|
||||
while True:
|
||||
for pos, exp in self._prefix:
|
||||
if pos >= len(buf):
|
||||
return True # 已匹配部分全对, 等更多字节
|
||||
if exp is not None and buf[pos] != exp:
|
||||
break # 失配 -> 丢 1 字节
|
||||
else:
|
||||
return True
|
||||
del buf[:1]
|
||||
if not buf:
|
||||
return False
|
||||
|
||||
def feed(self, data):
|
||||
"""喂入新收到的字节, 返回本次解析出的完整 Frame 列表"""
|
||||
self._buf += bytes(data)
|
||||
cfg, frames = self.cfg, []
|
||||
while self._aligned() and self._buf:
|
||||
if len(self._buf) < self._len_pos + 1:
|
||||
break # LEN 字节未到齐
|
||||
plen = self._buf[self._len_pos] - self._extra
|
||||
if plen < 0: # LEN 为负 -> 伪帧头, 逐出重同步
|
||||
del self._buf[:1]
|
||||
continue
|
||||
total = cfg.layout.total_size(plen, cfg.crc_size)
|
||||
if len(self._buf) < total:
|
||||
break # 整帧未收齐
|
||||
raw = bytes(self._buf[:total])
|
||||
if cfg.verify_frame(raw):
|
||||
frames.append(cfg.parse_frame(raw))
|
||||
del self._buf[:total]
|
||||
else:
|
||||
del self._buf[:1] # CRC 失配 -> 丢帧头重同步
|
||||
return frames
|
||||
|
||||
|
||||
# --- payload 数值提取(曲线通道用) --------------------------------------------
|
||||
|
||||
BIN_FORMATS = { # fmt -> (struct 格式, 字节数)
|
||||
'u8': ('B', 1), 's8': ('b', 1),
|
||||
'u16be': ('>H', 2), 'u16le': ('<H', 2),
|
||||
's16be': ('>h', 2), 's16le': ('<h', 2),
|
||||
'u32be': ('>I', 4), 'u32le': ('<I', 4),
|
||||
's32be': ('>i', 4), 's32le': ('<i', 4),
|
||||
'f32be': ('>f', 4), 'f32le': ('<f', 4),
|
||||
}
|
||||
NUM_RE = re.compile(r'-?\d+(?:\.\d+)?')
|
||||
|
||||
|
||||
def extract_value(payload, mode, *, offset=0, fmt='u16be', index=1,
|
||||
pattern=r'(?<=:)\d+', scale=1.0):
|
||||
"""从 payload 提取一个数值(已乘 scale); 提取失败返回 None。
|
||||
|
||||
bin 二进制定长提取: offset 起, 按 fmt(BIN_FORMATS 键)解
|
||||
num ASCII 文本第 index(1 基)个数字, 如 'okTL:246' -> 246
|
||||
re 正则第 1 捕获组(无组则整体), 如 '(?<=:)(\\d+)'
|
||||
"""
|
||||
payload = bytes(payload)
|
||||
try:
|
||||
if mode == 'bin':
|
||||
if fmt not in BIN_FORMATS:
|
||||
return None
|
||||
sfmt, size = BIN_FORMATS[fmt]
|
||||
if offset < 0 or len(payload) < offset + size:
|
||||
return None
|
||||
value = struct.unpack(sfmt, payload[offset:offset + size])[0]
|
||||
elif mode == 'num':
|
||||
nums = NUM_RE.findall(payload.decode('ascii', 'replace'))
|
||||
if not 1 <= index <= len(nums):
|
||||
return None
|
||||
value = float(nums[index - 1])
|
||||
elif mode == 're':
|
||||
m = re.search(pattern, payload.decode('ascii', 'replace'))
|
||||
if not m:
|
||||
return None
|
||||
value = float(m.group(1) if m.groups() else m.group(0))
|
||||
else:
|
||||
return None
|
||||
except (ValueError, re.error):
|
||||
return None
|
||||
return value * scale
|
||||
|
||||
|
||||
def hexs(data):
|
||||
"""bytes -> 'AA BB CC' 大写 hex, 日志统一格式"""
|
||||
return bytes(data).hex(' ').upper()
|
||||
|
||||
|
||||
def clean_text(s):
|
||||
"""控制字符 -> '·': 二进制按 ASCII 解出的 NUL 等会污染日志行,
|
||||
且 Windows 剪贴板遇 NUL 直接截断复制内容(DLPilot 同源教训)"""
|
||||
return ''.join(c if ord(c) >= 0x20 and ord(c) != 0x7F else '·' for c in s)
|
||||
@@ -0,0 +1,157 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
创建/更新 Gitea Release 并上传发布 zip(纯标准库, 无需安装依赖)
|
||||
================================================================
|
||||
用法:
|
||||
python publish_gitea.py SerialMate-v0.1.0-win64.zip
|
||||
python publish_gitea.py <zip> --notes-file release/notes-v0.1.0.md
|
||||
python publish_gitea.py <zip> --tag SerialMate-v0.1.0 --replace
|
||||
|
||||
行为:
|
||||
- tag 与版本默认取 zip 文件名(SerialMate-vX.Y.Z-win64.zip → SerialMate-vX.Y.Z)
|
||||
- release 不存在则创建(tag 不存在时 Gitea 在默认分支头自动建 tag);
|
||||
已存在则复用, 仅上传/更新资产
|
||||
- release 上已有同名资产时: 默认跳过, --replace 删旧再传
|
||||
- API token: --token 参数或环境变量 GITEA_TOKEN(需该仓库写权限)
|
||||
- 本实例 TLS 间歇抽风: 已内置连接级重试, 实际发布请加 --insecure
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import ssl
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
DEFAULT_NOTES = """## SerialMate {version}
|
||||
|
||||
RS485/串口调试助手(LT8619C 调试工具集配套): 串口收发 + 协议包解析,
|
||||
帧格式(包头/板ID/LEN/CRC 算法与覆盖范围)全可定制, 默认与 DLPilot 光机
|
||||
RS485 协议一致; payload 支持字符串/HEX 显示、4 通道曲线绘制, 发送框输入
|
||||
payload 自动包装组帧(多条用 / 分隔), 也可 HEX 直发与周期发送。
|
||||
|
||||
### 安装
|
||||
下载 `SerialMate-v{version}-win64.zip`, 解压后双击 `SerialMate.exe`
|
||||
(Windows 10/11 x64, 免安装; 串口功能需系统可访问串口, 程序自带 pyserial)。
|
||||
|
||||
### 包内容
|
||||
- `SerialMate.exe` — 主程序
|
||||
- `README.md` — 完整使用说明
|
||||
"""
|
||||
|
||||
|
||||
def git_remote_info():
|
||||
"""从 origin 远程地址解析 API base 与 owner/repo(ssh 端口不影响 https API)"""
|
||||
url = subprocess.check_output(['git', 'remote', 'get-url', 'origin']).decode().strip()
|
||||
m = re.match(r'(?:ssh://git@|git@)([\w.-]+)(?::\d+)?[:/]([\w.-]+)/([\w.-]+?)(?:\.git)?$', url)
|
||||
if not m:
|
||||
sys.exit(f'无法从 origin 解析 Gitea 地址: {url}')
|
||||
host, owner, repo = m.groups()
|
||||
return f'https://{host}', owner, repo
|
||||
|
||||
|
||||
class Api:
|
||||
def __init__(self, base, owner, repo, token, insecure):
|
||||
self.api = f'{base}/api/v1/repos/{owner}/{repo}'
|
||||
self.headers = {'Authorization': f'token {token}'}
|
||||
self.ctx = ssl._create_unverified_context() if insecure else None
|
||||
|
||||
def request(self, method, path, data=None, headers=None, retries=4):
|
||||
"""发请求; 实例 TLS 间歇性抽风(SSL UNEXPECTED_EOF), 连接级错误自动重试"""
|
||||
last = None
|
||||
for attempt in range(retries):
|
||||
req = urllib.request.Request(self.api + path, data=data, method=method,
|
||||
headers={**self.headers, **(headers or {})})
|
||||
try:
|
||||
with urllib.request.urlopen(req, context=self.ctx) as resp:
|
||||
body = resp.read()
|
||||
return resp.status, json.loads(body) if body else None
|
||||
except urllib.error.HTTPError as e:
|
||||
return e.code, json.loads(e.read() or b'{}')
|
||||
except (urllib.error.URLError, ssl.SSLError, OSError) as e:
|
||||
last = e
|
||||
time.sleep(1.0 + attempt)
|
||||
raise RuntimeError(f'Gitea 请求连续 {retries} 次失败: {last}')
|
||||
|
||||
def get_json(self, path):
|
||||
status, data = self.request('GET', path)
|
||||
return data if status == 200 else None
|
||||
|
||||
def upload_asset(self, release_id, filepath, replace):
|
||||
name = os.path.basename(filepath)
|
||||
status, rel = self.request('GET', f'/releases/{release_id}')
|
||||
if status != 200:
|
||||
sys.exit(f'读取 release {release_id} 失败: {status} {rel}')
|
||||
# Gitea 版本差异: assets 可能是 {items:[...]} 或纯数组
|
||||
assets = rel.get('assets') or []
|
||||
if isinstance(assets, dict):
|
||||
assets = assets.get('items', [])
|
||||
for asset in assets:
|
||||
if asset['name'] == name:
|
||||
if not replace:
|
||||
print(f'资产 {name} 已存在, 跳过(覆盖请加 --replace)')
|
||||
return
|
||||
self.request('DELETE', f"/releases/{release_id}/assets/{asset['id']}")
|
||||
print(f'已删除旧资产 {name}')
|
||||
boundary = uuid.uuid4().hex
|
||||
payload = open(filepath, 'rb').read()
|
||||
body = (f'--{boundary}\r\n'
|
||||
f'Content-Disposition: form-data; name="attachment"; filename="{name}"\r\n'
|
||||
f'Content-Type: application/octet-stream\r\n\r\n').encode() + payload + \
|
||||
f'\r\n--{boundary}--\r\n'.encode()
|
||||
status, data = self.request(
|
||||
'POST', f'/releases/{release_id}/assets?name={name}', body,
|
||||
{'Content-Type': f'multipart/form-data; boundary={boundary}'})
|
||||
if status != 201:
|
||||
sys.exit(f'上传资产失败: {status} {data}')
|
||||
print(f'已上传资产: {name} ({len(payload) / 1024 / 1024:.1f} MB)')
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument('zipfile_path', help='发布 zip(release/ 下)')
|
||||
ap.add_argument('--tag', help='release 标签, 默认取 zip 文件名去掉 -win64.zip')
|
||||
ap.add_argument('--notes-file', help='release 说明 markdown 文件, 默认用内置模板')
|
||||
ap.add_argument('--token', default=os.environ.get('GITEA_TOKEN'), help='API token')
|
||||
ap.add_argument('--replace', action='store_true', help='覆盖已存在的同名资产')
|
||||
ap.add_argument('--insecure', action='store_true', help='跳过 TLS 证书校验(自签证书)')
|
||||
args = ap.parse_args()
|
||||
|
||||
if not os.path.exists(args.zipfile_path):
|
||||
sys.exit(f'找不到 {args.zipfile_path}')
|
||||
if not args.token:
|
||||
sys.exit('缺少 API token: 传 --token 或设环境变量 GITEA_TOKEN')
|
||||
|
||||
tag = args.tag or re.sub(r'-win64\.zip$', '', os.path.basename(args.zipfile_path))
|
||||
version = tag.removeprefix('SerialMate-v')
|
||||
notes = open(args.notes_file, encoding='utf-8').read() if args.notes_file \
|
||||
else DEFAULT_NOTES.format(version=version)
|
||||
|
||||
base, owner, repo = git_remote_info()
|
||||
api = Api(base, owner, repo, args.token, args.insecure)
|
||||
print(f'目标: {base}/{owner}/{repo} tag={tag}')
|
||||
|
||||
rel = api.get_json(f'/releases/tags/{tag}')
|
||||
if rel:
|
||||
print(f'release 已存在(id={rel["id"]}), 复用')
|
||||
else:
|
||||
status, rel = api.request('POST', '/releases', json.dumps({
|
||||
'tag_name': tag,
|
||||
'name': f'SerialMate v{version}',
|
||||
'body': notes,
|
||||
}).encode(), {'Content-Type': 'application/json'})
|
||||
if status != 201:
|
||||
sys.exit(f'创建 release 失败: {status} {rel}')
|
||||
print(f'release 已创建(id={rel["id"]})')
|
||||
api.upload_asset(rel['id'], args.zipfile_path, args.replace)
|
||||
print(f'完成: {rel.get("html_url", "") or f"{base}/{owner}/{repo}/releases/tag/{tag}"}')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,228 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""protocol.py 协议自检(无硬件): DLPilot 实测帧向量回归 + CRC 标准校验值 + 解析同步
|
||||
|
||||
用法: python selftest.py (全部通过退出码 0)
|
||||
"""
|
||||
import random
|
||||
import sys
|
||||
|
||||
import protocol as d
|
||||
|
||||
fails = []
|
||||
|
||||
|
||||
def check(name, got, want):
|
||||
ok = got == want
|
||||
print(('PASS' if ok else 'FAIL'), name, '->', got, '' if ok else f'(期望 {want})')
|
||||
if not ok:
|
||||
fails.append(name)
|
||||
|
||||
|
||||
def check_close(name, got, want):
|
||||
ok = isinstance(got, float) and abs(got - want) < 1e-6
|
||||
print(('PASS' if ok else 'FAIL'), name, '->', got, '' if ok else f'(期望 {want})')
|
||||
if not ok:
|
||||
fails.append(name)
|
||||
|
||||
|
||||
cfg = d.ProtoConfig() # 默认即 DLPilot: D5/B5/01/SUM8/不含包头
|
||||
|
||||
# 1) 帧构造: DLPilot selftest_rs485.py 的实测向量(上游 protocols.md)
|
||||
for cmd, want in (
|
||||
('M730S0T2000', 'D5 01 0B 4D 37 33 30 53 30 54 32 30 30 30 8C'),
|
||||
('M303', 'D5 01 04 4D 33 30 33 E8'),
|
||||
('M304', 'D5 01 04 4D 33 30 34 E9'),
|
||||
('M305', 'D5 01 04 4D 33 30 35 EA'),
|
||||
):
|
||||
check(f'build_frame({cmd})', cfg.build_frame(cmd.encode('ascii')).hex(' ').upper(), want)
|
||||
|
||||
# 2) CRC 算法实现对照公开标准校验值 check("123456789")
|
||||
check('crc8 check', d._crc8(b'123456789'), 0xF4)
|
||||
check('crc16_modbus check', d._crc16_modbus(b'123456789'), 0x4B37)
|
||||
check('crc16_xmodem check', d._crc16_xmodem(b'123456789'), 0x31C3)
|
||||
check('crc16_ccitt check', d._crc16_ccitt(b'123456789'), 0x29B1)
|
||||
|
||||
# 3) CRC 覆盖范围(手算向量: 包头 5A 板ID 11, payload 01 02)
|
||||
for scope, want in (('payload_only', '5A 11 02 01 02 03'),
|
||||
('no_header', '5A 11 02 01 02 16'),
|
||||
('with_header', '5A 11 02 01 02 70')):
|
||||
c = d.ProtoConfig(tx_hdr=0x5A, board=0x11, crc_scope=scope)
|
||||
check(f'coverage {scope}', c.build_frame(b'\x01\x02').hex(' ').upper(), want)
|
||||
|
||||
# 4) CRC16 帧往返 + 坏帧拒绝(组帧/解析同头才能回路, 收发方向包头本就允许不同)
|
||||
c16 = d.ProtoConfig(tx_hdr=0x55, rx_hdr=0x55, crc_algo='CRC16_MODBUS', crc_scope='with_header')
|
||||
f = c16.build_frame(b'ABC')
|
||||
check('crc16 frame len', len(f), 3 + 3 + 2)
|
||||
check('crc16 verify', c16.verify_frame(f), True)
|
||||
bad = bytearray(f)
|
||||
bad[4] ^= 0xFF
|
||||
check('crc16 bad rejected', c16.verify_frame(bytes(bad)), False)
|
||||
frames = d.FrameParser(c16).feed(f)
|
||||
check('crc16 parse payload', (frames[0].payload, frames[0].code) if frames else None, (b'ABC', 0x01))
|
||||
|
||||
# 5) LEN=0 空载荷帧(心跳类): D5 01 00 + sum(01)=01
|
||||
check('empty payload frame', cfg.build_frame(b'').hex(' ').upper(), 'D5 01 00 01')
|
||||
loop = d.ProtoConfig(tx_hdr=0xB5) # tx/rx 同头以便回路
|
||||
check('empty payload parse', d.FrameParser(loop).feed(loop.build_frame(b''))[0].payload, b'')
|
||||
|
||||
# 6) 超长 payload 拒绝
|
||||
try:
|
||||
cfg.build_frame(b'\x00' * 256)
|
||||
fails.append('oversize-not-rejected')
|
||||
except ValueError:
|
||||
print('PASS 超长 payload 拒绝')
|
||||
|
||||
# 7) 增量解析: ok 应答逐字节喂入(拆到最碎) + 前导噪声
|
||||
good = bytes.fromhex('B5 F0 02 6F 6B CC')
|
||||
p = d.FrameParser(cfg)
|
||||
out = []
|
||||
for i in range(len(good)):
|
||||
out += p.feed(good[i:i + 1])
|
||||
check('byte-wise feed frames', len(out), 1)
|
||||
check('code', out[0].code, 0xF0)
|
||||
check('payload', out[0].payload, b'ok')
|
||||
check('raw', d.hexs(out[0].raw), 'B5 F0 02 6F 6B CC')
|
||||
out = d.FrameParser(cfg).feed(b'\x01\x02' + good)
|
||||
check('noise+frame count', len(out), 1)
|
||||
|
||||
# 8) CRC 坏帧被丢弃后继续同步到下一帧
|
||||
bad = bytearray(good)
|
||||
bad[-1] ^= 0xFF
|
||||
frames = d.FrameParser(cfg).feed(bytes(bad) + good)
|
||||
check('bad-crc then good', len(frames), 1)
|
||||
check('recovered code', frames[0].code, 0xF0)
|
||||
|
||||
# 9) 伪帧头与帧头字节入载荷
|
||||
# 9a 小 LEN 伪帧头(B5 AA 01 55,CRC 应为 00 而实为 01)失败后逐字节重同步到真帧
|
||||
frames = d.FrameParser(cfg).feed(b'\xB5\xAA\x01\x55\x01' + good)
|
||||
check('pseudo-header resync', len(frames), 1)
|
||||
check('resync payload', frames[0].payload, b'ok')
|
||||
# 9b payload 内含 0xB5(帧头字节)的合法帧: B5 F0 02 B5 F0 + sum(F0 02 B5 F0)=0x297→97
|
||||
frames = d.FrameParser(cfg).feed(bytes.fromhex('B5 F0 02 B5 F0 97') + good)
|
||||
check('rx-hdr-in-payload', len(frames), 2)
|
||||
check('rx-hdr-in-payload payload', frames[0].payload, b'\xB5\xF0')
|
||||
|
||||
# 10) 自定义包头/板ID 往返(非 DLPilot 值)
|
||||
cx = d.ProtoConfig(tx_hdr=0xC3, rx_hdr=0xC3, board=0x7A)
|
||||
frames = d.FrameParser(cx).feed(cx.build_frame(b'PING'))
|
||||
check('custom hdr frame count', len(frames), 1)
|
||||
check('custom hdr code', frames[0].code, 0x7A)
|
||||
check('custom hdr payload', frames[0].payload, b'PING')
|
||||
|
||||
# 11) 数值提取(曲线通道)
|
||||
check_close('bin u16be', d.extract_value(bytes.fromhex('00 34'), 'bin', fmt='u16be'), 52.0)
|
||||
check_close('bin s16le 负值', d.extract_value(bytes.fromhex('FE FF'), 'bin', fmt='s16le'), -2.0)
|
||||
check_close('bin f32be', d.extract_value(bytes.fromhex('3F 80 00 00'), 'bin', fmt='f32be'), 1.0)
|
||||
check_close('bin offset', d.extract_value(bytes.fromhex('AA 00 34'), 'bin', offset=1, fmt='u16be'), 52.0)
|
||||
check_close('num TL:246', d.extract_value(b'okTL:246', 'num'), 246.0)
|
||||
check_close('num scale 0.1', d.extract_value(b'okTL:246', 'num', scale=0.1), 24.6)
|
||||
check_close('num 第3个', d.extract_value(b'okR12G34B56', 'num', index=3), 56.0)
|
||||
check_close('re 捕获组', d.extract_value(b'okTL:246', 're'), 246.0)
|
||||
check('num 缺数字', d.extract_value(b'ok', 'num'), None)
|
||||
check('bin 越界', d.extract_value(b'\x01', 'bin', offset=2), None)
|
||||
check('re 非法表达式', d.extract_value(b'x', 're', pattern='('), None)
|
||||
check('未知模式', d.extract_value(b'x', 'bogus'), None)
|
||||
|
||||
# 12) 随机往返模糊: 三种算法 × 三种覆盖范围, 整喂/乱拆两种方式
|
||||
rng = random.Random(20260927)
|
||||
for algo in ('SUM8', 'CRC8', 'CRC16_MODBUS'):
|
||||
for scope in d.CRC_SCOPES:
|
||||
cf = d.ProtoConfig(tx_hdr=0xB5, crc_algo=algo, crc_scope=scope) # tx=rx 回路
|
||||
for _ in range(20):
|
||||
pl = bytes(rng.randrange(256) for _ in range(rng.randrange(41)))
|
||||
frame = cf.build_frame(pl)
|
||||
fp = d.FrameParser(cf)
|
||||
if rng.random() < 0.5:
|
||||
got = fp.feed(frame)
|
||||
else:
|
||||
cut = rng.randrange(len(frame) + 1)
|
||||
got = fp.feed(frame[:cut]) + fp.feed(frame[cut:])
|
||||
ok = len(got) == 1 and got[0].payload == pl and got[0].raw == frame
|
||||
if not ok:
|
||||
check(f'fuzz {algo}/{scope}', ok, True)
|
||||
break
|
||||
print(f'模糊往返 3 算法 × 3 覆盖 × 20 帧: {"PASS" if not fails else "见上"}')
|
||||
|
||||
# 13) 日志行安全: clean_text 把控制字符换成 '·'
|
||||
check('clean_text NUL', d.clean_text('a\x00b\x7f'), 'a·b·')
|
||||
|
||||
# 14) 帧构成定制(发送/接收同一布局, 收发方向仅包头值不同; 回路用例 rx_hdr=D5
|
||||
# 以便解析器接受 TX 包头, 字面量同步头布局则与包头值无关)
|
||||
# 14a 双字节字面量同步头 + 无板ID: AA 55 LEN payload CRC
|
||||
cl = d.FrameLayout.parse('AA 55 LEN PAYLOAD CRC')
|
||||
c = d.ProtoConfig(layout=cl)
|
||||
check('layout 字面量组帧', c.build_frame(b'\x01\x02').hex(' ').upper(), 'AA 55 02 01 02 5A')
|
||||
frames = d.FrameParser(c).feed(c.build_frame(b'\x01\x02'))
|
||||
check('layout 字面量解析', (len(frames), frames[0].code, frames[0].payload) if frames else None,
|
||||
(1, None, b'\x01\x02'))
|
||||
|
||||
# 14b 自定义字段顺序: HDR LEN ID PAYLOAD CRC(板ID 在 LEN 后)
|
||||
c = d.ProtoConfig(rx_hdr=0xD5, board=0x2A,
|
||||
layout=d.FrameLayout.parse('HDR LEN ID PAYLOAD CRC'))
|
||||
check('layout 换序组帧', c.build_frame(b'OK').hex(' ').upper(), 'D5 02 2A 4F 4B C6')
|
||||
frames = d.FrameParser(c).feed(c.build_frame(b'OK'))
|
||||
check('layout 换序解析', (frames[0].code, frames[0].payload) if frames else None, (0x2A, b'OK'))
|
||||
|
||||
# 14c 无板ID布局: HDR LEN PAYLOAD CRC, Frame.code 应为 None
|
||||
c = d.ProtoConfig(rx_hdr=0xD5, layout=d.FrameLayout.parse('HDR LEN PAYLOAD CRC'))
|
||||
check('layout 无ID组帧', c.build_frame(b'M1').hex(' ').upper(), 'D5 02 4D 31 80')
|
||||
frames = d.FrameParser(c).feed(c.build_frame(b'M1'))
|
||||
check('layout 无ID解析', (frames[0].code, frames[0].payload) if frames else None,
|
||||
(None, b'M1'))
|
||||
|
||||
# 14d LEN 口径: body(payload+CRC) / total(整帧) —— M303 载荷 4 字节
|
||||
c = d.ProtoConfig(rx_hdr=0xD5, len_sem='body')
|
||||
check('LEN 口径 body', c.build_frame(b'M303').hex(' ').upper(), 'D5 01 05 4D 33 30 33 E9')
|
||||
frames = d.FrameParser(c).feed(c.build_frame(b'M303'))
|
||||
check('LEN 口径 body 解析', frames[0].payload if frames else None, b'M303')
|
||||
c = d.ProtoConfig(rx_hdr=0xD5, len_sem='total')
|
||||
check('LEN 口径 total', c.build_frame(b'M303').hex(' ').upper(), 'D5 01 08 4D 33 30 33 EC')
|
||||
frames = d.FrameParser(c).feed(c.build_frame(b'M303'))
|
||||
check('LEN 口径 total 解析', frames[0].payload if frames else None, b'M303')
|
||||
|
||||
# 14e 仅 payload 覆盖 + 板ID 在 payload 后: 'HDR LEN PAYLOAD ID CRC'
|
||||
c = d.ProtoConfig(rx_hdr=0xD5, board=0x7B, crc_scope='payload_only',
|
||||
layout=d.FrameLayout.parse('HDR LEN PAYLOAD ID CRC'))
|
||||
check('layout ID在payload后', c.build_frame(b'\x01').hex(' ').upper(), 'D5 01 01 7B 01')
|
||||
frames = d.FrameParser(c).feed(c.build_frame(b'\x01'))
|
||||
check('layout ID在payload后解析', (frames[0].code, frames[0].payload) if frames else None,
|
||||
(0x7B, b'\x01'))
|
||||
|
||||
# 14f 非法帧构成一律拒绝
|
||||
for bad in ('HDR PAYLOAD CRC', # 缺 LEN
|
||||
'HDR LEN CRC PAYLOAD', # CRC 不在末尾
|
||||
'LEN PAYLOAD CRC', # 无同步头
|
||||
'HDR ID ID LEN PAYLOAD CRC', # ID 重复
|
||||
'HDR ID LEN FOO CRC', # 未知令牌
|
||||
'', # 空
|
||||
'HDR ID LEN PAYLOAD'): # 缺 CRC
|
||||
try:
|
||||
d.FrameLayout.parse(bad)
|
||||
fails.append(f'layout-not-rejected:{bad}')
|
||||
print('FAIL 帧构成未拒绝 ->', bad)
|
||||
except ValueError:
|
||||
pass
|
||||
print('非法帧构成拒绝: PASS')
|
||||
check('大小写/逗号容错', d.FrameLayout.parse('hdr, id, len, payload, crc').tokens,
|
||||
d.FrameLayout.parse(d.DEFAULT_LAYOUT).tokens)
|
||||
|
||||
# 15) 自定义布局模糊往返(乱拆喂入; 有 HDR 令牌的布局回路需 rx_hdr=tx_hdr)
|
||||
rng2 = random.Random(9527)
|
||||
for spec in ('AA 55 LEN PAYLOAD CRC', 'HDR LEN ID PAYLOAD CRC',
|
||||
'HDR LEN PAYLOAD ID CRC'):
|
||||
cf = d.ProtoConfig(rx_hdr=0xD5, layout=d.FrameLayout.parse(spec))
|
||||
for _ in range(10):
|
||||
pl = bytes(rng2.randrange(256) for _ in range(rng2.randrange(41)))
|
||||
frame = cf.build_frame(pl)
|
||||
fp = d.FrameParser(cf)
|
||||
cut = rng2.randrange(len(frame) + 1)
|
||||
got = fp.feed(frame[:cut]) + fp.feed(frame[cut:])
|
||||
ok = len(got) == 1 and got[0].payload == pl and got[0].raw == frame
|
||||
if not ok:
|
||||
check(f'layout fuzz {spec}', ok, True)
|
||||
break
|
||||
print(f'布局模糊往返 3 布局 × 10 帧: {"PASS" if not fails else "见上"}')
|
||||
|
||||
print('RESULT:', 'FAIL:' + ','.join(fails) if fails else 'ALL PASS')
|
||||
sys.exit(1 if fails else 0)
|
||||
@@ -0,0 +1,223 @@
|
||||
"""GUI 功能自检(offscreen 无窗口): 组帧预览/TX回显/帧日志/通道提取/显示切换/
|
||||
未连接丢弃/配置往返
|
||||
====================================================================
|
||||
不依赖硬件与串口, 直接驱动 Window 的事件处理函数; 改 main.py 后建议重跑。
|
||||
|
||||
用法: python selftest_gui.py (全部通过退出码 0)
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
os.environ['QT_QPA_PLATFORM'] = 'offscreen'
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from PySide6.QtWidgets import QApplication
|
||||
|
||||
import config
|
||||
import main as m
|
||||
import protocol as d
|
||||
from plot import fmt_val
|
||||
|
||||
fails = []
|
||||
|
||||
|
||||
def check(name, got, want):
|
||||
ok = got == want
|
||||
print(('PASS' if ok else 'FAIL'), name, '->', repr(got)[:120],
|
||||
'' if ok else f'(期望 {want!r})')
|
||||
if not ok:
|
||||
fails.append(name)
|
||||
|
||||
|
||||
# 0) 数值显示格式: 普通小数千分位(科学计数只留给 10 亿以上)
|
||||
check('fmt_val 千分位', fmt_val(28210029), '28,210,029')
|
||||
check('fmt_val 小数', fmt_val(24.6), '24.6')
|
||||
check('fmt_val 负小数', fmt_val(-0.25), '-0.25')
|
||||
check('fmt_val 零', fmt_val(0), '0')
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
win = m.Window()
|
||||
win.show() # offscreen 显示, isVisible 类断言前提
|
||||
check('倍率/偏移/序号框宽度(防数值裁切回归)',
|
||||
win._ch_ui[0]['scale'].width() >= 130
|
||||
and win._ch_ui[0]['off'].width() >= 110
|
||||
and win._ch_ui[0]['idx'].width() >= 110, True)
|
||||
win._apply_minimum_size()
|
||||
check('窗口最小高度(防曲线卡叠压)', win.minimumHeight() >= 800, True)
|
||||
|
||||
# 0b) 曲线时间窗换算(纯计算; 窗口不向数据收敛——选 5min 就显示 5min 宽的轴)
|
||||
p = win.plot
|
||||
p.set_view(10)
|
||||
p.set_offset(0)
|
||||
check('时间窗跟随右缘', p.visible_domain(0, 100) == (90.0, 100.0), True)
|
||||
p.set_offset(30)
|
||||
check('时间窗滞后回看', p.visible_domain(0, 100) == (60.0, 70.0), True)
|
||||
p.set_offset(90)
|
||||
check('时间窗滑到最旧', p.visible_domain(0, 100) == (0.0, 10.0), True)
|
||||
p.set_offset(500)
|
||||
check('滞后超史为空窗(保宽)', p.visible_domain(0, 100) == (-410.0, -400.0), True)
|
||||
p.set_view(300)
|
||||
p.set_offset(0)
|
||||
check('数据不足窗宽不收敛', p.visible_domain(0, 23) == (-277.0, 23.0), True)
|
||||
p.set_view(None)
|
||||
p.set_offset(0)
|
||||
check('跟随全部', p.visible_domain(0, 100) == (0.0, 100.0), True)
|
||||
|
||||
# 0c) 悬停读值: 最近采样命中/容差外未命中
|
||||
import time as _time
|
||||
base = _time.perf_counter()
|
||||
p._times[0].clear()
|
||||
p._vals[0].clear()
|
||||
p._names[0] = '甲'
|
||||
for tt, vv in ((0.0, 10.0), (1.0, 20.0), (2.0, 30.0)):
|
||||
p._times[0].append(base + tt)
|
||||
p._vals[0].append(vv)
|
||||
check('悬停命中最近点', p._nearest(0, base + 1.9, 0.2) == ('甲', 30.0), True)
|
||||
check('悬停容差外未命中', p._nearest(0, base + 1.4, 0.2) is None, True)
|
||||
lo, hi = p.auto_yrange()
|
||||
check('自动Y量程', abs(lo - 8.4) < 1e-9 and abs(hi - 31.6) < 1e-9, True)
|
||||
from PySide6.QtCore import QPoint
|
||||
p.resize(600, 300) # 未走布局, 手动给几何
|
||||
lines = p._hover_lines_at(QPoint(320, 150)) # x=绘图区正中 → t=base+1.0
|
||||
check('悬停框内容(位置固定也随数据刷新)',
|
||||
lines[:2] == [f'T = {base + 1.0:.2f}s', '甲 = 20'], True)
|
||||
|
||||
# 0d) 暂停渲染 vs 暂停采集: 冻结期间继续收数, 解冻后数据都在
|
||||
p.set_frozen(True)
|
||||
n0 = len(p._times[0])
|
||||
p.add(0, 42.0)
|
||||
check('冻结期间继续收数', p.is_frozen() and len(p._times[0]) == n0 + 1, True)
|
||||
p.set_frozen(False)
|
||||
p.set_paused(True)
|
||||
n1 = len(p._times[0])
|
||||
p.add(0, 43.0)
|
||||
check('暂停期间丢弃', p.is_paused() and len(p._times[0]) == n1, True)
|
||||
p.set_paused(False)
|
||||
|
||||
# 1) 组帧预览: 多条 '/' 分隔
|
||||
win.ed_tx.setText('M1/M731S0T1000')
|
||||
check('多条预览', win.lb_preview.text(),
|
||||
'D5 01 02 4D 31 81 | D5 01 0B 4D 37 33 31 53 30 54 31 30 30 30 8C')
|
||||
win.combo_tx_mode.setCurrentIndex(1)
|
||||
win.ed_tx.setText('D5 01 02 4D 31 81')
|
||||
check('HEX预览', win.lb_preview.text(), '→ D5 01 02 4D 31 81')
|
||||
win.combo_tx_mode.setCurrentIndex(0)
|
||||
win.ed_tx.setText('M1')
|
||||
|
||||
# 1b) 帧构成定制: 无板ID布局 + LEN 口径切换 + 非法输入提示
|
||||
win.ed_layout.setText('HDR LEN PAYLOAD CRC')
|
||||
check('帧构成无ID预览', win.lb_preview.text(), 'D5 02 4D 31 80')
|
||||
win.combo_len_sem.setCurrentIndex(1) # 到帧尾(payload+CRC)
|
||||
check('帧构成LEN口径预览', win.lb_preview.text(), 'D5 03 4D 31 81')
|
||||
win.combo_len_sem.setCurrentIndex(0)
|
||||
win.ed_layout.setText('HDR LEN') # 缺 PAYLOAD/CRC -> 拒绝
|
||||
check('帧构成非法提示', win.lb_preview.text().startswith('⚠'), True)
|
||||
check('帧构成非法详情', 'PAYLOAD 必须恰好出现一次' in win.lb_proto.text(), True)
|
||||
win.ed_layout.setText(d.DEFAULT_LAYOUT)
|
||||
check('帧构成还原预览', win.lb_preview.text(), 'D5 01 02 4D 31 81')
|
||||
|
||||
# 2) TX 回显行
|
||||
cfg = win._proto_from_ui()
|
||||
win._on_tx('M1', cfg.build_frame(b'M1'))
|
||||
check('TX日志行', 'M1 [D5 01 02 4D 31 81]' in win.text_rx.toPlainText(), True)
|
||||
|
||||
# 3) 帧事件: 日志 + 通道提取(文本数字×0.1)
|
||||
win.combo_display.setCurrentIndex(0)
|
||||
fr = d.FrameParser(d.ProtoConfig()).feed(
|
||||
bytes.fromhex('B5 F0 08 6F 6B 54 4C 3A 32 34 36 48'))[0]
|
||||
win._ch_ui[0]['chk'].setChecked(True)
|
||||
win._ch_ui[0]['mode'].setCurrentIndex(1) # 文本数字
|
||||
win._ch_ui[0]['idx'].setValue(1)
|
||||
win._ch_ui[0]['scale'].setValue(0.1)
|
||||
win._on_frame(fr)
|
||||
last = win.text_rx.toPlainText().splitlines()[-1]
|
||||
check('帧日志含 TYPE 名', 'B=0xF0 成功' in last and 'LEN=8' in last
|
||||
and 'okTL:246' in last, True)
|
||||
check('通道最新值', win._ch_ui[0]['val'].text(), '24.6')
|
||||
|
||||
# 4) HEX 显示切换 + 原始帧
|
||||
win.combo_display.setCurrentIndex(1)
|
||||
win._on_frame(fr)
|
||||
last = win.text_rx.toPlainText().splitlines()[-1]
|
||||
check('HEX显示+原始帧', '6F 6B 54 4C 3A 32 34 36' in last
|
||||
and '[B5 F0 08 6F 6B 54 4C 3A 32 34 36 48]' in last, True)
|
||||
|
||||
# 5) 透传行
|
||||
win.combo_display.setCurrentIndex(1)
|
||||
win._on_raw(b'\x11\x22')
|
||||
check('透传HEX行', win.text_rx.toPlainText().splitlines()[-1].endswith('11 22'), True)
|
||||
|
||||
# 6) 未连接发送丢弃
|
||||
win._send()
|
||||
win._drain_events()
|
||||
check('未连接发送提示', '未连接串口, 命令被丢弃' in win.text_rx.toPlainText(), True)
|
||||
|
||||
# 7) 配置往返
|
||||
store = win._store_from_ui()
|
||||
check('配置 proto', (store['proto']['tx_hdr'], store['proto']['crc_algo']),
|
||||
('D5', 'SUM8'))
|
||||
check('配置帧构成', (store['proto']['layout'], store['proto']['len_sem']),
|
||||
(d.DEFAULT_LAYOUT, 'payload'))
|
||||
check('配置通道1', (store['channels'][0]['enabled'], store['channels'][0]['scale']),
|
||||
(True, 0.1))
|
||||
|
||||
# 8) 板型档案: 保存/切换应用协议参数/存入命令/清理
|
||||
check('板型档案加载', win.combo_profile.count() >= 1, True)
|
||||
win.ed_board.setText('03')
|
||||
check('存为板型', win._save_profile_as('测试板'), True)
|
||||
names = [win.combo_profile.itemText(i) for i in range(win.combo_profile.count())]
|
||||
check('板型入列', '测试板' in names, True)
|
||||
pidx = names.index('测试板')
|
||||
win.combo_profile.setCurrentIndex(pidx)
|
||||
check('切换板型应用板ID', win.ed_board.text(), '03')
|
||||
win.combo_profile.setCurrentIndex(0)
|
||||
check('切回默认板ID', win.ed_board.text(), '01')
|
||||
win.ed_tx.setText('M1')
|
||||
win._add_command_named('M1', 'M1')
|
||||
check('存入命令去重', any(c[1] == 'M1' for c in win._commands)
|
||||
and len([c for c in win._commands if c[1] == 'M1']) == 1, True)
|
||||
check('预设显示名命令', win.combo_preset.itemText(0) == '握手 · M1', True)
|
||||
# 10) 新建板型(命令预设从空白开始) + 命令预设表编辑器
|
||||
check('新建板型', win._new_profile_named('新建测试板'), True)
|
||||
check('重名新建拒绝', win._new_profile_named('新建测试板'), False)
|
||||
check('新建板型命令为空', win._commands == [], True)
|
||||
dlg = m.ProfileEditDialog(win.profiles[win.combo_profile.currentIndex()])
|
||||
from PySide6.QtWidgets import QTableWidgetItem
|
||||
for r, (n, c) in enumerate((('甲', 'A1'), ('乙', 'B2'), ('丙', 'C3'))):
|
||||
dlg.table._add_row()
|
||||
dlg.table.setItem(r, 1, QTableWidgetItem(n))
|
||||
dlg.table.setItem(r, 2, QTableWidgetItem(c))
|
||||
check('编辑器收集', dlg.commands() == [['甲', 'A1'], ['乙', 'B2'], ['丙', 'C3']],
|
||||
True)
|
||||
dlg.table.move_row(0, 2) # 拖动排序: 首行挪到末尾
|
||||
check('拖动换序', dlg.commands() == [['乙', 'B2'], ['丙', 'C3'], ['甲', 'A1']],
|
||||
True)
|
||||
check('手柄列存在', dlg.table.item(0, 0).text() == '≡', True)
|
||||
dlg.table._add_row() # 重复命令与空命令行均被忽略
|
||||
dlg.table.setItem(3, 1, QTableWidgetItem('重复'))
|
||||
dlg.table.setItem(3, 2, QTableWidgetItem('B2'))
|
||||
check('编辑器去重空行', dlg.commands() == [['乙', 'B2'], ['丙', 'C3'], ['甲', 'A1']],
|
||||
True)
|
||||
# 11) 删除板型: 核函数删除/不存在拒绝/至少保留一个
|
||||
check('删除板型', win._delete_profile('新建测试板'), True)
|
||||
check('删除后不在列', all(p['name'] != '新建测试板' for p in win.profiles), True)
|
||||
check('删除不存在拒绝', win._delete_profile('不存在的板型'), False)
|
||||
# 清理运行态测试板型
|
||||
win.profiles[:] = [p for p in win.profiles if p['name'] != '测试板']
|
||||
config.save_profiles(win.profiles)
|
||||
win._reload_profiles()
|
||||
check('清理测试板型', all(p['name'] != '测试板' for p in win.profiles), True)
|
||||
|
||||
# 9) 曲线弹出/嵌回
|
||||
win._toggle_plot_window()
|
||||
check('曲线弹出', win._plot_win is not None
|
||||
and win.plot.window() is win._plot_win, True)
|
||||
win._toggle_plot_window()
|
||||
check('曲线嵌回', win._plot_win is None and win.plot.parent() is not None, True)
|
||||
win._pop_plot()
|
||||
win._dock_plot()
|
||||
check('弹出后关窗嵌回', win._plot_win is None and win.plot.isVisible(), True)
|
||||
|
||||
win.link.close()
|
||||
print('RESULT:', 'FAIL:' + ','.join(fails) if fails else 'ALL PASS')
|
||||
sys.exit(1 if fails else 0)
|
||||
@@ -0,0 +1,91 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""链路自检(无硬件): pyserial loop:// 虚拟回环验证 组帧→发送→接收→解析→曲线提取
|
||||
====================================================================
|
||||
缺 pyserial 时跳过(退出码 0)。loop:// 回环口写即读, 发出的帧同样回到
|
||||
接收侧, 但 TX 帧头(D5)不是 RX 帧头(B5), 会被解析器当噪声丢弃, 不影响判定。
|
||||
|
||||
用法: python selftest_loop.py (全部通过退出码 0)
|
||||
"""
|
||||
import sys
|
||||
import time
|
||||
|
||||
try:
|
||||
import serial # noqa: F401
|
||||
except ImportError:
|
||||
print('未安装 pyserial, 跳过回环自检')
|
||||
sys.exit(0)
|
||||
|
||||
import protocol as d
|
||||
from link import SerialLink
|
||||
|
||||
fails = []
|
||||
|
||||
|
||||
def check(name, got, want):
|
||||
ok = got == want
|
||||
print(('PASS' if ok else 'FAIL'), name, '->', got, '' if ok else f'(期望 {want})')
|
||||
if not ok:
|
||||
fails.append(name)
|
||||
|
||||
|
||||
def wait_for(pred, timeout=5.0):
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
if pred():
|
||||
return True
|
||||
time.sleep(0.05)
|
||||
return False
|
||||
|
||||
|
||||
got = {'status': [], 'frames': [], 'tx': [], 'raw': []}
|
||||
cfg = d.ProtoConfig()
|
||||
link = SerialLink(cfg, on_status=lambda s: got['status'].append(s))
|
||||
link.on_frame = lambda fr: got['frames'].append(fr)
|
||||
link.on_raw = lambda b: got['raw'].append(b)
|
||||
link.on_tx = lambda lb, data: got['tx'].append((lb, data))
|
||||
|
||||
# 1) 打开虚拟回环口
|
||||
link.open('loop://', 115200)
|
||||
check('连接 loop://', wait_for(lambda: got['status'] and
|
||||
got['status'][-1].startswith('已连接')), True)
|
||||
|
||||
# 2) 自动包装发送: TX 回显 + 回环字节
|
||||
payload = 'M731S0T1000'
|
||||
frame = cfg.build_frame(payload.encode('ascii'))
|
||||
link.send_frame(frame, label=payload)
|
||||
check('发送回显', wait_for(lambda: len(got['tx']) >= 1), True)
|
||||
if got['tx']:
|
||||
check('TX label', got['tx'][0][0], payload)
|
||||
check('TX frame', got['tx'][0][1], frame)
|
||||
|
||||
# 3) 模拟设备应答: okTL:246(温度 24.6°C) → 解析 + 曲线提取
|
||||
resp = bytes.fromhex('B5 F0 08 6F 6B 54 4C 3A 32 34 36 48') # sum(F0..36)=0x48
|
||||
check('响应帧向量自校验', cfg.verify_frame(resp), True)
|
||||
link.ser.write(resp)
|
||||
check('应答解析', wait_for(lambda: len(got['frames']) >= 1), True)
|
||||
if got['frames']:
|
||||
fr = got['frames'][0]
|
||||
check('TYPE 码', fr.code, 0xF0)
|
||||
check('payload', fr.payload, b'okTL:246')
|
||||
v = d.extract_value(fr.payload, 'num', index=1, scale=0.1)
|
||||
check('曲线提取(×0.1)', v is not None and abs(v - 24.6) < 1e-9, True)
|
||||
|
||||
# 4) 透传模式切换: 关解析后收到原始字节
|
||||
link.set_parse(False)
|
||||
raw = bytes([0x11, 0x22, 0xB5, 0x00])
|
||||
link.ser.write(raw)
|
||||
check('透传原始字节', wait_for(lambda: len(got['raw']) >= 1), True)
|
||||
if got['raw']:
|
||||
check('透传内容', got['raw'][0], raw)
|
||||
|
||||
# 5) 切回解析模式, 缓冲已清空, 正常再收一帧
|
||||
link.set_parse(True)
|
||||
link.ser.write(bytes.fromhex('B5 F0 02 6F 6B CC'))
|
||||
check('切回解析再收帧', wait_for(lambda: len(got['frames']) >= 2), True)
|
||||
|
||||
link.close()
|
||||
check('断开', wait_for(lambda: got['status'] and got['status'][-1] == '已断开'), True)
|
||||
|
||||
print('RESULT:', 'FAIL:' + ','.join(fails) if fails else 'ALL PASS')
|
||||
sys.exit(1 if fails else 0)
|
||||
@@ -0,0 +1,129 @@
|
||||
"""Theme persistence and Fluent Widgets integration."""
|
||||
|
||||
import ctypes
|
||||
import os
|
||||
|
||||
from PySide6.QtCore import QObject, QSettings, Qt
|
||||
from PySide6.QtGui import QColor, QPalette
|
||||
from qfluentwidgets import Theme, setTheme, setThemeColor
|
||||
|
||||
# Fluent widgets restyle themselves on setTheme(); QMainWindow, the native
|
||||
# QTabWidget/QStatusBar and the Windows title bar keep the OS palette unless
|
||||
# the palette, an app stylesheet and the DWM title bar attributes follow suit
|
||||
# (same approach as DLPilot app_fluent.py).
|
||||
_SURFACES = {
|
||||
False: {"window": "#f4f6f8", "base": "#ffffff", "text": "#1f2328",
|
||||
"dim": "#57606a", "border": "#d8dee4", "selected_text": "#ffffff"},
|
||||
True: {"window": "#202020", "base": "#2b2b2b", "text": "#f5f5f5",
|
||||
"dim": "#9a9a9a", "border": "#3a3a3a", "selected_text": "#1f2328"},
|
||||
}
|
||||
|
||||
|
||||
class ThemeManager(QObject):
|
||||
"""Keep the appearance choice persistent and synchronized with Qt."""
|
||||
|
||||
MODES = {"system", "light", "dark", "dark1", "dark2"}
|
||||
_THEMES = {
|
||||
"system": Theme.AUTO,
|
||||
"light": Theme.LIGHT,
|
||||
"dark": Theme.DARK,
|
||||
"dark1": Theme.DARK,
|
||||
"dark2": Theme.DARK,
|
||||
}
|
||||
_ACCENTS = {
|
||||
"light": "#0f6cbd",
|
||||
"dark": "#f2c94c",
|
||||
"dark1": "#5aa9e6",
|
||||
"dark2": "#35c6c8",
|
||||
}
|
||||
|
||||
def __init__(self, app):
|
||||
super().__init__(app)
|
||||
self.app = app
|
||||
self.window = None
|
||||
self.settings = QSettings("LT8619C-Tools", "SerialMate")
|
||||
mode = self.settings.value("appearance/theme", "system")
|
||||
self.mode = mode if mode in self.MODES else "system"
|
||||
self.apply()
|
||||
app.styleHints().colorSchemeChanged.connect(self.apply)
|
||||
|
||||
def set_window(self, window):
|
||||
"""Register the main window so the native title bar follows the theme."""
|
||||
self.window = window
|
||||
self._apply_titlebar()
|
||||
|
||||
def set_mode(self, mode):
|
||||
if mode not in self.MODES:
|
||||
raise ValueError(f"Unknown theme: {mode}")
|
||||
self.mode = mode
|
||||
self.settings.setValue("appearance/theme", mode)
|
||||
self.apply()
|
||||
|
||||
def is_dark(self):
|
||||
if self.mode in {"dark", "dark1", "dark2"}:
|
||||
return True
|
||||
return self.mode == "system" and self.app.styleHints().colorScheme() == Qt.ColorScheme.Dark
|
||||
|
||||
def apply(self, *_):
|
||||
is_dark = self.is_dark()
|
||||
setTheme(self._THEMES[self.mode])
|
||||
setThemeColor(self._ACCENTS.get(self.mode, self._ACCENTS["light"]))
|
||||
self._apply_palette(is_dark)
|
||||
self._apply_titlebar()
|
||||
|
||||
def _apply_palette(self, is_dark):
|
||||
"""Give native Qt surfaces the same colors as the Fluent widgets."""
|
||||
surface = _SURFACES[is_dark]
|
||||
accent = QColor(self._ACCENTS.get(self.mode, self._ACCENTS["light"]))
|
||||
palette = QPalette()
|
||||
for role, color in (
|
||||
(QPalette.Window, surface["window"]),
|
||||
(QPalette.Base, surface["base"]),
|
||||
(QPalette.AlternateBase, surface["base"]),
|
||||
(QPalette.Text, surface["text"]),
|
||||
(QPalette.WindowText, surface["text"]),
|
||||
(QPalette.Button, surface["base"]),
|
||||
(QPalette.ButtonText, surface["text"]),
|
||||
(QPalette.PlaceholderText, surface["dim"]),
|
||||
(QPalette.Highlight, accent),
|
||||
(QPalette.HighlightedText, surface["selected_text"]),
|
||||
(QPalette.ToolTipBase, surface["base"]),
|
||||
(QPalette.ToolTipText, surface["text"]),
|
||||
):
|
||||
palette.setColor(role, QColor(color))
|
||||
self.app.setPalette(palette)
|
||||
self.app.setStyleSheet(f"""
|
||||
QMainWindow, QWidget#root {{ background-color: {surface['window']}; color: {surface['text']}; }}
|
||||
QTabWidget::pane {{ background: {surface['base']}; border: 1px solid {surface['border']}; border-radius: 6px; }}
|
||||
QTabBar {{ qproperty-drawBase: 0; }}
|
||||
QTabBar::tab {{ background: transparent; color: {surface['dim']}; padding: 6px 16px;
|
||||
border: 1px solid transparent; border-top-left-radius: 6px;
|
||||
border-top-right-radius: 6px; }}
|
||||
QTabBar::tab:selected {{ background: {surface['base']}; color: {surface['text']};
|
||||
border: 1px solid {surface['border']}; }}
|
||||
QTabBar::tab:hover:!selected {{ background: {surface['base']}; }}
|
||||
QSplitter::handle {{ background: {surface['window']}; }}
|
||||
QStatusBar {{ background: {surface['window']}; color: {surface['dim']};
|
||||
border-top: 1px solid {surface['border']}; }}
|
||||
QToolTip {{ background-color: {surface['base']}; color: {surface['text']};
|
||||
border: 1px solid {surface['border']}; }}
|
||||
""")
|
||||
|
||||
def _apply_titlebar(self):
|
||||
"""Ask DWM for a title bar in the current surface color (Win10/11)."""
|
||||
if self.window is None or os.name != "nt" or not hasattr(ctypes, "windll"):
|
||||
return
|
||||
try:
|
||||
hwnd = int(self.window.winId())
|
||||
dwm = ctypes.windll.dwmapi
|
||||
dark = ctypes.c_int(1 if self.is_dark() else 0)
|
||||
dwm.DwmSetWindowAttribute(hwnd, 20, ctypes.byref(dark), ctypes.sizeof(dark))
|
||||
surface = _SURFACES[self.is_dark()]
|
||||
dwm.DwmSetWindowAttribute(hwnd, 35, ctypes.byref(self._colorref(QColor(surface["window"]))), 4)
|
||||
dwm.DwmSetWindowAttribute(hwnd, 36, ctypes.byref(self._colorref(QColor(surface["text"]))), 4)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def _colorref(color):
|
||||
return ctypes.c_int(color.red() | (color.green() << 8) | (color.blue() << 16))
|
||||
Reference in New Issue
Block a user