324 lines
11 KiB
Python
324 lines
11 KiB
Python
#!/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 _find_scripts(root):
|
||
"""在工具链根目录下找 scripts 目录, 兼容两种布局:
|
||
xpack 0.11(脚本在根: scripts/) 与 0.12(脚本在 openocd/scripts/)"""
|
||
for sub in ("openocd/scripts", "scripts"):
|
||
p = os.path.join(root, sub)
|
||
if os.path.isdir(p):
|
||
return p
|
||
return None
|
||
|
||
|
||
def bundled_toolchain():
|
||
"""内置工具链路径, 不存在返回 None"""
|
||
root = os.path.join(_app_base(), "toolchain", "openocd")
|
||
exe = os.path.join(root, "bin", "openocd.exe")
|
||
if os.path.isfile(exe):
|
||
scripts = _find_scripts(root)
|
||
return exe, (scripts or DEFAULT_SCRIPTS)
|
||
return None
|
||
|
||
|
||
def custom_cfg_dir():
|
||
"""用户自定义 cfg 目录: 程序目录旁 cfgs/(exe 模式在 exe 旁, 源码模式在 FlashTool/)"""
|
||
return os.path.join(_app_base(), "cfgs")
|
||
|
||
|
||
def list_custom_cfgs():
|
||
"""cfgs/ 下的 *.cfg 文件名列表(供界面下拉选择), 目录不存在返回空表"""
|
||
d = custom_cfg_dir()
|
||
if not os.path.isdir(d):
|
||
return []
|
||
return sorted(f for f in os.listdir(d)
|
||
if f.lower().endswith(".cfg") and os.path.isfile(os.path.join(d, f)))
|
||
|
||
|
||
def scan_cfgs(scripts_dir, kind):
|
||
"""扫描 scripts 目录下 kind(target/interface) 子目录里的 cfg 文件名。
|
||
|
||
返回 ['target/gd32e23x.cfg', ...]; scripts 无效/子目录不存在返回空表。
|
||
"""
|
||
if not scripts_dir or not os.path.isdir(scripts_dir):
|
||
return []
|
||
d = os.path.join(scripts_dir, kind)
|
||
if not os.path.isdir(d):
|
||
return []
|
||
return sorted(kind + "/" + f for f in os.listdir(d)
|
||
if f.lower().endswith(".cfg") and os.path.isfile(os.path.join(d, f)))
|
||
|
||
|
||
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 resolve_cfg(value, folder):
|
||
"""界面里的 cfg 引用 → OpenOCD -f 参数。
|
||
|
||
cfgs/<name>.cfg 是自定义 cfg, 转成绝对路径(不依赖 cwd);
|
||
其余(裸名/带路径/绝对路径)按原逻辑处理。
|
||
"""
|
||
value = value.strip().replace("\\", "/")
|
||
if value.startswith("cfgs/"):
|
||
p = os.path.join(custom_cfg_dir(), os.path.basename(value))
|
||
if os.path.isfile(p):
|
||
return p
|
||
return ensure_cfg_path(value, folder)
|
||
|
||
|
||
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=5, 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 选项。
|
||
interface 留空时跳过(配合自带 adapter driver 的自定义 target cfg 使用)"""
|
||
cmd = [openocd.strip()]
|
||
if scripts and scripts.strip():
|
||
cmd += ["-s", scripts.strip()]
|
||
if interface and interface.strip():
|
||
cmd += ["-f", resolve_cfg(interface, "interface")]
|
||
cmd += ["-f", resolve_cfg(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; %s; flash list; exit" % CONNECT_SEQ
|
||
|
||
|
||
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; %s; flash erase_address 0x%08X 0x%X; reset halt; exit"
|
||
% (CONNECT_SEQ, FLASH_BASE_DEFAULT, size))
|
||
|
||
|
||
# 复刻 VSCode tasks.json 的复位序列(0.11 语法, 与内置 xpack 0.11.0-3 匹配):
|
||
# init 连接(connect_assert_srst 由 cfg 声明)后显式释放 SRST 并等复位完成再 halt。
|
||
# 注意: 0.11 只认 "adapter deassert srst"; 0.12 的 "adapter srst deassert"
|
||
# 在 0.11 上是无效命令, 包 catch 会静默吞掉 → SRST 一直按住 → cannot read IDR。
|
||
# 全部操作(检测/烧录/擦除/读取/复位)都必须带本序列: 本目标必须按住复位连接,
|
||
# 否则 exam 失败(PARTNO 0x0), 后续任何访问都会 "Target not examined" 失败。
|
||
CONNECT_SEQ = "adapter deassert srst; sleep 200; catch {reset halt}; reset halt"
|
||
POST_RESET_PREFIX = "adapter deassert srst; sleep 200; "
|
||
|
||
|
||
def script_flash(firmware, base_text="", erase=True, verify=True, reset=True,
|
||
pre_reset=True, post_reset=True):
|
||
"""HEX/ELF 用文件内地址(base_text 留空), BIN 必须给基地址。
|
||
|
||
pre_reset - 烧录前硬件复位序列(SRST 释放 + halt), 复刻 VSCode 任务链
|
||
post_reset - 烧录后复位前先释放 SRST(硬件复位), 否则仅软件 reset
|
||
reset - 复位后运行(reset run); 不勾则复位后保持暂停(reset halt)
|
||
"""
|
||
ext = os.path.splitext(firmware)[1].lower()
|
||
script = "init; "
|
||
script += (CONNECT_SEQ if pre_reset else "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)
|
||
tail = "reset run" if reset else "reset halt"
|
||
script += (POST_RESET_PREFIX + tail if post_reset else tail) + "; 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; %s; dump_image %s 0x%08X 0x%X; exit" % (
|
||
CONNECT_SEQ, tcl_quote(out_path), FLASH_BASE_DEFAULT, size)
|
||
|
||
|
||
def script_reset():
|
||
"""对齐 VSCode Reset 任务: 释放 SRST 后复位运行"""
|
||
return "init; adapter deassert srst; sleep 200; reset; 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"
|