From 57872e59c18d1cafc2a3c2cfc91e0ac342b77176 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sat, 11 Jul 2026 16:47:22 -0600 Subject: [PATCH] Replace kicad-sch-api with internal SchDocument engine Eliminate the external kicad-sch-api dependency by migrating all MCP tools to the internal SchDocument class built on sexp_tree.py. This fixes three serialization bugs (dropped global labels, TypeError on local labels, mis-quoted property private keywords) and removes ~1,900 lines of workaround code. New modules: - sexp_tree.py: S-expression parser and round-trip serializer - sch_document.py: schematic read/write/mutate API - lib_resolver.py: symbol library search and resolution - sch_helpers.py: shared load/validate/expand helpers Migrated all 9 tool files, resources, and tests. Removed 5 workaround functions from sexp_parser.py. 531 tests pass, ruff + mypy clean. --- pyproject.toml | 4 +- src/mckicad/resources/schematic.py | 91 +- src/mckicad/tools/autowire.py | 62 +- src/mckicad/tools/batch.py | 193 +-- src/mckicad/tools/power_symbols.py | 50 +- src/mckicad/tools/schematic.py | 527 +++----- src/mckicad/tools/schematic_analysis.py | 167 +-- src/mckicad/tools/schematic_edit.py | 144 +-- src/mckicad/tools/schematic_patterns.py | 66 +- src/mckicad/utils/lib_resolver.py | 541 ++++++++ src/mckicad/utils/sch_document.py | 1147 ++++++++++++++++ src/mckicad/utils/sch_helpers.py | 40 + src/mckicad/utils/sexp_parser.py | 260 +--- src/mckicad/utils/sexp_tree.py | 410 ++++++ tests/conftest.py | 16 +- tests/test_autowire.py | 10 +- tests/test_batch.py | 174 +-- tests/test_lib_resolver.py | 458 +++++++ tests/test_patterns.py | 2 +- tests/test_power_symbols.py | 2 +- tests/test_sch_document.py | 1582 +++++++++++++++++++++++ tests/test_schematic.py | 16 +- tests/test_schematic_analysis.py | 16 +- tests/test_schematic_edit.py | 8 - tests/test_sexp_parser.py | 472 ++----- tests/test_sexp_tree.py | 668 ++++++++++ uv.lock | 83 +- 27 files changed, 5306 insertions(+), 1903 deletions(-) create mode 100644 src/mckicad/utils/lib_resolver.py create mode 100644 src/mckicad/utils/sch_document.py create mode 100644 src/mckicad/utils/sch_helpers.py create mode 100644 src/mckicad/utils/sexp_tree.py create mode 100644 tests/test_lib_resolver.py create mode 100644 tests/test_sch_document.py create mode 100644 tests/test_sexp_tree.py diff --git a/pyproject.toml b/pyproject.toml index 042c2f3..e6fe306 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,7 @@ dependencies = [ "pyyaml>=6.0.3", "defusedxml>=0.7.1", "kicad-python>=0.5.0", - "kicad-sch-api>=0.5.6", + "requests>=2.32.5", ] @@ -80,7 +80,7 @@ warn_unused_ignores = true show_error_codes = true [[tool.mypy.overrides]] -module = ["kipy.*", "kicad_sch_api.*", "requests.*"] +module = ["kipy.*", "requests.*"] ignore_missing_imports = true [tool.pytest.ini_options] diff --git a/src/mckicad/resources/schematic.py b/src/mckicad/resources/schematic.py index ac614fe..14e00b9 100644 --- a/src/mckicad/resources/schematic.py +++ b/src/mckicad/resources/schematic.py @@ -11,22 +11,11 @@ import os from typing import Any from mckicad.server import mcp +from mckicad.utils.sch_helpers import expand as _expand +from mckicad.utils.sch_helpers import load_schematic logger = logging.getLogger(__name__) -_HAS_SCH_API = False - -try: - from kicad_sch_api import load_schematic as _ksa_load - - _HAS_SCH_API = True -except ImportError: - pass - - -def _expand(path: str) -> str: - return os.path.abspath(os.path.expanduser(path)) - @mcp.resource("kicad://schematic/{path}/components") def schematic_components_resource(path: str) -> str: @@ -35,25 +24,22 @@ def schematic_components_resource(path: str) -> str: Returns a JSON array of all components with reference, lib_id, value, and position. """ - if not _HAS_SCH_API: - return json.dumps({"error": "kicad-sch-api not installed"}) - expanded = _expand(path) if not os.path.isfile(expanded): return json.dumps({"error": f"File not found: {expanded}"}) try: - sch = _ksa_load(expanded) + sch = load_schematic(expanded) components: list[dict[str, Any]] = [] for comp in sch.components: entry: dict[str, Any] = { - "reference": getattr(comp, "reference", None), - "lib_id": getattr(comp, "lib_id", None), - "value": getattr(comp, "value", None), + "reference": comp.reference, + "lib_id": comp.lib_id, + "value": comp.value, } - pos = getattr(comp, "position", None) - if pos is not None and isinstance(pos, (list, tuple)) and len(pos) >= 2: - entry["position"] = {"x": pos[0], "y": pos[1]} + pos = comp.position + if pos is not None: + entry["position"] = {"x": pos.x, "y": pos.y} components.append(entry) return json.dumps(components, indent=2) except Exception as e: @@ -66,36 +52,19 @@ def schematic_nets_resource(path: str) -> str: """Browsable net list for a KiCad schematic. Returns a JSON object mapping net names to lists of connected pins. + Uses the connectivity builder for accurate net resolution. """ - if not _HAS_SCH_API: - return json.dumps({"error": "kicad-sch-api not installed"}) - expanded = _expand(path) if not os.path.isfile(expanded): return json.dumps({"error": f"File not found: {expanded}"}) try: - sch = _ksa_load(expanded) - nets_attr = getattr(sch, "nets", None) + sch = load_schematic(expanded) - if nets_attr is None: - return json.dumps({"error": "Schematic has no nets attribute"}) + from mckicad.tools.schematic_analysis import _build_connectivity - net_data: dict[str, list[dict[str, str]]] = {} - if isinstance(nets_attr, dict): - for net_name, pins in nets_attr.items(): - pin_list = [] - for p in (pins if isinstance(pins, (list, tuple)) else []): - if isinstance(p, dict): - pin_list.append(p) - else: - pin_list.append({ - "reference": str(getattr(p, "reference", "")), - "pin": str(getattr(p, "pin", getattr(p, "number", ""))), - }) - net_data[str(net_name)] = pin_list - - return json.dumps(net_data, indent=2) + net_graph, _unconnected = _build_connectivity(sch, expanded) + return json.dumps(net_graph, indent=2) except Exception as e: logger.error("Failed to load schematic nets for resource: %s", e) return json.dumps({"error": str(e)}) @@ -108,36 +77,32 @@ def schematic_hierarchy_resource(path: str) -> str: Returns the hierarchical sheet tree with filenames and per-sheet component counts. Useful for understanding multi-sheet designs. """ - if not _HAS_SCH_API: - return json.dumps({"error": "kicad-sch-api not installed"}) - expanded = _expand(path) if not os.path.isfile(expanded): return json.dumps({"error": f"File not found: {expanded}"}) try: - sch = _ksa_load(expanded) + sch = load_schematic(expanded) def _build(sheet_sch: Any, name: str, filename: str) -> dict[str, Any]: info: dict[str, Any] = { "name": name, "filename": filename, - "component_count": len(list(sheet_sch.components)) if hasattr(sheet_sch, "components") else 0, + "component_count": len(list(sheet_sch.components)), } children: list[dict[str, Any]] = [] - if hasattr(sheet_sch, "sheets"): - for child in sheet_sch.sheets: - child_name = getattr(child, "name", "unknown") - child_file = getattr(child, "filename", "unknown") - child_path = os.path.join(os.path.dirname(expanded), child_file) - if os.path.isfile(child_path): - try: - child_sch = _ksa_load(child_path) - children.append(_build(child_sch, child_name, child_file)) - except Exception: - children.append({"name": child_name, "filename": child_file, "error": "load failed"}) - else: - children.append({"name": child_name, "filename": child_file, "note": "file not found"}) + for child in sheet_sch.sheets: + child_name = child.name + child_file = child.filename + child_path = os.path.join(os.path.dirname(expanded), child_file) + if os.path.isfile(child_path): + try: + child_sch = load_schematic(child_path) + children.append(_build(child_sch, child_name, child_file)) + except Exception: + children.append({"name": child_name, "filename": child_file, "error": "load failed"}) + else: + children.append({"name": child_name, "filename": child_file, "note": "file not found"}) if children: info["sheets"] = children return info diff --git a/src/mckicad/tools/autowire.py b/src/mckicad/tools/autowire.py index 2d12251..1d11840 100644 --- a/src/mckicad/tools/autowire.py +++ b/src/mckicad/tools/autowire.py @@ -23,41 +23,11 @@ from mckicad.autowire.strategy import ( from mckicad.config import TIMEOUT_CONSTANTS from mckicad.server import mcp from mckicad.utils.kicad_cli import find_kicad_cli +from mckicad.utils.sch_helpers import expand as _expand +from mckicad.utils.sch_helpers import validate_schematic_path as _validate_schematic_path logger = logging.getLogger(__name__) -_HAS_SCH_API = False -try: - from kicad_sch_api import load_schematic as _ksa_load - - _HAS_SCH_API = True -except ImportError: - pass - - -def _require_sch_api() -> dict[str, Any] | None: - if not _HAS_SCH_API: - return { - "success": False, - "error": "kicad-sch-api is not installed. Install it with: uv add kicad-sch-api", - } - return None - - -def _expand(path: str) -> str: - return os.path.abspath(os.path.expanduser(path)) - - -def _validate_schematic_path(path: str) -> dict[str, Any] | None: - if not path: - return {"success": False, "error": "Schematic path must be a non-empty string"} - expanded = os.path.expanduser(path) - if not expanded.endswith(".kicad_sch"): - return {"success": False, "error": f"Path must end with .kicad_sch, got: {path}"} - if not os.path.isfile(expanded): - return {"success": False, "error": f"Schematic file not found: {expanded}"} - return None - def _auto_export_netlist(schematic_path: str) -> tuple[str | None, str | None]: """Export a netlist via kicad-cli. Returns (path, error).""" @@ -163,10 +133,6 @@ def autowire_schematic( Dictionary with strategy summary (counts by method), the full plan, batch JSON, and apply results (when dry_run=False). """ - err = _require_sch_api() - if err: - return err - verr = _validate_schematic_path(schematic_path) if verr: return verr @@ -178,7 +144,8 @@ def autowire_schematic( try: # 1. Load schematic and build connectivity state - sch = _ksa_load(schematic_path) + from mckicad.utils.sch_helpers import load_schematic as _load_sch + sch = _load_sch(schematic_path) from mckicad.tools.schematic_analysis import _build_connectivity_state @@ -285,32 +252,15 @@ def autowire_schematic( # 12. Apply if not dry_run if not dry_run and batch_data: - from mckicad.tools.batch import ( - _apply_batch_operations, - _register_project_libraries, - ) - from mckicad.utils.sexp_parser import ( - fix_property_private_keywords, - insert_sexp_before_close, - ) + from mckicad.tools.batch import _apply_batch_operations # Re-load schematic fresh for application - sch = _ksa_load(schematic_path) - _register_project_libraries(batch_data, schematic_path) + sch = _load_sch(schematic_path) summary = _apply_batch_operations(sch, batch_data, schematic_path) sch.save(schematic_path) - private_fixes = fix_property_private_keywords(schematic_path) - if private_fixes: - summary["property_private_fixes"] = private_fixes - - pending_sexps = summary.pop("_pending_label_sexps", []) - if pending_sexps: - combined_sexp = "".join(pending_sexps) - insert_sexp_before_close(schematic_path, combined_sexp) - result["applied"] = { "components_placed": summary.get("components_placed", 0), "power_symbols_placed": summary.get("power_symbols_placed", 0), diff --git a/src/mckicad/tools/batch.py b/src/mckicad/tools/batch.py index 5d405b1..22e1748 100644 --- a/src/mckicad/tools/batch.py +++ b/src/mckicad/tools/batch.py @@ -17,58 +17,26 @@ from typing import Any from mckicad.config import BATCH_LIMITS, LABEL_DEFAULTS from mckicad.server import mcp +from mckicad.utils.sch_helpers import ( + expand as _expand, + load_schematic, + validate_schematic_path as _validate_schematic_path, +) logger = logging.getLogger(__name__) # Map user-facing direction names to the pin schematic rotation that # produces that label placement direction (label goes opposite to body). _DIRECTION_TO_ROTATION: dict[str, float] = { - "left": 0, # body right → label left - "right": 180, # body left → label right - "up": 90, # body down → label up - "down": 270, # body up → label down + "left": 0, # body right -> label left + "right": 180, # body left -> label right + "up": 90, # body down -> label up + "down": 270, # body up -> label down } -_HAS_SCH_API = False - -try: - from kicad_sch_api import load_schematic as _ksa_load - - _HAS_SCH_API = True -except ImportError: - logger.warning( - "kicad-sch-api not installed — batch tools will return helpful errors. " - "Install with: uv add kicad-sch-api" - ) - - -def _require_sch_api() -> dict[str, Any] | None: - if not _HAS_SCH_API: - return { - "success": False, - "error": "kicad-sch-api is not installed. Install it with: uv add kicad-sch-api", - "engine": "none", - } - return None - def _get_schematic_engine() -> str: - return "kicad-sch-api" if _HAS_SCH_API else "none" - - -def _validate_schematic_path(path: str, must_exist: bool = True) -> dict[str, Any] | None: - if not path: - return {"success": False, "error": "Schematic path must be a non-empty string"} - expanded = os.path.expanduser(path) - if not expanded.endswith(".kicad_sch"): - return {"success": False, "error": f"Path must end with .kicad_sch, got: {path}"} - if must_exist and not os.path.isfile(expanded): - return {"success": False, "error": f"Schematic file not found: {expanded}"} - return None - - -def _expand(path: str) -> str: - return os.path.abspath(os.path.expanduser(path)) + return "sch_document" def _resolve_batch_file(batch_file: str, schematic_path: str) -> str: @@ -277,75 +245,26 @@ def _validate_batch_data(data: dict[str, Any], sch: Any) -> list[str]: return errors -def _register_project_libraries( - data: dict[str, Any], schematic_path: str, -) -> list[str]: - """Register project-local symbol libraries with kicad-sch-api's cache. - - Scans the batch data for library names not already known to the global - ``SymbolLibraryCache``, locates the ``.kicad_sym`` files via mckicad's - own library search (which reads ``sym-lib-table``), and registers them - so that ``components.add()`` can resolve the lib_id. - - Returns list of newly registered library names. - """ - try: - from kicad_sch_api.library.cache import get_symbol_cache - except ImportError: - return [] - - from mckicad.utils.sexp_parser import _find_library_file - - cache = get_symbol_cache() - - # Collect unique library names from batch components - lib_names: set[str] = set() - for comp in data.get("components", []): - lib_id = comp.get("lib_id", "") - if ":" in lib_id: - lib_names.add(lib_id.split(":")[0]) - - registered: list[str] = [] - for lib_name in lib_names: - # Skip if already indexed - if lib_name in cache._library_index: - continue - - lib_path = _find_library_file(schematic_path, lib_name) - if lib_path and cache.add_library_path(lib_path): - logger.info("Registered project-local library: %s -> %s", lib_name, lib_path) - registered.append(lib_name) - elif lib_path is None: - logger.debug("Library '%s' not found in project paths", lib_name) - - return registered - - - def _apply_batch_operations( sch: Any, data: dict[str, Any], schematic_path: str, ) -> dict[str, Any]: """Apply all batch operations to a loaded schematic. Returns summary. - Labels are returned as pending sexp strings (``_pending_label_sexps``) - that must be inserted into the file *after* ``sch.save()``, because - kicad-sch-api's serializer drops global labels and raises TypeError - on local labels. + Labels are inserted directly into the sexp tree via + ``sch.add_label()`` / ``sch.add_global_label()`` and wire stubs via + ``sch.add_wire()``, so no post-save fixup is needed. Args: - sch: A kicad-sch-api SchematicDocument instance. + sch: A SchDocument instance. data: Parsed batch JSON data. - schematic_path: Path to the .kicad_sch file (needed for sexp - pin fallback and label insertion). + schematic_path: Path to the .kicad_sch file (needed for pin + position resolution). """ from mckicad.patterns._geometry import add_power_symbol_to_pin from mckicad.utils.sexp_parser import ( WireSegment, clamp_stub_length, compute_label_placement, - generate_global_label_sexp, - generate_label_sexp, - generate_wire_sexp, resolve_label_collision, resolve_pin_position, resolve_pin_position_and_orientation, @@ -357,7 +276,6 @@ def _apply_batch_operations( placed_wires: list[str] = [] placed_labels: list[str] = [] placed_no_connects = 0 - pending_label_sexps: list[str] = [] occupied_positions: dict[tuple[float, float], str] = {} placed_wire_segments: list[WireSegment] = [] collisions_resolved = 0 @@ -371,7 +289,7 @@ def _apply_batch_operations( except Exception: logger.warning("Could not set paper_size to '%s'", batch_paper) - # 1. Components (multi-unit supported natively by kicad-sch-api) + # 1. Components for comp in data.get("components", []): ref = comp.get("reference") unit = comp.get("unit", 1) @@ -426,10 +344,10 @@ def _apply_batch_operations( for wire in data.get("wires", []): if "from_ref" in wire: wire_id = sch.add_wire_between_pins( - component1_ref=wire["from_ref"], - pin1_number=wire["from_pin"], - component2_ref=wire["to_ref"], - pin2_number=wire["to_pin"], + from_ref=wire["from_ref"], + from_pin=wire["from_pin"], + to_ref=wire["to_ref"], + to_pin=wire["to_pin"], ) else: wire_id = sch.add_wire( @@ -481,7 +399,7 @@ def _apply_batch_operations( if _info: _register_obstacle(_nc["pin_ref"], _info) - # 4. Labels — generate sexp strings for post-save insertion + # 4. Labels -- added directly to the sexp tree via SchDocument for label in data.get("labels", []): is_global = label.get("global", False) rotation = label.get("rotation", 0) @@ -535,7 +453,7 @@ def _apply_batch_operations( lx, ly = placement["label_x"], placement["label_y"] rotation = placement["label_rotation"] - # Resolve collisions before generating sexp + # Resolve collisions before placing new_x, new_y = resolve_label_collision( lx, ly, rotation, label["text"], occupied_positions, ) @@ -544,17 +462,16 @@ def _apply_batch_operations( lx, ly = new_x, new_y if is_global: - sexp = generate_global_label_sexp( + sch.add_global_label( text=label["text"], x=lx, y=ly, rotation=rotation, shape=label.get("shape", "bidirectional"), ) else: - sexp = generate_label_sexp( + sch.add_label( text=label["text"], x=lx, y=ly, rotation=rotation, ) - pending_label_sexps.append(sexp) - # Wire stub from pin to (possibly shifted) label — with + # Wire stub from pin to (possibly shifted) label -- with # wire-level collision detection for collinear overlaps w_sx, w_sy, w_lx, w_ly = resolve_wire_collision( placement["stub_start_x"], placement["stub_start_y"], @@ -566,12 +483,11 @@ def _apply_batch_operations( ): wire_collisions_resolved += 1 lx, ly = w_lx, w_ly - wire_sexp = generate_wire_sexp(w_sx, w_sy, w_lx, w_ly) - pending_label_sexps.append(wire_sexp) + sch.add_wire(w_sx, w_sy, w_lx, w_ly) else: # Coordinate-based label (original path) if is_global: - sexp = generate_global_label_sexp( + sch.add_global_label( text=label["text"], x=label["x"], y=label["y"], @@ -579,17 +495,16 @@ def _apply_batch_operations( shape=label.get("shape", "bidirectional"), ) else: - sexp = generate_label_sexp( + sch.add_label( text=label["text"], x=label["x"], y=label["y"], rotation=rotation, ) - pending_label_sexps.append(sexp) placed_labels.append(label["text"]) - # 4b. Label connections — pin-ref labels sharing a net name + # 4b. Label connections -- pin-ref labels sharing a net name for lc in data.get("label_connections", []): net = lc["net"] is_global = lc.get("global", False) @@ -643,7 +558,7 @@ def _apply_batch_operations( lx, ly = placement["label_x"], placement["label_y"] rotation = placement["label_rotation"] - # Resolve collisions before generating sexp + # Resolve collisions before placing new_x, new_y = resolve_label_collision( lx, ly, rotation, net, occupied_positions, ) @@ -652,16 +567,15 @@ def _apply_batch_operations( lx, ly = new_x, new_y if is_global: - sexp = generate_global_label_sexp( + sch.add_global_label( text=net, x=lx, y=ly, rotation=rotation, shape=shape, ) else: - sexp = generate_label_sexp( + sch.add_label( text=net, x=lx, y=ly, rotation=rotation, ) - pending_label_sexps.append(sexp) - # Wire stub from pin to (possibly shifted) label — with + # Wire stub from pin to (possibly shifted) label -- with # wire-level collision detection for collinear overlaps w_sx, w_sy, w_lx, w_ly = resolve_wire_collision( placement["stub_start_x"], placement["stub_start_y"], @@ -673,8 +587,7 @@ def _apply_batch_operations( ): wire_collisions_resolved += 1 lx, ly = w_lx, w_ly - wire_sexp = generate_wire_sexp(w_sx, w_sy, w_lx, w_ly) - pending_label_sexps.append(wire_sexp) + sch.add_wire(w_sx, w_sy, w_lx, w_ly) placed_labels.append(net) # 5. No-connects (coordinate or pin-referenced) @@ -689,9 +602,9 @@ def _apply_batch_operations( nc["pin_ref"], nc["pin_number"], ) continue - sch.no_connects.add(position=pin_pos) + sch.add_no_connect(position=pin_pos) else: - sch.no_connects.add(position=(nc["x"], nc["y"])) + sch.add_no_connect(x=nc["x"], y=nc["y"]) placed_no_connects += 1 return { @@ -713,7 +626,6 @@ def _apply_batch_operations( + len(placed_labels) + placed_no_connects ), - "_pending_label_sexps": pending_label_sexps, } @@ -807,10 +719,6 @@ def apply_batch( Dictionary with ``success``, counts of each operation type applied, and wire/label IDs for the created elements. """ - err = _require_sch_api() - if err: - return err - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: @@ -846,11 +754,7 @@ def apply_batch( } try: - sch = _ksa_load(schematic_path) - - # Register project-local symbol libraries with the cache so - # components.add() can resolve non-standard lib_ids. - _register_project_libraries(data, schematic_path) + sch = load_schematic(schematic_path) # Set hierarchy context for sub-sheets so kicad-cli resolves # power symbol nets correctly during netlist export. @@ -898,27 +802,10 @@ def apply_batch( # Apply all operations summary = _apply_batch_operations(sch, data, schematic_path) - # Single save (components, power symbols, wires, no-connects) + # Single save -- labels and wires are already in the sexp tree, + # no post-save fixup needed. sch.save(schematic_path) - # Fix kicad-sch-api's mis-serialization of (property private ...) - # keywords before any further file modifications. - from mckicad.utils.sexp_parser import ( - fix_property_private_keywords, - insert_sexp_before_close, - ) - - private_fixes = fix_property_private_keywords(schematic_path) - if private_fixes: - summary["property_private_fixes"] = private_fixes - - # Insert labels via sexp AFTER save — kicad-sch-api's serializer - # drops global labels and raises TypeError on local labels. - pending_sexps = summary.pop("_pending_label_sexps", []) - if pending_sexps: - combined_sexp = "".join(pending_sexps) - insert_sexp_before_close(schematic_path, combined_sexp) - logger.info( "Batch applied: %d operations from %s to %s", summary["total_operations"], diff --git a/src/mckicad/tools/power_symbols.py b/src/mckicad/tools/power_symbols.py index 4febcfe..b600788 100644 --- a/src/mckicad/tools/power_symbols.py +++ b/src/mckicad/tools/power_symbols.py @@ -7,53 +7,17 @@ Uses the shared geometry helpers from the patterns library. """ import logging -import os from typing import Any from mckicad.server import mcp +from mckicad.utils.sch_helpers import expand as _expand +from mckicad.utils.sch_helpers import load_schematic, validate_schematic_path as _validate_schematic_path logger = logging.getLogger(__name__) -_HAS_SCH_API = False - -try: - from kicad_sch_api import load_schematic as _ksa_load - - _HAS_SCH_API = True -except ImportError: - logger.warning( - "kicad-sch-api not installed — power symbol tools will return helpful errors. " - "Install with: uv add kicad-sch-api" - ) - - -def _require_sch_api() -> dict[str, Any] | None: - if not _HAS_SCH_API: - return { - "success": False, - "error": "kicad-sch-api is not installed. Install it with: uv add kicad-sch-api", - "engine": "none", - } - return None - def _get_schematic_engine() -> str: - return "kicad-sch-api" if _HAS_SCH_API else "none" - - -def _validate_schematic_path(path: str, must_exist: bool = True) -> dict[str, Any] | None: - if not path: - return {"success": False, "error": "Schematic path must be a non-empty string"} - expanded = os.path.expanduser(path) - if not expanded.endswith(".kicad_sch"): - return {"success": False, "error": f"Path must end with .kicad_sch, got: {path}"} - if must_exist and not os.path.isfile(expanded): - return {"success": False, "error": f"Schematic file not found: {expanded}"} - return None - - -def _expand(path: str) -> str: - return os.path.abspath(os.path.expanduser(path)) + return "sch_document" @mcp.tool() @@ -97,14 +61,10 @@ def add_power_symbol( Dictionary with ``success``, placed ``reference``, ``lib_id``, symbol position, wire ID, and direction. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) if not net: return {"success": False, "error": "net must be a non-empty string"} @@ -117,7 +77,7 @@ def add_power_symbol( from mckicad.patterns._geometry import add_power_symbol_to_pin from mckicad.utils.sexp_parser import resolve_pin_position - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) # Look up the target pin position (with sexp fallback for custom symbols) pin_pos_tuple = resolve_pin_position(sch, schematic_path, pin_ref, pin_number) diff --git a/src/mckicad/tools/schematic.py b/src/mckicad/tools/schematic.py index 898d601..819dc23 100644 --- a/src/mckicad/tools/schematic.py +++ b/src/mckicad/tools/schematic.py @@ -1,31 +1,30 @@ """ Schematic creation and manipulation tools for the mckicad MCP server. -Wraps the kicad-sch-api library to provide schematic editing through MCP tools. +Wraps the SchDocument sexp-tree engine for schematic editing through MCP tools. Designed so that the underlying engine can be swapped to kipy IPC once KiCad exposes a schematic API over its IPC transport. """ -from collections import Counter import contextlib import logging import os import re +from collections import Counter from typing import Any from mckicad.config import INLINE_RESULT_THRESHOLD from mckicad.server import mcp from mckicad.utils.file_utils import write_detail_file +from mckicad.utils.lib_resolver import get_symbol_info as _lr_get_info +from mckicad.utils.lib_resolver import search_symbols as _lr_search +from mckicad.utils.sch_document import SchDocument +from mckicad.utils.sch_helpers import expand as _expand +from mckicad.utils.sch_helpers import load_schematic +from mckicad.utils.sch_helpers import validate_schematic_path as _validate_schematic_path from mckicad.utils.sexp_parser import ( compute_label_placement, - generate_global_label_sexp, - generate_label_sexp, - generate_wire_sexp, - insert_sexp_before_close, - parse_global_labels, - parse_lib_symbol_pins, resolve_pin_position_and_orientation, - transform_pin_to_schematic, ) logger = logging.getLogger(__name__) @@ -34,69 +33,14 @@ logger = logging.getLogger(__name__) # Engine abstraction — swap point for future kipy IPC schematic support # --------------------------------------------------------------------------- -_HAS_SCH_API = False - -try: - from kicad_sch_api import create_schematic as _ksa_create - from kicad_sch_api import get_symbol_info as _ksa_get_symbol_info - from kicad_sch_api import load_schematic as _ksa_load - from kicad_sch_api import search_symbols as _ksa_search - - _HAS_SCH_API = True -except ImportError: - logger.warning( - "kicad-sch-api not installed — schematic tools will return helpful errors. " - "Install with: uv add kicad-sch-api" - ) - def _get_schematic_engine() -> str: """Return the name of the active schematic manipulation engine. - Currently: ``kicad-sch-api`` (file-level manipulation) + Currently: ``sch_document`` (sexp-tree file-level manipulation) Future: ``kipy`` IPC when KiCad adds a schematic API over its IPC transport. """ - if _HAS_SCH_API: - return "kicad-sch-api" - return "none" - - -def _require_sch_api() -> dict[str, Any] | None: - """Return an error dict if kicad-sch-api is unavailable, else None.""" - if not _HAS_SCH_API: - return { - "success": False, - "error": ("kicad-sch-api is not installed. Install it with: uv add kicad-sch-api"), - "engine": "none", - } - return None - - -def _validate_schematic_path(path: str, must_exist: bool = True) -> dict[str, Any] | None: - """Validate a schematic file path. Returns an error dict on failure, else None.""" - if not path: - return {"success": False, "error": "Schematic path must be a non-empty string"} - - expanded = os.path.expanduser(path) - - if not expanded.endswith(".kicad_sch"): - return { - "success": False, - "error": f"Path must end with .kicad_sch, got: {path}", - } - - if must_exist and not os.path.isfile(expanded): - return { - "success": False, - "error": f"Schematic file not found: {expanded}", - } - - return None - - -def _expand(path: str) -> str: - """Expand ~ and return an absolute path.""" - return os.path.abspath(os.path.expanduser(path)) + return "sch_document" # --------------------------------------------------------------------------- @@ -129,10 +73,6 @@ def create_schematic( Returns: Dictionary with ``success``, ``path``, ``paper_size``, and ``engine`` keys. """ - err = _require_sch_api() - if err: - return err - output_path = _expand(output_path) if not output_path.endswith(".kicad_sch"): @@ -155,8 +95,8 @@ def create_schematic( if parent: os.makedirs(parent, exist_ok=True) - sch = _ksa_create(name) - sch.set_paper_size(paper_size) + sch = SchDocument.create(output_path, paper=paper_size) + sch.set_title_block(title=name) sch.save(output_path) logger.info("Created schematic '%s' (%s) at %s", name, paper_size, output_path) @@ -198,17 +138,13 @@ def add_component( Returns: Dictionary with ``success``, component ``reference``, and ``lib_id``. """ - err = _require_sch_api() - if err: - return err - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) sch.components.add( lib_id=lib_id, reference=reference, @@ -217,10 +153,6 @@ def add_component( ) sch.save(schematic_path) - from mckicad.utils.sexp_parser import fix_property_private_keywords - - fix_property_private_keywords(schematic_path) - logger.info("Added %s (%s) to %s at (%.1f, %.1f)", reference, lib_id, schematic_path, x, y) return { "success": True, @@ -257,12 +189,8 @@ def search_components(query: str, library: str | None = None) -> dict[str, Any]: Returns: Dictionary with ``success`` and a ``results`` list of matching symbols. """ - err = _require_sch_api() - if err: - return err - try: - raw_results = _ksa_search(query) + raw_results = _lr_search(query) # Filter by library when requested if library: @@ -287,7 +215,7 @@ def search_components(query: str, library: str | None = None) -> dict[str, Any]: logger.info("Symbol search for '%s' returned %d results", query, len(results)) - # Large result → sidecar file with top results inline + # Large result -> sidecar file with top results inline if len(results) > INLINE_RESULT_THRESHOLD: safe_query = re.sub(r"[^\w\-]", "_", query)[:30] detail_path = write_detail_file(None, f"search_{safe_query}.json", results) @@ -340,18 +268,14 @@ def add_wire( Returns: Dictionary with ``success`` and the wire ``id``. """ - err = _require_sch_api() - if err: - return err - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr try: - sch = _ksa_load(schematic_path) - wire_id = sch.add_wire(start=(start_x, start_y), end=(end_x, end_y)) + sch = load_schematic(schematic_path) + wire = sch.add_wire(start=(start_x, start_y), end=(end_x, end_y)) sch.save(schematic_path) logger.info( @@ -364,7 +288,7 @@ def add_wire( ) return { "success": True, - "wire_id": wire_id, + "wire_id": str(wire.uuid), "start": {"x": start_x, "y": start_y}, "end": {"x": end_x, "y": end_y}, "schematic_path": schematic_path, @@ -399,23 +323,14 @@ def connect_pins( Returns: Dictionary with ``success`` and the created wire ``id``. """ - err = _require_sch_api() - if err: - return err - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr try: - sch = _ksa_load(schematic_path) - wire_id = sch.add_wire_between_pins( - component1_ref=from_ref, - pin1_number=from_pin, - component2_ref=to_ref, - pin2_number=to_pin, - ) + sch = load_schematic(schematic_path) + wire = sch.add_wire_between_pins(from_ref, from_pin, to_ref, to_pin) sch.save(schematic_path) logger.info( @@ -428,7 +343,7 @@ def connect_pins( ) return { "success": True, - "wire_id": wire_id, + "wire_id": str(wire.uuid), "from": {"reference": from_ref, "pin": from_pin}, "to": {"reference": to_ref, "pin": to_pin}, "schematic_path": schematic_path, @@ -477,10 +392,6 @@ def add_label( automatically resolve position and rotation from a component pin. A wire stub is inserted connecting the pin to the label. - Uses direct s-expression file insertion to bypass kicad-sch-api - serializer bugs that silently drop global labels and raise TypeError - on local labels. - Args: schematic_path: Path to an existing .kicad_sch file. text: Label text (becomes the net name, e.g. ``GND``, ``SPI_CLK``). @@ -518,11 +429,10 @@ def add_label( } try: + sch = load_schematic(schematic_path) + if has_pin_ref: # Pin-referenced placement - from kicad_sch_api import load_schematic as _ksa_load - - sch = _ksa_load(schematic_path) pin_info = resolve_pin_position_and_orientation( sch, schematic_path, pin_ref, pin_number, # type: ignore[arg-type] ) @@ -541,42 +451,38 @@ def add_label( label_rotation = placement["label_rotation"] if global_label: - sexp = generate_global_label_sexp( - text, lx, ly, rotation=label_rotation, shape=shape, + gl = sch.add_global_label( + text, lx, ly, shape=shape, rotation=label_rotation, ) + label_id = gl.uuid label_type = "global" else: - sexp = generate_label_sexp(text, lx, ly, rotation=label_rotation) + lbl = sch.add_label(text, lx, ly, rotation=label_rotation) + label_id = lbl.uuid label_type = "local" - wire_sexp = generate_wire_sexp( + # Add wire stub + sch.add_wire( placement["stub_start_x"], placement["stub_start_y"], placement["stub_end_x"], placement["stub_end_y"], ) - - combined = sexp + wire_sexp - insert_sexp_before_close(schematic_path, combined) x, y = lx, ly else: - # Coordinate placement (original path) + # Coordinate placement if global_label: - sexp = generate_global_label_sexp( - text, x, y, rotation=rotation, shape=shape, # type: ignore[arg-type] + gl = sch.add_global_label( + text, x, y, shape=shape, rotation=rotation, # type: ignore[arg-type] ) + label_id = gl.uuid label_type = "global" else: - sexp = generate_label_sexp( + lbl = sch.add_label( text, x, y, rotation=rotation, # type: ignore[arg-type] ) + label_id = lbl.uuid label_type = "local" - insert_sexp_before_close(schematic_path, sexp) - - # Extract the UUID we generated from the sexp block - import re as _re - - uuid_match = _re.search(r'\(uuid "([^"]+)"\)', sexp) - label_id = uuid_match.group(1) if uuid_match else "unknown" + sch.save(schematic_path) logger.info( "Added %s label '%s' at (%.1f, %.1f) in %s", @@ -589,7 +495,7 @@ def add_label( "label_type": label_type, "position": {"x": x, "y": y}, "schematic_path": schematic_path, - "engine": "sexp-direct", + "engine": _get_schematic_engine(), } except Exception as e: logger.error("Failed to add label '%s' in %s: %s", text, schematic_path, e) @@ -625,10 +531,6 @@ def add_hierarchical_sheet( Returns: Dictionary with ``success`` and sheet metadata. """ - err = _require_sch_api() - if err: - return err - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: @@ -641,12 +543,14 @@ def add_hierarchical_sheet( } try: - sch = _ksa_load(schematic_path) - sheet_uuid = sch.add_sheet( + sch = load_schematic(schematic_path) + sheet_result = sch.add_sheet( name=name, filename=filename, - position=(x, y), - size=(width, height), + x=x, + y=y, + width=width, + height=height, ) parent_uuid = sch.uuid sch.save(schematic_path) @@ -665,7 +569,7 @@ def add_hierarchical_sheet( "success": True, "sheet_name": name, "sheet_filename": filename, - "sheet_uuid": sheet_uuid, + "sheet_uuid": sheet_result["uuid"], "parent_uuid": parent_uuid, "position": {"x": x, "y": y}, "size": {"width": width, "height": height}, @@ -687,7 +591,7 @@ def list_components( When called without ``reference``, returns all components. For large schematics (>{threshold} components), a compact summary is returned inline and the full list is written to ``.mckicad/components.json`` - — read that file for complete data. + -- read that file for complete data. When ``reference`` is provided, returns only that component's details (always inline, always compact). @@ -701,9 +605,6 @@ def list_components( Dictionary with ``success``, ``count``, and either inline ``components`` list or a ``detail_file`` path. """.replace("{threshold}", str(INLINE_RESULT_THRESHOLD)) - err = _require_sch_api() - if err: - return err schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) @@ -711,23 +612,16 @@ def list_components( return verr try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) components: list[dict[str, Any]] = [] for comp in sch.components: entry: dict[str, Any] = { - "reference": getattr(comp, "reference", None), - "lib_id": getattr(comp, "lib_id", None), - "value": getattr(comp, "value", None), + "reference": comp.reference, + "lib_id": comp.lib_id, + "value": comp.value, + "position": {"x": comp.position.x, "y": comp.position.y}, } - - pos = getattr(comp, "position", None) - if pos is not None: - if isinstance(pos, (list, tuple)) and len(pos) >= 2: - entry["position"] = {"x": pos[0], "y": pos[1]} - else: - entry["position"] = str(pos) - components.append(entry) # Single-component lookup @@ -749,7 +643,7 @@ def list_components( logger.info("Listed %d components in %s", len(components), schematic_path) - # Large result → sidecar file + # Large result -> sidecar file if len(components) > INLINE_RESULT_THRESHOLD: prefix_counts = Counter( re.match(r"[A-Za-z]+", c.get("reference", "") or "").group() # type: ignore[union-attr] @@ -785,7 +679,7 @@ def get_schematic_info(schematic_path: str) -> dict[str, Any]: Returns statistics and a validation summary inline. For schematics with many unique symbols, full symbol details and validation issues are - written to ``.mckicad/schematic_info.json`` — read that file when you + written to ``.mckicad/schematic_info.json`` -- read that file when you need per-symbol data. Args: @@ -795,83 +689,43 @@ def get_schematic_info(schematic_path: str) -> dict[str, Any]: Dictionary with ``success``, ``statistics``, ``validation`` summary, ``unique_symbol_count``, and optionally a ``detail_file`` path. """ - err = _require_sch_api() - if err: - return err - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) # Gather statistics stats = sch.get_statistics() - # Supplement label statistics — sch.get_statistics() often reports 0 - # for labels because the raw data parser doesn't count them. - # Use sch.labels.get_statistics() for accurate counts. - with contextlib.suppress(Exception): - label_stats = sch.labels.get_statistics() - if isinstance(label_stats, dict): - if "text_elements" not in stats or not isinstance(stats.get("text_elements"), dict): - stats["text_elements"] = {} - label_count = label_stats.get("total_labels", label_stats.get("item_count", 0)) - stats["text_elements"]["label"] = label_count - - with contextlib.suppress(Exception): - hier_labels = list(getattr(sch, "hierarchical_labels", [])) - if "text_elements" not in stats or not isinstance(stats.get("text_elements"), dict): - stats["text_elements"] = {} - stats["text_elements"]["global_label"] = len(hier_labels) - - # Fallback: parse (global_label ...) nodes from raw file when - # kicad-sch-api reports 0 — it has no sch.global_labels attribute. - if isinstance(stats.get("text_elements"), dict): - if stats["text_elements"].get("global_label", 0) == 0: - raw_global = parse_global_labels(schematic_path) - if raw_global: - stats["text_elements"]["global_label"] = len(raw_global) - - stats["text_elements"]["total_text_elements"] = ( - stats["text_elements"].get("label", 0) - + stats["text_elements"].get("global_label", 0) - ) + # Supplement text_elements for backwards compatibility + stats["text_elements"] = { + "label": stats.get("labels", 0), + "global_label": stats.get("global_labels", 0), + "total_text_elements": stats.get("labels", 0) + stats.get("global_labels", 0), + } # Run validation issues = sch.validate() - # Collect symbol details (for sidecar file) + # Collect symbol details lib_ids_seen: set[str] = set() symbol_details: list[dict[str, Any]] = [] for comp in sch.components: - lid = getattr(comp, "lib_id", None) + lid = comp.lib_id if lid and lid not in lib_ids_seen: lib_ids_seen.add(lid) try: - info = _ksa_get_symbol_info(lid) - if isinstance(info, dict): + info = _lr_get_info(lid) + if info is not None: symbol_details.append(info) else: - symbol_details.append( - { - "lib_id": lid, - "name": getattr(info, "name", str(info)), - "description": getattr(info, "description", None), - "pin_count": getattr(info, "pin_count", None), - } - ) + symbol_details.append({"lib_id": lid, "lookup_failed": True}) except Exception: symbol_details.append({"lib_id": lid, "lookup_failed": True}) - # Normalise stats and issues - if not isinstance(stats, dict): - stats = {"raw": str(stats)} - if not isinstance(issues, list): - issues = [str(issues)] if issues else [] - validation_passed = len(issues) == 0 logger.info("Retrieved info for %s: %d issues", schematic_path, len(issues)) @@ -928,17 +782,13 @@ def get_component_detail(schematic_path: str, reference: str) -> dict[str, Any]: Dictionary with ``success`` and detailed component data including properties, pin list, footprint, and validation results. """ - err = _require_sch_api() - if err: - return err - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) comp = sch.components.get(reference) if comp is None: return { @@ -949,119 +799,43 @@ def get_component_detail(schematic_path: str, reference: str) -> dict[str, Any]: detail: dict[str, Any] = { "reference": reference, - "lib_id": getattr(comp, "lib_id", None), - "value": getattr(comp, "value", None), - "footprint": getattr(comp, "footprint", None), + "lib_id": comp.lib_id, + "value": comp.value, + "footprint": comp.footprint, + "position": {"x": comp.position.x, "y": comp.position.y}, + "rotation": comp.rotation, } - # Position - pos = getattr(comp, "position", None) - if pos is not None: - if isinstance(pos, (list, tuple)) and len(pos) >= 2: - detail["position"] = {"x": pos[0], "y": pos[1]} - else: - detail["position"] = str(pos) - - # Rotation - detail["rotation"] = getattr(comp, "rotation", None) - # Properties - try: - if hasattr(comp, "to_dict"): - comp_dict = comp.to_dict() - detail["properties"] = comp_dict.get("properties", {}) - elif hasattr(comp, "properties"): - detail["properties"] = ( - dict(comp.properties.items()) - if hasattr(comp.properties, "items") - else str(comp.properties) - ) - except Exception: - detail["properties"] = None + detail["properties"] = comp.properties - # Pins — use get_symbol_definition().pins for metadata and - # sch.list_component_pins() for schematic-transformed positions. - # comp.pins is empty on loaded schematics (kicad-sch-api gap). + # Pins via SchDocument.list_component_pins() pins: list[dict[str, Any]] = [] + pin_data = sch.list_component_pins(reference) + for pin_num, pos in pin_data: + pins.append({ + "number": pin_num, + "position": {"x": pos.x, "y": pos.y}, + }) - pin_positions: dict[str, Any] = {} - with contextlib.suppress(Exception): - for pin_num, pos in sch.list_component_pins(reference): - pin_positions[str(pin_num)] = pos - - raw_pins: list[Any] | None = None - with contextlib.suppress(Exception): - sym_def = comp.get_symbol_definition() - if sym_def is not None: - raw_pins = list(sym_def.pins) - - if not raw_pins: - if hasattr(comp, "list_pins"): - with contextlib.suppress(Exception): - raw_pins = list(comp.list_pins()) - if not raw_pins: - raw_pins = list(getattr(comp, "pins", [])) - - # Last resort: parse pins from the raw (lib_symbols ...) section. - # Handles custom library symbols where get_symbol_definition() fails. - if not raw_pins: - lib_id = getattr(comp, "lib_id", None) - if lib_id: - sexp_pins = parse_lib_symbol_pins(schematic_path, str(lib_id)) - if sexp_pins: - comp_pos = getattr(comp, "position", None) - comp_rot = float(getattr(comp, "rotation", 0) or 0) - comp_mirror = getattr(comp, "mirror", None) - mirror_x = comp_mirror in ("x", True) if comp_mirror else False - cx = float(comp_pos.x) if comp_pos is not None and hasattr(comp_pos, "x") else 0.0 - cy = float(comp_pos.y) if comp_pos is not None and hasattr(comp_pos, "y") else 0.0 - - for sp in sexp_pins: - sx, sy = transform_pin_to_schematic( - sp["x"], sp["y"], cx, cy, comp_rot, mirror_x - ) - pins.append({ - "number": sp["number"], - "name": sp["name"], - "type": sp["type"], - "position": {"x": sx, "y": sy}, - }) - - for p in (raw_pins or []): - if isinstance(p, dict): - pins.append(p) - else: - pin_num = str(getattr(p, "number", getattr(p, "pin_number", ""))) - pin_entry: dict[str, Any] = { - "number": pin_num, - "name": getattr(p, "name", getattr(p, "pin_name", None)), - "type": str(getattr(p, "pin_type", getattr(p, "electrical_type", getattr(p, "type", None)))), - } - if pin_num in pin_positions: - spos = pin_positions[pin_num] - pin_entry["position"] = {"x": spos.x, "y": spos.y} - else: - ppos = getattr(p, "position", None) - if ppos is not None: - if hasattr(ppos, "x"): - pin_entry["position"] = {"x": ppos.x, "y": ppos.y} - else: - pin_entry["position"] = str(ppos) - pins.append(pin_entry) + # Enrich pins with name/type from lib_resolver if available + lib_info = _lr_get_info(comp.lib_id) + if lib_info and "pins" in lib_info: + lib_pins_by_num = { + str(p["number"]): p for p in lib_info["pins"] + } + for pin in pins: + lib_pin = lib_pins_by_num.get(str(pin["number"])) + if lib_pin: + pin["name"] = lib_pin.get("name") + pin["type"] = lib_pin.get("type") detail["pins"] = pins detail["pin_count"] = len(pins) # BOM/board flags - detail["in_bom"] = getattr(comp, "in_bom", None) - detail["on_board"] = getattr(comp, "on_board", None) - - # Validation - try: - if hasattr(comp, "validate"): - detail["validation"] = comp.validate() - except Exception: - detail["validation"] = None + detail["in_bom"] = comp.in_bom + detail["on_board"] = comp.on_board logger.info("Got detail for %s in %s", reference, schematic_path) return { @@ -1080,7 +854,7 @@ def get_schematic_hierarchy(schematic_path: str) -> dict[str, Any]: """Get the hierarchical sheet tree of a KiCad schematic. Returns the sheet structure with filenames and per-sheet component - counts. Essential for multi-sheet designs — use this to understand + counts. Essential for multi-sheet designs -- use this to understand the design's organization before diving into specific sheets. Args: @@ -1089,17 +863,13 @@ def get_schematic_hierarchy(schematic_path: str) -> dict[str, Any]: Returns: Dictionary with ``success`` and a ``hierarchy`` tree of sheets. """ - err = _require_sch_api() - if err: - return err - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) schematic_dir = os.path.dirname(schematic_path) def _count_components(filepath: str) -> int: @@ -1107,77 +877,68 @@ def get_schematic_hierarchy(schematic_path: str) -> dict[str, Any]: if not os.path.isfile(filepath): return -1 try: - sheet_sch = _ksa_load(filepath) - return len(list(sheet_sch.components)) + sheet_sch = load_schematic(filepath) + return len(sheet_sch.components) except Exception: return -1 - def _walk_hierarchy(node: dict[str, Any], base_dir: str) -> dict[str, Any]: - """Convert a SheetManager hierarchy node into our output format.""" - filename = node.get("filename", "") - info: dict[str, Any] = { - "name": node.get("name", "unknown"), - "filename": filename, - } + def _walk_sheets(sheets: list, base_dir: str) -> list[dict[str, Any]]: + """Recursively walk sheets to build the hierarchy tree.""" + result_sheets: list[dict[str, Any]] = [] + for sheet in sheets: + filename = sheet.filename + info: dict[str, Any] = { + "name": sheet.name, + "filename": filename, + } - if filename: - filepath = os.path.join(base_dir, filename) - count = _count_components(filepath) - if count >= 0: - info["component_count"] = count + if filename: + filepath = os.path.join(base_dir, filename) + count = _count_components(filepath) + if count >= 0: + info["component_count"] = count + else: + info["component_count"] = 0 + info["note"] = "file not found or failed to load" + + # Recurse into child sheets + if os.path.isfile(filepath): + with contextlib.suppress(Exception): + child_sch = load_schematic(filepath) + if child_sch.sheets: + child_dir = os.path.dirname(filepath) + info["sheets"] = _walk_sheets(child_sch.sheets, child_dir) else: info["component_count"] = 0 - info["note"] = "file not found or failed to load" - else: - info["component_count"] = 0 - children = node.get("children", []) - if children: - info["sheets"] = [_walk_hierarchy(child, base_dir) for child in children] - info["total_sheets"] = 1 + sum( - c.get("total_sheets", 1) for c in info["sheets"] - ) - else: - info["total_sheets"] = 1 + # Count total sheets + child_sheets = info.get("sheets", []) + if child_sheets: + info["total_sheets"] = 1 + sum( + c.get("total_sheets", 1) for c in child_sheets + ) + else: + info["total_sheets"] = 1 - return info + result_sheets.append(info) - # Use SheetManager API — sch.sheets.get_sheet_hierarchy() returns a - # tree with children already identified from the S-expression data. - hierarchy_data = None - with contextlib.suppress(Exception): - hierarchy_data = sch.sheets.get_sheet_hierarchy() + return result_sheets - if hierarchy_data and isinstance(hierarchy_data, dict) and "root" in hierarchy_data: - root_node = hierarchy_data["root"] + root_info: dict[str, Any] = { + "name": "root", + "filename": os.path.basename(schematic_path), + "component_count": len(sch.components), + } - root_info: dict[str, Any] = { - "name": "root", - "filename": os.path.basename(schematic_path), - "component_count": len(list(sch.components)), - } - - children = root_node.get("children", []) - if children: - root_info["sheets"] = [ - _walk_hierarchy(child, schematic_dir) for child in children - ] - root_info["total_sheets"] = 1 + sum( - c.get("total_sheets", 1) for c in root_info["sheets"] - ) - else: - root_info["total_sheets"] = 1 - - hierarchy = root_info + if sch.sheets: + root_info["sheets"] = _walk_sheets(sch.sheets, schematic_dir) + root_info["total_sheets"] = 1 + sum( + c.get("total_sheets", 1) for c in root_info["sheets"] + ) else: - # Fallback: report just the root sheet - hierarchy = { - "name": "root", - "filename": os.path.basename(schematic_path), - "component_count": len(list(sch.components)), - "total_sheets": 1, - "note": "Sheet hierarchy API unavailable — only root sheet reported", - } + root_info["total_sheets"] = 1 + + hierarchy = root_info logger.info("Got hierarchy for %s: %d total sheets", schematic_path, hierarchy["total_sheets"]) return { diff --git a/src/mckicad/tools/schematic_analysis.py b/src/mckicad/tools/schematic_analysis.py index b1a2e47..e3d43c9 100644 --- a/src/mckicad/tools/schematic_analysis.py +++ b/src/mckicad/tools/schematic_analysis.py @@ -2,9 +2,8 @@ Schematic analysis and export tools for the mckicad MCP server. Provides ERC checking, connectivity analysis, pin-level queries, netlist -export, and PDF export. Uses kicad-sch-api for structural analysis and -kicad-cli for rule checks and file exports, falling back gracefully when -either dependency is unavailable. +export, and PDF export. Uses SchDocument for structural analysis and +kicad-cli for rule checks and file exports. """ import contextlib @@ -18,8 +17,9 @@ from mckicad.config import INLINE_RESULT_THRESHOLD, TIMEOUT_CONSTANTS from mckicad.server import mcp from mckicad.utils.file_utils import write_detail_file from mckicad.utils.kicad_cli import find_kicad_cli +from mckicad.utils.sch_helpers import expand as _expand +from mckicad.utils.sch_helpers import load_schematic, validate_schematic_path as _validate_schematic_path from mckicad.utils.sexp_parser import ( - parse_global_labels, parse_lib_symbol_pins, transform_pin_to_schematic, ) @@ -29,55 +29,9 @@ from mckicad.utils.sexp_parser import ( logger = logging.getLogger(__name__) -# --------------------------------------------------------------------------- -# Engine detection -# --------------------------------------------------------------------------- - -_HAS_SCH_API = False - -try: - from kicad_sch_api import load_schematic as _ksa_load - - _HAS_SCH_API = True -except ImportError: - logger.warning( - "kicad-sch-api not installed -- schematic analysis tools will return helpful errors. " - "Install with: uv add kicad-sch-api" - ) - - -def _require_sch_api() -> dict[str, Any] | None: - """Return an error dict if kicad-sch-api is unavailable, else None.""" - if not _HAS_SCH_API: - return { - "success": False, - "error": "kicad-sch-api is not installed. Install it with: uv add kicad-sch-api", - "engine": "none", - } - return None - def _get_schematic_engine() -> str: - if _HAS_SCH_API: - return "kicad-sch-api" - return "none" - - -def _validate_schematic_path(path: str, must_exist: bool = True) -> dict[str, Any] | None: - """Validate a schematic file path. Returns an error dict on failure, else None.""" - if not path: - return {"success": False, "error": "Schematic path must be a non-empty string"} - expanded = os.path.expanduser(path) - if not expanded.endswith(".kicad_sch"): - return {"success": False, "error": f"Path must end with .kicad_sch, got: {path}"} - if must_exist and not os.path.isfile(expanded): - return {"success": False, "error": f"Schematic file not found: {expanded}"} - return None - - -def _expand(path: str) -> str: - """Expand ~ and return an absolute path.""" - return os.path.abspath(os.path.expanduser(path)) + return "sch_document" def _resolve_root_schematic(schematic_path: str) -> str | None: @@ -315,14 +269,17 @@ def _build_connectivity_state( except Exception: pass - # Global labels from raw file — kicad-sch-api has no sch.global_labels - # attribute, so (global_label ...) nodes are invisible to the API. - # Parse them directly from the S-expression when the file path is known. - if schematic_path: - for gl in parse_global_labels(schematic_path): - coord = _rc(gl["x"], gl["y"]) - _find(coord) - label_at[coord] = gl["text"] + # Global labels — SchDocument exposes these natively + try: + for gl in getattr(sch, "global_labels", []): + text = getattr(gl, "text", None) + pos = getattr(gl, "position", None) + if text and pos: + coord = _rc(*_coord_from_point(pos)) + _find(coord) + label_at[coord] = str(text) + except Exception: + pass # Power symbols — their value acts as a net name (e.g. GND, VCC) for comp in getattr(sch, "components", []): @@ -470,40 +427,13 @@ def run_schematic_erc( "error": f"Invalid severity filter: {severity}. Must be 'all', 'error', or 'warning'.", } - # --- Attempt 1: kicad-sch-api --- - if _HAS_SCH_API: - try: - from kicad_sch_api import ElectricalRulesChecker # type: ignore[attr-defined] - - sch = _ksa_load(schematic_path) - checker = ElectricalRulesChecker(sch) - violations = checker.run() - - logger.info("ERC via kicad-sch-api found %d violation(s)", len(violations)) - return _format_erc_result(schematic_path, violations, severity, "kicad-sch-api") - - except (ImportError, AttributeError) as exc: - logger.info( - "kicad-sch-api ElectricalRulesChecker not available (%s), trying kicad-cli", - exc, - ) - except Exception as exc: - logger.warning("kicad-sch-api ERC failed (%s), trying kicad-cli", exc) - - # --- Attempt 2: kicad-cli --- + # ERC via kicad-cli cli_path, cli_err = _require_kicad_cli() if cli_err: - # Neither engine is available - api_err = _require_sch_api() - if api_err: - return { - "success": False, - "error": ( - "No ERC engine available. Install kicad-sch-api (uv add kicad-sch-api) " - "or ensure kicad-cli is on your PATH." - ), - } - return cli_err + return { + "success": False, + "error": "kicad-cli not found. Ensure kicad-cli is on your PATH.", + } assert cli_path is not None try: @@ -630,18 +560,13 @@ def analyze_connectivity(schematic_path: str) -> dict[str, Any]: Dictionary with ``net_count``, ``connection_count``, ``unconnected_pins``, and ``detail_file``. """ - err = _require_sch_api() - if err: - return err - verr = _validate_schematic_path(schematic_path) if verr: return verr - schematic_path = _expand(schematic_path) try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) net_graph, unconnected = _build_connectivity(sch, schematic_path) total_connections = sum(len(pins) for pins in net_graph.values()) @@ -693,18 +618,13 @@ def check_pin_connection( Dictionary with ``net`` name and ``connected_to`` list of ``{reference, pin}`` dicts for every other pin on the same net. """ - err = _require_sch_api() - if err: - return err - verr = _validate_schematic_path(schematic_path) if verr: return verr - schematic_path = _expand(schematic_path) try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) net_graph, _unconnected = _build_connectivity(sch, schematic_path) # Find which net this pin belongs to @@ -775,18 +695,13 @@ def verify_pins_connected( Dictionary with ``connected`` boolean, plus ``from`` and ``to`` endpoint descriptions. """ - err = _require_sch_api() - if err: - return err - verr = _validate_schematic_path(schematic_path) if verr: return verr - schematic_path = _expand(schematic_path) try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) net_graph, _unconnected = _build_connectivity(sch, schematic_path) # Find the net for each pin @@ -851,18 +766,13 @@ def get_component_pins( Dictionary with ``reference``, ``pin_count``, and ``pins`` list where each entry describes one pin. """ - err = _require_sch_api() - if err: - return err - verr = _validate_schematic_path(schematic_path) if verr: return verr - schematic_path = _expand(schematic_path) try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) # Locate the component comp = None @@ -1239,10 +1149,6 @@ def audit_wiring( (unconnected pins, high-fanout nets, auto-named nets), and optionally ``detail_file`` when wire data is offloaded. """ - err = _require_sch_api() - if err: - return err - verr = _validate_schematic_path(schematic_path) if verr: return verr @@ -1253,7 +1159,7 @@ def audit_wiring( schematic_path = _expand(schematic_path) try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) state = _build_connectivity_state(sch, schematic_path) find_fn = state["find"] @@ -1430,10 +1336,6 @@ def verify_connectivity( Dictionary with ``verified`` count, ``failed`` count, and ``results`` list with per-net status (``match``, ``mismatch``, ``missing_net``). """ - err = _require_sch_api() - if err: - return err - verr = _validate_schematic_path(schematic_path) if verr: return verr @@ -1444,7 +1346,7 @@ def verify_connectivity( schematic_path = _expand(schematic_path) try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) state = _build_connectivity_state(sch, schematic_path) net_graph = state["net_graph"] @@ -1666,18 +1568,9 @@ def _run_connectivity_raw(schematic_path: str) -> dict[str, Any]: except Exception as exc: logger.warning("Netlist-based connectivity failed: %s", exc) - # Fallback: single-file kicad-sch-api analysis - if not _HAS_SCH_API: - return { - "success": False, - "error": "kicad-cli and kicad-sch-api both unavailable", - "net_count": 0, - "connection_count": 0, - "unconnected_pins": 0, - } - + # Fallback: single-file SchDocument analysis try: - sch = _ksa_load(expanded) + sch = load_schematic(expanded) net_graph, unconnected = _build_connectivity(sch, expanded) total_connections = sum(len(pins) for pins in net_graph.values()) return { @@ -1685,7 +1578,7 @@ def _run_connectivity_raw(schematic_path: str) -> dict[str, Any]: "net_count": len(net_graph), "connection_count": total_connections, "unconnected_pins": len(unconnected), - "engine": "kicad-sch-api", + "engine": "sch_document", } except Exception as exc: return { diff --git a/src/mckicad/tools/schematic_edit.py b/src/mckicad/tools/schematic_edit.py index 120d041..25b0da4 100644 --- a/src/mckicad/tools/schematic_edit.py +++ b/src/mckicad/tools/schematic_edit.py @@ -2,18 +2,19 @@ Schematic editing tools for modifying existing elements in KiCad schematics. Complements the creation-oriented tools in schematic.py with modification, -removal, and annotation capabilities. Uses the same kicad-sch-api engine -and follows the same swap-point pattern for future kipy IPC support. +removal, and annotation capabilities. Also includes s-expression-direct tools (e.g. ``remove_wires_by_criteria``) -that bypass kicad-sch-api for operations it doesn't support well. +that bypass the schematic engine for operations that work directly on the +s-expression tree. """ import logging -import os from typing import Any from mckicad.server import mcp +from mckicad.utils.sch_helpers import expand as _expand +from mckicad.utils.sch_helpers import load_schematic, validate_schematic_path as _validate_schematic_path from mckicad.utils.sexp_parser import ( parse_wire_segments, remove_sexp_blocks_by_uuid, @@ -21,66 +22,9 @@ from mckicad.utils.sexp_parser import ( logger = logging.getLogger(__name__) -# --------------------------------------------------------------------------- -# Engine abstraction — mirrors schematic.py setup -# --------------------------------------------------------------------------- - -_HAS_SCH_API = False - -try: - from kicad_sch_api import load_schematic as _ksa_load - - _HAS_SCH_API = True -except ImportError: - logger.warning( - "kicad-sch-api not installed — schematic edit tools will return helpful errors. " - "Install with: uv add kicad-sch-api" - ) - def _get_schematic_engine() -> str: - """Return the name of the active schematic manipulation engine.""" - if _HAS_SCH_API: - return "kicad-sch-api" - return "none" - - -def _require_sch_api() -> dict[str, Any] | None: - """Return an error dict if kicad-sch-api is unavailable, else None.""" - if not _HAS_SCH_API: - return { - "success": False, - "error": "kicad-sch-api is not installed. Install it with: uv add kicad-sch-api", - "engine": "none", - } - return None - - -def _validate_schematic_path(path: str, must_exist: bool = True) -> dict[str, Any] | None: - """Validate a schematic file path. Returns an error dict on failure, else None.""" - if not path: - return {"success": False, "error": "Schematic path must be a non-empty string"} - - expanded = os.path.expanduser(path) - - if not expanded.endswith(".kicad_sch"): - return { - "success": False, - "error": f"Path must end with .kicad_sch, got: {path}", - } - - if must_exist and not os.path.isfile(expanded): - return { - "success": False, - "error": f"Schematic file not found: {expanded}", - } - - return None - - -def _expand(path: str) -> str: - """Expand ~ and return an absolute path.""" - return os.path.abspath(os.path.expanduser(path)) + return "sch_document" # --------------------------------------------------------------------------- @@ -123,14 +67,10 @@ def modify_component( Returns: Dictionary with ``success``, ``reference``, ``modified`` list, and ``engine``. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) if not reference: return {"success": False, "error": "reference must be a non-empty string"} @@ -143,7 +83,7 @@ def modify_component( } try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) comp = sch.components.get(reference) if comp is None: return { @@ -162,11 +102,11 @@ def modify_component( modified.append(f"rotation={rotation}") if value is not None: - comp.value = value + comp.set_value(value) modified.append(f"value={value}") if footprint is not None: - comp.footprint = footprint + comp.set_footprint(footprint) modified.append(f"footprint={footprint}") if properties: @@ -175,11 +115,11 @@ def modify_component( modified.append(f"property[{key}]={val}") if in_bom is not None: - comp.in_bom = in_bom + comp.set_in_bom(in_bom) modified.append(f"in_bom={in_bom}") if on_board is not None: - comp.on_board = on_board + comp.set_on_board(on_board) modified.append(f"on_board={on_board}") if not modified: @@ -227,20 +167,16 @@ def remove_component(schematic_path: str, reference: str) -> dict[str, Any]: Returns: Dictionary with ``success``, removed ``reference``, and ``engine``. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) if not reference: return {"success": False, "error": "reference must be a non-empty string"} try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) removed = sch.components.remove(reference) if removed is False: return { @@ -283,20 +219,16 @@ def remove_wire(schematic_path: str, wire_id: str) -> dict[str, Any]: Returns: Dictionary with ``success``, removed ``wire_id``, and ``engine``. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) if not wire_id: return {"success": False, "error": "wire_id must be a non-empty string"} try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) sch.remove_wire(wire_id) sch.save(schematic_path) @@ -332,20 +264,16 @@ def remove_label(schematic_path: str, label_id: str) -> dict[str, Any]: Returns: Dictionary with ``success``, removed ``label_id``, and ``engine``. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) if not label_id: return {"success": False, "error": "label_id must be a non-empty string"} try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) sch.remove_label(label_id) sch.save(schematic_path) @@ -392,14 +320,10 @@ def set_title_block( Returns: Dictionary with ``success``, ``fields_set`` list, and ``engine``. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) # Build kwargs for kicad-sch-api's set_title_block(). # API signature: set_title_block(title, date, rev, company, comments) @@ -432,7 +356,7 @@ def set_title_block( } try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) sch.set_title_block(**kwargs) sch.save(schematic_path) @@ -470,21 +394,17 @@ def add_text_annotation( Returns: Dictionary with ``success``, ``text_id``, ``position``, and ``engine``. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) if not text: return {"success": False, "error": "text must be a non-empty string"} try: - sch = _ksa_load(schematic_path) - text_id = sch.add_text(text=text, position=(x, y)) + sch = load_schematic(schematic_path) + text_id = sch.add_text(text, x, y) sch.save(schematic_path) logger.info("Added text annotation at (%.1f, %.1f) in %s", x, y, schematic_path) @@ -521,18 +441,14 @@ def add_no_connect( Returns: Dictionary with ``success``, ``position``, and ``engine``. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) try: - sch = _ksa_load(schematic_path) - sch.no_connects.add(position=(x, y)) + sch = load_schematic(schematic_path) + sch.add_no_connect(x, y) sch.save(schematic_path) logger.info("Added no-connect at (%.1f, %.1f) in %s", x, y, schematic_path) @@ -562,17 +478,13 @@ def backup_schematic(schematic_path: str) -> dict[str, Any]: Returns: Dictionary with ``success``, ``backup_path``, and ``engine``. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) try: - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) backup_path = sch.backup() logger.info("Backed up %s to %s", schematic_path, backup_path) diff --git a/src/mckicad/tools/schematic_patterns.py b/src/mckicad/tools/schematic_patterns.py index 6173724..f67b512 100644 --- a/src/mckicad/tools/schematic_patterns.py +++ b/src/mckicad/tools/schematic_patterns.py @@ -8,53 +8,17 @@ for use in standalone scripts without MCP. """ import logging -import os from typing import Any from mckicad.server import mcp +from mckicad.utils.sch_helpers import expand as _expand +from mckicad.utils.sch_helpers import load_schematic, validate_schematic_path as _validate_schematic_path logger = logging.getLogger(__name__) -_HAS_SCH_API = False - -try: - from kicad_sch_api import load_schematic as _ksa_load - - _HAS_SCH_API = True -except ImportError: - logger.warning( - "kicad-sch-api not installed — pattern tools will return helpful errors. " - "Install with: uv add kicad-sch-api" - ) - - -def _require_sch_api() -> dict[str, Any] | None: - if not _HAS_SCH_API: - return { - "success": False, - "error": "kicad-sch-api is not installed. Install it with: uv add kicad-sch-api", - "engine": "none", - } - return None - def _get_schematic_engine() -> str: - return "kicad-sch-api" if _HAS_SCH_API else "none" - - -def _validate_schematic_path(path: str, must_exist: bool = True) -> dict[str, Any] | None: - if not path: - return {"success": False, "error": "Schematic path must be a non-empty string"} - expanded = os.path.expanduser(path) - if not expanded.endswith(".kicad_sch"): - return {"success": False, "error": f"Path must end with .kicad_sch, got: {path}"} - if must_exist and not os.path.isfile(expanded): - return {"success": False, "error": f"Schematic file not found: {expanded}"} - return None - - -def _expand(path: str) -> str: - return os.path.abspath(os.path.expanduser(path)) + return "sch_document" @mcp.tool() @@ -93,14 +57,10 @@ def place_decoupling_bank_pattern( Dictionary with placed component references, power symbol details, and grid bounds. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) # Parse comma-separated values into cap specs values = [v.strip() for v in cap_values.split(",") if v.strip()] @@ -112,7 +72,7 @@ def place_decoupling_bank_pattern( try: from mckicad.patterns.decoupling_bank import place_decoupling_bank - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) result = place_decoupling_bank( sch=sch, caps=caps, @@ -173,19 +133,15 @@ def place_pull_resistor_pattern( Returns: Dictionary with resistor reference, wire ID, and power symbol details. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) try: from mckicad.patterns.pull_resistor import place_pull_resistor - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) result = place_pull_resistor( sch=sch, signal_ref=signal_ref, @@ -253,19 +209,15 @@ def place_crystal_pattern( Dictionary with crystal and capacitor references, wire IDs, and ground symbol details. """ - err = _require_sch_api() - if err: - return err - - schematic_path = _expand(schematic_path) verr = _validate_schematic_path(schematic_path) if verr: return verr + schematic_path = _expand(schematic_path) try: from mckicad.patterns.crystal_oscillator import place_crystal_with_caps - sch = _ksa_load(schematic_path) + sch = load_schematic(schematic_path) result = place_crystal_with_caps( sch=sch, xtal_value=xtal_value, diff --git a/src/mckicad/utils/lib_resolver.py b/src/mckicad/utils/lib_resolver.py new file mode 100644 index 0000000..38ff381 --- /dev/null +++ b/src/mckicad/utils/lib_resolver.py @@ -0,0 +1,541 @@ +"""Unified symbol library resolution for KiCad. + +Given a lib_id like ``"Device:R"``, finds the ``.kicad_sym`` file and +extracts the full symbol sexp tree. Reads both global and project +``sym-lib-table`` files, resolving KiCad path variables. + +This module is the eventual replacement for the ad-hoc library search +in sexp_parser.py, providing proper sym-lib-table parsing with variable +substitution and mtime-based cache invalidation. +""" + +from __future__ import annotations + +import logging +import os +from typing import Any + +from mckicad.config import get_kicad_user_dir +from mckicad.utils.sexp_tree import ( + SexpNode, + find, + find_named, + find_one, + find_recursive, + get_values, + parse, + parse_file, + text_at, +) + +logger = logging.getLogger(__name__) + +# Cache: path -> (mtime, parsed table dict) +_lib_table_cache: dict[str, tuple[float, dict[str, str]]] = {} + + +def resolve_symbol( + lib_id: str, *, project_dir: str | None = None, +) -> SexpNode | None: + """Resolve a lib_id to its parsed symbol node. + + Args: + lib_id: Full library identifier (e.g. ``"Device:R"``). + project_dir: Project directory for ``${KIPRJMOD}`` substitution. + + Returns: + The parsed SexpNode for the symbol, or None if not found. + """ + if ":" not in lib_id: + return None + + library_name, symbol_name = lib_id.split(":", 1) + if not library_name or not symbol_name: + return None + + lib_path = resolve_library_path(library_name, project_dir=project_dir) + if lib_path is None: + return None + + try: + tree = parse_file(lib_path) + except (OSError, UnicodeDecodeError): + logger.debug("Failed to parse library file: %s", lib_path) + return None + + return find_named(tree, "symbol", symbol_name) + + +def resolve_symbol_with_tree( + lib_id: str, *, project_dir: str | None = None, +) -> tuple[SexpNode | None, SexpNode | None]: + """Resolve a lib_id, returning both the symbol node and its library tree. + + The library tree is needed for ``extract_symbol_pins`` and + ``extract_pin_unit_map`` when the symbol uses ``(extends ...)`` + to inherit pin geometry from a parent symbol. + + Returns: + ``(symbol_node, lib_tree)`` tuple. Both are None if resolution fails. + """ + if ":" not in lib_id: + return None, None + + library_name, symbol_name = lib_id.split(":", 1) + if not library_name or not symbol_name: + return None, None + + lib_path = resolve_library_path(library_name, project_dir=project_dir) + if lib_path is None: + return None, None + + try: + tree = parse_file(lib_path) + except (OSError, UnicodeDecodeError): + logger.debug("Failed to parse library file: %s", lib_path) + return None, None + + sym = find_named(tree, "symbol", symbol_name) + return sym, tree + + +def resolve_library_path( + library_name: str, *, project_dir: str | None = None, +) -> str | None: + """Resolve a library name to its ``.kicad_sym`` file path. + + Search order: + 1. Project sym-lib-table (if project_dir given) + 2. Global sym-lib-table (KiCad 9, then 8) + 3. Direct file search (project dir, libs/, ../libs/) + """ + # 1. Project sym-lib-table + if project_dir: + table_path = os.path.join(project_dir, "sym-lib-table") + if os.path.isfile(table_path): + table = load_sym_lib_table(table_path) + if library_name in table: + uri = _substitute_variables(table[library_name], project_dir) + if os.path.isfile(uri): + return uri + + # 2. Global sym-lib-tables (KiCad 9 first, then 8) + for version_dir in ("9.0", "8.0"): + config_dir = os.path.expanduser(f"~/.config/kicad/{version_dir}") + table_path = os.path.join(config_dir, "sym-lib-table") + if os.path.isfile(table_path): + table = load_sym_lib_table(table_path) + if library_name in table: + uri = _substitute_variables(table[library_name], project_dir) + if os.path.isfile(uri): + return uri + + # 3. Direct file search (fallback for projects without sym-lib-table) + if project_dir: + filename = f"{library_name}.kicad_sym" + candidates = [ + os.path.join(project_dir, filename), + os.path.join(project_dir, "libs", filename), + os.path.join(project_dir, os.pardir, "libs", filename), + ] + for candidate in candidates: + resolved = os.path.normpath(candidate) + if os.path.isfile(resolved): + return resolved + + return None + + +def extract_symbol_pins( + symbol_node: SexpNode, + *, + lib_tree: SexpNode | None = None, +) -> list[dict[str, Any]]: + """Extract pin dicts from a parsed symbol node. + + Output format matches ``sexp_parser.parse_lib_symbol_pins()``: + ``{number, name, type, shape, x, y, rotation, length}`` + + Searches through sub-symbols (children with tag "symbol") to find + all pin definitions. If the symbol uses ``(extends "Parent")``, + follows the chain to find pin definitions in the parent symbol + (requires ``lib_tree`` — the parsed library file root). + """ + # Check for extends chain + effective = _resolve_extends(symbol_node, lib_tree) + + pins: list[dict[str, Any]] = [] + all_pin_nodes = find_recursive(effective, "pin") + + for pin_node in all_pin_nodes: + pin_dict = _parse_pin_node(pin_node) + if pin_dict is not None: + pins.append(pin_dict) + + return pins + + +def extract_pin_unit_map( + symbol_node: SexpNode, + *, + lib_tree: SexpNode | None = None, +) -> dict[str, int]: + """Build pin_number -> unit_number map from sub-symbols. + + KiCad sub-symbols are named ``__