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 ``__