sync: HP_BOWEI 2026-09-09
This commit is contained in:
Myfro9
@@ -0,0 +1,337 @@
|
||||
"""ak_flow.py — 驱动 Altium→KiCad 迁移的 LangGraph 图（进程内，无 langgraph dev）。
|
||||
|
||||
图本体在 ~/skills_env.md 的 ALTIUM_KICAD_GRAPH_ROOT（app/main_graph.py），本脚本只做三件事：
|
||||
1) 用 SqliteSaver 打开持久化 checkpoint，按 thread_id 续跑；
|
||||
2) 把当前挂起的 interrupt 以结构化 JSON 打印给 Claude 会话；
|
||||
3) 把 Claude 执行完真实动作后的结论 Command(resume=...) 回填、推进到下一个 interrupt。
|
||||
|
||||
为什么不直接用工程里的 run.py：run.py 的 resume 只接受一个位置参数 JSON 字符串，
|
||||
在 PowerShell 里转义引号极易出错；且 thread 固定为 AK_THREAD 单值，多工程并发会串。
|
||||
本脚本用 --set key=value 免转义回填，thread 由 target_dir 推导，DB 沿用
|
||||
runs/skill_checkpoint.sqlite（已跑完的 ak-xmod-v1p0 / ak-gq-exct7g-v1 仍可续）。
|
||||
|
||||
退出码：0 = 图已结束(done)；2 = 停在 interrupt 等回填(waiting)；1 = 出错。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
def _env_lookup(key: str, default: str | None = None) -> str:
|
||||
"""环境变量优先，其次 ~/skills_env.md 的 `KEY = value` 行（机器相关参数的真源）。"""
|
||||
v = os.environ.get(key)
|
||||
if v:
|
||||
return v
|
||||
path = os.path.join(os.path.expanduser("~"), "skills_env.md")
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
for line in f:
|
||||
m = re.match(rf"\s*{key}\s*=\s*(\S+)", line)
|
||||
if m:
|
||||
return m.group(1)
|
||||
except OSError:
|
||||
pass
|
||||
if default is None:
|
||||
sys.exit(f"missing {key}: add it to ~/skills_env.md or set it as an env var")
|
||||
return default
|
||||
|
||||
|
||||
# 图工程根是机器相关的：真源是 ~/skills_env.md 的 ALTIUM_KICAD_GRAPH_ROOT。
|
||||
DEFAULT_PROJECT_ROOT = _env_lookup("ALTIUM_KICAD_GRAPH_ROOT", r"E:\LangGraph_Design\Altium-Kicad")
|
||||
DB_RELATIVE = ("runs", "skill_checkpoint.sqlite")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- utils
|
||||
def _out(*parts: str) -> None:
|
||||
print(*parts, flush=True)
|
||||
|
||||
|
||||
def _fatal(msg: str) -> "None":
|
||||
_out(f"[ERROR] {msg}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def slug_for(target_dir: str) -> str:
|
||||
"""E:\\Kicad_design\\XMOD_V1p0 -> ak-xmod-v1p0 (与既有线程命名一致)。"""
|
||||
name = Path(target_dir.rstrip("\\/")).name or "run"
|
||||
name = re.sub(r"[^0-9A-Za-z]+", "-", name).strip("-").lower()
|
||||
return f"ak-{name or 'run'}"
|
||||
|
||||
|
||||
def coerce(raw: str):
|
||||
"""--set 的值做最小类型推断：true/false→bool，纯数字→int，其余原样字符串。"""
|
||||
low = raw.strip().lower()
|
||||
if low in ("true", "yes", "1"):
|
||||
return True
|
||||
if low in ("false", "no", "0"):
|
||||
return False
|
||||
if re.fullmatch(r"-?\d+", raw.strip()):
|
||||
return int(raw.strip())
|
||||
return raw
|
||||
|
||||
|
||||
def pending_interrupt(snap):
|
||||
for task in snap.tasks:
|
||||
if task.interrupts:
|
||||
return task.interrupts[0].value
|
||||
return None
|
||||
|
||||
|
||||
def load_graph(project_root: Path):
|
||||
if not (project_root / "app" / "main_graph.py").is_file():
|
||||
_fatal(f"工程根不对：{project_root} 下没有 app\\main_graph.py。"
|
||||
f"用 --project-root 指定，设环境变量 AK_PROJECT_ROOT，或改 ~/skills_env.md 的 ALTIUM_KICAD_GRAPH_ROOT。")
|
||||
sys.path.insert(0, str(project_root))
|
||||
from app.main_graph import build_graph # noqa: E402
|
||||
return build_graph
|
||||
|
||||
|
||||
RETRY_LIMITS = {"bridge": 3, "import": 3, "netlist": 5, "gerber": 5}
|
||||
|
||||
|
||||
def init_state(source_prj: str, target_dir: str, limits: dict[str, int]) -> dict:
|
||||
state = {
|
||||
"source_prj": source_prj,
|
||||
"target_dir": target_dir,
|
||||
"migration_dir": str(Path(target_dir) / "_migration"),
|
||||
"history": [],
|
||||
}
|
||||
for name, default in RETRY_LIMITS.items():
|
||||
state[f"{name}_retry"] = 0
|
||||
state[f"max_{name}_retry"] = limits.get(name) or default
|
||||
return state
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- report
|
||||
def report(graph, thread: dict, before_len: int) -> int:
|
||||
"""打印当前状态；返回进程退出码。"""
|
||||
snap = graph.get_state(thread)
|
||||
values = snap.values or {}
|
||||
history = values.get("history", [])
|
||||
walked = [h.get("node") for h in history[before_len:]]
|
||||
if walked:
|
||||
_out("[NODES] " + " -> ".join(str(n) for n in walked))
|
||||
|
||||
_out(f"[THREAD] {thread['configurable']['thread_id']}")
|
||||
if values.get("source_prj"):
|
||||
_out(f"[PATHS] source_prj = {values['source_prj']}")
|
||||
_out(f"[PATHS] target_dir = {values['target_dir']}")
|
||||
_out(f"[PATHS] migration = {values.get('migration_dir', '')}")
|
||||
|
||||
retries = {k: values.get(k) for k in
|
||||
("bridge_retry", "import_retry", "netlist_retry", "gerber_retry")
|
||||
if values.get(k)}
|
||||
if retries:
|
||||
_out("[RETRY] " + json.dumps(retries))
|
||||
|
||||
if snap.next:
|
||||
payload = pending_interrupt(snap)
|
||||
_out(f"[WAITING] next node: {', '.join(snap.next)}")
|
||||
if payload is not None:
|
||||
_out("[INTERRUPT] " + json.dumps(payload, ensure_ascii=False, indent=2))
|
||||
else:
|
||||
_out("[INTERRUPT] (无 payload — 直接 resume 推进)")
|
||||
return 2
|
||||
|
||||
if any(h.get("node") == "safety_exit" for h in history):
|
||||
_out("[SAFETY-END] 某个判定回环超过重试上限，图已优雅退出。"
|
||||
"先修好卡住的那一步，再用 restart 重跑，或调高对应 max_*_retry。")
|
||||
return 0
|
||||
|
||||
_out("[DONE] 图已跑完。")
|
||||
_out(f"[REPORT] {values.get('report_path', '(未回填 report_path)')}")
|
||||
return 0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- commands
|
||||
def cmd_start(args, graph_builder, db_path: Path):
|
||||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||||
|
||||
source = args.source_prj
|
||||
target = args.target_dir
|
||||
if not source or not target:
|
||||
_fatal("start 需要 --source-prj 与 --target-dir（均为绝对路径，无默认值）。")
|
||||
if not Path(source).is_file():
|
||||
_fatal(f"源工程不存在：{source}（应指向 .PrjPCB 文件本身）")
|
||||
Path(target).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
thread_id = args.thread or slug_for(target)
|
||||
thread = {"configurable": {"thread_id": thread_id}}
|
||||
|
||||
with SqliteSaver.from_conn_string(str(db_path)) as cp:
|
||||
graph = graph_builder(checkpointer=cp)
|
||||
existing = graph.get_state(thread)
|
||||
if (existing.next or existing.values) and not args.restart:
|
||||
_out(f"[INFO] 线程 {thread_id} 已存在，续跑（不重置）。要从头开始请加 --restart。")
|
||||
before = len(existing.values.get("history", []))
|
||||
if existing.next and pending_interrupt(existing) is not None:
|
||||
# 已经停在 interrupt 上，什么都不推进，直接把 payload 亮出来
|
||||
return report(graph, thread, before)
|
||||
graph.invoke(None, thread)
|
||||
return report(graph, thread, before)
|
||||
|
||||
if args.restart:
|
||||
_out(f"[INFO] --restart：清空线程 {thread_id} 的 checkpoint。")
|
||||
wipe_thread(db_path, thread_id)
|
||||
|
||||
limits = {name: getattr(args, f"max_{name}_retry", None) for name in RETRY_LIMITS}
|
||||
_out(f"[INFO] 新建线程 {thread_id}")
|
||||
graph.invoke(init_state(source, target, limits), thread)
|
||||
return report(graph, thread, 0)
|
||||
|
||||
|
||||
def cmd_status(args, graph_builder, db_path: Path):
|
||||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||||
|
||||
thread_id = resolve_thread(args)
|
||||
thread = {"configurable": {"thread_id": thread_id}}
|
||||
with SqliteSaver.from_conn_string(str(db_path)) as cp:
|
||||
graph = graph_builder(checkpointer=cp)
|
||||
snap = graph.get_state(thread)
|
||||
if not snap.values and not snap.next:
|
||||
_fatal(f"线程 {thread_id} 没有任何 checkpoint。先跑 start。")
|
||||
return report(graph, thread, len(snap.values.get("history", [])))
|
||||
|
||||
|
||||
def cmd_resume(args, graph_builder, db_path: Path):
|
||||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||||
from langgraph.types import Command
|
||||
|
||||
payload: dict = {}
|
||||
if args.json:
|
||||
try:
|
||||
payload = json.loads(args.json)
|
||||
except json.JSONDecodeError as exc:
|
||||
_fatal(f"--json 不是合法 JSON：{exc}")
|
||||
for item in args.set or []:
|
||||
if "=" not in item:
|
||||
_fatal(f"--set 要写成 key=value，收到：{item}")
|
||||
key, raw = item.split("=", 1)
|
||||
payload[key.strip()] = coerce(raw)
|
||||
|
||||
if not payload:
|
||||
_fatal("resume 的回填不能为空——langgraph 会把 falsy 的 resume 当作『没提供』"
|
||||
"从而在同一个节点再次中断。至少给一个键，例如 --set acknowledged=true。")
|
||||
|
||||
thread_id = resolve_thread(args)
|
||||
thread = {"configurable": {"thread_id": thread_id}}
|
||||
with SqliteSaver.from_conn_string(str(db_path)) as cp:
|
||||
graph = graph_builder(checkpointer=cp)
|
||||
snap = graph.get_state(thread)
|
||||
if not snap.next:
|
||||
_fatal(f"线程 {thread_id} 没有挂起的 interrupt（图已结束或未启动）。先跑 status。")
|
||||
before = len(snap.values.get("history", []))
|
||||
_out("[RESUME] " + json.dumps(payload, ensure_ascii=False))
|
||||
graph.invoke(Command(resume=payload), thread)
|
||||
return report(graph, thread, before)
|
||||
|
||||
|
||||
def cmd_list(args, graph_builder, db_path: Path):
|
||||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||||
|
||||
if not db_path.is_file():
|
||||
_out("[INFO] 还没有任何 checkpoint 数据库。")
|
||||
return 0
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
try:
|
||||
threads = sorted({r[0] for r in conn.execute("select distinct thread_id from checkpoints")})
|
||||
finally:
|
||||
conn.close()
|
||||
if not threads:
|
||||
_out("[INFO] 数据库里没有线程。")
|
||||
return 0
|
||||
with SqliteSaver.from_conn_string(str(db_path)) as cp:
|
||||
graph = graph_builder(checkpointer=cp)
|
||||
for tid in threads:
|
||||
snap = graph.get_state({"configurable": {"thread_id": tid}})
|
||||
v = snap.values or {}
|
||||
state = f"waiting@{','.join(snap.next)}" if snap.next else "done"
|
||||
_out(f"[THREAD] {tid:<24} {state:<28} target={v.get('target_dir', '?')}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_reset(args, graph_builder, db_path: Path):
|
||||
thread_id = resolve_thread(args)
|
||||
n = wipe_thread(db_path, thread_id)
|
||||
_out(f"[INFO] 已删除线程 {thread_id} 的 {n} 条 checkpoint 记录。")
|
||||
return 0
|
||||
|
||||
|
||||
def wipe_thread(db_path: Path, thread_id: str) -> int:
|
||||
if not db_path.is_file():
|
||||
return 0
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
try:
|
||||
n = conn.execute("delete from checkpoints where thread_id=?", (thread_id,)).rowcount
|
||||
conn.execute("delete from writes where thread_id=?", (thread_id,))
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
return n
|
||||
|
||||
|
||||
def resolve_thread(args) -> str:
|
||||
if getattr(args, "thread", None):
|
||||
return args.thread
|
||||
if getattr(args, "target_dir", None):
|
||||
return slug_for(args.target_dir)
|
||||
_fatal("需要 --thread <id> 或 --target-dir <目标 KiCad 目录> 来定位线程。用 list 看所有线程。")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- main
|
||||
def main() -> int:
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
ap = argparse.ArgumentParser(description="Drive the Altium->KiCad migration LangGraph flow.")
|
||||
ap.add_argument("--project-root", default=os.environ.get("AK_PROJECT_ROOT", DEFAULT_PROJECT_ROOT))
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
p = sub.add_parser("start", help="新建或续跑一次迁移")
|
||||
p.add_argument("--source-prj", help="源 Altium .PrjPCB 绝对路径")
|
||||
p.add_argument("--target-dir", help="目标 KiCad 工程目录绝对路径")
|
||||
p.add_argument("--thread", help="线程 id（默认由 target-dir 推导）")
|
||||
p.add_argument("--restart", action="store_true", help="清空该线程重新开始")
|
||||
for name, default in RETRY_LIMITS.items():
|
||||
p.add_argument(f"--max-{name}-retry", type=int,
|
||||
help=f"{name} 回环的重试上限（默认 {default}，超限走 safety_end）")
|
||||
p.set_defaults(func=cmd_start)
|
||||
|
||||
p = sub.add_parser("status", help="看当前停在哪个 interrupt")
|
||||
p.add_argument("--thread")
|
||||
p.add_argument("--target-dir")
|
||||
p.set_defaults(func=cmd_status)
|
||||
|
||||
p = sub.add_parser("resume", help="回填结论并推进到下一个 interrupt")
|
||||
p.add_argument("--thread")
|
||||
p.add_argument("--target-dir")
|
||||
p.add_argument("--set", action="append", metavar="KEY=VALUE",
|
||||
help="免转义回填，可重复；true/false→bool，纯数字→int")
|
||||
p.add_argument("--json", help="直接给 JSON 对象（与 --set 可混用）")
|
||||
p.set_defaults(func=cmd_resume)
|
||||
|
||||
p = sub.add_parser("list", help="列出所有迁移线程")
|
||||
p.set_defaults(func=cmd_list)
|
||||
|
||||
p = sub.add_parser("reset", help="删除某个线程的 checkpoint")
|
||||
p.add_argument("--thread")
|
||||
p.add_argument("--target-dir")
|
||||
p.set_defaults(func=cmd_reset)
|
||||
|
||||
args = ap.parse_args()
|
||||
root = Path(args.project_root)
|
||||
db_path = root.joinpath(*DB_RELATIVE)
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
graph_builder = load_graph(root)
|
||||
return args.func(args, graph_builder, db_path)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,200 @@
|
||||
"""cmp_vec.py — Altium 与 KiCad 的离散版图几何等价判定（阶段 3 的决定性证据）。
|
||||
|
||||
为什么这个比栅格 Gerber XOR 更权威：跨工具的 Gerber 不可能像素级一致
|
||||
（aperture / 坐标格式 / 填充算法 / 字体 / 负片都不同），栅格 XOR 只能给出
|
||||
「像什么程度」。而走线/过孔/焊盘是离散图元，可以逐条逐个匹配 —— 匹配上了
|
||||
就是同一块板，匹配不上就是真差异，没有解释空间。
|
||||
|
||||
坐标变换只有两个自由参数，不是曲线拟合：
|
||||
kx = s*ax + tx
|
||||
ky = -s*ay + ty (Altium y 向上，KiCad y 向下)
|
||||
s 由单位固定（Altium API 返回整数 mil -> mm 就是 0.0254），(tx,ty) 由两侧过孔
|
||||
质心解出。用 2 个参数去对上几千个过孔+焊盘+走线，对上了就是实证。
|
||||
|
||||
容差默认 30um：eda-agent 读出的 Altium 坐标是**整数 mil**，读数精度本身就有
|
||||
±12.7um，两端各一次就接近 25um。KiCad 侧是精确的（导入器直接解析二进制 PcbDoc）。
|
||||
|
||||
用法:
|
||||
python cmp_vec.py --altium-vec .../altium_vec --kicad-vec .../kicad_vec \\
|
||||
--layer-map .../layer_map.json [--tol-mm 0.030] [--report r.json]
|
||||
退出码 0 = 全部图元匹配；1 = 有未匹配图元。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import collections
|
||||
import json
|
||||
import math
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def load(path: Path, *containers: str):
|
||||
"""容忍两种外壳：裸 list，或 {"vias": [...]} / {"objects": [...]}。"""
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
if isinstance(data, dict):
|
||||
for key in containers + ("objects", "vias", "items"):
|
||||
if key in data and isinstance(data[key], list):
|
||||
return data[key]
|
||||
raise SystemExit(f"[ERROR] {path} 是 dict 但找不到列表字段（试过 {containers + ('objects',)}）")
|
||||
return data
|
||||
|
||||
|
||||
def num(v) -> float:
|
||||
return float(v)
|
||||
|
||||
|
||||
def match(a_pts, k_pts, tol, label, out_lines):
|
||||
"""网格加速的最近邻一对一匹配。返回 (未匹配的A, 未匹配的K)。"""
|
||||
if not a_pts and not k_pts:
|
||||
return 0, 0
|
||||
cell = max(tol * 4, 0.05)
|
||||
grid = collections.defaultdict(list)
|
||||
for i, p in enumerate(k_pts):
|
||||
grid[(int(p[0] / cell), int(p[1] / cell))].append(i)
|
||||
used, res, unmatched = set(), [], 0
|
||||
for p in a_pts:
|
||||
gx, gy = int(p[0] / cell), int(p[1] / cell)
|
||||
best, bd = None, 1e9
|
||||
for dx in (-1, 0, 1):
|
||||
for dy in (-1, 0, 1):
|
||||
for i in grid.get((gx + dx, gy + dy), ()):
|
||||
if i in used:
|
||||
continue
|
||||
d = math.hypot(k_pts[i][0] - p[0], k_pts[i][1] - p[1])
|
||||
if d < bd:
|
||||
bd, best = d, i
|
||||
if best is None or bd > tol:
|
||||
unmatched += 1
|
||||
else:
|
||||
used.add(best)
|
||||
res.append(bd)
|
||||
line = (f"{label}: altium={len(a_pts)} kicad={len(k_pts)} matched={len(res)} "
|
||||
f"unmatched_altium={unmatched} unmatched_kicad={len(k_pts) - len(used)}")
|
||||
if res:
|
||||
res.sort()
|
||||
line += (f"\n residual mm: max={res[-1]:.6f} "
|
||||
f"p99={res[min(int(len(res) * 0.99), len(res) - 1)]:.6f} "
|
||||
f"median={res[len(res) // 2]:.6f}")
|
||||
print("\n" + line)
|
||||
out_lines.append(line)
|
||||
return unmatched, len(k_pts) - len(used)
|
||||
|
||||
|
||||
def suggest_layer_map(a_tracks, k_tracks):
|
||||
"""没给 layer_map 时，按每层走线条数给个候选映射，供人核对。"""
|
||||
a_cnt = collections.Counter(str(t.get("Layer", t.get("layer"))) for t in a_tracks)
|
||||
k_cnt = collections.Counter(t["layer"] for t in k_tracks)
|
||||
print("\n[layer] Altium 各层走线数: " + json.dumps(a_cnt.most_common(), ensure_ascii=False))
|
||||
print("[layer] KiCad 各层走线数: " + json.dumps(k_cnt.most_common(), ensure_ascii=False))
|
||||
print("[layer] 没有 --layer-map。请对照上面两组计数写一个 layer_map.json，例如\n"
|
||||
' {"TopLayer": "F.Cu", "MidLayer1": "In2.Cu", "BottomLayer": "B.Cu"}\n'
|
||||
" 注意层序按叠层走，不要想当然 MidLayerN -> InN.Cu。")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="Vector geometry equivalence: Altium vs KiCad")
|
||||
ap.add_argument("--altium-vec", required=True)
|
||||
ap.add_argument("--kicad-vec", required=True)
|
||||
ap.add_argument("--layer-map", help="{Altium层名: KiCad层名} 的 JSON；不给则只比过孔和焊盘并给出建议")
|
||||
ap.add_argument("--tol-mm", type=float, default=0.030)
|
||||
ap.add_argument("--scale-to-mm", type=float,
|
||||
help="Altium 原始单位 -> mm。默认读 altium_vec/meta.json，再默认 0.0254(mil)")
|
||||
ap.add_argument("--report", help="把结论写成 JSON")
|
||||
args = ap.parse_args()
|
||||
|
||||
av, kv = Path(args.altium_vec), Path(args.kicad_vec)
|
||||
scale = args.scale_to_mm
|
||||
if scale is None:
|
||||
meta_path = av / "meta.json"
|
||||
if meta_path.is_file():
|
||||
scale = json.loads(meta_path.read_text(encoding="utf-8")).get("scale_to_mm")
|
||||
scale = scale or 0.0254
|
||||
print(f"scale_to_mm : {scale} tol_mm : {args.tol_mm}")
|
||||
|
||||
a_vias = load(av / "vias.json", "vias")
|
||||
k_vias = load(kv / "vias.json", "vias")
|
||||
print(f"vias: altium={len(a_vias)} kicad={len(k_vias)}")
|
||||
if not a_vias or not k_vias:
|
||||
raise SystemExit("[ERROR] 两侧都必须有过孔才能标定平移（过孔是最干净的基准）。")
|
||||
if len(a_vias) != len(k_vias):
|
||||
print("[WARN] 两侧过孔数不同 —— 质心标定会被拉偏。先查清是导入丢了过孔，"
|
||||
"还是 Altium 侧导出漏了层，再比。")
|
||||
|
||||
ax = sum(num(v["x"]) * scale for v in a_vias) / len(a_vias)
|
||||
ay = sum(num(v["y"]) * scale for v in a_vias) / len(a_vias)
|
||||
kx = sum(v["x"] for v in k_vias) / len(k_vias)
|
||||
ky = sum(v["y"] for v in k_vias) / len(k_vias)
|
||||
TX, TY = kx - ax, ky + ay
|
||||
print(f"transform : kx = ax*{scale} + {TX:.6f} ky = -ay*{scale} + {TY:.6f}")
|
||||
|
||||
def T(o, kx_="x", ky_="y"):
|
||||
return (num(o[kx_]) * scale + TX, TY - num(o[ky_]) * scale)
|
||||
|
||||
lines: list[str] = []
|
||||
bad = 0
|
||||
skipped_tracks = False
|
||||
bad += sum(match([T(v) for v in a_vias],
|
||||
[(v["x"], v["y"]) for v in k_vias], args.tol_mm, "VIAS", lines))
|
||||
|
||||
a_pads_p, k_pads_p = av / "pads.json", kv / "pads.json"
|
||||
if a_pads_p.is_file() and k_pads_p.is_file():
|
||||
a_pads, k_pads = load(a_pads_p), load(k_pads_p)
|
||||
bad += sum(match([T(p, "X", "Y") for p in a_pads],
|
||||
[(p["x"], p["y"]) for p in k_pads], args.tol_mm, "PADS", lines))
|
||||
|
||||
a_tr_p, k_tr_p = av / "tracks.json", kv / "tracks.json"
|
||||
if a_tr_p.is_file() and k_tr_p.is_file():
|
||||
a_tr, k_tr = load(a_tr_p), load(k_tr_p)
|
||||
if not args.layer_map:
|
||||
skipped_tracks = True
|
||||
suggest_layer_map(a_tr, k_tr)
|
||||
else:
|
||||
laymap = json.loads(Path(args.layer_map).read_text(encoding="utf-8"))
|
||||
a_cu = [t for t in a_tr if str(t.get("Layer")) in laymap]
|
||||
print(f"\naltium tracks total={len(a_tr)} on mapped copper={len(a_cu)} "
|
||||
f"kicad copper tracks={len(k_tr)}")
|
||||
|
||||
def mid_a(t):
|
||||
p1, p2 = T(t, "X1", "Y1"), T(t, "X2", "Y2")
|
||||
return ((p1[0] + p2[0]) / 2, (p1[1] + p2[1]) / 2)
|
||||
|
||||
for alay, klay in sorted(laymap.items()):
|
||||
# 逐层比，否则相邻层上位置重合的走线会互相错配
|
||||
aa = [mid_a(t) for t in a_cu if str(t.get("Layer")) == alay]
|
||||
kk = [((t["x1"] + t["x2"]) / 2, (t["y1"] + t["y2"]) / 2)
|
||||
for t in k_tr if t["layer"] == klay]
|
||||
if not aa and not kk:
|
||||
continue
|
||||
bad += sum(match(aa, kk, args.tol_mm, f"TRACKS {alay} -> {klay}", lines))
|
||||
|
||||
equivalent = bad == 0 and not skipped_tracks
|
||||
print("\n" + "=" * 60)
|
||||
print(f"TOTAL unmatched primitives: {bad}"
|
||||
+ (" (走线未比对)" if skipped_tracks else ""))
|
||||
print("=" * 60)
|
||||
if skipped_tracks:
|
||||
print("[INCOMPLETE] 没给 --layer-map，走线整个跳过了 —— 上面这个 0 只覆盖过孔和焊盘，"
|
||||
"不能当作等价结论。按上面的层计数写好 layer_map.json 再跑一次。")
|
||||
elif equivalent:
|
||||
print("离散几何 100% 等价 —— 走线/过孔/焊盘逐条对上。铺铜纹理、丝印字形、"
|
||||
"负片内电层的差异是工具固有项，不计入本判定（理由见 references\\playbook.md）。")
|
||||
|
||||
if args.report:
|
||||
Path(args.report).write_text(json.dumps({
|
||||
"unmatched_total": bad, "equivalent": equivalent,
|
||||
"tracks_compared": not skipped_tracks,
|
||||
"scale_to_mm": scale, "tol_mm": args.tol_mm,
|
||||
"transform": {"tx": TX, "ty": TY},
|
||||
"details": lines,
|
||||
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(f"[report] {args.report}")
|
||||
return 0 if equivalent else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,100 @@
|
||||
"""dump_kicad_vec.py — 把 KiCad 板的离散几何导成 JSON，供 cmp_vec.py 比对。必须用 KiCad python 跑。
|
||||
|
||||
输出（单位一律 mm，KiCad 原生坐标系：y 向下）:
|
||||
vias.json [{x, y, drill, top, bot, net}]
|
||||
pads.json [{x, y, layer, net, ref, pad, drill}]
|
||||
tracks.json [{x1, y1, x2, y2, width, layer, net}]
|
||||
meta.json {counts, layers}
|
||||
|
||||
用法（KiCad python 路径见 ~/skills_env.md 的 KICAD_PYTHON）:
|
||||
D:\\KiCad\\10.0\\bin\\python.exe dump_kicad_vec.py \\
|
||||
--board E:\\Kicad_design\\FOO\\FOO.kicad_pcb \\
|
||||
--out E:\\Kicad_design\\FOO\\_migration\\gerber\\kicad_vec
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="Dump KiCad board discrete geometry to JSON (mm)")
|
||||
ap.add_argument("--board", required=True)
|
||||
ap.add_argument("--out", required=True)
|
||||
args = ap.parse_args()
|
||||
|
||||
try:
|
||||
import pcbnew
|
||||
except ImportError:
|
||||
print("[ERROR] 导不到 pcbnew —— 用 KiCad 自带 python 跑（~\\skills_env.md 的 KICAD_PYTHON）。")
|
||||
return 1
|
||||
|
||||
out = Path(args.out)
|
||||
out.mkdir(parents=True, exist_ok=True)
|
||||
board = pcbnew.LoadBoard(args.board)
|
||||
mm = pcbnew.ToMM
|
||||
|
||||
def via_size(v) -> float:
|
||||
"""KiCad 10 起 PCB_VIA::GetWidth() 必须带层参数，不带会断言中止进程。
|
||||
老版本又只接受无参形式，所以两种都试。size 只是给人看的，取不到就算了。"""
|
||||
for call in (lambda: v.GetWidth(v.TopLayer()), lambda: v.GetWidth()):
|
||||
try:
|
||||
return mm(call())
|
||||
except Exception:
|
||||
continue
|
||||
return 0.0
|
||||
|
||||
vias, tracks = [], []
|
||||
for t in board.GetTracks():
|
||||
if isinstance(t, pcbnew.PCB_VIA):
|
||||
vias.append({
|
||||
"x": mm(t.GetPosition().x), "y": mm(t.GetPosition().y),
|
||||
"drill": mm(t.GetDrill()), "size": via_size(t),
|
||||
"top": board.GetLayerName(t.TopLayer()),
|
||||
"bot": board.GetLayerName(t.BottomLayer()),
|
||||
"net": t.GetNetname(),
|
||||
})
|
||||
else:
|
||||
tracks.append({
|
||||
"x1": mm(t.GetStart().x), "y1": mm(t.GetStart().y),
|
||||
"x2": mm(t.GetEnd().x), "y2": mm(t.GetEnd().y),
|
||||
"width": mm(t.GetWidth()),
|
||||
"layer": board.GetLayerName(t.GetLayer()),
|
||||
"net": t.GetNetname(),
|
||||
})
|
||||
|
||||
pads = []
|
||||
for fp in board.GetFootprints():
|
||||
ref = fp.GetReference()
|
||||
for p in fp.Pads():
|
||||
pads.append({
|
||||
"x": mm(p.GetPosition().x), "y": mm(p.GetPosition().y),
|
||||
"layer": board.GetLayerName(p.GetLayer()),
|
||||
"net": p.GetNetname(), "ref": ref, "pad": p.GetNumber(),
|
||||
"drill": mm(p.GetDrillSizeX()),
|
||||
})
|
||||
|
||||
layers = sorted({t["layer"] for t in tracks})
|
||||
for name, data in (("vias", vias), ("tracks", tracks), ("pads", pads)):
|
||||
(out / f"{name}.json").write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
||||
(out / "meta.json").write_text(json.dumps({
|
||||
"units": "mm", "y_axis": "down (KiCad native)",
|
||||
"counts": {"vias": len(vias), "tracks": len(tracks), "pads": len(pads)},
|
||||
"copper_layers_with_tracks": layers,
|
||||
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
|
||||
print(f"[dump] vias={len(vias)} tracks={len(tracks)} pads={len(pads)} -> {out}")
|
||||
print(f"[layers] 有走线的铜层：{', '.join(layers)}")
|
||||
print("[next] Altium 侧用 eda-agent 导同名 JSON 到 altium_vec/（格式见 "
|
||||
"references\\vector-compare.md），然后跑 cmp_vec.py。")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,183 @@
|
||||
"""fix_pcb.py — 导入后 PCB 修复（阶段 1.5 的两个硬骨头）。必须用 KiCad 自带 python 跑。
|
||||
|
||||
A) BGA / 平面焊盘丢网络回填
|
||||
Altium 用负片内电层隐式连接电源地，KiCad 导入时这些焊盘拿不到网络，
|
||||
表现为成百上千个 "GND vs 空网络" 假短路。用阶段 0 导出的 Altium 网表真值
|
||||
(component,pin,net) 把空网络焊盘重新指派回去。
|
||||
|
||||
B) 覆铜 zone 的本地 clearance
|
||||
Altium 导入器把【每块 zone 的本地 Clearance 写成 KiCad 默认 0.5mm】
|
||||
((zone ... (connect_pads (clearance 0.5))))，而本地值会盖过网络类 clearance
|
||||
—— 这就是导入后异 net 避让间隔明显偏大的根因。只改 .kicad_pro 的网络类
|
||||
或板级规则没有用，必须逐块 zone 改。目标值取 Altium 的 InPolygon 规则
|
||||
（XMOD/GQ 两个工程实测都是 0.254mm，但别照抄，去基线 pcb_design_rules.json 里查）。
|
||||
热焊盘 thermal_gap 通常已经对上 Altium PolygonConnect AirGap(0.165mm)，不要动。
|
||||
|
||||
本脚本改完不 refill。大板上 SWIG refill 容易超时甚至段错误，去 KiCad GUI 里按 B
|
||||
填充，或走 IPC 后端。
|
||||
|
||||
用法（KICAD_PYTHON 见 ~\\skills_env.md）:
|
||||
# 先干跑看看要动多少
|
||||
D:\\KiCad\\10.0\\bin\\python.exe fix_pcb.py --board X.kicad_pcb --netlist-csv t.csv \\
|
||||
--zone-clearance-mm 0.254 --dry-run
|
||||
# 确认后真改（会先备份 .kicad_pcb 到 <board>.bak_fixpcb）
|
||||
D:\\KiCad\\10.0\\bin\\python.exe fix_pcb.py --board X.kicad_pcb --netlist-csv t.csv \\
|
||||
--zone-clearance-mm 0.254
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import shutil
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
COMP_KEYS = ("component", "designator", "ref", "reference")
|
||||
PIN_KEYS = ("pin", "pin_number", "pad", "pin designator")
|
||||
NET_KEYS = ("net", "netname", "net name")
|
||||
|
||||
|
||||
def pick(fieldnames, candidates) -> str | None:
|
||||
lowered = {(f or "").strip().lower(): f for f in fieldnames}
|
||||
for c in candidates:
|
||||
if c in lowered:
|
||||
return lowered[c]
|
||||
return None
|
||||
|
||||
|
||||
def load_truth(csv_path: Path) -> dict[tuple[str, str], str]:
|
||||
"""Altium 网表真值 (component, pin) -> net。
|
||||
|
||||
Altium 的 tabular 导出会把隐藏的电源/平面引脚重复列成 net='?' 的行，
|
||||
这些行没有信息量，直接丢掉。
|
||||
"""
|
||||
with csv_path.open(newline="", encoding="utf-8-sig") as f:
|
||||
reader = csv.DictReader(f)
|
||||
c_key = pick(reader.fieldnames or [], COMP_KEYS)
|
||||
p_key = pick(reader.fieldnames or [], PIN_KEYS)
|
||||
n_key = pick(reader.fieldnames or [], NET_KEYS)
|
||||
if not (c_key and p_key and n_key):
|
||||
raise SystemExit(
|
||||
f"[ERROR] 认不出 CSV 的列名：{reader.fieldnames}\n"
|
||||
f" 需要 component / pin / net 三列（大小写和别名可容忍）。")
|
||||
truth: dict[tuple[str, str], str] = {}
|
||||
dropped = 0
|
||||
for row in reader:
|
||||
comp = (row.get(c_key) or "").strip()
|
||||
pin = (row.get(p_key) or "").strip()
|
||||
net = (row.get(n_key) or "").strip()
|
||||
if not (comp and pin):
|
||||
continue
|
||||
if not net or net == "?":
|
||||
dropped += 1
|
||||
continue
|
||||
truth[(comp, pin)] = net
|
||||
print(f"[csv] 真值引脚 {len(truth)} 条（丢弃 net='?'/空 的重复行 {dropped} 条）")
|
||||
return truth
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="Post-import KiCad PCB repair (nets + zone clearance)")
|
||||
ap.add_argument("--board", required=True, help=".kicad_pcb 路径")
|
||||
ap.add_argument("--netlist-csv", required=True,
|
||||
help="Altium 侧网表真值 CSV（阶段0 proj_export_netlist tabular）")
|
||||
ap.add_argument("--zone-clearance-mm", type=float, required=True,
|
||||
help="Altium InPolygon 规则的 polygon clearance，去基线 pcb_design_rules.json 查")
|
||||
ap.add_argument("--dry-run", action="store_true", help="只统计不落盘")
|
||||
ap.add_argument("--report", help="把统计写成 JSON")
|
||||
ap.add_argument("--skip-nets", action="store_true", help="只改 zone clearance，不动焊盘网络")
|
||||
ap.add_argument("--skip-zones", action="store_true", help="只回填焊盘网络，不动 zone clearance")
|
||||
args = ap.parse_args()
|
||||
|
||||
try:
|
||||
import pcbnew
|
||||
except ImportError:
|
||||
print("[ERROR] 导不到 pcbnew —— 本脚本必须用 KiCad 自带的 python 跑，"
|
||||
"例如 D:\\KiCad\\10.0\\bin\\python.exe（见 ~\\skills_env.md 的 KICAD_PYTHON）。")
|
||||
return 1
|
||||
|
||||
board_path = Path(args.board)
|
||||
if not board_path.is_file():
|
||||
print(f"[ERROR] 板文件不存在：{board_path}")
|
||||
return 1
|
||||
|
||||
truth = load_truth(Path(args.netlist_csv)) if not args.skip_nets else {}
|
||||
board = pcbnew.LoadBoard(str(board_path))
|
||||
|
||||
stats = {"assigned": 0, "no_truth": 0, "no_ref": 0, "already": 0,
|
||||
"zones_fixed": 0, "zones_skipped": 0}
|
||||
missing_examples: list[str] = []
|
||||
|
||||
# ---- A) 焊盘网络回填 ----
|
||||
if not args.skip_nets:
|
||||
def get_net(name):
|
||||
ni = board.FindNet(name)
|
||||
if ni is None:
|
||||
ni = pcbnew.NETINFO_ITEM(board, name)
|
||||
board.Add(ni)
|
||||
return ni
|
||||
|
||||
for fp in board.GetFootprints():
|
||||
ref = fp.GetReference()
|
||||
for pad in fp.Pads():
|
||||
if pad.GetNetCode() != 0 and pad.GetNetname() != "":
|
||||
stats["already"] += 1
|
||||
continue
|
||||
if not ref:
|
||||
stats["no_ref"] += 1
|
||||
continue
|
||||
net = truth.get((ref, pad.GetNumber()))
|
||||
if not net:
|
||||
stats["no_truth"] += 1
|
||||
if len(missing_examples) < 20:
|
||||
missing_examples.append(f"{ref}.{pad.GetNumber()}")
|
||||
continue
|
||||
if not args.dry_run:
|
||||
pad.SetNet(get_net(net))
|
||||
stats["assigned"] += 1
|
||||
print(f"[nets] 待回填={stats['assigned']} 真值里查不到={stats['no_truth']} "
|
||||
f"无位号={stats['no_ref']} 本来就有网络={stats['already']}")
|
||||
if missing_examples:
|
||||
print(f"[nets] 查不到真值的空网络焊盘举例：{', '.join(missing_examples)}")
|
||||
print(" 多为机械/屏蔽/安装孔焊盘。确认属实的话，按需单独指派 GND。")
|
||||
|
||||
# ---- B) zone 本地 clearance ----
|
||||
if not args.skip_zones:
|
||||
clr = pcbnew.FromMM(args.zone_clearance_mm)
|
||||
for z in board.Zones():
|
||||
if z.GetIsRuleArea() or not z.IsOnCopperLayer():
|
||||
stats["zones_skipped"] += 1
|
||||
continue
|
||||
if not args.dry_run:
|
||||
z.SetLocalClearance(clr)
|
||||
stats["zones_fixed"] += 1
|
||||
print(f"[zones] 铺铜 zone 本地 clearance -> {args.zone_clearance_mm}mm "
|
||||
f"共 {stats['zones_fixed']} 块；跳过 {stats['zones_skipped']} 块（规则区/非铜层）")
|
||||
|
||||
if args.dry_run:
|
||||
print("[dry-run] 没有写盘。去掉 --dry-run 才会真正修改。")
|
||||
else:
|
||||
backup = board_path.with_suffix(board_path.suffix + ".bak_fixpcb")
|
||||
shutil.copy2(board_path, backup)
|
||||
board.Save(str(board_path))
|
||||
print(f"[save] 已写回 {board_path}（原文件备份为 {backup.name}）")
|
||||
print("[next] 还要重填铺铜：KiCad GUI 里按 B，或走 IPC 后端。"
|
||||
"大板不要用 SWIG refill，会超时/段错误。")
|
||||
|
||||
if args.report:
|
||||
Path(args.report).write_text(
|
||||
json.dumps({**stats, "zone_clearance_mm": args.zone_clearance_mm,
|
||||
"dry_run": args.dry_run,
|
||||
"no_truth_examples": missing_examples},
|
||||
ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
print(f"[report] {args.report}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,223 @@
|
||||
"""netcmp.py — 原理图网表比对（阶段 2）。判定「电气连接是否与 Altium 完全一致」。
|
||||
|
||||
比对方法（两次迁移验证过，别改成比网络名）：
|
||||
两个工具的网络命名规则不同，逐名比对必然一片红。真正要比的是 **net partition**：
|
||||
把每份网表归约成「每个引脚 -> 它所在网络的全部节点集合」，两侧的等价类集合相等
|
||||
即电气连接一致。
|
||||
|
||||
三个必须做的归一化，少一个就会出假差异：
|
||||
1. KiCad 网名带前导 '/'（层次路径），Altium 不带 -> 剥掉前导 '/'。
|
||||
2. Altium tabular 导出把隐藏的电源/平面引脚重复列成 net='?' 的行 -> 去掉。
|
||||
3. KiCad 扁平导入后，整体导出网表只能拿到根图的一小部分 -> 必须**逐页**
|
||||
`kicad-cli sch export netlist --format kicadxml` 再按网名合并。
|
||||
（格式必须是 kicadxml；PADS 格式会漏掉单节点网。）
|
||||
|
||||
用法:
|
||||
python netcmp.py --kicad-dir E:\\Kicad_design\\FOO \\
|
||||
--netlist-csv E:\\Kicad_design\\FOO\\_migration\\reference\\netlist_tabular.csv \\
|
||||
--out E:\\Kicad_design\\FOO\\_migration\\netcmp
|
||||
退出码 0 = 100% 一致；1 = 有差异（详情看 diff.json / 终端前 30 条）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import collections
|
||||
import csv
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import xml.etree.ElementTree as ET
|
||||
from pathlib import Path
|
||||
|
||||
COMP_KEYS = ("component", "designator", "ref", "reference")
|
||||
PIN_KEYS = ("pin", "pin_number", "pad", "pin designator")
|
||||
NET_KEYS = ("net", "netname", "net name")
|
||||
|
||||
|
||||
def pick(fieldnames, candidates):
|
||||
lowered = {(f or "").strip().lower(): f for f in fieldnames}
|
||||
for c in candidates:
|
||||
if c in lowered:
|
||||
return lowered[c]
|
||||
return None
|
||||
|
||||
|
||||
def norm_net(name: str | None) -> str | None:
|
||||
"""剥掉 KiCad 层次网名的前导 '/'，让两侧命名可比。"""
|
||||
if name is None:
|
||||
return None
|
||||
return name.strip().lstrip("/")
|
||||
|
||||
|
||||
def _env_lookup(key: str, default: str | None = None) -> str:
|
||||
"""环境变量优先，其次 ~/skills_env.md 的 `KEY = value` 行（机器相关参数的真源）。"""
|
||||
v = os.environ.get(key)
|
||||
if v:
|
||||
return v
|
||||
path = os.path.join(os.path.expanduser("~"), "skills_env.md")
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
for line in f:
|
||||
m = re.match(rf"\s*{key}\s*=\s*(\S+)", line)
|
||||
if m:
|
||||
return m.group(1)
|
||||
except OSError:
|
||||
pass
|
||||
if default is None:
|
||||
sys.exit(f"missing {key}: add it to ~/skills_env.md or set it as an env var")
|
||||
return default
|
||||
|
||||
|
||||
def find_kicad_cli(explicit: str | None) -> str:
|
||||
if explicit:
|
||||
return explicit
|
||||
# 安装路径是机器相关的：真源是 ~/skills_env.md 的 KICAD_ROOT。
|
||||
kicad_root = _env_lookup("KICAD_ROOT", "")
|
||||
for cand in (os.environ.get("KICAD_CLI"),
|
||||
os.path.join(kicad_root, "bin", "kicad-cli.exe") if kicad_root else None,
|
||||
"kicad-cli"):
|
||||
if not cand:
|
||||
continue
|
||||
if cand == "kicad-cli" or Path(cand).is_file():
|
||||
return cand
|
||||
return "kicad-cli"
|
||||
|
||||
|
||||
def export_pages(kicad_dir: Path, out_dir: Path, cli: str) -> list[Path]:
|
||||
"""逐页导出 kicadxml。整体导出在扁平结构下会漏页，所以必须一页一页来。"""
|
||||
sheets = sorted(kicad_dir.glob("*.kicad_sch"))
|
||||
if not sheets:
|
||||
raise SystemExit(f"[ERROR] {kicad_dir} 下没有 .kicad_sch")
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
made = []
|
||||
for sch in sheets:
|
||||
dst = out_dir / f"xml_{sch.stem}.xml"
|
||||
cmd = [cli, "sch", "export", "netlist", "--format", "kicadxml",
|
||||
"-o", str(dst), str(sch)]
|
||||
# 不能用 text=True：那会按系统区域编码(中文 Windows 是 GBK)解码，
|
||||
# 而 kicad-cli 输出 UTF-8，遇到非 ASCII 直接 UnicodeDecodeError。
|
||||
proc = subprocess.run(cmd, capture_output=True,
|
||||
encoding="utf-8", errors="replace")
|
||||
if proc.returncode != 0 or not dst.is_file():
|
||||
print(f"[WARN] 导出失败 {sch.name}: {proc.stderr.strip()[:200]}")
|
||||
continue
|
||||
made.append(dst)
|
||||
print(f"[export] {len(made)}/{len(sheets)} 页导出成功 -> {out_dir}")
|
||||
return made
|
||||
|
||||
|
||||
def load_altium(csv_path: Path) -> dict[str, str]:
|
||||
with csv_path.open(newline="", encoding="utf-8-sig") as f:
|
||||
reader = csv.DictReader(f)
|
||||
c_key = pick(reader.fieldnames or [], COMP_KEYS)
|
||||
p_key = pick(reader.fieldnames or [], PIN_KEYS)
|
||||
n_key = pick(reader.fieldnames or [], NET_KEYS)
|
||||
if not (c_key and p_key and n_key):
|
||||
raise SystemExit(f"[ERROR] 认不出 CSV 列名：{reader.fieldnames}（需要 component/pin/net）")
|
||||
real = collections.defaultdict(set)
|
||||
rows = 0
|
||||
for row in reader:
|
||||
rows += 1
|
||||
comp = (row.get(c_key) or "").strip()
|
||||
pin = (row.get(p_key) or "").strip()
|
||||
net = (row.get(n_key) or "").strip()
|
||||
if comp and pin and net and net != "?":
|
||||
real[(comp, pin)].add(norm_net(net))
|
||||
print(f"[altium] 原始 {rows} 行 -> 去 '?' 重复后 {len(real)} 个唯一节点")
|
||||
return {f"{c}.{p}": sorted(v)[0] for (c, p), v in real.items()}
|
||||
|
||||
|
||||
def load_kicad(xml_paths: list[Path]) -> dict[str, str]:
|
||||
nn = collections.defaultdict(set)
|
||||
for fn in xml_paths:
|
||||
root = ET.parse(fn).getroot()
|
||||
for net in root.iter("net"):
|
||||
nm = norm_net(net.get("name"))
|
||||
for nd in net.findall("node"):
|
||||
nn[f"{nd.get('ref')}.{nd.get('pin')}"].add(nm)
|
||||
print(f"[kicad] 合并 {len(xml_paths)} 页 -> {len(nn)} 个节点")
|
||||
# 同一节点在多页上出现在同名网络里是正常的；名字冲突时保留全部以便暴露问题
|
||||
return {nd: (sorted(names)[0] if len(names) == 1 else "|".join(sorted(names)))
|
||||
for nd, names in nn.items()}
|
||||
|
||||
|
||||
def partition(node_net: dict[str, str]):
|
||||
groups = collections.defaultdict(set)
|
||||
for node, net in node_net.items():
|
||||
groups[net].add(node)
|
||||
return {node: frozenset(groups[net]) for node, net in node_net.items()}, groups
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="Altium vs KiCad schematic netlist partition compare")
|
||||
ap.add_argument("--kicad-dir", required=True, help="KiCad 工程目录（含 *.kicad_sch）")
|
||||
ap.add_argument("--netlist-csv", required=True, help="Altium tabular 网表真值 CSV")
|
||||
ap.add_argument("--out", required=True, help="输出目录，放 xml_*.xml / diff.json / 报告")
|
||||
ap.add_argument("--kicad-cli", help="kicad-cli 路径，默认自动找")
|
||||
ap.add_argument("--skip-export", action="store_true", help="复用 --out 下已有的 xml_*.xml")
|
||||
args = ap.parse_args()
|
||||
|
||||
out_dir = Path(args.out)
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if args.skip_export:
|
||||
xmls = [Path(p) for p in sorted(glob.glob(str(out_dir / "xml_*.xml")))]
|
||||
if not xmls:
|
||||
raise SystemExit(f"[ERROR] --skip-export 但 {out_dir} 下没有 xml_*.xml")
|
||||
print(f"[export] 跳过，复用已有 {len(xmls)} 个 xml")
|
||||
else:
|
||||
xmls = export_pages(Path(args.kicad_dir), out_dir, find_kicad_cli(args.kicad_cli))
|
||||
|
||||
node_net_a = load_altium(Path(args.netlist_csv))
|
||||
node_net_k = load_kicad(xmls)
|
||||
|
||||
pa, ga = partition(node_net_a)
|
||||
pk, gk = partition(node_net_k)
|
||||
na, nk = set(node_net_a), set(node_net_k)
|
||||
only_a, only_k = sorted(na - nk), sorted(nk - na)
|
||||
mismatch = sorted(nd for nd in (na & nk) if pa[nd] != pk[nd])
|
||||
|
||||
classes_a = {frozenset(v) for v in ga.values()}
|
||||
classes_k = {frozenset(v) for v in gk.values()}
|
||||
identical = (classes_a == classes_k) and not only_a and not only_k
|
||||
|
||||
print(f"\n节点 Altium={len(na)} KiCad={len(nk)} 只在A={len(only_a)} 只在K={len(only_k)}")
|
||||
print(f"网络 Altium={len(ga)} KiCad={len(gk)} 等价类集合相等={classes_a == classes_k}")
|
||||
print(f"同伴集不同的节点：{len(mismatch)}")
|
||||
for nd in only_a[:15]:
|
||||
print(f" 只在Altium {nd}: net={node_net_a[nd]}")
|
||||
for nd in only_k[:15]:
|
||||
print(f" 只在KiCad {nd}: net={node_net_k[nd]}")
|
||||
for nd in mismatch[:30]:
|
||||
am, km = pa[nd] - {nd}, pk[nd] - {nd}
|
||||
print(f" {nd}: A={node_net_a[nd]} K={node_net_k[nd]} | "
|
||||
f"只A的同伴:{sorted(am - km)[:5]} 只K的同伴:{sorted(km - am)[:5]}")
|
||||
|
||||
diff_path = out_dir / "diff.json"
|
||||
diff_path.write_text(json.dumps({
|
||||
"identical": identical,
|
||||
"altium_nodes": len(na), "kicad_nodes": len(nk),
|
||||
"altium_nets": len(ga), "kicad_nets": len(gk),
|
||||
"onlyA_nodes": only_a, "onlyK_nodes": only_k,
|
||||
"mismatch_nodes": mismatch,
|
||||
"onlyA_classes": len(classes_a - classes_k),
|
||||
"onlyK_classes": len(classes_k - classes_a),
|
||||
}, ensure_ascii=False, indent=1), encoding="utf-8")
|
||||
|
||||
print(f"\n[diff] {diff_path}")
|
||||
print(f"identical: {identical}")
|
||||
if not identical:
|
||||
print("差异不一定是真错：先确认 KiCad 网名前导 '/' 已剥、Altium 的 '?' 行已去，"
|
||||
"再看是不是某页 kicadxml 没导出成功（上面 [export] 的比例）。")
|
||||
return 0 if identical else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,208 @@
|
||||
"""stage_ascii.py — convert_ascii 这一步的机械部分（阶段 1.2）。
|
||||
|
||||
做三件确定性的事，剩下的「在 Altium GUI 里跑脚本」只能人来点：
|
||||
stage 把整个 Altium 工程复制到 <工程目录>_ascii staging，并按实际枚举到的
|
||||
.SchDoc 生成 convert_to_ascii.pas（DelphiScript 的 FindFirst 在
|
||||
Altium 里枚举不可靠，所以文件名必须在生成期硬编码进脚本）。
|
||||
verify 检查 staging 里每个 .SchDoc 的文件头，判断是否真的转成了 ASCII。
|
||||
ASCII 版以 "|HEADER=Protel for Windows" 开头；AD26 二进制是 CFB
|
||||
复合文档，以 D0 CF 11 E0 开头 —— 后者会让 KiCad 导入器报
|
||||
"Header already parsed" 而失败。
|
||||
|
||||
原始工程绝不改动：所有操作只作用于 staging 副本。
|
||||
|
||||
用法:
|
||||
python stage_ascii.py stage --source-prj E:\\Altium_design\\FOO\\FOO.PrjPCB \\
|
||||
--migration-dir E:\\Kicad_design\\FOO\\_migration
|
||||
python stage_ascii.py verify --staging-dir E:\\Altium_design\\FOO_ascii
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import shutil
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ASCII_MAGIC = b"|HEADER=Protel for Windows"
|
||||
CFB_MAGIC = b"\xd0\xcf\x11\xe0"
|
||||
|
||||
PAS_TEMPLATE = """{{ convert_to_ascii.pas — 由 stage_ascii.py 生成，请勿手改文件列表。
|
||||
|
||||
把 staging 副本里的 {count} 个 .SchDoc 全部另存为 ASCII 格式，绕开 KiCad 导入器
|
||||
对 AD26 二进制原理图的 "Header already parsed" 失败。
|
||||
|
||||
运行方式（必须在 Altium GUI 内手动跑，不要经 eda-agent/MCP 驱动——
|
||||
MCP 的 StartMCPServer 与本脚本会互相冲突挂起）：
|
||||
先停掉 eda-agent -> Altium: File > Run Script > 选本文件 > ConvertSchToAscii > Run
|
||||
|
||||
只作用于 staging：{dir}
|
||||
原始工程不受影响。
|
||||
}}
|
||||
|
||||
Procedure ConvertSchToAscii;
|
||||
Var
|
||||
Names : TStringList;
|
||||
LogList : TStringList;
|
||||
Dir : String;
|
||||
Path : String;
|
||||
SD : IServerDocument;
|
||||
i : Integer;
|
||||
Found : Integer;
|
||||
Failed : Integer;
|
||||
Begin
|
||||
Dir := '{dir}\\';
|
||||
|
||||
Names := TStringList.Create;
|
||||
{names}
|
||||
LogList := TStringList.Create;
|
||||
Found := 0;
|
||||
Failed := 0;
|
||||
|
||||
For i := 0 To Names.Count - 1 Do
|
||||
Begin
|
||||
Path := Dir + Names[i];
|
||||
Try
|
||||
SD := Client.OpenDocument('SCH', Path);
|
||||
If SD <> Nil Then
|
||||
Begin
|
||||
Client.ShowDocument(SD);
|
||||
SD.SetModified(True);
|
||||
SD.DoFileSave('ASCII');
|
||||
Client.CloseDocument(SD);
|
||||
LogList.Add('OK ' + Names[i]);
|
||||
Inc(Found);
|
||||
End
|
||||
Else
|
||||
Begin
|
||||
LogList.Add('FAIL(open) ' + Names[i]);
|
||||
Inc(Failed);
|
||||
End;
|
||||
Except
|
||||
LogList.Add('FAIL(exc) ' + Names[i]);
|
||||
Inc(Failed);
|
||||
End;
|
||||
End;
|
||||
|
||||
LogList.Insert(0, 'ASCII conversion: ' + IntToStr(Found) + ' OK, ' + IntToStr(Failed) + ' failed');
|
||||
LogList.SaveToFile('{log}');
|
||||
Names.Free;
|
||||
ShowMessage('ASCII done: ' + IntToStr(Found) + ' OK, ' + IntToStr(Failed) + ' failed');
|
||||
LogList.Free;
|
||||
End;
|
||||
"""
|
||||
|
||||
|
||||
def sch_docs(folder: Path) -> list[Path]:
|
||||
"""staging 里的原理图页，按名字排序（含子目录，Altium 工程偶尔分目录放）。"""
|
||||
return sorted(folder.rglob("*.SchDoc"), key=lambda p: str(p).lower())
|
||||
|
||||
|
||||
def cmd_stage(args) -> int:
|
||||
source = Path(args.source_prj)
|
||||
if not source.is_file():
|
||||
print(f"[ERROR] 源工程不存在：{source}")
|
||||
return 1
|
||||
src_dir = source.parent
|
||||
staging = Path(args.staging_dir) if args.staging_dir else src_dir.parent / (src_dir.name + "_ascii")
|
||||
migration = Path(args.migration_dir)
|
||||
migration.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if staging.exists():
|
||||
if not args.force:
|
||||
print(f"[ERROR] staging 已存在：{staging}\n"
|
||||
f" 加 --force 覆盖重建（会先删除该目录），或用 --staging-dir 换个位置。")
|
||||
return 1
|
||||
print(f"[INFO] --force：删除已有 staging {staging}")
|
||||
shutil.rmtree(staging)
|
||||
|
||||
print(f"[COPY] {src_dir} -> {staging}")
|
||||
shutil.copytree(src_dir, staging)
|
||||
|
||||
docs = sch_docs(staging)
|
||||
if not docs:
|
||||
print(f"[ERROR] staging 里没找到任何 .SchDoc：{staging}")
|
||||
return 1
|
||||
|
||||
names = "".join(
|
||||
f" Names.Add('{d.relative_to(staging).as_posix().replace('/', chr(92))}');\n"
|
||||
for d in docs
|
||||
)
|
||||
log_path = migration / "ascii_convert_log.txt"
|
||||
# DelphiScript 的字符串字面量不做反斜杠转义，路径按原样写进去
|
||||
pas = PAS_TEMPLATE.format(count=len(docs), dir=str(staging), names=names,
|
||||
log=str(log_path))
|
||||
pas_path = migration / "convert_to_ascii.pas"
|
||||
pas_path.write_text(pas, encoding="utf-8")
|
||||
|
||||
print(f"[SCHDOC] 枚举到 {len(docs)} 张原理图页：")
|
||||
for d in docs:
|
||||
print(f" {d.relative_to(staging)}")
|
||||
print(f"[SCRIPT] {pas_path}")
|
||||
print(f"[LOG] 脚本跑完会写到 {log_path}")
|
||||
print(f"[STAGING] {staging}")
|
||||
print("\n下一步（必须人工在 Altium GUI 内做，不要走 MCP）："
|
||||
"\n 1. 停掉 eda-agent（避免与 StartMCPServer 脚本冲突挂起）"
|
||||
f"\n 2. Altium: File > Run Script > 选 {pas_path} > ConvertSchToAscii > Run"
|
||||
"\n 3. 回来跑 `python stage_ascii.py verify --staging-dir \"%s\"` 确认" % staging)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_verify(args) -> int:
|
||||
staging = Path(args.staging_dir)
|
||||
if not staging.is_dir():
|
||||
print(f"[ERROR] staging 目录不存在：{staging}")
|
||||
return 1
|
||||
docs = sch_docs(staging)
|
||||
if not docs:
|
||||
print(f"[ERROR] staging 里没有 .SchDoc：{staging}")
|
||||
return 1
|
||||
|
||||
ascii_n, binary_n, unknown = 0, 0, []
|
||||
for d in docs:
|
||||
head = d.open("rb").read(64)
|
||||
if head.startswith(ASCII_MAGIC):
|
||||
ascii_n += 1
|
||||
elif head.startswith(CFB_MAGIC):
|
||||
binary_n += 1
|
||||
unknown.append((d, "binary CFB — 未转换"))
|
||||
else:
|
||||
unknown.append((d, f"未知文件头 {head[:12]!r}"))
|
||||
|
||||
print(f"[VERIFY] {staging}")
|
||||
print(f" ASCII: {ascii_n} / {len(docs)}")
|
||||
for d, why in unknown:
|
||||
print(f" [!] {d.relative_to(staging)} {why}")
|
||||
|
||||
if ascii_n == len(docs):
|
||||
print("[OK] 全部原理图已是 ASCII，可以进 KiCad 导入了。")
|
||||
return 0
|
||||
print("[FAIL] 还有页没转成 ASCII —— 直接导入会报 'Header already parsed'。"
|
||||
"\n 回 Altium 重跑 convert_to_ascii.pas（注意 eda-agent 要停掉）。")
|
||||
return 1
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
p = sub.add_parser("stage", help="复制到 _ascii staging 并生成 convert_to_ascii.pas")
|
||||
p.add_argument("--source-prj", required=True)
|
||||
p.add_argument("--migration-dir", required=True)
|
||||
p.add_argument("--staging-dir", help="默认 <工程目录>_ascii，与原工程同级")
|
||||
p.add_argument("--force", action="store_true", help="staging 已存在时先删除重建")
|
||||
p.set_defaults(func=cmd_stage)
|
||||
|
||||
p = sub.add_parser("verify", help="检查 staging 里 .SchDoc 是否都成了 ASCII")
|
||||
p.add_argument("--staging-dir", required=True)
|
||||
p.set_defaults(func=cmd_verify)
|
||||
|
||||
args = ap.parse_args()
|
||||
return args.func(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
except Exception:
|
||||
pass
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user