Openocd

{
  "exe": "openocd",
  "scripts_dir": "",
  "gdb_port": 3333,
  "telnet_port": 4444,
  "gdb_exe": "",
  "operation_mode": 1
}

"""openocd skill 私有 GDB 工具。"""

from __future__ import annotations

import re
import subprocess
import sys
from pathlib import Path
from typing import Any


SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from openocd_runtime import hidden_subprocess_kwargs


INTROSPECTION_ACTIONS = {
    "backtrace",
    "locals",
    "break",
    "frame",
    "print",
    "watch",
    "threads",
    "disassemble",
    "crash-report",
}


def run_gdb_commands(gdb_exe: str, elf_file: str, target_remote: str, commands: list[str], timeout: int = 30) -> dict:
    gdb_init = ["set pagination off", "set confirm off", "set width 0"]
    if elf_file:
        gdb_init.append(f"file {elf_file}")
    gdb_init.append(f"target remote {target_remote}")

    cmd = [gdb_exe, "--batch", "--nx"]
    for item in gdb_init + commands + ["quit"]:
        cmd.extend(["-ex", item])

    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
            encoding="utf-8",
            errors="replace",
            **hidden_subprocess_kwargs(),
        )
        return {
            "status": "ok" if proc.returncode == 0 else "error",
            "stdout": proc.stdout,
            "stderr": proc.stderr,
            "returncode": proc.returncode,
        }
    except subprocess.TimeoutExpired:
        return {"status": "error", "error": f"GDB 执行超时({timeout}s)"}
    except Exception as exc:  # pragma: no cover - 兜底异常
        return {"status": "error", "error": str(exc)}


def require_action_expr(action: str, expr: str | None, hint: str) -> str:
    if not expr:
        raise ValueError(f"{action} 必须提供 {hint}")
    return expr


def build_gdb_commands(action: str, expr: str | None = None, *, halt_before: bool = True) -> list[str]:
    commands: list[str] = []
    if halt_before and action in INTROSPECTION_ACTIONS | {"next", "step", "finish", "until"}:
        commands.append("monitor halt")

    if action == "run":
        raise ValueError("run 需要由调用方直接提供 commands")
    if action == "backtrace":
        commands.append("backtrace")
    elif action == "locals":
        commands.append("info locals")
    elif action == "break":
        commands.extend([f"break {require_action_expr(action, expr, '--expr')}", "info breakpoints"])
    elif action == "continue":
        commands.append("monitor resume")
    elif action == "next":
        commands.append("next")
    elif action == "step":
        commands.append("step")
    elif action == "finish":
        commands.append("finish")
    elif action == "until":
        commands.append(f"until {expr}" if expr else "until")
    elif action == "frame":
        commands.append(f"frame {require_action_expr(action, expr, '--expr <帧号>')}")
    elif action == "print":
        commands.append(f"print {require_action_expr(action, expr, '--expr')}")
    elif action == "watch":
        commands.extend([f"watch {require_action_expr(action, expr, '--expr')}", "info breakpoints"])
    elif action == "disassemble":
        commands.append(f"disassemble {expr}" if expr else "disassemble")
    elif action == "threads":
        commands.extend(["info threads", "thread apply all backtrace 1"])
    elif action == "crash-report":
        commands.extend(
            [
                "backtrace full",
                "info registers",
                "frame 0",
                "info locals",
                "info threads",
                "disassemble /m $pc,$pc+32",
            ]
        )
    else:
        raise ValueError(f"未知 GDB 子命令: {action}")

    return commands


def _parse_frames(stdout: str) -> list[dict[str, Any]]:
    frames = []
    for line in stdout.splitlines():
        match = re.match(
            r"#(?P<index>\d+)\s+(?:(?P<address>0x[0-9a-fA-F]+)\s+in\s+)?(?P<function>[^\s(]+)?\s*\((?P<args>[^)]*)\)(?:\s+at\s+(?P<location>.+))?",
            line.strip(),
        )
        if not match:
            continue
        frame = {
            "frame": int(match.group("index")),
            "function": match.group("function") or "??",
        }
        if match.group("address"):
            frame["address"] = match.group("address")
        if match.group("args"):
            frame["args"] = match.group("args").strip()
        if match.group("location"):
            frame["location"] = match.group("location").strip()
        frames.append(frame)
    return frames


def _parse_variables(stdout: str) -> dict[str, str]:
    variables: dict[str, str] = {}
    for line in stdout.splitlines():
        match = re.match(r"^([A-Za-z_][\w.\->\[\]]*)\s*=\s*(.+)$", line.strip())
        if match:
            variables[match.group(1)] = match.group(2).strip()
    return variables


def _parse_registers(stdout: str) -> dict[str, str]:
    registers: dict[str, str] = {}
    for line in stdout.splitlines():
        match = re.match(r"^([A-Za-z_][\w]*)\s+(0x[0-9a-fA-F]+)\b(.*)$", line.strip())
        if match:
            registers[match.group(1)] = match.group(2)
    return registers


def _parse_threads(stdout: str) -> list[dict[str, Any]]:
    threads: list[dict[str, Any]] = []
    for line in stdout.splitlines():
        match = re.match(r"^([* ])\s*(\d+)\s+Thread\s+(.+)$", line.strip())
        if match:
            threads.append(
                {
                    "selected": match.group(1) == "*",
                    "id": int(match.group(2)),
                    "description": match.group(3).strip(),
                }
            )
    return threads


def _parse_disassembly(stdout: str) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    for line in stdout.splitlines():
        match = re.match(r"^(=>)?\s*(0x[0-9a-fA-F]+)(?:\s+<([^>]+)>)?:\s+(.+)$", line.strip())
        if match:
            item = {
                "address": match.group(2),
                "instruction": match.group(4).strip(),
            }
            if match.group(1):
                item["selected"] = "true"
            if match.group(3):
                item["symbol"] = match.group(3).strip()
            items.append(item)
    return items


def _extract_source_location(stdout: str, frames: list[dict[str, Any]]) -> str:
    for frame in frames:
        location = frame.get("location", "")
        if location:
            return location
    match = re.search(r'at\s+([A-Za-z]:)?[^:\n]+\:\d+', stdout)
    return match.group(0).replace("at ", "").strip() if match else ""


def _parse_selected_frame(stdout: str) -> dict[str, Any]:
    match = re.search(r"#(?P<index>\d+)\s+.+?(?:at\s+(?P<location>.+))?$", stdout, re.MULTILINE)
    if not match:
        return {}
    selected = {"frame": int(match.group("index"))}
    if match.group("location"):
        selected["location"] = match.group("location").strip()
    return selected


def parse_gdb_output(stdout: str, action: str) -> dict:
    frames = _parse_frames(stdout)
    variables = _parse_variables(stdout)
    registers = _parse_registers(stdout)
    threads = _parse_threads(stdout)
    disassembly = _parse_disassembly(stdout)

    parsed: dict[str, Any] = {"output": stdout}
    if frames:
        parsed["frames"] = frames
    if variables:
        parsed["variables"] = variables
    if registers:
        parsed["registers"] = registers
    if threads:
        parsed["threads"] = threads
    if disassembly:
        parsed["disassembly"] = disassembly

    selected_frame = _parse_selected_frame(stdout)
    if not selected_frame and frames:
        selected_frame = frames[0]
    if selected_frame:
        parsed["selected_frame"] = selected_frame

    source_location = _extract_source_location(stdout, frames)
    if source_location:
        parsed["source_location"] = source_location

    if action == "print":
        match = re.search(r"\$\d+\s*=\s*(.+)", stdout)
        if match:
            parsed["value"] = match.group(1).strip()

    return parsed

"""OpenOCD GDB Server 启动与 one-shot 调试。"""

from __future__ import annotations

import argparse
import os
import signal
import subprocess
import sys
import time
from pathlib import Path
from shutil import which


ROOT_DIR = Path(__file__).resolve().parents[2]
if str(ROOT_DIR) not in sys.path:
    sys.path.insert(0, str(ROOT_DIR))

from openocd_gdb_common import build_gdb_commands, parse_gdb_output, run_gdb_commands  # noqa: E402
from openocd_runtime import (  # noqa: E402
    build_artifacts,
    default_config_path,
    get_state_entry,
    hidden_subprocess_kwargs,
    is_missing,
    load_json_file,
    load_local_config,
    load_project_config,
    load_workspace_state,
    make_result,
    make_timing,
    normalize_path,
    now_iso,
    output_json,
    parameter_context,
    resolve_param,
    save_project_config,
    update_state_entry,
    workspace_root,
)


GDB_ACTIONS = [
    "server",
    "run",
    "backtrace",
    "locals",
    "break",
    "continue",
    "next",
    "step",
    "finish",
    "until",
    "frame",
    "print",
    "watch",
    "disassemble",
    "threads",
    "crash-report",
]


def build_openocd_cmd(
    exe: str,
    board: str = "",
    interface: str = "",
    target: str = "",
    search: str = "",
    adapter_speed: str = "",
    transport: str = "",
    gdb_port: int = 3333,
    telnet_port: int = 4444,
) -> list[str]:
    cmd = [exe]
    if search:
        cmd.extend(["-s", search])
    if board:
        cmd.extend(["-f", board])
    else:
        if interface:
            cmd.extend(["-f", interface])
        if target:
            cmd.extend(["-f", target])
    if adapter_speed:
        cmd.extend(["-c", f"adapter speed {adapter_speed}"])
    if transport:
        cmd.extend(["-c", f"transport select {transport}"])
    cmd.extend(["-c", f"gdb_port {gdb_port}"])
    cmd.extend(["-c", f"telnet_port {telnet_port}"])
    return cmd


def start_openocd_server(cmd: list[str]) -> subprocess.Popen:
    popen_kwargs = hidden_subprocess_kwargs()
    creationflags = subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0
    if popen_kwargs.get("creationflags"):
        creationflags |= popen_kwargs["creationflags"]
    return subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        encoding="utf-8",
        errors="replace",
        creationflags=creationflags,
        startupinfo=popen_kwargs.get("startupinfo"),
    )


def wait_server_ready(proc: subprocess.Popen, gdb_port: int, timeout: int = 15) -> tuple[bool, list[str]]:
    started = time.time()
    errors: list[str] = []
    ready = False

    while time.time() - started < timeout:
        if proc.poll() is not None:
            remaining = proc.stderr.read()
            for line in remaining.splitlines():
                if "Error:" in line:
                    errors.append(line.strip())
            return False, errors

        line = proc.stderr.readline()
        if not line:
            time.sleep(0.1)
            continue
        stripped = line.strip()
        if "Error:" in stripped:
            errors.append(stripped)
        if f"Listening on port {gdb_port}" in stripped or "listening on" in stripped.lower():
            ready = True
            break

    if not ready:
        return False, errors

    critical = [
        "open failed",
        "init mode failed",
        "no device found",
        "cannot connect",
        "error connecting dp",
        "examination failed",
        "failed to read memory",
        "failed to write memory",
        "cannot read idr",
        "polling failed",
    ]
    critical_errors = [item for item in errors if any(keyword in item.lower() for keyword in critical)]
    if critical_errors:
        return False, critical_errors
    return True, errors


def cleanup(proc: subprocess.Popen | None) -> None:
    if proc and proc.poll() is None:
        try:
            if sys.platform == "win32":
                proc.terminate()
            else:
                proc.send_signal(signal.SIGTERM)
            proc.wait(timeout=5)
        except (subprocess.TimeoutExpired, OSError):
            proc.kill()


def add_common_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--exe", default=None, help="openocd 路径")
    parser.add_argument("--board", default=None, help="board 配置文件")
    parser.add_argument("--interface", default=None, help="interface 配置文件")
    parser.add_argument("--target", default=None, help="target 配置文件")
    parser.add_argument("--search", default=None, help="额外配置脚本搜索目录")
    parser.add_argument("--adapter-speed", default=None, help="调试速率 kHz")
    parser.add_argument("--transport", default=None, choices=["", "swd", "jtag"], help="传输协议")
    parser.add_argument("--gdb-port", type=int, default=None, help="GDB 端口")
    parser.add_argument("--telnet-port", type=int, default=None, help="Telnet 端口")
    parser.add_argument("--gdb-exe", default=None, help="arm-none-eabi-gdb 路径")
    parser.add_argument("--elf", default=None, help="ELF 文件路径")
    parser.add_argument("--config", default=None, help="skill config.json 路径")
    parser.add_argument("--workspace", default=None, help="workspace 根目录，默认当前目录")
    parser.add_argument("--json", action="store_true", dest="as_json")


def build_parser(legacy_server: bool) -> argparse.ArgumentParser:
    if legacy_server:
        parser = argparse.ArgumentParser(description="OpenOCD GDB Server 启动")
        add_common_args(parser)
        return parser

    parser = argparse.ArgumentParser(description="OpenOCD GDB Server 启动与调试")
    sub = parser.add_subparsers(dest="command")
    for name in GDB_ACTIONS:
        sub_parser = sub.add_parser(name, help=f"GDB {name}")
        add_common_args(sub_parser)
        if name == "run":
            sub_parser.add_argument("--commands", nargs="+", required=True, help="GDB 命令序列")
        elif name in {"break", "frame", "print", "watch"}:
            sub_parser.add_argument("--expr", required=True, help="表达式或参数")
        elif name in {"until", "disassemble"}:
            sub_parser.add_argument("--expr", default=None, help="表达式或参数")
    return parser


def _legacy_mode() -> bool:
    return not (len(sys.argv) > 1 and sys.argv[1] in GDB_ACTIONS)


def _state_lookup(state: dict) -> dict:
    last_build = get_state_entry(state, "last_build")
    last_flash = get_state_entry(state, "last_flash")
    last_debug = get_state_entry(state, "last_debug")
    artifacts = last_build.get("artifacts", {})
    return {
        "board": last_debug.get("board") or last_flash.get("board"),
        "interface": last_debug.get("interface") or last_flash.get("interface"),
        "target": last_debug.get("target") or last_flash.get("target"),
        "search": last_debug.get("search"),
        "adapter_speed": last_debug.get("adapter_speed") or last_flash.get("adapter_speed"),
        "transport": last_debug.get("transport") or last_flash.get("transport"),
        "gdb_port": last_debug.get("gdb_port"),
        "telnet_port": last_debug.get("telnet_port"),
        "elf_file": last_build.get("debug_file") or last_build.get("elf_file") or artifacts.get("debug_file"),
        "debug_file": last_build.get("debug_file") or artifacts.get("debug_file"),
    }


def resolve_openocd_params(args, project_config: dict, state_lookup: dict) -> dict:
    """解析 OpenOCD 工程级参数，优先级: CLI > 工程配置 > state.json"""
    # board: CLI > 工程配置 > state
    board = args.board
    board_source = "cli"
    if is_missing(board):
        board = project_config.get("board")
        board_source = "project_config"
    if is_missing(board):
        board = state_lookup.get("board")
        board_source = "state"

    # interface: CLI > 工程配置 > state
    interface = args.interface
    interface_source = "cli"
    if is_missing(interface):
        interface = project_config.get("interface")
        interface_source = "project_config"
    if is_missing(interface):
        interface = state_lookup.get("interface")
        interface_source = "state"

    # target: CLI > 工程配置 > state
    target = args.target
    target_source = "cli"
    if is_missing(target):
        target = project_config.get("target")
        target_source = "project_config"
    if is_missing(target):
        target = state_lookup.get("target")
        target_source = "state"

    # adapter_speed: CLI > 工程配置 > state
    adapter_speed = args.adapter_speed
    adapter_speed_source = "cli"
    if is_missing(adapter_speed):
        adapter_speed = project_config.get("adapter_speed")
        adapter_speed_source = "project_config"
    if is_missing(adapter_speed):
        adapter_speed = state_lookup.get("adapter_speed")
        adapter_speed_source = "state"

    # transport: CLI > 工程配置 > state
    transport = args.transport
    transport_source = "cli"
    if is_missing(transport):
        transport = project_config.get("transport")
        transport_source = "project_config"
    if is_missing(transport):
        transport = state_lookup.get("transport")
        transport_source = "state"

    return {
        "board": board,
        "board_source": board_source,
        "interface": interface,
        "interface_source": interface_source,
        "target": target,
        "target_source": target_source,
        "adapter_speed": adapter_speed,
        "adapter_speed_source": adapter_speed_source,
        "transport": transport,
        "transport_source": transport_source,
    }


def _summary(command: str, parsed: dict) -> str:
    if command == "server":
        return "gdb server 已就绪"
    if command == "backtrace" and parsed.get("frames"):
        return f"backtrace 完成，frames={len(parsed['frames'])}"
    if command == "locals" and parsed.get("variables"):
        return f"locals 完成，variables={len(parsed['variables'])}"
    if command == "threads" and parsed.get("threads"):
        return f"threads 完成，threads={len(parsed['threads'])}"
    if command == "print" and parsed.get("value"):
        return f"print 完成，value={parsed['value']}"
    return f"gdb {command} 完成"


def _metrics(parsed: dict) -> dict:
    metrics: dict[str, int] = {}
    if parsed.get("frames"):
        metrics["frames"] = len(parsed["frames"])
    if parsed.get("variables"):
        metrics["variables"] = len(parsed["variables"])
    if parsed.get("registers"):
        metrics["registers"] = len(parsed["registers"])
    if parsed.get("threads"):
        metrics["threads"] = len(parsed["threads"])
    if parsed.get("disassembly"):
        metrics["instructions"] = len(parsed["disassembly"])
    return metrics


def main() -> None:
    legacy_server = _legacy_mode()
    parser = build_parser(legacy_server)
    args = parser.parse_args()
    if legacy_server:
        args.command = "server"

    started_at = now_iso()
    started_ts = time.time()
    workspace = workspace_root(args.workspace)
    config_path = normalize_path(args.config or str(default_config_path(__file__)))
    config = load_json_file(config_path)
    state = load_workspace_state(str(workspace))
    state_lookup = _state_lookup(state)
    project_config = load_project_config(str(workspace))

    # 解析 OpenOCD 工程级参数
    oc_params = resolve_openocd_params(args, project_config, state_lookup)

    parameter_sources: dict[str, str] = {}
    try:
        exe, parameter_sources["exe"] = resolve_param(
            "exe",
            args.exe,
            config=config,
            config_keys=["exe"],
            required=True,
        )

        # 从工程配置或 state 解析 board/interface/target
        board = oc_params["board"]
        parameter_sources["board"] = oc_params["board_source"]
        interface = oc_params["interface"]
        parameter_sources["interface"] = oc_params["interface_source"]
        target = oc_params["target"]
        parameter_sources["target"] = oc_params["target_source"]
        adapter_speed = oc_params["adapter_speed"]
        parameter_sources["adapter_speed"] = oc_params["adapter_speed_source"]
        transport = oc_params["transport"]
        parameter_sources["transport"] = oc_params["transport_source"]

        search, parameter_sources["search"] = resolve_param(
            "search",
            args.search,
            config=config,
            config_keys=["scripts_dir"],
            state_record=state_lookup,
            state_keys=["search"],
        )
        gdb_port, parameter_sources["gdb_port"] = resolve_param(
            "gdb_port",
            args.gdb_port,
            config=config,
            config_keys=["gdb_port"],
            state_record=state_lookup,
            state_keys=["gdb_port"],
        )
        telnet_port, parameter_sources["telnet_port"] = resolve_param(
            "telnet_port",
            args.telnet_port,
            config=config,
            config_keys=["telnet_port"],
            state_record=state_lookup,
            state_keys=["telnet_port"],
        )
    except ValueError as exc:
        result = make_result(
            status="error",
            action=getattr(args, "command", "server"),
            summary=str(exc),
            details={},
            context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "missing_param", "message": str(exc)},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {exc}", file=sys.stderr)
        sys.exit(1)

    if not board and not interface and not target:
        message = "必须提供 --board 或 --interface + --target"
        result = make_result(
            status="error",
            action=args.command,
            summary=message,
            details={},
            context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "missing_config", "message": message},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {message}", file=sys.stderr)
        sys.exit(1)

    gdb_exe = None
    elf_file = None
    if args.command != "server":
        try:
            gdb_exe, parameter_sources["gdb_exe"] = resolve_param(
                "gdb_exe",
                args.gdb_exe,
                config=config,
                config_keys=["gdb_exe"],
                normalize_as_path=True,
            )
            if is_missing(gdb_exe) and not is_missing(project_config.get("gdb_exe")):
                gdb_exe = normalize_path(project_config.get("gdb_exe"))
                parameter_sources["gdb_exe"] = "project_config:gdb_exe"
            if is_missing(gdb_exe):
                discovered = which("arm-none-eabi-gdb") or which("arm-none-eabi-gdb.exe")
                if discovered:
                    gdb_exe = normalize_path(discovered)
                    parameter_sources["gdb_exe"] = "path"
            if is_missing(gdb_exe):
                raise ValueError("缺少必要参数: gdb_exe")
            elf_file, parameter_sources["elf"] = resolve_param(
                "elf",
                args.elf,
                config=config,
                config_keys=["default_elf"],
                state_record=state_lookup,
                state_keys=["elf_file", "debug_file"],
                normalize_as_path=True,
            )
        except ValueError as exc:
            result = make_result(
                status="error",
                action=args.command,
                summary=str(exc),
                details={},
                context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                error={"code": "missing_param", "message": str(exc)},
                timing=make_timing(started_at, (time.time() - started_ts) * 1000),
            )
            if args.as_json:
                output_json(result)
            else:
                print(f"错误: {exc}", file=sys.stderr)
            sys.exit(1)

        if not os.path.isfile(gdb_exe):
            message = f"arm-none-eabi-gdb 不存在: {gdb_exe}"
            result = make_result(
                status="error",
                action=args.command,
                summary=message,
                details={},
                context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                error={"code": "gdb_not_found", "message": message},
                timing=make_timing(started_at, (time.time() - started_ts) * 1000),
            )
            if args.as_json:
                output_json(result)
            else:
                print(f"错误: {message}", file=sys.stderr)
            sys.exit(1)

    cmd = build_openocd_cmd(
        exe=exe,
        board=board or "",
        interface=interface or "",
        target=target or "",
        search=search or "",
        adapter_speed=str(adapter_speed or ""),
        transport=transport or "",
        gdb_port=int(gdb_port or 3333),
        telnet_port=int(telnet_port or 4444),
    )

    proc: subprocess.Popen | None = None
    try:
        proc = start_openocd_server(cmd)
        ready, errors = wait_server_ready(proc, int(gdb_port or 3333))
        if not ready:
            message = "; ".join(errors) if errors else "GDB Server 启动失败或超时"
            result = make_result(
                status="error",
                action=args.command,
                summary="GDB Server 启动失败",
                details={"errors": errors},
                context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                error={"code": "gdbserver_failed", "message": message},
                timing=make_timing(started_at, (time.time() - started_ts) * 1000),
            )
            if args.as_json:
                output_json(result)
            else:
                print(f"[{args.command}] 失败 — {message}", file=sys.stderr)
            sys.exit(1)

        if args.command == "server":
            state_info = update_state_entry(
                "last_debug",
                {
                    "provider": "openocd",
                    "action": "server",
                    "board": board or "",
                    "interface": interface or "",
                    "target": target or "",
                    "search": search or "",
                    "adapter_speed": adapter_speed or "",
                    "transport": transport or "",
                    "gdb_port": int(gdb_port or 3333),
                    "telnet_port": int(telnet_port or 4444),
                },
                str(workspace),
            )
            result = make_result(
                status="ok",
                action="server",
                summary=_summary("server", {}),
                details={
                    "gdb_port": int(gdb_port or 3333),
                    "telnet_port": int(telnet_port or 4444),
                    "pid": proc.pid,
                    "connect_cmd": f"arm-none-eabi-gdb -ex 'target remote localhost:{int(gdb_port or 3333)}'",
                    "warnings": errors,
                },
                context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                state=state_info,
                timing=make_timing(started_at, (time.time() - started_ts) * 1000),
            )
            if args.as_json:
                output_json(result)
            else:
                print("[gdb-server] GDB Server 已就绪")
                print(f"  GDB 端口: {int(gdb_port or 3333)}")
                print(f"  Telnet 端口: {int(telnet_port or 4444)}")
                print(f"  PID: {proc.pid}")
            try:
                proc.wait()
            except KeyboardInterrupt:
                pass
            return

        try:
            if args.command == "run":
                gdb_commands = list(args.commands)
            else:
                gdb_commands = build_gdb_commands(args.command, getattr(args, "expr", None))
        except ValueError as exc:
            result = make_result(
                status="error",
                action=args.command,
                summary=str(exc),
                details={},
                context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                error={"code": "invalid_args", "message": str(exc)},
                timing=make_timing(started_at, (time.time() - started_ts) * 1000),
            )
            if args.as_json:
                output_json(result)
            else:
                print(f"错误: {exc}", file=sys.stderr)
            sys.exit(1)

        gdb_result = run_gdb_commands(gdb_exe, elf_file or "", f"localhost:{int(gdb_port or 3333)}", gdb_commands)
        elapsed_ms = (time.time() - started_ts) * 1000
        if gdb_result["status"] == "error":
            result = make_result(
                status="error",
                action=args.command,
                summary="GDB 执行失败",
                details={"gdb_port": int(gdb_port or 3333), "errors": errors},
                context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                artifacts=build_artifacts(debug_file=elf_file),
                error={"code": "gdb_error", "message": gdb_result.get("error", gdb_result.get("stderr", "GDB 执行失败"))},
                timing=make_timing(started_at, elapsed_ms),
            )
        else:
            parsed = parse_gdb_output(gdb_result["stdout"], args.command)
            artifacts = build_artifacts(debug_file=elf_file)
            state_info = update_state_entry(
                "last_debug",
                {
                    "provider": "openocd",
                    "action": args.command,
                    "board": board or "",
                    "interface": interface or "",
                    "target": target or "",
                    "search": search or "",
                    "adapter_speed": adapter_speed or "",
                    "transport": transport or "",
                    "gdb_port": int(gdb_port or 3333),
                    "telnet_port": int(telnet_port or 4444),
                    "debug_file": elf_file or "",
                    "artifacts": artifacts,
                },
                str(workspace),
            )
            # 写回确认过的参数到工程配置
            save_project_config(str(workspace), {
                "board": board or "",
                "interface": interface or "",
                "target": target or "",
                "adapter_speed": adapter_speed or "",
                "transport": transport or "",
            })
            result = make_result(
                status="ok",
                action=args.command,
                summary=_summary(args.command, parsed),
                details={
                    "gdb_port": int(gdb_port or 3333),
                    "telnet_port": int(telnet_port or 4444),
                    "commands": gdb_commands,
                    "output": parsed.get("output", ""),
                    "returncode": gdb_result.get("returncode", 0),
                    "warnings": errors,
                    **{key: value for key, value in parsed.items() if key != "output"},
                },
                context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                artifacts=artifacts,
                metrics=_metrics(parsed),
                state=state_info,
                next_actions=["可继续基于 last_debug 复用 cfg 组合和 debug_file"],
                timing=make_timing(started_at, elapsed_ms),
            )

        if args.as_json:
            output_json(result)
        elif result["status"] == "ok":
            print(f"[gdb-{args.command}] {result['summary']}")
            output = result.get("details", {}).get("output", "")
            if output:
                print(output)
        else:
            print(f"[gdb-{args.command}] 失败 — {result['error']['message']}", file=sys.stderr)
            sys.exit(1)
    except FileNotFoundError:
        message = f"openocd 不存在或不在 PATH 中: {exe}"
        result = make_result(
            status="error",
            action=args.command,
            summary=message,
            details={},
            context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "exe_not_found", "message": message},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {message}", file=sys.stderr)
        sys.exit(1)
    finally:
        if args.command != "server":
            cleanup(proc)


if __name__ == "__main__":
    main()

"""OpenOCD ITM/SWO 观测。

基于 OpenOCD 官方 TPIU/SWO 命令：
- $tpiu_name configure -protocol uart -output :<port> -traceclk <Hz> [-pin-freq <Hz>]
- $tpiu_name enable
- itm port <n> on / itm ports on
"""

from __future__ import annotations

import argparse
import signal
import socket
import subprocess
import sys
import time
from pathlib import Path


ROOT_DIR = Path(__file__).resolve().parents[2]
if str(ROOT_DIR) not in sys.path:
    sys.path.insert(0, str(ROOT_DIR))

from openocd_runtime import (  # noqa: E402
    default_config_path,
    emit_stream_record,
    get_state_entry,
    hidden_subprocess_kwargs,
    is_missing,
    load_json_file,
    load_local_config,
    load_project_config,
    load_workspace_state,
    make_result,
    make_timing,
    normalize_path,
    now_iso,
    output_json,
    parameter_context,
    resolve_param,
    save_project_config,
    update_state_entry,
    workspace_root,
)


def build_openocd_cmd(
    exe: str,
    board: str = "",
    interface: str = "",
    target: str = "",
    search: str = "",
    adapter_speed: str = "",
    transport: str = "",
    gdb_port: int = 3333,
    telnet_port: int = 4444,
    trace_port: int = 3443,
    tpiu_name: str = "",
    traceclk: str = "",
    pin_freq: str = "",
    itm_ports: list[str] | None = None,
) -> list[str]:
    cmd = [exe]
    if search:
        cmd.extend(["-s", search])
    if board:
        cmd.extend(["-f", board])
    else:
        if interface:
            cmd.extend(["-f", interface])
        if target:
            cmd.extend(["-f", target])
    if adapter_speed:
        cmd.extend(["-c", f"adapter speed {adapter_speed}"])
    if transport:
        cmd.extend(["-c", f"transport select {transport}"])
    cmd.extend(["-c", f"gdb_port {gdb_port}", "-c", f"telnet_port {telnet_port}"])
    tpiu_cmd = f"{tpiu_name} configure -protocol uart -output :{trace_port} -traceclk {traceclk}"
    if pin_freq:
        tpiu_cmd += f" -pin-freq {pin_freq}"
    cmd.extend(["-c", tpiu_cmd, "-c", "init", "-c", "reset init", "-c", f"{tpiu_name} enable"])
    if itm_ports:
        for port in itm_ports:
            cmd.extend(["-c", f"itm port {port} on"])
    else:
        cmd.extend(["-c", "itm ports on"])
    return cmd


def start_openocd_server(cmd: list[str]) -> subprocess.Popen:
    popen_kwargs = hidden_subprocess_kwargs()
    creationflags = subprocess.CREATE_NEW_PROCESS_GROUP if sys.platform == "win32" else 0
    if popen_kwargs.get("creationflags"):
        creationflags |= popen_kwargs["creationflags"]
    return subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        encoding="utf-8",
        errors="replace",
        creationflags=creationflags,
        startupinfo=popen_kwargs.get("startupinfo"),
    )


def wait_server_ready(proc: subprocess.Popen, trace_port: int, timeout: int = 15) -> tuple[bool, list[str]]:
    started = time.time()
    lines: list[str] = []
    ready = False
    ready_deadline = 0.0
    critical_keywords = [
        "error:",
        "failed to start adapter's trace",
        "not supported by the device",
    ]
    while time.time() - started < timeout:
        if proc.poll() is not None:
            lines.extend(proc.stderr.read().splitlines())
            return False, lines
        line = proc.stderr.readline()
        if not line:
            time.sleep(0.1)
            if ready and time.time() >= ready_deadline:
                return True, lines
            continue
        stripped = line.strip()
        lines.append(stripped)
        lowered = stripped.lower()
        if any(keyword in lowered for keyword in critical_keywords):
            return False, lines
        if f"port {trace_port}" in lowered or "trace data" in lowered:
            ready = True
            ready_deadline = time.time() + 1.0
    return ready, lines


def cleanup(proc: subprocess.Popen | None) -> None:
    if proc and proc.poll() is None:
        try:
            if sys.platform == "win32":
                proc.terminate()
            else:
                proc.send_signal(signal.SIGTERM)
            proc.wait(timeout=5)
        except (subprocess.TimeoutExpired, OSError):
            proc.kill()


def _state_lookup(state: dict) -> dict:
    last_debug = get_state_entry(state, "last_debug")
    last_flash = get_state_entry(state, "last_flash")
    return {
        "board": last_debug.get("board") or last_flash.get("board"),
        "interface": last_debug.get("interface") or last_flash.get("interface"),
        "target": last_debug.get("target") or last_flash.get("target"),
        "search": last_debug.get("search"),
        "adapter_speed": last_debug.get("adapter_speed") or last_flash.get("adapter_speed"),
        "transport": last_debug.get("transport") or last_flash.get("transport"),
    }


def resolve_openocd_params(args, project_config: dict, state_lookup: dict) -> dict:
    """解析 OpenOCD 工程级参数，优先级: CLI > 工程配置 > state.json"""
    # board: CLI > 工程配置 > state
    board = args.board
    board_source = "cli"
    if is_missing(board):
        board = project_config.get("board")
        board_source = "project_config"
    if is_missing(board):
        board = state_lookup.get("board")
        board_source = "state"

    # interface: CLI > 工程配置 > state
    interface = args.interface
    interface_source = "cli"
    if is_missing(interface):
        interface = project_config.get("interface")
        interface_source = "project_config"
    if is_missing(interface):
        interface = state_lookup.get("interface")
        interface_source = "state"

    # target: CLI > 工程配置 > state
    target = args.target
    target_source = "cli"
    if is_missing(target):
        target = project_config.get("target")
        target_source = "project_config"
    if is_missing(target):
        target = state_lookup.get("target")
        target_source = "state"

    # adapter_speed: CLI > 工程配置 > state
    adapter_speed = args.adapter_speed
    adapter_speed_source = "cli"
    if is_missing(adapter_speed):
        adapter_speed = project_config.get("adapter_speed")
        adapter_speed_source = "project_config"
    if is_missing(adapter_speed):
        adapter_speed = state_lookup.get("adapter_speed")
        adapter_speed_source = "state"

    # transport: CLI > 工程配置 > state
    transport = args.transport
    transport_source = "cli"
    if is_missing(transport):
        transport = project_config.get("transport")
        transport_source = "project_config"
    if is_missing(transport):
        transport = state_lookup.get("transport")
        transport_source = "state"

    # tpiu_name: CLI > 工程配置
    tpiu_name = args.tpiu_name
    tpiu_name_source = "cli"
    if is_missing(tpiu_name):
        tpiu_name = project_config.get("tpiu_name")
        tpiu_name_source = "project_config"

    # traceclk: CLI > 工程配置
    traceclk = args.traceclk
    traceclk_source = "cli"
    if is_missing(traceclk):
        traceclk = project_config.get("traceclk")
        traceclk_source = "project_config"

    # pin_freq: CLI > 工程配置
    pin_freq = args.pin_freq
    pin_freq_source = "cli"
    if is_missing(pin_freq):
        pin_freq = project_config.get("pin_freq")
        pin_freq_source = "project_config"

    return {
        "board": board,
        "board_source": board_source,
        "interface": interface,
        "interface_source": interface_source,
        "target": target,
        "target_source": target_source,
        "adapter_speed": adapter_speed,
        "adapter_speed_source": adapter_speed_source,
        "transport": transport,
        "transport_source": transport_source,
        "tpiu_name": tpiu_name,
        "tpiu_name_source": tpiu_name_source,
        "traceclk": traceclk,
        "traceclk_source": traceclk_source,
        "pin_freq": pin_freq,
        "pin_freq_source": pin_freq_source,
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="OpenOCD ITM 输出捕获")
    parser.add_argument("--exe", default=None, help="openocd 路径")
    parser.add_argument("--board", default=None, help="board 配置文件")
    parser.add_argument("--interface", default=None, help="interface 配置文件")
    parser.add_argument("--target", default=None, help="target 配置文件")
    parser.add_argument("--search", default=None, help="额外配置脚本搜索目录")
    parser.add_argument("--adapter-speed", default=None, help="调试速率 kHz")
    parser.add_argument("--transport", default=None, choices=["", "swd", "jtag"], help="传输协议")
    parser.add_argument("--gdb-port", type=int, default=None, help="GDB 端口")
    parser.add_argument("--telnet-port", type=int, default=None, help="Telnet 端口")
    parser.add_argument("--trace-port", type=int, default=3443, help="OpenOCD trace TCP 端口")
    parser.add_argument("--tpiu-name", default=None, help="TPIU/SWO 对象名，例如 stm32l1.tpiu")
    parser.add_argument("--traceclk", default=None, help="TRACECLKIN 频率 Hz")
    parser.add_argument("--pin-freq", default=None, help="SWO pin 频率 Hz")
    parser.add_argument("--itm-port", action="append", dest="itm_ports", help="启用的 ITM stimulus port，可多次传入")
    parser.add_argument("--workspace", default=None, help="workspace 根目录，默认当前目录")
    parser.add_argument("--config", default=None, help="skill config.json 路径")
    parser.add_argument("--json", action="store_true", dest="as_json")
    args = parser.parse_args()

    started_at = now_iso()
    started_ts = time.time()
    workspace = workspace_root(args.workspace)
    config_path = normalize_path(args.config or str(default_config_path(__file__)))
    config = load_json_file(config_path)
    state = load_workspace_state(str(workspace))
    state_lookup = _state_lookup(state)
    project_config = load_project_config(str(workspace))

    # 解析 OpenOCD 工程级参数
    oc_params = resolve_openocd_params(args, project_config, state_lookup)

    parameter_sources: dict[str, str] = {}
    try:
        exe, parameter_sources["exe"] = resolve_param("exe", args.exe, config=config, config_keys=["exe"], required=True)

        # 从工程配置或 state 解析 board/interface/target
        board = oc_params["board"]
        parameter_sources["board"] = oc_params["board_source"]
        interface = oc_params["interface"]
        parameter_sources["interface"] = oc_params["interface_source"]
        target = oc_params["target"]
        parameter_sources["target"] = oc_params["target_source"]
        adapter_speed = oc_params["adapter_speed"]
        parameter_sources["adapter_speed"] = oc_params["adapter_speed_source"]
        transport = oc_params["transport"]
        parameter_sources["transport"] = oc_params["transport_source"]

        search, parameter_sources["search"] = resolve_param("search", args.search, config=config, config_keys=["scripts_dir"], state_record=state_lookup, state_keys=["search"])

        # tpiu_name, traceclk, pin_freq 从工程配置解析
        tpiu_name = oc_params["tpiu_name"]
        parameter_sources["tpiu_name"] = oc_params["tpiu_name_source"]
        traceclk = oc_params["traceclk"]
        parameter_sources["traceclk"] = oc_params["traceclk_source"]
        pin_freq = oc_params["pin_freq"]
        parameter_sources["pin_freq"] = oc_params["pin_freq_source"]
    except ValueError as exc:
        result = make_result(
            status="error",
            action="itm",
            summary=str(exc),
            details={},
            context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "missing_param", "message": str(exc)},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {exc}", file=sys.stderr)
        sys.exit(1)

    if not board and not interface and not target:
        message = "必须提供 --board 或 --interface + --target"
        result = make_result(
            status="error",
            action="itm",
            summary=message,
            details={},
            context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "missing_config", "message": message},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {message}", file=sys.stderr)
        sys.exit(1)

    proc = None
    trace_sock = None
    try:
        proc = start_openocd_server(
            build_openocd_cmd(
                exe=exe,
                board=board or "",
                interface=interface or "",
                target=target or "",
                search=search or "",
                adapter_speed=str(adapter_speed or ""),
                transport=transport or "",
                gdb_port=int(args.gdb_port or config.get("gdb_port", 3333)),
                telnet_port=int(args.telnet_port or config.get("telnet_port", 4444)),
                trace_port=args.trace_port,
                tpiu_name=tpiu_name,
                traceclk=traceclk,
                pin_freq=pin_freq or "",
                itm_ports=args.itm_ports,
            )
        )
        ready, lines = wait_server_ready(proc, args.trace_port)
        if not ready:
            message = "; ".join(line for line in lines if line) or "OpenOCD ITM 初始化失败"
            result = make_result(
                status="error",
                action="itm",
                summary=message,
                details={"server_log": lines},
                context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                error={"code": "server_failed", "message": message},
                timing=make_timing(started_at, (time.time() - started_ts) * 1000),
            )
            if args.as_json:
                output_json(result)
            else:
                print(f"错误: {message}", file=sys.stderr)
            sys.exit(1)

        last_trace_error: OSError | None = None
        trace_deadline = time.time() + 5.0
        while time.time() < trace_deadline:
            try:
                trace_sock = socket.create_connection(("127.0.0.1", args.trace_port), timeout=1.0)
                trace_sock.settimeout(0.5)
                last_trace_error = None
                break
            except OSError as exc:
                last_trace_error = exc
                if proc.poll() is not None:
                    break
                time.sleep(0.2)
        if trace_sock is None:
            raise OSError(last_trace_error or "trace socket not ready")

        update_state_entry(
            "last_observe",
            {
                "provider": "openocd",
                "action": "itm",
                "board": board or "",
                "interface": interface or "",
                "target": target or "",
                "search": search or "",
                "adapter_speed": adapter_speed or "",
                "transport": transport or "",
                "channel_type": "itm",
                "stream_type": "binary",
                "source": "openocd",
            },
            str(workspace),
        )
        # 写回确认过的参数到工程配置
        save_project_config(str(workspace), {
            "board": board or "",
            "interface": interface or "",
            "target": target or "",
            "adapter_speed": adapter_speed or "",
            "transport": transport or "",
            "tpiu_name": tpiu_name or "",
            "traceclk": traceclk or "",
            "pin_freq": pin_freq or "",
        })

        while True:
            try:
                chunk = trace_sock.recv(4096)
            except socket.timeout:
                if proc.poll() is not None:
                    break
                continue
            if not chunk:
                if proc.poll() is not None:
                    break
                continue
            text = chunk.decode("utf-8", errors="replace")
            emit_stream_record(source="openocd", channel_type="itm", text=text, as_json=args.as_json, stream_type="event")

    except FileNotFoundError:
        message = f"openocd 不存在或不在 PATH 中: {exe}"
        result = make_result(
            status="error",
            action="itm",
            summary=message,
            details={},
            context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "exe_not_found", "message": message},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {message}", file=sys.stderr)
        sys.exit(1)
    except OSError as exc:
        message = f"ITM trace 连接失败: {exc}"
        result = make_result(
            status="error",
            action="itm",
            summary=message,
            details={},
            context=parameter_context(provider="openocd", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "trace_connect_failed", "message": message},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {message}", file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        pass
    finally:
        if trace_sock:
            trace_sock.close()
        cleanup(proc)


if __name__ == "__main__":
    main()