refactor(flash-gui): 对齐 DLPilot 重构 GUI 并内置 OpenOCD 发布链
This commit is contained in:
@@ -0,0 +1,249 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
OpenOCD 子进程封装: 工具链定位 / 烧录脚本构建 / 后台运行
|
||||
================================================================
|
||||
工具链定位顺序(locate_toolchain):
|
||||
1. 内置 <程序目录>/toolchain/openocd/bin/openocd.exe
|
||||
(发布包自带; PyInstaller frozen 取 exe 旁, 源码模式取 FlashTool/ 下)
|
||||
2. 默认 D:/toolchain/openocd(开发机本机安装)
|
||||
3. PATH 里的 openocd
|
||||
返回值带来源标记, GUI 在界面上显示"工具链: 内置/本机/PATH"。
|
||||
|
||||
GUI 与命令行共用 build_cmd/script_* 这组函数, 保证两边生成的命令一致。
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
|
||||
from fw_image import CHIP_PROFILES, FLASH_BASE_DEFAULT, parse_int
|
||||
|
||||
DEFAULT_OPENOCD = r"D:/toolchain/openocd/bin/openocd.exe"
|
||||
DEFAULT_SCRIPTS = r"D:/toolchain/openocd/openocd/scripts"
|
||||
|
||||
# 探针型号 → OpenOCD interface cfg(空白 = 自定义, 不改手填值)
|
||||
PROBE_PROFILES = {
|
||||
"DAPLink / CMSIS-DAP": "interface/cmsis-dap.cfg",
|
||||
"GD-Link (CMSIS-DAP)": "interface/cmsis-dap.cfg",
|
||||
"ST-Link": "interface/stlink.cfg",
|
||||
"J-Link": "interface/jlink.cfg",
|
||||
"CMSIS-DAP (TCP)": "interface/cmsis-dap-tcp.cfg",
|
||||
"Custom": "",
|
||||
}
|
||||
|
||||
|
||||
def _app_base():
|
||||
"""程序所在目录: frozen 取 exe 旁, 源码模式取 FlashTool/"""
|
||||
if getattr(sys, "frozen", False):
|
||||
return os.path.dirname(os.path.abspath(sys.executable))
|
||||
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def bundled_toolchain():
|
||||
"""内置工具链路径, 不存在返回 None"""
|
||||
exe = os.path.join(_app_base(), "toolchain", "openocd", "bin", "openocd.exe")
|
||||
if os.path.isfile(exe):
|
||||
scripts = os.path.join(_app_base(), "toolchain", "openocd", "openocd", "scripts")
|
||||
return exe, scripts if os.path.isdir(scripts) else DEFAULT_SCRIPTS
|
||||
return None
|
||||
|
||||
|
||||
def locate_toolchain():
|
||||
"""返回 (openocd_exe, scripts_dir, source)，source ∈ 内置/本机/PATH"""
|
||||
bundled = bundled_toolchain()
|
||||
if bundled:
|
||||
return bundled[0], bundled[1], "内置"
|
||||
if os.path.isfile(DEFAULT_OPENOCD):
|
||||
return DEFAULT_OPENOCD, DEFAULT_SCRIPTS, "本机"
|
||||
which = shutil.which("openocd")
|
||||
if which:
|
||||
return which, "", "PATH"
|
||||
return DEFAULT_OPENOCD, DEFAULT_SCRIPTS, "本机"
|
||||
|
||||
|
||||
def strip_ansi(text):
|
||||
return re.sub(r"\x1b\[[0-9;]*[A-Za-z]", "", text)
|
||||
|
||||
|
||||
def tcl_quote(path):
|
||||
return '"' + path.replace("\\", "/").replace('"', '\\"') + '"'
|
||||
|
||||
|
||||
def ensure_cfg_path(value, folder):
|
||||
"""裸 cfg 文件名(如 gd32e23x.cfg)补上所属目录前缀"""
|
||||
value = value.strip()
|
||||
if not value:
|
||||
return value
|
||||
if "/" in value or "\\" in value or not value.lower().endswith(".cfg"):
|
||||
return value
|
||||
return folder + "/" + value
|
||||
|
||||
|
||||
def version(exe):
|
||||
"""openocd --version 首行(如 Open On-Chip Debugger 0.12.0), 失败返回 None"""
|
||||
try:
|
||||
out = subprocess.run([exe, "--version"], capture_output=True, text=True,
|
||||
timeout=10, creationflags=subprocess.CREATE_NO_WINDOW
|
||||
if os.name == "nt" else 0)
|
||||
line = (out.stdout or out.stderr or "").strip().splitlines()
|
||||
return line[0].strip() if line else None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 烧录脚本构建(GUI 与 CLI 共用)
|
||||
# ============================================================
|
||||
def build_cmd(openocd, scripts, interface, target, speed="", serial="", extra="", script=""):
|
||||
"""拼 OpenOCD 命令行: -s 搜索目录 + interface/target cfg + -c 选项"""
|
||||
cmd = [openocd.strip()]
|
||||
if scripts and scripts.strip():
|
||||
cmd += ["-s", scripts.strip()]
|
||||
cmd += ["-f", ensure_cfg_path(interface, "interface"),
|
||||
"-f", ensure_cfg_path(target, "target")]
|
||||
if speed and speed.strip():
|
||||
cmd += ["-c", "adapter speed " + speed.strip()]
|
||||
if serial and serial.strip() and "cmsis-dap" in interface.lower():
|
||||
cmd += ["-c", "cmsis_dap_serial " + serial.strip()]
|
||||
if extra and extra.strip():
|
||||
cmd += ["-c", extra.strip()]
|
||||
if script:
|
||||
cmd += ["-c", script]
|
||||
return cmd
|
||||
|
||||
|
||||
def script_detect():
|
||||
return "init; reset halt; flash list; exit"
|
||||
|
||||
|
||||
def script_erase(flash_size_text):
|
||||
size = parse_int(flash_size_text)
|
||||
if size is None or size <= 0:
|
||||
raise ValueError("Flash 大小无效: %s" % flash_size_text)
|
||||
return ("init; reset halt; flash erase_address 0x%08X 0x%X; reset halt; exit"
|
||||
% (FLASH_BASE_DEFAULT, size))
|
||||
|
||||
|
||||
def script_flash(firmware, base_text="", erase=True, verify=True, reset=True):
|
||||
"""HEX/ELF 用文件内地址(base_text 留空), BIN 必须给基地址"""
|
||||
ext = os.path.splitext(firmware)[1].lower()
|
||||
script = "init; reset halt; "
|
||||
addr = ""
|
||||
if ext == ".bin":
|
||||
base = parse_int(base_text)
|
||||
if base is None:
|
||||
raise ValueError("BIN 基地址无效: %s" % base_text)
|
||||
addr = " 0x%08X" % base
|
||||
quoted = tcl_quote(firmware)
|
||||
script += ("flash write_image erase %s%s; " if erase
|
||||
else "flash write_image %s%s; ") % (quoted, addr)
|
||||
if verify:
|
||||
script += "verify_image %s%s; " % (quoted, addr)
|
||||
script += ("reset run" if reset else "reset halt") + "; exit"
|
||||
return script
|
||||
|
||||
|
||||
def script_read(out_path, flash_size_text):
|
||||
size = parse_int(flash_size_text)
|
||||
if size is None or size <= 0:
|
||||
raise ValueError("Flash 大小无效: %s" % flash_size_text)
|
||||
return "init; reset halt; dump_image %s 0x%08X 0x%X; exit" % (
|
||||
tcl_quote(out_path), FLASH_BASE_DEFAULT, size)
|
||||
|
||||
|
||||
def script_reset():
|
||||
return "init; reset run; exit"
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 进程运行
|
||||
# ============================================================
|
||||
class OcdProcess:
|
||||
"""后台跑一条 openocd 命令。
|
||||
|
||||
on_line(line)/on_done(code) 在工作线程里回调, GUI 侧自行经队列
|
||||
投递回主线程; CLI 可直接在回调里 print。
|
||||
"""
|
||||
|
||||
def __init__(self, cmd, on_line, on_done):
|
||||
self.cmd = cmd
|
||||
self._on_line = on_line
|
||||
self._on_done = on_done
|
||||
self.proc = None
|
||||
self._thread = None
|
||||
|
||||
@property
|
||||
def running(self):
|
||||
return self.proc is not None
|
||||
|
||||
def start(self):
|
||||
self._thread = threading.Thread(target=self._worker, daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def terminate(self):
|
||||
if self.proc is not None:
|
||||
try:
|
||||
self.proc.terminate()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _worker(self):
|
||||
creation = subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0
|
||||
try:
|
||||
proc = subprocess.Popen(
|
||||
self.cmd,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
bufsize=1,
|
||||
creationflags=creation,
|
||||
)
|
||||
except Exception as exc:
|
||||
self._on_line("启动 OpenOCD 失败: %s" % exc)
|
||||
self._on_done(-1)
|
||||
return
|
||||
|
||||
self.proc = proc
|
||||
try:
|
||||
for line in proc.stdout:
|
||||
line = strip_ansi(line.rstrip("\r\n"))
|
||||
if line.strip():
|
||||
self._on_line(line)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
code = proc.wait()
|
||||
self.proc = None
|
||||
self._on_done(code)
|
||||
|
||||
|
||||
def run_sync(cmd):
|
||||
"""阻塞运行一条 openocd 命令, 返回 (exit_code, 输出行)"""
|
||||
lines = []
|
||||
done = []
|
||||
|
||||
def on_line(line):
|
||||
lines.append(line)
|
||||
|
||||
def on_done(code):
|
||||
done.append(code)
|
||||
|
||||
job = OcdProcess(cmd, on_line, on_done)
|
||||
job.start()
|
||||
job._thread.join()
|
||||
return (done[0] if done else -1), lines
|
||||
|
||||
|
||||
def chip_target(chip_name):
|
||||
return CHIP_PROFILES.get(chip_name, {}).get("target", "target/gd32e23x.cfg")
|
||||
|
||||
|
||||
def chip_flash_size(chip_name):
|
||||
profile = CHIP_PROFILES.get(chip_name)
|
||||
return "0x%X" % profile["flash"] if profile else "0x10000"
|
||||
Reference in New Issue
Block a user