Keil

{
  "uv4_exe": "C:\\Keil_v5\\UV4\\UV4.exe",
  "operation_mode": 1
}

"""Keil MDK 构建 / 重建 / 清理 / 烧录。"""

from __future__ import annotations

import argparse
import os
import re
import subprocess
import sys
import time
import xml.etree.ElementTree as ET
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 keil_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,
)


ERRORLEVEL_MAP = {
    0: ("ok", "无错误或警告"),
    1: ("ok", "有警告"),
    2: ("error", "有错误"),
    3: ("error", "致命错误"),
    11: ("error", "无法打开工程文件"),
    12: ("error", "设备数据库缺失"),
    13: ("error", "写入错误"),
    15: ("error", "UV4 访问错误"),
    20: ("error", "未知错误"),
}

ACTION_FLAG = {
    "build": "-b",
    "rebuild": "-r",
    "clean": "-c",
    "flash": "-f",
}

UV4_TIMEOUT_SEC = 1800

GENERATED_OUTPUT_SUFFIXES = {
    ".axf",
    ".elf",
    ".hex",
    ".bin",
    ".o",
    ".d",
    ".crf",
    ".dep",
    ".htm",
    ".lnp",
    ".iex",
}


def parse_log(log_path: str) -> dict:
    metrics = {"errors": 0, "warnings": 0, "flash_bytes": 0, "ram_bytes": 0}
    if not os.path.isfile(log_path):
        return metrics

    with open(log_path, "r", encoding="utf-8", errors="replace") as file_obj:
        content = file_obj.read()

    error_match = re.search(r"(\d+)\s+Error\(s\)\s*,\s*(\d+)\s+Warning\(s\)", content)
    if error_match:
        metrics["errors"] = int(error_match.group(1))
        metrics["warnings"] = int(error_match.group(2))

    size_match = re.search(
        r"Program Size:\s+Code=(\d+)\s+RO-data=(\d+)\s+RW-data=(\d+)\s+ZI-data=(\d+)",
        content,
    )
    if size_match:
        code_size = int(size_match.group(1))
        ro_data = int(size_match.group(2))
        rw_data = int(size_match.group(3))
        zi_data = int(size_match.group(4))
        metrics["flash_bytes"] = code_size + ro_data + rw_data
        metrics["ram_bytes"] = rw_data + zi_data

    return metrics


def _resolve_path(base_dir: Path, raw_path: str) -> Path:
    raw_path = (raw_path or "").strip()
    if not raw_path:
        return base_dir
    path = Path(raw_path)
    if path.is_absolute():
        return path
    return (base_dir / path).resolve()


def _resolve_workspace_path(workspace: Path, raw_path: str | None, default: str) -> str:
    value = default if is_missing(raw_path) else str(raw_path)
    path = Path(value)
    return str(path.resolve() if path.is_absolute() else (workspace / path).resolve())


def _resolve_project_path(workspace: Path, raw_path: str | None) -> str:
    if is_missing(raw_path):
        return ""
    path = Path(str(raw_path)).expanduser()
    return str(path.resolve() if path.is_absolute() else (workspace / path).resolve())


def _make_relative_to_workspace(workspace: Path, path: str) -> str:
    """将绝对路径转换为相对于 workspace 的相对路径"""
    try:
        p = Path(path).resolve()
        ws = workspace.resolve()
        rel = p.relative_to(ws)
        return str(rel).replace("\\", "/")
    except ValueError:
        return path


def _collect_target_artifacts(project_path: Path, target: str) -> dict:
    if project_path.suffix.lower() != ".uvprojx":
        return {}

    try:
        root = ET.parse(str(project_path)).getroot()
    except (ET.ParseError, OSError):
        return {}

    target_el = None
    fallback_target = None
    for item in root.iter("Target"):
        name_el = item.find("TargetName")
        if name_el is None or not name_el.text:
            continue
        if fallback_target is None:
            fallback_target = item
        if target and name_el.text.strip() == target:
            target_el = item
            break

    if target_el is None:
        target_el = fallback_target
    if target_el is None:
        return {}

    common = target_el.find("TargetOption/TargetCommonOption")
    if common is None:
        return {}

    output_dir = _resolve_path(project_path.parent, common.findtext("OutputDirectory", default=""))
    output_name = common.findtext("OutputName", default="").strip() or project_path.stem

    candidates = {
        "axf_file": output_dir / f"{output_name}.axf",
        "elf_file": output_dir / f"{output_name}.elf",
        "hex_file": output_dir / f"{output_name}.hex",
        "bin_file": output_dir / f"{output_name}.bin",
    }

    details: dict[str, str] = {}
    for key, file_path in candidates.items():
        if file_path.exists():
            details[key] = str(file_path.resolve())

    if not details and output_dir.exists():
        for suffix, key in (
            (".axf", "axf_file"),
            (".elf", "elf_file"),
            (".hex", "hex_file"),
            (".bin", "bin_file"),
        ):
            matches = sorted(output_dir.rglob(f"{output_name}*{suffix}"), key=lambda path: path.stat().st_mtime, reverse=True)
            if matches:
                details[key] = str(matches[0].resolve())

    debug_file = details.get("elf_file") or details.get("axf_file")
    flash_file = details.get("hex_file") or details.get("bin_file") or debug_file
    if debug_file:
        details["debug_file"] = debug_file
    if flash_file:
        details["flash_file"] = flash_file
    if output_dir.exists():
        details["output_dir"] = str(output_dir.resolve())
    return details


def _target_common_option(project_path: Path, target: str) -> tuple[Path | None, str]:
    if project_path.suffix.lower() != ".uvprojx":
        return None, ""

    try:
        root = ET.parse(str(project_path)).getroot()
    except (ET.ParseError, OSError):
        return None, ""

    target_el = None
    fallback_target = None
    for item in root.iter("Target"):
        name_el = item.find("TargetName")
        if name_el is None or not name_el.text:
            continue
        if fallback_target is None:
            fallback_target = item
        if target and name_el.text.strip() == target:
            target_el = item
            break

    if target_el is None:
        target_el = fallback_target
    if target_el is None:
        return None, ""

    common = target_el.find("TargetOption/TargetCommonOption")
    if common is None:
        return None, ""

    output_dir = _resolve_path(project_path.parent, common.findtext("OutputDirectory", default=""))
    output_name = common.findtext("OutputName", default="").strip() or project_path.stem
    return output_dir if output_dir.exists() else None, output_name


def _cleanup_generated_outputs(project_path: Path, target: str) -> dict:
    output_dir, output_name = _target_common_option(project_path, target)
    if output_dir is None:
        return {}

    removed: list[str] = []
    prefix = output_name.lower()
    for item in output_dir.iterdir():
        if not item.is_file():
            continue
        suffix = item.suffix.lower()
        name = item.name.lower()
        should_remove = suffix in GENERATED_OUTPUT_SUFFIXES or name.startswith(prefix)
        if not should_remove:
            continue
        try:
            item.unlink()
            removed.append(str(item.resolve()))
        except OSError:
            continue

    details: dict[str, object] = {
        "output_dir": str(output_dir.resolve()),
        "removed_files": removed,
        "removed_count": len(removed),
    }
    return details


def _build_summary(action: str, status: str, metrics: dict) -> str:
    errors = metrics.get("errors", 0)
    warnings = metrics.get("warnings", 0)
    if status == "error":
        return f"{action} 失败，errors={errors} warnings={warnings}"
    if action in ("build", "rebuild"):
        return f"{action} 成功，errors={errors} warnings={warnings}"
    return f"{action} 成功"


def _next_actions(action: str, artifacts: dict) -> list[str]:
    actions: list[str] = []
    if action in ("build", "rebuild") and artifacts.get("flash_file"):
        actions.append("可直接复用 artifacts.flash_file 继续 flash")
    if action in ("build", "rebuild") and artifacts.get("debug_file"):
        actions.append("可直接复用 artifacts.debug_file 继续 gdb 调试")
    return actions


def _extract_uv4_error(action: str, proc: subprocess.CompletedProcess[str], log_file: Path, errorlevel_desc: str) -> dict:
    log_tail = ""
    if log_file.exists():
        try:
            lines = log_file.read_text(encoding="utf-8", errors="replace").splitlines()
            log_tail = "\n".join(lines[-40:])
        except OSError:
            log_tail = ""

    combined = "\n".join(part for part in (proc.stdout, proc.stderr, log_tail) if part).strip()
    if action == "flash":
        flash_failed = re.search(r"(Flash Download failed\s+-\s+.+)", combined, re.IGNORECASE)
        if flash_failed:
            return {"code": "flash_failed", "message": flash_failed.group(1).strip()}
        if re.search(r"Programming Failed!", combined, re.IGNORECASE):
            return {"code": "flash_failed", "message": "Programming Failed"}

    patterns = [
        (r"\*\*\*\s*error\s+\d+:\s+(.+)", "uv4_error"),
        (r"Error:\s+(.+)", "uv4_error"),
    ]
    for pattern, code in patterns:
        match = re.search(pattern, combined, re.IGNORECASE)
        if match:
            return {"code": code, "message": match.group(1).strip()}

    return {
        "code": f"{action}_failed",
        "message": errorlevel_desc or "UV4 执行失败",
    }


def run_uv4(uv4_exe: str, action: str, project: str, target: str, log_dir: str, clean_first: bool = False) -> dict:
    project_path = Path(project).resolve()
    if not project_path.exists():
        return {
            "status": "error",
            "action": action,
            "error": {"code": "project_not_found", "message": f"工程文件不存在: {project_path}"},
        }

    if not os.path.isfile(uv4_exe):
        return {
            "status": "error",
            "action": action,
            "error": {"code": "uv4_not_found", "message": f"UV4.exe 不存在: {uv4_exe}"},
        }

    log_path = Path(log_dir).resolve()
    log_path.mkdir(parents=True, exist_ok=True)
    log_file = log_path / f"{project_path.stem}-{target or 'default'}-{action}.log"

    flag = ACTION_FLAG[action]
    if action == "rebuild" and clean_first:
        flag = "-cr"

    cmd = [uv4_exe, flag, str(project_path), "-j0", "-o", str(log_file)]
    if target:
        cmd.extend(["-t", target])

    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=UV4_TIMEOUT_SEC,
            cwd=str(project_path.parent),
            encoding="utf-8",
            errors="replace",
            **hidden_subprocess_kwargs(),
        )
    except subprocess.TimeoutExpired:
        return {
            "status": "error",
            "action": action,
            "error": {"code": "timeout", "message": f"UV4.exe 执行超时({UV4_TIMEOUT_SEC}s)"},
        }
    except Exception as exc:  # pragma: no cover - 兜底异常
        return {
            "status": "error",
            "action": action,
            "error": {"code": "exec_error", "message": str(exc)},
        }

    if action == "clean":
        cleanup_details = _cleanup_generated_outputs(project_path, target)
    else:
        cleanup_details = {}

    metrics = parse_log(str(log_file))
    _, errorlevel_desc = ERRORLEVEL_MAP.get(proc.returncode, ("error", f"未知返回码: {proc.returncode}"))
    status = "error" if proc.returncode >= 2 or metrics["errors"] > 0 else "ok"

    details = {
        "project": str(project_path),
        "target": target,
        "log_file": str(log_file.resolve()),
        "errorlevel": proc.returncode,
        "errorlevel_desc": errorlevel_desc,
        **_collect_target_artifacts(project_path, target),
        **cleanup_details,
    }

    result = {
        "status": status,
        "action": action,
        "metrics": metrics,
        "details": details,
    }
    if status == "error":
        result["error"] = _extract_uv4_error(action, proc, log_file, errorlevel_desc)

    return result


def check_last_build_ok(log_dir: str, project: str, target: str) -> bool:
    log_path = Path(log_dir).resolve()
    stem = Path(project).stem
    for action in ("build", "rebuild"):
        log_file = log_path / f"{stem}-{target or 'default'}-{action}.log"
        if log_file.exists():
            metrics = parse_log(str(log_file))
            if metrics["errors"] == 0:
                return True
    return False


def main() -> None:
    parser = argparse.ArgumentParser(description="Keil MDK 构建/重建/清理/烧录")
    parser.add_argument("action", choices=["build", "rebuild", "clean", "flash"])
    parser.add_argument("--uv4", default=None, help="UV4.exe 路径")
    parser.add_argument("--project", default=None, help="工程文件路径")
    parser.add_argument("--target", default=None, help="Target 名称")
    parser.add_argument("--log-dir", default=None, help="日志输出目录")
    parser.add_argument("--clean-first", action="store_true", help="rebuild 时先 clean")
    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")
    args = parser.parse_args()

    started_at = now_iso()
    started_ts = time.time()
    workspace = workspace_root(args.workspace)
    
    # 加载三层配置：环境级、工程级、状态
    local_config = load_local_config(__file__)
    project_config = load_project_config(str(workspace))
    state = load_workspace_state(str(workspace))
    last_build = get_state_entry(state, "last_build")

    parameter_sources: dict[str, str] = {}
    try:
        # uv4_exe: CLI > 环境级配置 > 必需
        uv4_exe, parameter_sources["uv4"] = resolve_param(
            "uv4",
            args.uv4,
            config=local_config,
            config_keys=["uv4_exe"],
            required=True,
            normalize_as_path=True,
            workspace=str(workspace),
        )
        
        # project: CLI > 环境级配置 > 工程级配置 > state.json > 必需
        project, parameter_sources["project"] = resolve_param(
            "project",
            args.project,
            config=local_config,
            config_keys=["default_project"],
            normalize_as_path=True,
            workspace=str(workspace),
        )
        # 工程级配置（优先于 state）
        if is_missing(project) and not is_missing(project_config.get("project")):
            project = _resolve_project_path(workspace, project_config.get("project"))
            parameter_sources["project"] = "project_config:project"
        # state.json（最后 fallback）
        if is_missing(project) and not is_missing(last_build.get("project")):
            project = _resolve_project_path(workspace, str(last_build.get("project")))
            parameter_sources["project"] = "state:project"
        if is_missing(project):
            raise ValueError("缺少必要参数: project")
        
        # target: CLI > 环境级配置 > 工程级配置 > state.json
        target, parameter_sources["target"] = resolve_param(
            "target",
            args.target,
            config=local_config,
            config_keys=["default_target"],
        )
        # 工程级配置（优先于 state）
        if is_missing(target) and not is_missing(project_config.get("target")):
            target = project_config.get("target")
            parameter_sources["target"] = "project_config:target"
        # state.json（最后 fallback）
        if is_missing(target) and not is_missing(last_build.get("target")):
            target = last_build.get("target")
            parameter_sources["target"] = "state:target"
        
        # log_dir: CLI > 工程级配置 > 环境级配置 > 默认值(.embeddedskills/build)
        log_dir_raw = args.log_dir or project_config.get("log_dir") or local_config.get("log_dir")
        log_dir = _resolve_workspace_path(workspace, log_dir_raw, ".embeddedskills/build")
        if args.log_dir:
            parameter_sources["log_dir"] = "cli"
        elif project_config.get("log_dir"):
            parameter_sources["log_dir"] = "project_config:log_dir"
        elif local_config.get("log_dir"):
            parameter_sources["log_dir"] = "config:log_dir"
        else:
            parameter_sources["log_dir"] = "default"
    except ValueError as exc:
        result = make_result(
            status="error",
            action=args.action,
            summary=str(exc),
            details={},
            context=parameter_context(
                provider="keil",
                workspace=str(workspace),
                parameter_sources=parameter_sources,
            ),
            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 args.action == "flash" and not check_last_build_ok(log_dir, project, target or ""):
        result = make_result(
            status="error",
            action="flash",
            summary="最近构建不可用于 flash",
            details={"project": project, "target": target, "log_dir": log_dir},
            context=parameter_context(
                provider="keil",
                workspace=str(workspace),
                parameter_sources=parameter_sources,
            ),
            error={
                "code": "build_not_clean",
                "message": "最近一次构建存在错误或无构建记录，禁止继续烧录。请先执行 build 并确认无错误。",
            },
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {result['error']['message']}", file=sys.stderr)
        sys.exit(1)

    raw_result = run_uv4(
        uv4_exe=uv4_exe,
        action=args.action,
        project=project,
        target=target or "",
        log_dir=log_dir,
        clean_first=args.clean_first,
    )
    elapsed_ms = (time.time() - started_ts) * 1000

    if raw_result["status"] == "error":
        result = make_result(
            status="error",
            action=args.action,
            summary=raw_result["error"]["message"],
            details=raw_result.get("details", {}),
            context=parameter_context(
                provider="keil",
                workspace=str(workspace),
                parameter_sources=parameter_sources,
            ),
            error=raw_result["error"],
            timing=make_timing(started_at, elapsed_ms),
        )
    else:
        details = raw_result["details"]
        artifacts = build_artifacts(
            axf_file=details.get("axf_file"),
            elf_file=details.get("elf_file"),
            hex_file=details.get("hex_file"),
            bin_file=details.get("bin_file"),
            flash_file=details.get("flash_file"),
            debug_file=details.get("debug_file"),
            output_dir=details.get("output_dir"),
            log_file=details.get("log_file"),
        )
        summary = _build_summary(args.action, raw_result["status"], raw_result["metrics"])
        state_info = None
        if args.action in ("build", "rebuild") and raw_result["status"] == "ok":
            state_info = update_state_entry(
                "last_build",
                {
                    "provider": "keil",
                    "action": args.action,
                    "project": project,
                    "target": target,
                    "log_dir": log_dir,
                    "artifacts": artifacts,
                    **artifacts,
                },
                str(workspace),
            )
        elif args.action == "flash" and raw_result["status"] == "ok":
            state_info = update_state_entry(
                "last_flash",
                {
                    "provider": "keil",
                    "action": args.action,
                    "project": project,
                    "target": target,
                    "artifacts": artifacts,
                    **artifacts,
                },
                str(workspace),
            )

        # 构建成功后，将确认过的参数写回工程级配置
        if raw_result["status"] == "ok":
            project_rel = _make_relative_to_workspace(workspace, project)
            save_project_config(
                str(workspace),
                {
                    "project": project_rel,
                    "target": target or "",
                    "log_dir": _make_relative_to_workspace(workspace, log_dir),
                },
            )

        result = make_result(
            status=raw_result["status"],
            action=args.action,
            summary=summary,
            details=details,
            context=parameter_context(
                provider="keil",
                workspace=str(workspace),
                parameter_sources=parameter_sources,
            ),
            artifacts=artifacts,
            metrics=raw_result["metrics"],
            state=state_info,
            next_actions=_next_actions(args.action, artifacts),
            timing=make_timing(started_at, elapsed_ms),
        )

    if args.as_json:
        output_json(result)
        return

    if result["status"] == "ok":
        print(f"[{args.action}] {result['summary']}")
        if result.get("artifacts", {}).get("log_file"):
            print(f"  日志: {result['artifacts']['log_file']}")
        if result.get("artifacts", {}).get("flash_file"):
            print(f"  Flash: {result['artifacts']['flash_file']}")
        if result.get("artifacts", {}).get("debug_file"):
            print(f"  Debug: {result['artifacts']['debug_file']}")
    else:
        error = result.get("error", {})
        print(f"[{args.action}] 失败 — {error.get('message', result['summary'])}", file=sys.stderr)
        if result.get("details", {}).get("log_file"):
            print(f"  日志: {result['details']['log_file']}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

"""Keil MDK 工程扫描与 Target 枚举"""

import argparse
import json
import os
import sys
import xml.etree.ElementTree as ET
from pathlib import Path


def scan_projects(root: str) -> list[dict]:
    """递归搜索 .uvprojx 和 .uvmpw 文件"""
    root_path = Path(root).resolve()
    projects = []
    for ext in ("*.uvprojx", "*.uvmpw"):
        for p in root_path.rglob(ext):
            projects.append({
                "path": str(p),
                "name": p.stem,
                "type": "workspace" if p.suffix == ".uvmpw" else "project",
            })
    projects.sort(key=lambda x: x["path"])
    return projects


def list_targets(project_path: str) -> list[dict]:
    """解析 .uvprojx 中的 TargetName"""
    p = Path(project_path).resolve()
    if not p.exists():
        raise FileNotFoundError(f"工程文件不存在: {p}")
    if p.suffix != ".uvprojx":
        raise ValueError(f"仅支持 .uvprojx 文件，当前: {p.suffix}")

    tree = ET.parse(str(p))
    root = tree.getroot()
    targets = []
    for target_el in root.iter("Target"):
        name_el = target_el.find("TargetName")
        if name_el is not None and name_el.text:
            targets.append({"name": name_el.text.strip()})
    return targets


def output_json(data: dict):
    sys.stdout.reconfigure(encoding="utf-8")
    print(json.dumps(data, ensure_ascii=False, indent=2))


def main():
    parser = argparse.ArgumentParser(description="Keil 工程扫描与 Target 枚举")
    sub = parser.add_subparsers(dest="command")

    scan_p = sub.add_parser("scan", help="搜索工程文件")
    scan_p.add_argument("--root", default=".", help="搜索根目录")
    scan_p.add_argument("--json", action="store_true", dest="as_json")

    targets_p = sub.add_parser("targets", help="枚举 Target")
    targets_p.add_argument("--project", required=True, help="工程文件路径")
    targets_p.add_argument("--json", action="store_true", dest="as_json")

    args = parser.parse_args()

    if args.command == "scan":
        projects = scan_projects(args.root)
        result = {
            "status": "ok",
            "action": "scan",
            "details": {"projects": projects, "count": len(projects)},
        }
        if args.as_json:
            output_json(result)
        else:
            if not projects:
                print("未找到 Keil 工程文件")
            else:
                print(f"找到 {len(projects)} 个工程：")
                for i, p in enumerate(projects, 1):
                    print(f"  {i}. [{p['type']}] {p['name']} — {p['path']}")

    elif args.command == "targets":
        try:
            targets = list_targets(args.project)
            result = {
                "status": "ok",
                "action": "targets",
                "details": {
                    "project": args.project,
                    "targets": targets,
                    "count": len(targets),
                },
            }
            if args.as_json:
                output_json(result)
            else:
                if not targets:
                    print("未找到 Target")
                else:
                    print(f"工程 {args.project} 包含 {len(targets)} 个 Target：")
                    for i, t in enumerate(targets, 1):
                        print(f"  {i}. {t['name']}")
        except (FileNotFoundError, ValueError) as e:
            result = {
                "status": "error",
                "action": "targets",
                "error": {"code": "invalid_project", "message": str(e)},
            }
            if args.as_json:
                output_json(result)
            else:
                print(f"错误: {e}", file=sys.stderr)
            sys.exit(1)

    else:
        parser.print_help()
        sys.exit(1)


if __name__ == "__main__":
    main()

"""keil skill 私有运行时工具。"""

from __future__ import annotations

import json
import os
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from shutil import which
from typing import Any


STATE_DIR_NAME = ".embeddedskills"
STATE_FILE_NAME = "state.json"
PROJECT_CONFIG_FILE_NAME = "config.json"

# Skill name for project config
SKILL_NAME = "keil"


def now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def default_config_path(script_file: str) -> Path:
    return Path(script_file).resolve().parents[1] / "config.json"


def load_local_config(script_file: str | None = None) -> dict:
    """加载 skill/config.json（环境级配置）
    
    路径：当前脚本所在 skill 目录下的 config.json
    """
    if script_file is None:
        # 获取调用者的文件路径
        import inspect
        frame = inspect.currentframe()
        if frame and frame.f_back:
            script_file = frame.f_back.f_globals.get("__file__", "")
        if not script_file:
            return {}
    config_path = default_config_path(script_file)
    return load_json_file(config_path)


def save_local_config(data: dict, script_file: str | None = None) -> Path | None:
    """保存环境级配置到 skill/config.json"""
    if script_file is None:
        import inspect
        frame = inspect.currentframe()
        if frame and frame.f_back:
            script_file = frame.f_back.f_globals.get("__file__", "")
        if not script_file:
            return None
    config_path = default_config_path(script_file)
    existing = load_json_file(config_path)
    existing.update(data)
    save_json_file(config_path, existing)
    return config_path


def load_project_config(workspace: str | None = None) -> dict:
    """从 workspace/.embeddedskills/config.json 读取本 skill 的工程级配置
    
    参数: workspace - 工作区路径，None 时使用 cwd
    返回: 该 skill 对应的配置字典（如 config["keil"] 或 config["gcc"]）
    """
    ws = workspace_root(workspace)
    config_file = ws / STATE_DIR_NAME / PROJECT_CONFIG_FILE_NAME
    data = load_json_file(config_file)
    return data.get(SKILL_NAME, {})


def save_project_config(workspace: str | None = None, values: dict | None = None) -> Path | None:
    """写回工程级配置到 workspace/.embeddedskills/config.json
    
    - 只更新本 skill 的配置部分，不覆盖其他 skill 的配置
    - 目录不存在时自动创建 .embeddedskills/
    """
    if values is None:
        values = {}
    ws = workspace_root(workspace)
    config_file = ws / STATE_DIR_NAME / PROJECT_CONFIG_FILE_NAME
    data = load_json_file(config_file)
    data[SKILL_NAME] = {**(data.get(SKILL_NAME, {})), **values}
    save_json_file(config_file, data)
    return config_file


def output_json(data: dict, *, indent: int = 2) -> None:
    sys.stdout.reconfigure(encoding="utf-8")
    print(json.dumps(data, ensure_ascii=False, indent=indent), flush=True)


def is_missing(value: Any) -> bool:
    return value is None or value == ""


def normalize_path(value: str | None) -> str:
    if is_missing(value):
        return ""
    return str(Path(str(value)).expanduser().resolve())


def normalize_path_with_base(value: str | None, base: str | Path | None = None) -> str:
    if is_missing(value):
        return ""
    path = Path(str(value)).expanduser()
    if base and not path.is_absolute():
        path = Path(base) / path
    return str(path.resolve())


def _serialize_state_value(value: Any, workspace: Path) -> Any:
    if isinstance(value, dict):
        return {key: _serialize_state_value(item, workspace) for key, item in value.items()}
    if isinstance(value, list):
        return [_serialize_state_value(item, workspace) for item in value]
    if not isinstance(value, str) or "://" in value:
        return value

    path = Path(value).expanduser()
    if not path.is_absolute():
        return value
    try:
        return Path(os.path.relpath(path.resolve(), workspace)).as_posix()
    except ValueError:
        return value


def hidden_subprocess_kwargs() -> dict:
    if sys.platform != "win32":
        return {}

    startupinfo = subprocess.STARTUPINFO()
    startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
    startupinfo.wShowWindow = getattr(subprocess, "SW_HIDE", 0)
    return {
        "creationflags": getattr(subprocess, "CREATE_NO_WINDOW", 0),
        "startupinfo": startupinfo,
    }


def load_json_file(path: str | Path) -> dict:
    file_path = Path(path)
    if not file_path.exists():
        return {}
    try:
        return json.loads(file_path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return {}


def save_json_file(path: str | Path, data: dict) -> None:
    file_path = Path(path)
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def workspace_root(workspace: str | None = None) -> Path:
    if not is_missing(workspace):
        return Path(str(workspace)).expanduser().resolve()
    return Path.cwd().resolve()


def load_workspace_state(workspace: str | None = None) -> dict:
    return load_json_file(workspace_root(workspace) / STATE_DIR_NAME / STATE_FILE_NAME)


def save_workspace_state(state: dict, workspace: str | None = None) -> Path:
    ws = workspace_root(workspace)
    file_path = ws / STATE_DIR_NAME / STATE_FILE_NAME
    save_json_file(file_path, _serialize_state_value(state, ws))
    return file_path


def get_state_entry(state: dict | None, key: str) -> dict:
    if not isinstance(state, dict):
        return {}
    value = state.get(key, {})
    return value if isinstance(value, dict) else {}


def update_state_entry(category: str, record: dict, workspace: str | None = None) -> dict:
    ws = workspace_root(workspace)
    state = load_workspace_state(workspace)
    state[category] = _serialize_state_value({**record, "timestamp": record.get("timestamp") or now_iso()}, ws)
    file_path = save_workspace_state(state, workspace)
    return {
        "workspace": str(ws),
        "file": str(file_path),
        "updated_keys": [category],
        category: state[category],
    }


def _first_resolved(mapping: dict, keys: list[str]) -> tuple[Any, str | None]:
    for key in keys:
        value = mapping.get(key)
        if not is_missing(value):
            return value, key
    return None, None


def _auto_detect_uv4() -> str:
    candidates = [
        which("UV4.exe"),
        which("UV4"),
        r"C:\Keil_v5\UV4\UV4.exe",
        r"C:\Keil_v5\ARM\UV4\UV4.exe",
    ]
    keil_root = os.environ.get("KEIL_ROOT", "")
    if keil_root:
        candidates.append(str(Path(keil_root) / "UV4" / "UV4.exe"))
    for candidate in candidates:
        if candidate and Path(candidate).is_file():
            return str(Path(candidate).resolve())
    return ""


def resolve_param(
    name: str,
    cli_value: Any,
    *,
    config: dict | None = None,
    config_keys: list[str] | None = None,
    state_record: dict | None = None,
    state_keys: list[str] | None = None,
    required: bool = False,
    normalize_as_path: bool = False,
    workspace: str | None = None,
) -> tuple[Any, str]:
    if not is_missing(cli_value):
        value = cli_value
        source = "cli"
    else:
        value = None
        source = ""
        if config and config_keys:
            value, config_key = _first_resolved(config, config_keys)
            if not is_missing(value):
                source = f"config:{config_key}"
        if is_missing(value) and state_record and state_keys:
            value, state_key = _first_resolved(state_record, state_keys)
            if not is_missing(value):
                source = f"state:{state_key}"
        if is_missing(value) and name == "uv4":
            value = _auto_detect_uv4()
            if not is_missing(value):
                source = "auto:uv4"
    if normalize_as_path and not is_missing(value):
        value = normalize_path_with_base(str(value), workspace_root(workspace))
    if required and is_missing(value):
        raise ValueError(f"缺少必要参数: {name}")
    return value, source


def compact_dict(data: dict | None) -> dict:
    if not isinstance(data, dict):
        return {}
    return {key: value for key, value in data.items() if value not in (None, "", [], {})}


def build_artifacts(**paths: str) -> dict:
    return {key: normalize_path(str(value)) for key, value in paths.items() if not is_missing(value)}


def make_result(
    *,
    status: str,
    action: str,
    summary: str,
    details: dict | None = None,
    context: dict | None = None,
    artifacts: dict | None = None,
    metrics: dict | None = None,
    state: dict | None = None,
    next_actions: list[str] | None = None,
    timing: dict | None = None,
    error: dict | None = None,
) -> dict:
    result = {"status": status, "action": action, "summary": summary, "details": compact_dict(details)}
    optional = {
        "context": compact_dict(context),
        "artifacts": compact_dict(artifacts),
        "metrics": compact_dict(metrics),
        "state": compact_dict(state),
        "timing": compact_dict(timing),
    }
    for key, value in optional.items():
        if value:
            result[key] = value
    if next_actions:
        result["next_actions"] = [item for item in next_actions if item]
    if error:
        result["error"] = compact_dict(error)
    return result


def make_timing(started_at: str, elapsed_ms: int | float) -> dict:
    return {"started_at": started_at, "finished_at": now_iso(), "elapsed_ms": int(elapsed_ms)}


def parameter_context(*, provider: str, workspace: str | None = None, parameter_sources: dict | None = None, config_path: str | None = None) -> dict:
    context = {"provider": provider, "workspace": str(workspace_root(workspace))}
    if parameter_sources:
        context["parameter_sources"] = compact_dict(parameter_sources)
    if not is_missing(config_path):
        context["config_path"] = normalize_path(str(config_path))
    return context