app
KiCad - the KiCad Bridge
Public Made by Adomby adom
Reference implementation of the KiCad bridge: multi-instance Python server, forward path via kicad-cli, reverse path via in-process plugin. Most complex of the three bundled bridges.
← Commit history
tour/tour.pydeleted−716@@ -1,716 +0,0 @@-"""KiCad Feature Tour — main orchestrator.--Usage:- python tour.py [--speed fast|normal|slow] [--start N] [--no-ipc]-"""--import argparse-import json-import sys-import threading-import time--from narration import (- NarrationPanel, GREEN, RED, YELLOW, BLUE, ACCENT,- TEXT_MUTED, TEXT_SECONDARY, FG_DIM, FG_SECONDARY,-)-from steps import (- TOUR_STEPS, DISPLAY, BRIDGE, BRIDGE_VISUAL, IPC_TEST, IPC_LAYER,- WIN_CLOSE,-)-import bridge_client-import ipc_client-import win_manager---# ── Speed multipliers ────────────────────────────────────────────-SPEED_MULT = {"fast": 0.5, "normal": 1.0, "slow": 1.5}---# ── Data formatters ──────────────────────────────────────────────--def _extract_result(data: dict):- """Extract the meaningful payload from a bridge response."""- if "data" in data:- return data["data"]- if "result" in data:- return data["result"]- return data---def format_json_response(data: dict, indent: int = 2) -> str:- """Format a bridge response dict into readable text."""- if not data:- return "(no data)"- if "error" in data and not data.get("success", True):- return f"Error: {data['error']}"- result = _extract_result(data)- if isinstance(result, str):- return result- if isinstance(result, dict):- return json.dumps(result, indent=indent, default=str)- if isinstance(result, list):- lines = []- for i, item in enumerate(result[:30]):- if isinstance(item, dict):- line = json.dumps(item, default=str)- if len(line) > 80:- line = line[:77] + "..."- lines.append(f" {line}")- else:- lines.append(f" {item}")- if len(result) > 30:- lines.append(f" ... and {len(result) - 30} more")- return f"[{len(result)} items]\n" + "\n".join(lines)- return str(result)---def format_pcb_stats(data: dict) -> str:- r = _extract_result(data)- if isinstance(r, str):- try:- r = json.loads(r)- except (json.JSONDecodeError, TypeError):- return str(r)- if not isinstance(r, dict):- return format_json_response(data)- lines = []- if "board_dimensions" in r:- d = r["board_dimensions"]- lines.append(f"Board Size: {d.get('width_mm', '?')} x {d.get('height_mm', '?')} mm")- for key, label in [("layer_count", "Copper Layers"), ("track_count", "Tracks"),- ("via_count", "Vias"), ("pad_count", "Pads"),- ("footprint_count", "Footprints"), ("net_count", "Nets"),- ("zone_count", "Zones")]:- if key in r:- val = r[key]- lines.append(f"{label + ':':16s} {val:,}" if isinstance(val, int) else f"{label + ':':16s} {val}")- return "\n".join(lines) if lines else format_json_response(data)---def format_schematic_stats(data: dict) -> str:- r = _extract_result(data)- if isinstance(r, str):- try:- r = json.loads(r)- except (json.JSONDecodeError, TypeError):- return str(r)- if not isinstance(r, dict):- return format_json_response(data)- lines = []- for key, label in [- ("component_count", "Components"), ("symbol_count", "Symbols"),- ("wire_count", "Wires"), ("label_count", "Labels"),- ("global_label_count", "Global Labels"),- ("hierarchical_label_count", "Hier. Labels"),- ("sheet_count", "Sub-sheets"), ("junction_count", "Junctions"),- ("no_connect_count", "No-connects"), ("bus_count", "Buses"),- ("power_port_count", "Power Ports"),- ]:- if key in r:- lines.append(f"{label + ':':20s} {r[key]}")- return "\n".join(lines) if lines else format_json_response(data)---def format_hierarchy(data: dict) -> str:- r = _extract_result(data)- if isinstance(r, str):- try:- r = json.loads(r)- except (json.JSONDecodeError, TypeError):- return str(r)-- def _tree(node, prefix="", is_last=True):- lines = []- connector = "+-- " if is_last else "|-- "- name = node.get("name", node.get("sheet_name", "?"))- filename = node.get("filename", node.get("file", ""))- label = f"{name}"- if filename:- label += f" ({filename})"- lines.append(f"{prefix}{connector}{label}")- children = node.get("children", node.get("sub_sheets", []))- child_prefix = prefix + (" " if is_last else "| ")- for i, child in enumerate(children):- lines.extend(_tree(child, child_prefix, i == len(children) - 1))- return lines-- if isinstance(r, dict):- name = r.get("name", r.get("sheet_name", "Root"))- children = r.get("children", r.get("sub_sheets", []))- lines = [f"[Root] {name}"]- for i, child in enumerate(children):- lines.extend(_tree(child, " ", i == len(children) - 1))- return "\n".join(lines)- elif isinstance(r, list):- return "\n".join(- format_hierarchy({"result": item}) if isinstance(item, dict) else str(item)- for item in r- )- return format_json_response(data)---def format_components(data: dict) -> str:- """Format schematic component list."""- r = _extract_result(data)- if isinstance(r, str):- try:- r = json.loads(r)- except (json.JSONDecodeError, TypeError):- return str(r)- items = r if isinstance(r, list) else r.get("components", []) if isinstance(r, dict) else []- lines = [f"Total components: {len(items)}", ""]- lines.append(f"{'Ref':8s} {'Value':15s} {'Library Symbol'}")- lines.append("-" * 50)- for item in items[:25]:- ref = str(item.get("reference", item.get("ref", "?")))[:7]- val = str(item.get("value", "?"))[:14]- lib = str(item.get("library_symbol", item.get("symbol", item.get("lib_id", "?"))))- if ":" in lib:- lib = lib.split(":")[-1]- lines.append(f"{ref:8s} {val:15s} {lib[:25]}")- if len(items) > 25:- lines.append(f"... and {len(items) - 25} more")- return "\n".join(lines)---def format_bom(data: dict) -> str:- r = _extract_result(data)- if isinstance(r, str):- try:- r = json.loads(r)- except (json.JSONDecodeError, TypeError):- return str(r)- if isinstance(r, dict):- items = r.get("items", r.get("components", []))- summary = r.get("summary", {})- elif isinstance(r, list):- items = r- summary = {}- else:- return format_json_response(data)- lines = []- if summary:- lines.append(f"Unique parts: {summary.get('unique_count', len(items))}")- lines.append(f"Total placed: {summary.get('total_count', '?')}")- lines.append("")- lines.append(f"{'Ref':10s} {'Value':15s} {'Footprint':25s}")- lines.append("-" * 52)- for item in items[:25]:- ref = str(item.get("reference", item.get("ref", "?")))[:9]- val = str(item.get("value", "?"))[:14]- fp = str(item.get("footprint", "?")).split(":")[-1][:24]- lines.append(f"{ref:10s} {val:15s} {fp:25s}")- if len(items) > 25:- lines.append(f"... and {len(items) - 25} more components")- return "\n".join(lines)---def format_nets(data: dict) -> str:- r = _extract_result(data)- if isinstance(r, str):- try:- r = json.loads(r)- except (json.JSONDecodeError, TypeError):- return str(r)- nets = r if isinstance(r, list) else r.get("nets", []) if isinstance(r, dict) else []- lines = [f"Total nets: {len(nets)}", ""]- lines.append(f"{'ID':6s} {'Net Name'}")- lines.append("-" * 40)- for net in nets[:30]:- if isinstance(net, dict):- nid = str(net.get("id", net.get("number", "?")))- name = net.get("name", "unnamed")- else:- nid = "-"- name = str(net)- lines.append(f"{nid:6s} {name}")- if len(nets) > 30:- lines.append(f"... and {len(nets) - 30} more nets")- return "\n".join(lines)---def format_search_results(data: dict) -> str:- r = _extract_result(data)- if isinstance(r, str):- try:- r = json.loads(r)- except (json.JSONDecodeError, TypeError):- return str(r)- results = r if isinstance(r, list) else r.get("results", r.get("symbols", [])) if isinstance(r, dict) else []- lines = [f"Found {len(results)} matching symbols:", ""]- for item in results[:20]:- if isinstance(item, dict):- name = item.get("name", item.get("symbol_name", "?"))- lib = item.get("library", item.get("lib_name", ""))- desc = item.get("description", "")- lines.append(f" {lib}:{name}" if lib else f" {name}")- if desc:- lines.append(f" {desc[:60]}")- else:- lines.append(f" {item}")- return "\n".join(lines)---def format_symbol_info(data: dict) -> str:- r = _extract_result(data)- if isinstance(r, str):- try:- r = json.loads(r)- except (json.JSONDecodeError, TypeError):- return str(r)- if not isinstance(r, dict):- return format_json_response(data)- lines = []- name = r.get("name", r.get("symbol_name", "?"))- lines.append(f"Symbol: {name}")- if "description" in r:- lines.append(f" {r['description']}")- lines.append("")- pins = r.get("pins", [])- if pins:- lines.append(f"{'Pin':6s} {'Name':15s} {'Type':12s} {'Dir'}")- lines.append("-" * 45)- for pin in pins[:20]:- num = str(pin.get("number", pin.get("num", "?")))[:5]- pname = str(pin.get("name", "?"))[:14]- ptype = str(pin.get("electrical_type", pin.get("type", "?")))[:11]- direction = str(pin.get("direction", pin.get("orientation", "")))- lines.append(f"{num:6s} {pname:15s} {ptype:12s} {direction}")- if len(pins) > 20:- lines.append(f"... and {len(pins) - 20} more pins")- return "\n".join(lines)---def format_drc_erc(data: dict, check_type: str = "DRC") -> str:- r = _extract_result(data)- if isinstance(r, str):- try:- r = json.loads(r)- except (json.JSONDecodeError, TypeError):- return str(r)- if not isinstance(r, dict):- return format_json_response(data)- lines = []- violations = r.get("violations", r.get("errors", []))- warnings = r.get("warnings", [])- total = len(violations) + len(warnings)- if total == 0:- lines.append(f"{check_type} passed with no violations!")- else:- lines.append(f"{check_type}: {len(violations)} violations, {len(warnings)} warnings")- lines.append("")- for v in violations[:15]:- if isinstance(v, dict):- sev = v.get("severity", "error")- msg = v.get("message", v.get("description", str(v)))- lines.append(f" [{sev}] {msg[:70]}")- else:- lines.append(f" {str(v)[:70]}")- for w in warnings[:10]:- if isinstance(w, dict):- msg = w.get("message", w.get("description", str(w)))- lines.append(f" [warn] {msg[:70]}")- else:- lines.append(f" [warn] {str(w)[:70]}")- remaining = total - min(len(violations), 15) - min(len(warnings), 10)- if remaining > 0:- lines.append(f" ... and {remaining} more")- return "\n".join(lines)---# Command-specific formatters-FORMATTERS = {- "get_pcb_statistics": format_pcb_stats,- "get_schematic_stats": format_schematic_stats,- "get_schematic_hierarchy": format_hierarchy,- "list_schematic_components": format_components,- "generate_bom": format_bom,- "list_pcb_nets": format_nets,- "search_symbols": format_search_results,- "get_symbol_info": format_symbol_info,- "run_drc": lambda d: format_drc_erc(d, "DRC"),- "run_erc": lambda d: format_drc_erc(d, "ERC"),-}---class TourRunner:- """Runs the tour steps on a background thread, updates the narration panel."""-- def __init__(self, panel: NarrationPanel, speed: float = 1.0,- start_step: int = 1, skip_ipc: bool = False):- self.panel = panel- self.speed = speed- self.start_step = start_step- self.skip_ipc = skip_ipc-- self._pause_event = threading.Event()- self._pause_event.set()- self._skip_flag = False- self._quit_flag = False- self._ipc_cache: dict = {}-- def start(self):- t = threading.Thread(target=self._run, daemon=True)- t.start()-- def pause(self):- self._pause_event.clear()-- def resume(self):- self._pause_event.set()-- def skip(self):- self._skip_flag = True- self._pause_event.set()-- def quit(self):- self._quit_flag = True- self._pause_event.set()-- # ── Internal ─────────────────────────────────────────────────-- def _sleep(self, seconds: float):- end = time.monotonic() + seconds * self.speed- while time.monotonic() < end:- if self._quit_flag or self._skip_flag:- return- self._pause_event.wait(timeout=0.2)-- def _update_panel(self, func, *args, **kwargs):- self.panel.root.after(0, lambda: func(*args, **kwargs))-- def _run(self):- total = len(TOUR_STEPS)- self._preflight()-- for i, step in enumerate(TOUR_STEPS):- step_num = i + 1- if step_num < self.start_step:- continue- if self._quit_flag:- break-- self._skip_flag = False- self._update_panel(- self.panel.set_step, step_num, total,- step["title"], step["description"], step.get("act", ""),- )-- self._sleep(step.get("delay_before", 1.0))- if self._quit_flag:- break- if self._skip_flag:- continue-- # Bring KiCad to foreground before every step so user sees the action- action = step["action"]- if action not in (DISPLAY,):- try:- win_manager.position_kicad_left(self.panel.width)- except Exception:- pass- time.sleep(0.3)-- try:- self._execute_step(step)- except Exception as e:- self._update_panel(self.panel.set_status, f"Error: {e}", RED)- self._update_panel(self.panel.set_data, str(e))- self._update_panel(self.panel.set_data_color, RED)-- if self._quit_flag:- break- if self._skip_flag:- continue-- self._sleep(step.get("delay_after", 3.0))-- if not self._quit_flag:- self._update_panel(self.panel.set_status, "Tour complete!", GREEN)-- def _preflight(self):- self._update_panel(self.panel.set_status, "Preflight checks...", YELLOW)-- health = bridge_client.health_check()- if health:- self._update_panel(- self.panel.set_data,- f"Bridge server: ONLINE\n URL: {bridge_client.BRIDGE_URL}"- )- self._update_panel(self.panel.set_data_color, GREEN)- else:- self._update_panel(- self.panel.set_data,- f"Bridge server: OFFLINE\n URL: {bridge_client.BRIDGE_URL}\n"- " Start with: python plugins/kicad/server.py"- )- self._update_panel(self.panel.set_data_color, RED)-- time.sleep(1.5)-- if not self.skip_ipc:- has_test = ipc_client.has_test_binary()- has_layer = ipc_client.has_layer_binary()- self._update_panel(- self.panel.set_data,- f"IPC test binary: {'FOUND' if has_test else 'MISSING'}\n"- f"IPC layer binary: {'FOUND' if has_layer else 'MISSING'}"- )- self._update_panel(- self.panel.set_data_color, GREEN if (has_test and has_layer) else YELLOW- )- time.sleep(1.5)-- self._update_panel(self.panel.set_status, "Starting tour...", GREEN)- time.sleep(1.0)-- def _execute_step(self, step: dict):- action = step["action"]-- if action == DISPLAY:- self._execute_display(step)- elif action == BRIDGE:- self._execute_bridge(step)- elif action == BRIDGE_VISUAL:- self._execute_bridge_visual(step)- elif action == IPC_TEST:- self._execute_ipc_test(step)- elif action == IPC_LAYER:- self._execute_ipc_layer(step)- elif action == WIN_CLOSE:- self._execute_win_close(step)-- def _execute_display(self, step: dict):- self._update_panel(self.panel.set_status, "", TEXT_MUTED)- if "Tour Complete" in step["title"]:- self._update_panel(- self.panel.set_data,- "GALLIA FEATURE SUMMARY\n"- "======================\n\n"- "Track A Python Bridge (37 commands):\n"- " Schematic: open, stats, hierarchy, components\n"- " PCB: open, stats, nets, BOM, DRC, ERC\n"- " Library: search, symbol info, footprints\n"- " Editors: symbol editor, footprint editor, 3D\n"- " Export: Gerber, drill, STEP, BOM\n\n"- "Track B Rust IPC (15 commands):\n"- " Connection: ping, version, get_api_version\n"- " Board: layers, stackup, nets, items\n"- " Editor: active layer, selection, refresh\n\n"- "All commands available to AI agents\n"- "via the Gallia bridge."- )- self._update_panel(self.panel.set_data_color, GREEN)-- def _execute_bridge(self, step: dict):- """Execute a data-only bridge command and show results."""- command = step["command"]- args = step.get("args", {})-- self._update_panel(self.panel.set_status, f"bridge: {command}", ACCENT)-- result = bridge_client.send_command(command, args)- formatter = FORMATTERS.get(command, format_json_response)- display_text = formatter(result)-- is_ok = result.get("success", True) and "error" not in result- self._update_panel(self.panel.set_data_color, GREEN if is_ok else RED)- self._update_panel(- self.panel.set_status,- f"{command} OK" if is_ok else f"{command} Error",- GREEN if is_ok else RED,- )- self._update_panel(self.panel.set_data, display_text)-- def _execute_bridge_visual(self, step: dict):- """Execute a bridge command that opens/changes KiCad visually.-- Supports:- close_first: window title to close before opening- command/args: bridge command to send- wait_window: window title to wait for and position- post_command: optional second bridge command to run after (e.g. run_drc)- """- command = step["command"]- args = step.get("args", {})-- # Close previous editor if requested- close_first = step.get("close_first")- if close_first:- self._update_panel(self.panel.set_status, f"Closing {close_first}...", TEXT_SECONDARY)- win_manager.close_window_by_title(close_first)- time.sleep(1.5)-- # Send the visual command- self._update_panel(self.panel.set_status, f"bridge: {command}", ACCENT)- result = bridge_client.send_command(command, args)-- if result.get("error"):- self._update_panel(self.panel.set_data, f"Error: {result['error']}")- self._update_panel(self.panel.set_data_color, RED)- self._update_panel(self.panel.set_status, f"{command} failed", RED)- return-- # Wait for the window to appear- wait_window = step.get("wait_window")- if wait_window:- self._update_panel(self.panel.set_status, f"Waiting for {wait_window}...", YELLOW)- time.sleep(3.0)- hwnd = win_manager.position_kicad_left(self.panel.width)- if hwnd:- self._update_panel(- self.panel.set_status, f"{wait_window} ready", GREEN,- )- else:- self._update_panel(- self.panel.set_status, f"{wait_window} (window not positioned)", YELLOW,- )-- # Run post-command if specified (e.g. run_drc after opening the board)- post_cmd = step.get("post_command")- if post_cmd:- post_args = step.get("post_args", {})- self._update_panel(self.panel.set_status, f"bridge: {post_cmd}", ACCENT)- post_result = bridge_client.send_command(post_cmd, post_args)- formatter = FORMATTERS.get(post_cmd, format_json_response)- display_text = formatter(post_result)- is_ok = post_result.get("success", True) and "error" not in post_result- self._update_panel(self.panel.set_data_color, GREEN if is_ok else RED)- self._update_panel(- self.panel.set_status,- f"{post_cmd} OK" if is_ok else f"{post_cmd} Error",- GREEN if is_ok else RED,- )- self._update_panel(self.panel.set_data, display_text)- else:- # Show the open result- formatter = FORMATTERS.get(command, format_json_response)- self._update_panel(self.panel.set_data, formatter(result))- self._update_panel(self.panel.set_data_color, GREEN)-- def _execute_ipc_test(self, step: dict):- if self.skip_ipc:- self._update_panel(self.panel.set_status, "IPC skipped (--no-ipc)", YELLOW)- self._update_panel(self.panel.set_data, "IPC steps skipped.")- return-- if not self._ipc_cache:- self._update_panel(self.panel.set_status, "Running IPC test suite...", ACCENT)- self._ipc_cache = ipc_client.run_ipc_test()- if not self._ipc_cache:- self._update_panel(self.panel.set_status, "IPC binary not found", RED)- self._update_panel(self.panel.set_data, "test_kicad_ipc.exe not found.")- self._update_panel(self.panel.set_data_color, RED)- self.skip_ipc = True- return-- section = step.get("section", 1)- text = self._ipc_cache.get(section, f"(Section [{section}] not found)")- self._update_panel(self.panel.set_data, text)- self._update_panel(self.panel.set_data_color, GREEN)- self._update_panel(self.panel.set_status, f"IPC section [{section}]", GREEN)-- def _execute_ipc_layer(self, step: dict):- if self.skip_ipc:- self._update_panel(self.panel.set_status, "IPC skipped", YELLOW)- return-- layer_id = step["layer_id"]- layer_name = step.get("layer_name", f"layer {layer_id}")-- self._update_panel(- self.panel.set_status,- f"IPC: set_active_layer({layer_id}) {layer_name}", ACCENT,- )-- # Bring KiCad to front so the user sees the change- win_manager.position_kicad_left(self.panel.width)- time.sleep(0.5)-- result = ipc_client.set_active_layer(layer_id)- if result.get("success"):- self._update_panel(- self.panel.set_data,- f"Active layer: {layer_name}\n\n{result.get('output', '')}",- )- self._update_panel(self.panel.set_data_color, GREEN)- self._update_panel(self.panel.set_status, f"Layer: {layer_name}", GREEN)- else:- self._update_panel(- self.panel.set_data,- f"Layer switch failed:\n{result.get('output', 'Unknown error')}",- )- self._update_panel(self.panel.set_data_color, RED)- self._update_panel(self.panel.set_status, "Layer switch failed", RED)-- def _execute_win_close(self, step: dict):- title = step["window_title"]- self._update_panel(self.panel.set_status, f"Closing {title}...", TEXT_SECONDARY)- success = win_manager.close_window_by_title(title)- if success:- self._update_panel(self.panel.set_data, f"{title} closed.")- self._update_panel(self.panel.set_data_color, GREEN)- self._update_panel(self.panel.set_status, "Closed", GREEN)- else:- self._update_panel(self.panel.set_data, f"'{title}' not found (may already be closed).")- self._update_panel(self.panel.set_data_color, YELLOW)- self._update_panel(self.panel.set_status, "Window not found", YELLOW)---# ── Entry point ──────────────────────────────────────────────────--def main():- parser = argparse.ArgumentParser(description="KiCad Feature Tour")- parser.add_argument(- "--speed", choices=["fast", "normal", "slow"], default="normal",- help="Tour speed (fast=0.5x delays, slow=1.5x delays)",- )- parser.add_argument(- "--start", type=int, default=1,- help="Start at step N (1-based)",- )- parser.add_argument(- "--no-ipc", action="store_true",- help="Skip IPC steps (if binaries not available)",- )- args = parser.parse_args()- speed = SPEED_MULT[args.speed]-- runner = None-- def on_pause():- if runner:- runner.pause()-- def on_resume():- if runner:- runner.resume()-- def on_skip():- if runner:- runner.skip()-- def on_quit():- if runner:- runner.quit()-- panel = NarrationPanel(- width=480,- on_pause=on_pause,- on_resume=on_resume,- on_skip=on_skip,- on_quit=on_quit,- )-- runner = TourRunner(- panel=panel, speed=speed,- start_step=args.start, skip_ipc=args.no_ipc,- )- runner.start()-- panel.root.mainloop()---if __name__ == "__main__":- main()