Loading files
This file is pinned to the selected release. Download the original to open it in your design tool.
670d2a9c4This file is pinned to the selected release. Download the original to open it in your design tool.
"""Schematic-domain command handlers for KiCADInterface.
Extracted verbatim from kicad_interface.py as a mixin to keep that dispatcher
file manageable. These methods are mixed into KiCADInterface, so ``self`` is the
full interface instance: every ``self.board``, ``self._auto_save_board()``,
cross-handler call, and the command_routes table resolve exactly as before.
No behavior change — pure relocation.
"""
import base64
import glob
import json
import logging
import os
import re
import shutil
import subprocess
import tempfile
import traceback
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
import pcbnew
import sexpdata
from commands.library_schematic import LibraryManager as SchematicLibraryManager
from commands.schematic import SchematicLoadError, SchematicManager
from commands.wire_manager import WireManager
from utils.board_items import clone_footprint
from utils.interactive_schematic import reload_kicad_schematic
from utils.kicad_cli import kicad_cli_not_found_message, resolve_kicad_cli
from utils.project_settings_guard import preserve_project_settings
from utils.sexpr_format import QUOTED_VALUE
from utils.sexpr_format import dumps as kicad_dumps
from utils.sexpr_format import escape_sexpr_string, unescape_sexpr_string
from utils.symbol_instances import annotate_sheet, instance_report
logger = logging.getLogger("kicad_interface")
def _find_placed_symbol_position(content: str, reference: str) -> Optional[Tuple[float, float]]:
"""Locate the ``(at x y ...)`` origin of the symbol instance ``reference``.
Matches by reference, not by lib_id: with several instances of the same
symbol in one schematic (two resistors), a first-match lib_id search
reports the coordinates of the WRONG instance for every placement after
the first. Scans symbol headers and picks the block whose
``(property "Reference" ...)`` is the requested one. Returns ``None`` if
no block matches (best-effort caller metadata, not a load-bearing path).
"""
headers = list(
re.finditer(
r'\(symbol\s+\(lib_id\s+"[^"]+"\)\s+\(at\s+(-?[\d.]+)\s+(-?[\d.]+)',
content,
)
)
ref_needle = f'(property "Reference" "{reference}"'
for i, header in enumerate(headers):
block_end = headers[i + 1].start() if i + 1 < len(headers) else len(content)
if ref_needle in content[header.start() : block_end]:
return float(header.group(1)), float(header.group(2))
return None
def _pick_root_svg(svg_dir: str, schematic_path: str) -> Optional[str]:
"""Return the root-page SVG produced by ``kicad-cli sch export svg``.
kicad-cli names the root page ``<schematic-stem>.svg`` and hierarchical
sub-sheet pages ``<stem>-<SheetName>.svg``. Returns the absolute path of
the stem-matched root SVG when present, the sole SVG when exactly one was
produced, or None when the root page cannot be identified.
"""
stem = Path(schematic_path).stem
candidate = os.path.join(svg_dir, stem + ".svg")
if os.path.isfile(candidate):
return candidate
svg_files = sorted(glob.glob(os.path.join(svg_dir, "*.svg")))
if len(svg_files) == 1:
return svg_files[0]
return None
def _svg_to_png(svg_path: str, width: int, height: int) -> Optional[bytes]:
"""Convert SVG to PNG at the requested pixel dimensions.
Priority:
1. pymupdf (fitz) — the declared dependency (see requirements.txt).
Bundles the MuPDF renderer with binary wheels on all three
platforms, so `pip install -r requirements.txt` alone guarantees a
working converter — cairosvg cannot promise that on Windows, where
the native cairo runtime is typically absent (#274 review).
2. cairosvg — honors output_width/output_height; fine wherever the
cairo runtime exists.
3. Inkscape CLI — accurate KiCAD SVG rendering, if installed.
4. ImageMagick convert — broad availability fallback.
Returns PNG bytes or None if all converters fail.
Rasterizing here is essential (#274): returning the raw KiCAD SVG
instead produces a full-sheet vector document (title block, grid, every
path and font) whose text easily exceeds an MCP client's inline size
cap, and width/height have no effect on it.
"""
import subprocess
import tempfile
try:
import fitz
doc = fitz.open(svg_path)
page = doc[0]
mat = fitz.Matrix(width / page.rect.width, height / page.rect.height)
return page.get_pixmap(matrix=mat).tobytes("png")
except Exception:
pass
try:
import cairosvg
return cairosvg.svg2png(
url=svg_path,
output_width=int(width),
output_height=int(height),
)
except Exception:
pass
out_path = Path(tempfile.mkdtemp()) / "out.png"
try:
r = subprocess.run(
[
"inkscape",
svg_path,
"--export-type=png",
f"--export-width={width}",
f"--export-height={height}",
f"--export-filename={out_path}",
],
capture_output=True,
timeout=60,
)
if r.returncode == 0 and out_path.exists():
with open(out_path, "rb") as f:
return f.read()
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
try:
r = subprocess.run(
["convert", "-density", "150", svg_path, "-resize", f"{width}x{height}", str(out_path)],
capture_output=True,
timeout=60,
)
if r.returncode == 0 and out_path.exists():
with open(out_path, "rb") as f:
return f.read()
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
return None
class SchematicHandlersMixin:
"""Schematic-domain handlers mixed into KiCADInterface."""
# Provided by the host KiCADInterface this mixin is composed into. Declared here so
# mypy can resolve ``self.board`` when type-checking this module in isolation (the
# module docstring notes the mixin relies on the host's ``self.board``).
board: Any
def _reload_kicad_schematic(self) -> None:
"""Opt-in: reload open Schematic Editor after external file edits (Windows)."""
reload_kicad_schematic()
def _handle_create_schematic(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new schematic"""
logger.info("Creating schematic")
try:
# Support multiple parameter naming conventions for compatibility:
# - TypeScript tools use: name, path
# - Python schema uses: filename, title
# - Legacy uses: projectName, path, metadata
project_name = params.get("projectName") or params.get("name") or params.get("title")
# Handle filename parameter - it may contain full path
filename = params.get("filename")
if filename:
# If filename provided, extract name and path from it
if filename.endswith(".kicad_sch"):
filename = filename[:-10] # Remove .kicad_sch extension
filename_p = Path(filename)
path = str(filename_p.parent) if str(filename_p.parent) != "" else "."
project_name = project_name or filename_p.name
else:
path = params.get("path", ".")
metadata = params.get("metadata", {})
if not project_name:
return {
"success": False,
"message": "Schematic name is required. Provide 'name', 'projectName', or 'filename' parameter.",
}
sch_path = path if path and path != "." else None
schematic = SchematicManager.create_schematic(
project_name, path=sch_path, metadata=metadata
)
# Resolve the saved file path the same way create_schematic does: when
# `path` is already a full ".kicad_sch" file path, use it directly;
# otherwise treat it as a directory and append the file name. This keeps
# the save target in step with the created file and avoids doubling the
# name into ".../V4.kicad_sch/V4.kicad_sch" (issue #242).
if sch_path and sch_path.endswith(".kicad_sch"):
file_path = sch_path
else:
base_name = (
project_name
if project_name.endswith(".kicad_sch")
else f"{project_name}.kicad_sch"
)
normalized_path = path or "."
file_path = str(Path(normalized_path) / base_name)
success = SchematicManager.save_schematic(schematic, file_path)
return {"success": success, "file_path": file_path}
except Exception as e:
logger.error(f"Error creating schematic: {str(e)}")
return {"success": False, "message": str(e)}
def _handle_load_schematic(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Load an existing schematic"""
logger.info("Loading schematic")
try:
filename = params.get("filename")
if not filename:
return {"success": False, "message": "Filename is required"}
try:
schematic = SchematicManager.load_schematic(filename)
except SchematicLoadError as e:
return e.to_response()
metadata = SchematicManager.get_schematic_metadata(schematic)
return {"success": True, "metadata": metadata}
except Exception as e:
logger.error(f"Error loading schematic: {str(e)}")
return {"success": False, "message": str(e)}
def _handle_add_schematic_component(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Add a component to a schematic using text-based injection (no sexpdata)"""
logger.info("Adding component to schematic")
try:
from pathlib import Path
from commands.dynamic_symbol_loader import DynamicSymbolLoader
schematic_path = params.get("schematicPath")
component = params.get("component", {})
if not schematic_path:
return {"success": False, "message": "Schematic path is required"}
if not component:
return {"success": False, "message": "Component definition is required"}
comp_type = component.get("type", "R")
library = component.get("library", "Device")
reference = component.get("reference", "X?")
value = component.get("value", comp_type)
footprint = component.get("footprint", "")
x = component.get("x", 0)
y = component.get("y", 0)
unit = component.get("unit", 1)
# The TS layer puts these inside `component` alongside x/y/unit
# (src/tools/schematic.ts). Dropping them here silently ignored the
# documented angle/mirrorY arguments, so every symbol landed at 0°
# and callers had to follow up with rotate_schematic_component.
angle = component.get("angle", 0)
mirror_y = bool(component.get("mirrorY", False))
# Derive project path from schematic path for project-local library resolution.
# Walk up from the schematic file to find the directory that owns the project
# (contains sym-lib-table or a .kicad_pro file). Schematics stored in a
# sub-folder (e.g. sheets/) would otherwise resolve to the wrong directory and
# miss any project-local sym-lib-table entries.
schematic_file = Path(schematic_path)
derived_project_path = schematic_file.parent
for ancestor in schematic_file.parents:
if (ancestor / "sym-lib-table").exists() or list(ancestor.glob("*.kicad_pro")):
derived_project_path = ancestor
break
loader = DynamicSymbolLoader(project_path=derived_project_path)
loader.add_component(
schematic_file,
library,
comp_type,
reference=reference,
value=value,
footprint=footprint,
x=x,
y=y,
unit=unit,
angle=angle,
mirror_y=mirror_y,
project_path=derived_project_path,
)
self._reload_kicad_schematic()
# Read back the actual placed position so callers can see grid-snapped
# coordinates when they differ from the request.
response: Dict[str, Any] = {
"success": True,
"component_reference": reference,
"symbol_source": f"{library}:{comp_type}",
}
response.update(instance_report(loader.placed_instances))
try:
content = schematic_file.read_text(encoding="utf-8")
placed = _find_placed_symbol_position(content, reference)
if placed is not None:
placed_x, placed_y = placed
response["placed_at"] = {"x": placed_x, "y": placed_y}
if (placed_x, placed_y) != (x, y):
response["snapped"] = True
response["requested_at"] = {"x": x, "y": y}
except Exception:
pass # Best-effort read; the component was placed successfully.
return response
except Exception as e:
logger.error(f"Error adding component to schematic: {str(e)}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_delete_schematic_component(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Remove a placed symbol from a schematic using text-based manipulation (no skip writes)"""
logger.info("Deleting schematic component")
try:
import re
from pathlib import Path
schematic_path = params.get("schematicPath")
reference = params.get("reference")
delete_attached_labels = bool(params.get("deleteAttachedLabels", False))
unit = params.get("unit")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
if not reference:
return {"success": False, "message": "reference is required"}
sch_file = Path(schematic_path)
if not sch_file.exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
with open(sch_file, "r", encoding="utf-8") as f:
content = f.read()
# String-aware paren matcher (see _find_matching_paren): a naive
# counter over-runs on unescaped parens inside quoted strings (e.g.
# MCU pin names like "PA13(JTMS"), which would extend lib_symbols to
# EOF and make every placed-symbol lookup fail.
find_matching_paren = self._find_matching_paren
# Skip lib_symbols section
lib_sym_pos = content.find("(lib_symbols")
lib_sym_end = find_matching_paren(content, lib_sym_pos) if lib_sym_pos >= 0 else -1
# Find ALL placed symbol blocks matching the reference (handles duplicates).
# Use content-string search so multi-line KiCAD format is handled correctly:
# KiCAD writes (symbol\n\t\t(lib_id "...") across two lines, which a
# line-by-line regex would never match.
blocks_to_delete = [] # list of (char_start, char_end) into content
search_start = 0
# Match the opening of any placed-symbol block. KiCAD may emit the
# children of (symbol ...) in any order — most commonly
# `(symbol (lib_id "..."))`, but symbols whose library entry has been
# rescued / customised carry an additional `(lib_name "...")` first:
# `(symbol (lib_name "...") (lib_id "...") ...)`. Matching just
# `(symbol\s+(` covers both, and the lib_symbols range check below
# still excludes library-definition symbols (which use the
# `(symbol "name" ...)` form with a quoted string, not a paren).
pattern = re.compile(r"\(symbol\s+\(")
while True:
m = pattern.search(content, search_start)
if not m:
break
pos = m.start()
# Skip blocks inside lib_symbols
if lib_sym_pos >= 0 and lib_sym_pos <= pos <= lib_sym_end:
search_start = lib_sym_end + 1
continue
end = find_matching_paren(content, pos)
if end < 0:
search_start = pos + 1
continue
block_text = content[pos : end + 1]
if re.search(
r'\(property\s+"Reference"\s+"' + re.escape(reference) + r'"',
block_text,
):
# A multi-unit part places each unit as its own block under
# the same reference, so deleting by reference alone takes
# the whole part. `unit` narrows it to one placement.
if unit is not None:
um = re.search(r"\(unit\s+(\d+)\)", block_text)
if not um or int(um.group(1)) != int(unit):
search_start = end + 1
continue
blocks_to_delete.append((pos, end))
search_start = end + 1
if not blocks_to_delete:
if unit is not None:
return {
"success": False,
"message": (
f"Component '{reference}' has no unit {unit} in this schematic "
"(note: this tool removes schematic symbols, use delete_component "
"for PCB footprints)"
),
}
return {
"success": False,
"message": f"Component '{reference}' not found in schematic (note: this tool removes schematic symbols, use delete_component for PCB footprints)",
}
# Compute the doomed symbol's pin world positions BEFORE removal so
# the optional label cleanup knows where its labels sat.
target_pin_positions: List[Tuple[float, float]] = []
label_cleanup_warning: Optional[str] = None
if delete_attached_labels:
try:
target_pin_positions = self._pin_positions_for_reference(
content, reference, unit
)
if not target_pin_positions:
logger.warning(
"deleteAttachedLabels: no pin positions resolvable "
"for %s (missing lib_symbols entry?); skipping "
"label cleanup",
reference,
)
except Exception as e:
label_cleanup_warning = f"could not compute pin positions: {e}"
logger.warning("deleteAttachedLabels: %s", label_cleanup_warning)
# Delete from back to front to preserve character offsets
for b_start, b_end in sorted(blocks_to_delete, reverse=True):
# Include any leading newline/whitespace before the block
trim_start = b_start
while trim_start > 0 and content[trim_start - 1] in (" ", "\t"):
trim_start -= 1
if trim_start > 0 and content[trim_start - 1] == "\n":
trim_start -= 1
content = content[:trim_start] + content[b_end + 1 :]
with open(sch_file, "w", encoding="utf-8") as f:
f.write(content)
deleted_count = len(blocks_to_delete)
logger.info(f"Deleted {deleted_count} instance(s) of {reference} from {sch_file.name}")
result = {
"success": True,
"reference": reference,
"deleted_count": deleted_count,
"schematic": str(sch_file),
}
if delete_attached_labels:
if label_cleanup_warning is not None:
result["deleted_labels"] = []
result["deleted_label_count"] = 0
result["labelCleanupWarning"] = label_cleanup_warning
else:
result.update(self._delete_dangling_labels_at(sch_file, target_pin_positions))
return result
except Exception as e:
logger.error(f"Error deleting schematic component: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
@staticmethod
def _pin_positions_for_reference(
content: str, reference: str, unit: Optional[int] = None
) -> List[Tuple[float, float]]:
"""World (x, y) positions of every pin of every placed instance whose
Reference property equals ``reference``.
``unit`` restricts this to one placement of a multi-unit part, matching
the units the caller actually deleted.
Builds a mini document of [lib_symbols] + [matching placed symbols]
and reuses WireManager._collect_pin_positions, which applies the
authoritative unit-aware y-negate/mirror/rotate/translate transform.
Returns [] when the symbol has no lib_symbols entry.
"""
sym = sexpdata.Symbol("symbol")
lib_symbols = sexpdata.Symbol("lib_symbols")
prop = sexpdata.Symbol("property")
unit_sym = sexpdata.Symbol("unit")
def _instance_unit(instance: list) -> Optional[int]:
for part in instance[1:]:
if isinstance(part, list) and len(part) >= 2 and part[0] == unit_sym:
try:
return int(part[1])
except (TypeError, ValueError):
return None
return None
data = sexpdata.loads(content)
mini_doc: list = []
for item in data:
if not (isinstance(item, list) and item):
continue
if item[0] == lib_symbols:
mini_doc.append(item)
continue
if item[0] != sym:
continue
# Placed instance whose (property "Reference" "<ref>") matches
for part in item[1:]:
if (
isinstance(part, list)
and len(part) >= 3
and part[0] == prop
and str(part[1]) == "Reference"
and str(part[2]) == reference
):
if unit is None or _instance_unit(item) == int(unit):
mini_doc.append(item)
break
return WireManager._collect_pin_positions(mini_doc)
@staticmethod
def _delete_dangling_labels_at(
sch_file: Path,
target_positions: List[Tuple[float, float]],
tolerance: float = 0.5,
) -> Dict[str, Any]:
"""Delete net labels left dangling at a removed symbol's pin positions.
Re-reads ``sch_file`` (the symbol is already gone) and removes every
label/global_label/hierarchical_label whose anchor lies within
``tolerance`` mm of one of ``target_positions`` — unless the label is
still attached to something else: coincident with a remaining
component pin, coincident with a wire endpoint, or lying strictly on
a wire segment. Never raises; cleanup failure is reported via
``labelCleanupWarning``.
"""
try:
if not target_positions:
return {"deleted_labels": [], "deleted_label_count": 0}
with open(sch_file, "r", encoding="utf-8") as f:
sch_content = f.read()
sch_data = sexpdata.loads(sch_content)
remaining_pins = WireManager._collect_pin_positions(sch_data)
wire_endpoints = WireManager._collect_wire_endpoints(sch_data)
wire_segments = [
parsed
for parsed in (WireManager._parse_wire(item) for item in sch_data)
if parsed is not None
]
def _near(x: float, y: float, points: List[Tuple[float, float]]) -> bool:
return any(
abs(x - px) <= tolerance and abs(y - py) <= tolerance for px, py in points
)
label_types = {
sexpdata.Symbol("label"),
sexpdata.Symbol("global_label"),
sexpdata.Symbol("hierarchical_label"),
}
at_sym = sexpdata.Symbol("at")
deleted: List[Dict[str, Any]] = []
doomed_indices: List[int] = []
for i, item in enumerate(sch_data):
if not (isinstance(item, list) and item and item[0] in label_types):
continue
at_entry = next(
(p for p in item[1:] if isinstance(p, list) and len(p) >= 3 and p[0] == at_sym),
None,
)
if at_entry is None:
continue
x, y = float(at_entry[1]), float(at_entry[2])
if not _near(x, y, target_positions):
continue
# Attachment guards: keep the label if it still serves live
# connectivity.
if _near(x, y, remaining_pins):
continue
if _near(x, y, wire_endpoints):
continue
if any(
WireManager._point_strictly_on_wire(x, y, x1, y1, x2, y2)
for (x1, y1), (x2, y2), _w, _t in wire_segments
):
continue
doomed_indices.append(i)
deleted.append(
{
"text": str(item[1]) if len(item) > 1 else "",
"type": str(item[0]),
"position": {"x": x, "y": y},
}
)
if doomed_indices:
for i in reversed(doomed_indices):
del sch_data[i]
with open(sch_file, "w", encoding="utf-8") as f:
f.write(kicad_dumps(sch_data))
logger.info(
"Deleted %d dangling label(s) at removed pin positions",
len(deleted),
)
return {"deleted_labels": deleted, "deleted_label_count": len(deleted)}
except Exception as e:
logger.warning(f"Label cleanup after component delete failed: {e}")
return {
"deleted_labels": [],
"deleted_label_count": 0,
"labelCleanupWarning": str(e),
}
@staticmethod
def _escape_sexpr_string(value: str) -> str:
"""Escape a string for safe insertion into an S-expression double-quoted token.
Delegates to the shared helper so escaping and unescaping cannot drift
apart (#336). Kept as a method because callers use it via ``self``.
"""
return escape_sexpr_string(value)
@staticmethod
def _find_matching_paren(s: str, start: int) -> int:
"""Return the index of the closing paren matching the opening paren at `start`.
String-aware: parens inside double-quoted tokens are ignored. KiCAD does
NOT backslash-escape bare parens inside quoted strings — e.g. MCU pin
names like "PA13(JTMS" or descriptions like "Vin(fwd) 40V" appear raw in
.kicad_sch / .kicad_sym files. A naive depth counter treats such an
in-string "(" as real structure, so it never rebalances and runs to EOF.
When that happens to the (lib_symbols ...) block, every placed symbol —
which follows lib_symbols — looks like it lives *inside* it and gets
skipped, so reference lookups silently fail for the whole schematic.
Returns -1 if no match is found.
"""
depth = 0
i = start
in_string = False
while i < len(s):
ch = s[i]
if in_string:
if ch == "\\":
i += 2 # skip escaped char (e.g. \" or \\)
continue
if ch == '"':
in_string = False
elif ch == '"':
in_string = True
elif ch == "(":
depth += 1
elif ch == ")":
depth -= 1
if depth == 0:
return i
i += 1
return -1
def _handle_edit_schematic_component(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Update properties of a placed symbol in a schematic.
Supports updating the standard fields (footprint / value / reference rename),
repositioning field labels, and managing **arbitrary custom properties**
(MPN, Manufacturer, Distributor part numbers, Voltage, Dielectric, Tolerance,
LCSC, etc.) used by BOM/CPL exporters and JLCPCB / Digi-Key sourcing.
Uses text-based in-place editing — preserves position, UUID, and all
unrelated fields.
"""
logger.info("Editing schematic component")
try:
import re
from pathlib import Path
schematic_path = params.get("schematicPath")
reference = params.get("reference")
new_footprint = params.get("footprint")
new_value = params.get("value")
new_reference = params.get("newReference")
# dict: {"Reference": {"x": 1, "y": 2, "angle": 0}}
field_positions = params.get("fieldPositions")
# dict: {"MPN": "RC0603FR-0710KL"} OR {"MPN": {"value": "...", "hide": true}}
properties = params.get("properties")
# list[str]: ["OldField"] — protected built-ins are rejected
remove_properties = params.get("removeProperties")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
if not reference:
return {"success": False, "message": "reference is required"}
if not any(
[
new_footprint is not None,
new_value is not None,
new_reference is not None,
field_positions is not None,
properties is not None,
remove_properties is not None,
]
):
return {
"success": False,
"message": (
"At least one of footprint, value, newReference, fieldPositions, "
"properties, or removeProperties must be provided"
),
}
# Reject removal attempts targeting protected built-in fields up-front
if remove_properties:
blocked = [n for n in remove_properties if n in self._PROTECTED_PROPERTY_FIELDS]
if blocked:
return {
"success": False,
"message": (
f"Cannot remove built-in field(s) {blocked}: use the dedicated "
"value/footprint/newReference parameters or set the value to ''"
),
}
sch_file = Path(schematic_path)
if not sch_file.exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
with open(sch_file, "r", encoding="utf-8") as f:
content = f.read()
# Skip lib_symbols section
lib_sym_pos = content.find("(lib_symbols")
lib_sym_end = (
self._find_matching_paren(content, lib_sym_pos) if lib_sym_pos >= 0 else -1
)
# Find placed symbol blocks that match the reference. KiCAD may
# serialise the children of (symbol ...) in different orders —
# `(symbol (lib_id "..."))` is the common case but rescued or
# locally-customised symbols carry an extra `(lib_name "...")`
# before the lib_id: `(symbol (lib_name "...") (lib_id "..."))`.
# Match any opening paren after `(symbol`; the lib_symbols range
# check below excludes library-definition symbols, which use the
# `(symbol "name" ...)` form (quoted string, not paren).
block_start = block_end = None
search_start = 0
pattern = re.compile(r"\(symbol\s+\(")
while True:
m = pattern.search(content, search_start)
if not m:
break
pos = m.start()
# Skip if inside lib_symbols section
if lib_sym_pos >= 0 and lib_sym_pos <= pos <= lib_sym_end:
search_start = lib_sym_end + 1
continue
end = self._find_matching_paren(content, pos)
if end < 0:
search_start = pos + 1
continue
block_text = content[pos : end + 1]
if re.search(
r'\(property\s+"Reference"\s+"' + re.escape(reference) + r'"',
block_text,
):
block_start, block_end = pos, end
break
search_start = end + 1
if block_start is None or block_end is None:
return {
"success": False,
"message": f"Component '{reference}' not found in schematic",
}
# Apply property replacements within the found block
block_text = content[block_start : block_end + 1]
# Determine the parent symbol position so that newly-added properties
# default to a sensible location (anchored near the component).
# KiCAD always emits the symbol's own (at x y angle) before any
# (property ...) child blocks, so the FIRST (at ...) inside the
# symbol block is the symbol origin regardless of whether
# (lib_name ...) precedes (lib_id ...).
comp_at = re.search(
r"\(at\s+([\d\.\-]+)\s+([\d\.\-]+)",
block_text,
)
comp_origin: Tuple[float, float] = (
(float(comp_at.group(1)), float(comp_at.group(2))) if comp_at else (0.0, 0.0)
)
if new_footprint is not None:
escaped_fp = self._escape_sexpr_string(str(new_footprint))
block_text = re.sub(
r'(\(property\s+"Footprint"\s+)"[^"]*"',
rf'\1"{escaped_fp}"',
block_text,
)
if new_value is not None:
escaped_v = self._escape_sexpr_string(str(new_value))
block_text = re.sub(
r'(\(property\s+"Value"\s+)"[^"]*"',
rf'\1"{escaped_v}"',
block_text,
)
if new_reference is not None:
escaped_r = self._escape_sexpr_string(str(new_reference))
block_text = re.sub(
r'(\(property\s+"Reference"\s+)"[^"]*"',
rf'\1"{escaped_r}"',
block_text,
)
# Also update the (reference "...") leaves inside the symbol's
# (instances) → (project) → (path) subtree. KiCad reads those
# entries — not the (property "Reference" ...) field — when
# generating netlists and syncing the PCB via "Update PCB from
# Schematic", so leaving them stale produces a silent
# reference mismatch where eeschema shows the new ref but ERC
# / netlist export / PCB sync all use the old one. See #126.
instances_pos = block_text.find("(instances")
if instances_pos >= 0:
instances_end = self._find_matching_paren(block_text, instances_pos)
if instances_end >= 0:
instances_block = block_text[instances_pos : instances_end + 1]
updated_instances = re.sub(
r'(\(reference\s+)"' + re.escape(reference) + r'"',
rf'\1"{escaped_r}"',
instances_block,
)
block_text = (
block_text[:instances_pos]
+ updated_instances
+ block_text[instances_end + 1 :]
)
if field_positions is not None:
for field_name, pos in field_positions.items():
x = pos.get("x", 0)
y = pos.get("y", 0)
angle = pos.get("angle", 0)
block_text = re.sub(
r'(\(property\s+"'
+ re.escape(field_name)
+ r'"\s+"[^"]*"\s+)\(at\s+[\d\.\-]+\s+[\d\.\-]+\s+[\d\.\-]+\s*\)',
rf"\1(at {x} {y} {angle})",
block_text,
)
justify = pos.get("justify")
if justify is not None:
block_text = self._set_justify_on_property(
block_text, field_name, str(justify)
)
properties_added: Dict[str, Any] = {}
properties_updated: Dict[str, Any] = {}
if properties:
if not isinstance(properties, dict):
return {
"success": False,
"message": "properties must be a dict mapping property name -> value or spec",
}
for name, spec in properties.items():
if not isinstance(name, str) or not name:
return {
"success": False,
"message": f"Invalid property name: {name!r}",
}
# Normalise scalar values to a spec dict with just {"value": ...}
if not isinstance(spec, dict):
spec = {"value": spec}
try:
block_text, action = self._set_property_in_block(
block_text, name, spec, comp_origin
)
except ValueError as ve:
return {"success": False, "message": str(ve)}
target = properties_added if action == "added" else properties_updated
target[name] = spec.get("value")
properties_removed: list = []
if remove_properties:
if not isinstance(remove_properties, list):
return {
"success": False,
"message": "removeProperties must be a list of property names",
}
for name in remove_properties:
block_text, removed = self._remove_property_from_block(block_text, name)
if removed:
properties_removed.append(name)
content = content[:block_start] + block_text + content[block_end + 1 :]
with open(sch_file, "w", encoding="utf-8") as f:
f.write(content)
changes: Dict[str, Any] = {
k: v
for k, v in {
"footprint": new_footprint,
"value": new_value,
"reference": new_reference,
}.items()
if v is not None
}
if field_positions is not None:
changes["fieldPositions"] = field_positions
if properties_added:
changes["propertiesAdded"] = properties_added
if properties_updated:
changes["propertiesUpdated"] = properties_updated
if properties_removed:
changes["propertiesRemoved"] = properties_removed
logger.info(f"Edited schematic component {reference}: {changes}")
self._reload_kicad_schematic()
return {"success": True, "reference": reference, "updated": changes}
except Exception as e:
logger.error(f"Error editing schematic component: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_set_schematic_component_property(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Add or update a single property on a placed schematic symbol.
Convenience wrapper around `edit_schematic_component` for the very common
case of setting one BOM/sourcing field at a time. The property is created
if it does not already exist, otherwise its value (and optionally its
position / visibility) is updated in place.
"""
logger.info("Setting schematic component property")
name = params.get("name")
if not isinstance(name, str) or not name:
return {"success": False, "message": "name is required"}
if "value" not in params:
return {"success": False, "message": "value is required"}
spec: Dict[str, Any] = {"value": params["value"]}
for key in ("x", "y", "angle", "hide", "fontSize", "justify"):
if params.get(key) is not None:
spec[key] = params[key]
return self._handle_edit_schematic_component(
{
"schematicPath": params.get("schematicPath"),
"reference": params.get("reference"),
"properties": {name: spec},
}
)
def _handle_remove_schematic_component_property(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Remove a single custom property from a placed schematic symbol.
Built-in fields (Reference, Value, Footprint, Datasheet) cannot be
removed — use `edit_schematic_component` to clear them instead.
"""
logger.info("Removing schematic component property")
name = params.get("name")
if not isinstance(name, str) or not name:
return {"success": False, "message": "name is required"}
return self._handle_edit_schematic_component(
{
"schematicPath": params.get("schematicPath"),
"reference": params.get("reference"),
"removeProperties": [name],
}
)
def _handle_get_schematic_component(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Return full component info: position and all field values with their (at x y angle) positions."""
logger.info("Getting schematic component info")
try:
import re
from pathlib import Path
schematic_path = params.get("schematicPath")
reference = params.get("reference")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
if not reference:
return {"success": False, "message": "reference is required"}
sch_file = Path(schematic_path)
if not sch_file.exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
with open(sch_file, "r", encoding="utf-8") as f:
content = f.read()
# String-aware paren matcher (see _find_matching_paren): a naive
# counter over-runs on unescaped parens inside quoted strings (e.g.
# MCU pin names like "PA13(JTMS"), which would extend lib_symbols to
# EOF and make every placed-symbol lookup fail.
find_matching_paren = self._find_matching_paren
# Skip lib_symbols section
lib_sym_pos = content.find("(lib_symbols")
lib_sym_end = find_matching_paren(content, lib_sym_pos) if lib_sym_pos >= 0 else -1
# Find the placed symbol block for this reference. KiCAD may emit
# the children of (symbol ...) in different orders — most commonly
# `(symbol (lib_id "..."))`, but symbols whose library entry has
# been rescued / customised carry an extra `(lib_name "...")` first
# (`(symbol (lib_name "...") (lib_id "..."))`). Match `(symbol\s+(`
# — any opening paren — to handle both. The lib_symbols range check
# below excludes library-definition symbols, which use the
# `(symbol "name" ...)` form (quoted string, not paren).
block_start = block_end = None
search_start = 0
pattern = re.compile(r"\(symbol\s+\(")
while True:
m = pattern.search(content, search_start)
if not m:
break
pos = m.start()
if lib_sym_pos >= 0 and lib_sym_pos <= pos <= lib_sym_end:
search_start = lib_sym_end + 1
continue
end = find_matching_paren(content, pos)
if end < 0:
search_start = pos + 1
continue
block_text = content[pos : end + 1]
if re.search(
r'\(property\s+"Reference"\s+"' + re.escape(reference) + r'"',
block_text,
):
block_start, block_end = pos, end
break
search_start = end + 1
if block_start is None or block_end is None:
return {
"success": False,
"message": f"Component '{reference}' not found in schematic",
}
block_text = content[block_start : block_end + 1]
# Extract component position: the first (at x y angle) inside the
# symbol block. KiCAD always writes the symbol's own (at) before
# any (property ...) child blocks, so the first match is the
# symbol origin regardless of the (lib_name)/(lib_id) ordering.
comp_at = re.search(
r"\(at\s+([\d\.\-]+)\s+([\d\.\-]+)\s+([\d\.\-]+)\s*\)",
block_text,
)
if comp_at:
comp_pos = {
"x": float(comp_at.group(1)),
"y": float(comp_at.group(2)),
"angle": float(comp_at.group(3)),
}
else:
comp_pos = None
# Extract all properties with their at positions. Name and value
# are read escape-aware and decoded (#336): with "([^"]*)" a value
# holding an escaped quote did not match and the field went
# missing, and a line break came back as a backslash and an "n".
prop_pattern = re.compile(
r"\(property\s+"
+ QUOTED_VALUE
+ r"\s+"
+ QUOTED_VALUE
+ r"\s+\(at\s+([\d\.\-]+)\s+([\d\.\-]+)\s+([\d\.\-]+)\s*\)"
)
fields = {}
for m in prop_pattern.finditer(block_text):
name, value, x, y, angle = (
unescape_sexpr_string(m.group(1)),
unescape_sexpr_string(m.group(2)),
m.group(3),
m.group(4),
m.group(5),
)
fields[name] = {
"value": value,
"x": float(x),
"y": float(y),
"angle": float(angle),
}
return {
"success": True,
"reference": reference,
"position": comp_pos,
"fields": fields,
}
except Exception as e:
logger.error(f"Error getting schematic component: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_add_schematic_wire(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Add a wire to a schematic using WireManager, with optional pin snapping"""
logger.info("Adding wire to schematic")
try:
from pathlib import Path
from commands.wire_manager import WireManager
schematic_path = params.get("schematicPath")
points = params.get("waypoints")
properties = params.get("properties", {})
snap_to_pins = params.get("snapToPins", True)
snap_tolerance = params.get("snapTolerance", 1.0)
if not schematic_path:
return {"success": False, "message": "Schematic path is required"}
if not points or len(points) < 2:
return {
"success": False,
"message": "At least 2 waypoints are required",
}
# Make a mutable copy of points
points = [list(p) for p in points]
# Pin snapping: adjust first and last endpoints to nearest pin.
# The response says what happened to each endpoint. Snapping
# nothing used to be silent, so a wire drawn from a miscalculated
# pin position was reported as a success while floating next to
# the pin it was meant for (#404).
snapped_info: List[str] = []
warnings: List[str] = []
endpoints: Dict[str, Dict[str, Any]] = {
"start": {"position": list(points[0]), "snapped": False},
"end": {"position": list(points[-1]), "snapped": False},
}
if snap_to_pins:
from commands.pin_locator import PinLocator
locator = PinLocator()
sch_path = Path(schematic_path)
# Load schematic via the guarded loader so a parse failure
# surfaces a diagnosed SchematicLoadError instead of a crash.
sch = SchematicManager.load_schematic(str(sch_path))
# Collect all pin locations: list of (ref, pin_num, [x, y])
all_pins = []
for symbol in sch.symbol:
if not hasattr(symbol.property, "Reference"):
continue
ref = symbol.property.Reference.value
if ref.startswith("_TEMPLATE"):
continue
pin_locs = locator.get_all_symbol_pins(sch_path, ref)
for pin_num, coords in pin_locs.items():
all_pins.append((ref, pin_num, coords))
def find_nearest_pin(point: Any) -> Any:
"""Return (ref, pin_num, coords, distance) of the closest pin."""
best = None
for ref, pin_num, coords in all_pins:
dx = point[0] - coords[0]
dy = point[1] - coords[1]
dist = (dx * dx + dy * dy) ** 0.5
if best is None or dist < best[3]:
best = (ref, pin_num, coords, dist)
return best
for label, index in (("start", 0), ("end", -1)):
requested = points[index]
info = endpoints[label]
match = find_nearest_pin(requested)
if match is None:
warnings.append(
f"{label} point [{requested[0]}, {requested[1]}] was not snapped: "
"the schematic has no symbol pins"
)
continue
ref, pin_num, coords, dist = match
info["nearestPin"] = {
"reference": ref,
"pin": pin_num,
"position": [coords[0], coords[1]],
}
info["nearestPinDistance"] = round(dist, 4)
if dist <= snap_tolerance:
logger.info(
f"Snapped {label} point {requested} -> {coords} ({ref}/{pin_num})"
)
snapped_info.append(
f"{label} snapped to {ref}/{pin_num} at [{coords[0]}, {coords[1]}]"
)
points[index] = list(coords)
info["position"] = list(coords)
info["snapped"] = True
else:
warnings.append(
f"{label} point [{requested[0]}, {requested[1]}] is {dist:.2f} mm from "
f"the nearest pin {ref}/{pin_num} at [{coords[0]}, {coords[1]}], "
f"outside snapTolerance {snap_tolerance} mm; the endpoint was left "
"where it was and the wire may be floating"
)
# Extract wire properties
stroke_width = properties.get("stroke_width", 0)
stroke_type = properties.get("stroke_type", "default")
# Use WireManager for S-expression manipulation
if len(points) == 2:
success = WireManager.add_wire(
Path(schematic_path),
points[0],
points[1],
stroke_width=stroke_width,
stroke_type=stroke_type,
)
else:
success = WireManager.add_polyline_wire(
Path(schematic_path),
points,
stroke_width=stroke_width,
stroke_type=stroke_type,
)
if success:
message = "Wire added successfully"
if snapped_info:
message += "; " + "; ".join(snapped_info)
if warnings:
message += ". Warning: " + " ".join(warnings)
self._reload_kicad_schematic()
result: Dict[str, Any] = {
"success": True,
"message": message,
"endpoints": endpoints,
}
if warnings:
result["warnings"] = warnings
return result
else:
return {"success": False, "message": "Failed to add wire"}
except SchematicLoadError as e:
return e.to_response()
except Exception as e:
logger.error(f"Error adding wire to schematic: {str(e)}")
import traceback
logger.error(traceback.format_exc())
return {
"success": False,
"message": str(e),
"errorDetails": traceback.format_exc(),
}
def _handle_list_schematic_libraries(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""List available symbol libraries"""
logger.info("Listing schematic libraries")
try:
search_paths = params.get("searchPaths")
libraries = SchematicLibraryManager.list_available_libraries(search_paths)
return {"success": True, "libraries": libraries}
except Exception as e:
logger.error(f"Error listing schematic libraries: {str(e)}")
return {"success": False, "message": str(e)}
def _handle_check_wire_collisions(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Detect wires passing through component bodies without connecting to pins"""
logger.info("Checking wire collisions")
try:
from commands.schematic_analysis import check_wire_collisions
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
result = check_wire_collisions(schematic_path)
return {"success": True, **result}
except ImportError:
return {
"success": False,
"message": "schematic_analysis module not available",
}
except Exception as e:
logger.error(f"Error checking wire collisions: {e}")
return {"success": False, "message": str(e)}
def _handle_export_schematic_pdf(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Export schematic to PDF"""
logger.info("Exporting schematic to PDF")
try:
schematic_path = params.get("schematicPath")
output_path = params.get("outputPath")
if not schematic_path:
return {"success": False, "message": "Schematic path is required"}
if not output_path:
return {"success": False, "message": "Output path is required"}
if not Path(schematic_path).exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
import subprocess
kicad_cli = resolve_kicad_cli()
if not kicad_cli:
return {"success": False, "message": kicad_cli_not_found_message()}
cmd = [
kicad_cli,
"sch",
"export",
"pdf",
"--output",
output_path,
schematic_path,
]
if params.get("blackAndWhite"):
cmd.insert(-1, "--black-and-white")
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
if result.returncode == 0:
return {"success": True, "file": {"path": output_path}}
else:
return {
"success": False,
"message": f"kicad-cli failed: {result.stderr}",
}
except FileNotFoundError:
return {"success": False, "message": kicad_cli_not_found_message()}
except Exception as e:
logger.error(f"Error exporting schematic to PDF: {str(e)}")
return {"success": False, "message": str(e)}
def _handle_add_schematic_net_label(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Add a net label to schematic using WireManager.
When componentRef and pinNumber are supplied the label is placed at the
exact pin endpoint retrieved via PinLocator, ignoring the provided
position. The response includes the actual coordinates used and
whether the label landed on a pin endpoint.
"""
logger.info("Adding net label to schematic")
try:
from pathlib import Path
from commands.wire_manager import WireManager
schematic_path = params.get("schematicPath")
net_name = params.get("netName")
position = params.get("position")
label_type = params.get("labelType", "label")
orientation_param = params.get("orientation")
orientation = 0 if orientation_param is None else orientation_param
component_ref = params.get("componentRef")
pin_number = params.get("pinNumber")
if not all([schematic_path, net_name]):
return {
"success": False,
"message": "Missing required parameters: schematicPath, netName",
}
snapped_to_pin: Optional[Dict[str, Any]] = None
if component_ref and pin_number:
# Snap position to exact pin endpoint using PinLocator
from commands.pin_locator import PinLocator
locator = PinLocator()
pin_loc = locator.get_pin_location(
Path(schematic_path), component_ref, str(pin_number)
)
if pin_loc is None:
return {
"success": False,
"message": (
f"Could not locate pin {pin_number} on {component_ref}. "
"Check the reference and pin number."
),
}
position = pin_loc
snapped_to_pin = {"component": component_ref, "pin": str(pin_number)}
logger.info(
f"Snapped label '{net_name}' to pin {component_ref}/{pin_number} at {position}"
)
# Orient the label along the pin unless the caller asked for a
# specific angle. A label inherits its text direction from its
# angle: 0 grows right, 180 grows left (WireManager.add_label
# pairs the matching justify). Defaulting to 0 on a left-facing
# pin — or 180 on a right-facing one — lays the text back over
# the symbol body, and since `justify` is not separately
# settable, callers could not fix it without editing the file.
if orientation_param is None:
pin_angle = locator.get_pin_angle(
Path(schematic_path), component_ref, str(pin_number)
)
if pin_angle is not None:
# Snap the outward bearing (0=right, 90=up, 180=left,
# 270=down) to the four orientations KiCad allows.
orientation = int(round(pin_angle / 90.0) * 90) % 360
logger.info(
f"Derived label orientation {orientation}° from pin "
f"{component_ref}/{pin_number} outward angle {pin_angle:.1f}°"
)
elif position is None:
return {
"success": False,
"message": (
"Missing position. Either provide position [x, y] or "
"componentRef + pinNumber to snap to a pin endpoint."
),
}
# Collect existing net names BEFORE adding the new label so we can
# detect case-mismatch collisions against pre-existing nets only.
existing_net_names: List[str] = []
try:
pre_schematic = SchematicManager.load_schematic(schematic_path)
if pre_schematic is not None:
if hasattr(pre_schematic, "label"):
for lbl in pre_schematic.label:
if hasattr(lbl, "value"):
existing_net_names.append(lbl.value)
if hasattr(pre_schematic, "global_label"):
for lbl in pre_schematic.global_label:
if hasattr(lbl, "value"):
existing_net_names.append(lbl.value)
except Exception:
# Non-fatal: if we can't read existing nets, skip the warning
existing_net_names = []
# Use WireManager for S-expression manipulation
success = WireManager.add_label(
Path(schematic_path),
net_name,
position,
label_type=label_type,
orientation=orientation,
)
if not success:
return {"success": False, "message": "Failed to add net label"}
# Compute case-mismatch warnings against pre-existing net names.
# A collision is: existing name != new name, but lowercases match.
new_name_lower = net_name.lower()
case_warnings: List[str] = [
f"Net '{existing}' already exists — label '{net_name}' may be a case mismatch."
for existing in existing_net_names
if existing.lower() == new_name_lower and existing != net_name
]
response: Dict[str, Any] = {
"success": True,
"message": f"Added net label '{net_name}' at {position}",
"actual_position": position,
"orientation": orientation,
}
if snapped_to_pin:
response["snapped_to_pin"] = snapped_to_pin
response["message"] = (
f"Added net label '{net_name}' at exact pin endpoint "
f"{component_ref}/{pin_number} ({position[0]}, {position[1]})"
)
if case_warnings:
response["case_warnings"] = case_warnings
return response
except Exception as e:
logger.error(f"Error adding net label: {str(e)}")
import traceback
logger.error(traceback.format_exc())
return {
"success": False,
"message": str(e),
"errorDetails": traceback.format_exc(),
}
def _handle_get_schematic_pin_locations(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Return exact pin endpoint coordinates for a schematic component"""
logger.info("Getting schematic pin locations")
try:
from pathlib import Path
from commands.pin_locator import PinLocator
schematic_path = params.get("schematicPath")
reference = params.get("reference")
if not all([schematic_path, reference]):
return {
"success": False,
"message": "Missing required parameters: schematicPath, reference",
}
locator = PinLocator()
all_pins = locator.get_all_symbol_pins(Path(schematic_path), reference)
if not all_pins:
return {
"success": False,
"message": f"No pins found for {reference} — check reference and schematic path",
}
# Enrich with pin names and angles from the symbol definition
pins_def = (
locator.get_symbol_pins(
Path(schematic_path),
locator._get_lib_id(Path(schematic_path), reference),
)
if hasattr(locator, "_get_lib_id")
else {}
)
result = {}
for pin_num, coords in all_pins.items():
entry = {"x": coords[0], "y": coords[1]}
if pin_num in pins_def:
entry["name"] = pins_def[pin_num].get("name", pin_num)
entry["angle"] = (
locator.get_pin_angle(Path(schematic_path), reference, pin_num) or 0
)
result[pin_num] = entry
return {"success": True, "reference": reference, "pins": result}
except Exception as e:
logger.error(f"Error getting pin locations: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_get_schematic_view(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Get a rasterised image of the schematic (SVG export → optional PNG conversion)"""
logger.info("Getting schematic view")
import base64
import subprocess
import tempfile
try:
schematic_path = params.get("schematicPath")
if not schematic_path or not Path(schematic_path).exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
fmt = params.get("format", "png")
width = params.get("width", 1200)
height = params.get("height", 900)
# Step 1: Export schematic to SVG via kicad-cli
kicad_cli = resolve_kicad_cli()
if not kicad_cli:
return {"success": False, "message": kicad_cli_not_found_message()}
with tempfile.TemporaryDirectory() as tmpdir:
svg_path = str(Path(tmpdir) / "schematic.svg")
cmd = [
kicad_cli,
"sch",
"export",
"svg",
"--output",
tmpdir,
"--no-background-color",
schematic_path,
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
if result.returncode != 0:
return {
"success": False,
"message": f"kicad-cli SVG export failed: {result.stderr}",
}
# kicad-cli names one file per page after the schematic/sheet
# names — pick the root page deterministically.
svg_files = glob.glob(os.path.join(tmpdir, "*.svg"))
if not svg_files:
return {
"success": False,
"message": "No SVG file produced by kicad-cli",
}
picked = _pick_root_svg(tmpdir, schematic_path)
if picked is None:
picked = sorted(svg_files)[0]
logger.warning(
"Could not identify root SVG page for %s; " "falling back to %s",
schematic_path,
picked,
)
svg_path = picked
if fmt == "svg":
with open(svg_path, "r", encoding="utf-8") as f:
svg_data = f.read()
return {"success": True, "imageData": svg_data, "format": "svg"}
# Step 2: Convert SVG to PNG (cffi-free)
png_data = _svg_to_png(svg_path, width, height)
if png_data is None:
with open(svg_path, "r", encoding="utf-8") as f:
svg_data = f.read()
return {
"success": True,
"imageData": svg_data,
"format": "svg",
"message": "No PNG converter available — returning SVG. Install pymupdf, inkscape, or imagemagick.",
}
return {
"success": True,
"imageData": base64.b64encode(png_data).decode("utf-8"),
"format": "png",
"width": width,
"height": height,
}
except FileNotFoundError:
return {"success": False, "message": kicad_cli_not_found_message()}
except Exception as e:
logger.error(f"Error getting schematic view: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_list_schematic_components(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""List all components in a schematic"""
logger.info("Listing schematic components")
try:
from pathlib import Path
from commands.pin_locator import PinLocator
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
sch_file = Path(schematic_path)
if not sch_file.exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
try:
schematic = SchematicManager.load_schematic(schematic_path)
except SchematicLoadError as e:
return e.to_response()
# Optional filters
filter_params = params.get("filter", {})
lib_id_filter = filter_params.get("libId", "")
ref_prefix_filter = filter_params.get("referencePrefix", "")
locator = PinLocator()
components = []
for symbol in schematic.symbol:
if not hasattr(symbol.property, "Reference"):
continue
ref = symbol.property.Reference.value
# Skip template symbols
if ref.startswith("_TEMPLATE"):
continue
lib_id = symbol.lib_id.value if hasattr(symbol, "lib_id") else ""
# Apply filters
if lib_id_filter and lib_id_filter not in lib_id:
continue
if ref_prefix_filter and not ref.startswith(ref_prefix_filter):
continue
value = symbol.property.Value.value if hasattr(symbol.property, "Value") else ""
footprint = (
symbol.property.Footprint.value if hasattr(symbol.property, "Footprint") else ""
)
position = symbol.at.value if hasattr(symbol, "at") else [0, 0, 0]
uuid_val = symbol.uuid.value if hasattr(symbol, "uuid") else ""
comp = {
"reference": ref,
"libId": lib_id,
"value": value,
"footprint": footprint,
"position": {"x": float(position[0]), "y": float(position[1])},
"rotation": float(position[2]) if len(position) > 2 else 0,
"uuid": str(uuid_val),
}
# Get pins if available
try:
all_pins = locator.get_all_symbol_pins(sch_file, ref)
if all_pins:
pins_def = locator.get_symbol_pins(sch_file, lib_id) or {}
pin_list = []
for pin_num, coords in all_pins.items():
pin_info = {
"number": pin_num,
"position": {"x": coords[0], "y": coords[1]},
}
if pin_num in pins_def:
pin_info["name"] = pins_def[pin_num].get("name", pin_num)
pin_list.append(pin_info)
comp["pins"] = pin_list
except Exception:
pass # Pin lookup is best-effort
components.append(comp)
return {"success": True, "components": components, "count": len(components)}
except Exception as e:
logger.error(f"Error listing schematic components: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_list_schematic_nets(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""List all nets in a schematic with their connections"""
logger.info("Listing schematic nets")
try:
from commands.pin_locator import PinLocator
from commands.wire_connectivity import (
_build_adjacency,
_discover_sub_sheets,
_load_sexp,
_parse_labels_sexp,
_parse_virtual_connections,
_parse_wires,
count_pins_on_net,
get_connections_for_nets,
)
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
try:
schematic = SchematicManager.load_schematic(schematic_path)
except SchematicLoadError as e:
return e.to_response()
# Collect net names from the top-level sheet using sexpdata.
# Falls back to kicad-skip's label collections when the file
# cannot be read (e.g. mocked schematics in unit tests).
net_names: set = set()
sexp_loaded = False
try:
sexp = _load_sexp(schematic_path)
sexp_loaded = True
_, label_to_points = _parse_labels_sexp(sexp)
net_names.update(label_to_points.keys())
except Exception as e:
logger.debug(
f"Could not parse labels from {schematic_path} via sexp ({e}); "
"falling back to kicad-skip label collections"
)
for attr in ("label", "global_label"):
if not hasattr(schematic, attr):
continue
for label in getattr(schematic, attr):
if hasattr(label, "value"):
net_names.add(label.value)
# Collect net names from all sub-sheets (only when the parent
# sheet was readable; fake/mock paths skip recursion entirely).
if sexp_loaded:
sub_sheets = _discover_sub_sheets(schematic_path)
for sub_path in sub_sheets:
try:
sub_sexp = _load_sexp(sub_path)
_, sub_label_to_points = _parse_labels_sexp(sub_sexp)
net_names.update(sub_label_to_points.keys())
except Exception as e:
logger.warning(f"Error reading sub-sheet {sub_path}: {e}")
# Pre-build shared wire graph structures for efficiency
all_wires = _parse_wires(schematic)
if all_wires:
adjacency, iu_to_wires = _build_adjacency(all_wires)
else:
adjacency, iu_to_wires = [], {}
point_to_label, label_to_points = _parse_virtual_connections(schematic, schematic_path)
# One shared locator + one batched connectivity pass: every sheet
# is parsed and pin-located once for the whole net list, instead of
# once per net (which made this handler O(nets × sheet size) and
# took minutes on large flat schematics).
sorted_net_names = sorted(net_names)
locator = PinLocator(memoize_pins=True) # this request only
connections_by_net = get_connections_for_nets(
schematic, schematic_path, sorted_net_names, locator=locator
)
nets = []
for net_name in sorted_net_names:
connections = connections_by_net[net_name]
pin_count = count_pins_on_net(
schematic,
schematic_path,
net_name,
all_wires,
iu_to_wires,
adjacency,
point_to_label,
label_to_points,
locator=locator,
)
nets.append(
{
"name": net_name,
"connections": connections,
"connected_pin_count": pin_count,
}
)
return {"success": True, "nets": nets, "count": len(nets)}
except Exception as e:
logger.error(f"Error listing schematic nets: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_list_schematic_wires(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""List all wires in a schematic"""
logger.info("Listing schematic wires")
try:
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
try:
schematic = SchematicManager.load_schematic(schematic_path)
except SchematicLoadError as e:
return e.to_response()
wires = []
if hasattr(schematic, "wire"):
for wire in schematic.wire:
if hasattr(wire, "pts") and hasattr(wire.pts, "xy"):
points = []
for point in wire.pts.xy:
if hasattr(point, "value"):
points.append(
{
"x": float(point.value[0]),
"y": float(point.value[1]),
}
)
if len(points) >= 2:
wires.append(
{
"start": points[0],
"end": points[-1],
}
)
return {"success": True, "wires": wires, "count": len(wires)}
except Exception as e:
logger.error(f"Error listing schematic wires: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_list_schematic_labels(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""List all net labels and power flags in a schematic"""
logger.info("Listing schematic labels")
try:
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
net_name = params.get("netName")
label_type = params.get("labelType")
_valid_label_types = {"net", "global", "power"}
if label_type is not None and label_type not in _valid_label_types:
return {"success": False, "message": "labelType must be one of: net, global, power"}
try:
schematic = SchematicManager.load_schematic(schematic_path)
except SchematicLoadError as e:
return e.to_response()
labels = []
# Regular labels
if hasattr(schematic, "label"):
for label in schematic.label:
if hasattr(label, "value"):
pos = (
label.at.value
if hasattr(label, "at") and hasattr(label.at, "value")
else [0, 0]
)
labels.append(
{
"name": label.value,
"type": "net",
"position": {"x": float(pos[0]), "y": float(pos[1])},
}
)
# Global labels
if hasattr(schematic, "global_label"):
for label in schematic.global_label:
if hasattr(label, "value"):
pos = (
label.at.value
if hasattr(label, "at") and hasattr(label.at, "value")
else [0, 0]
)
labels.append(
{
"name": label.value,
"type": "global",
"position": {"x": float(pos[0]), "y": float(pos[1])},
}
)
# Power symbols (components with power flag)
if hasattr(schematic, "symbol"):
for symbol in schematic.symbol:
if not hasattr(symbol.property, "Reference"):
continue
ref = symbol.property.Reference.value
if ref.startswith("_TEMPLATE"):
continue
if not ref.startswith("#PWR"):
continue
value = (
symbol.property.Value.value if hasattr(symbol.property, "Value") else ref
)
pos = symbol.at.value if hasattr(symbol, "at") else [0, 0, 0]
labels.append(
{
"name": value,
"type": "power",
"position": {"x": float(pos[0]), "y": float(pos[1])},
}
)
# Apply filters
if net_name is not None:
labels = [lbl for lbl in labels if lbl["name"] == net_name]
if label_type is not None:
labels = [lbl for lbl in labels if lbl["type"] == label_type]
return {"success": True, "labels": labels, "count": len(labels)}
except Exception as e:
logger.error(f"Error listing schematic labels: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
@staticmethod
def _unit_not_found_message(sch_data: list, reference: str, unit: Any) -> str:
"""Explain a failed symbol lookup, naming the units that do exist."""
from commands.wire_dragger import WireDragger
if unit is None:
return f"Component {reference} not found"
units = WireDragger.list_symbol_units(sch_data, reference)
if not units:
return f"Component {reference} not found"
placed = ", ".join(str(u) for u in sorted(set(units)))
return f"Component {reference} has no unit {unit}; units placed here: {placed}"
def _handle_move_schematic_component(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Move a schematic component to a new position, dragging connected wires."""
logger.info("Moving schematic component")
try:
from commands.wire_dragger import WireDragger
schematic_path = params.get("schematicPath")
reference = params.get("reference")
position = params.get("position", {})
new_x = position.get("x")
new_y = position.get("y")
preserve_wires = params.get("preserveWires", True)
straighten = params.get("straightenWires", True)
unit = params.get("unit")
if not schematic_path or not reference:
return {
"success": False,
"message": "schematicPath and reference are required",
}
if new_x is None or new_y is None:
return {
"success": False,
"message": "position with x and y is required",
}
with open(schematic_path, "r", encoding="utf-8") as f:
sch_data = sexpdata.loads(f.read())
# Find symbol and record old position
found = WireDragger.find_symbol(sch_data, reference, unit)
if found is None:
return {
"success": False,
"message": self._unit_not_found_message(sch_data, reference, unit),
}
_, old_x, old_y = found[0], found[1], found[2]
old_position = {"x": old_x, "y": old_y}
drag_summary = {}
if preserve_wires:
# Compute pin world positions before and after the move
pin_positions = WireDragger.compute_pin_positions(
sch_data, reference, float(new_x), float(new_y), unit
)
# Build old→new coordinate map (deduplicate coincident pins)
old_to_new = {}
for _pin, (old_xy, new_xy) in pin_positions.items():
if old_xy in old_to_new:
logger.warning(
f"move_schematic_component: pin {_pin!r} of {reference!r} "
f"shares old position {old_xy} with another pin; "
f"keeping first entry, skipping duplicate"
)
continue
old_to_new[old_xy] = new_xy
# Everything that must stay put while bends are straightened:
# the pins of every other symbol, plus this symbol's own pins at
# their new positions (they are the ends we just dragged).
anchors = set(
WireDragger.get_all_stationary_pin_positions(sch_data, reference, unit).keys()
)
anchors.update(new_xy for (_old, new_xy) in pin_positions.values())
drag_summary = WireDragger.drag_wires(
sch_data, old_to_new, anchor_points=anchors, straighten=straighten
)
# Synthesize wires for touching-pin connections after dragging,
# so drag_wires doesn't accidentally move and collapse the new wire.
wires_synthesized = WireDragger.synthesize_touching_pin_wires(
sch_data, reference, pin_positions, moved_unit=unit
)
drag_summary["wires_synthesized"] = wires_synthesized
# Update symbol position
WireDragger.update_symbol_position(
sch_data, reference, float(new_x), float(new_y), unit
)
WireManager.sync_junctions(sch_data)
with open(schematic_path, "w", encoding="utf-8") as f:
f.write(kicad_dumps(sch_data))
return {
"success": True,
"oldPosition": old_position,
"newPosition": {"x": new_x, "y": new_y},
"unit": unit,
"wiresMoved": drag_summary.get("endpoints_moved", 0),
"wiresRemoved": drag_summary.get("wires_removed", 0),
"wiresSynthesized": drag_summary.get("wires_synthesized", 0),
"labelsMoved": drag_summary.get("labels_moved", 0),
"noConnectsMoved": drag_summary.get("no_connects_moved", 0),
"wiresStraightened": drag_summary.get("wires_straightened", 0),
"wiresLeftDiagonal": drag_summary.get("wires_left_diagonal", 0),
}
except Exception as e:
logger.error(f"Error moving schematic component: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_rotate_schematic_component(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Rotate and/or mirror a schematic component, dragging connected wires."""
logger.info("Rotating schematic component")
try:
import sexpdata as _sexpdata
from commands.wire_dragger import WireDragger
schematic_path = params.get("schematicPath")
reference = params.get("reference")
angle = params.get("angle", 0)
mirror = params.get("mirror") # "x", "y", or None
straighten = params.get("straightenWires", True)
unit = params.get("unit")
if not schematic_path or not reference:
return {
"success": False,
"message": "schematicPath and reference are required",
}
with open(schematic_path, "r", encoding="utf-8") as f:
sch_data = _sexpdata.loads(f.read())
found = WireDragger.find_symbol(sch_data, reference, unit)
if found is None:
return {
"success": False,
"message": self._unit_not_found_message(sch_data, reference, unit),
}
# Determine new mirror state: explicit param overrides; None preserves existing
_, _, _, _, _, old_mirror_x, old_mirror_y = found
if mirror is None:
new_mirror_x = old_mirror_x
new_mirror_y = old_mirror_y
effective_mirror = "x" if old_mirror_x else ("y" if old_mirror_y else None)
else:
new_mirror_x = mirror == "x"
new_mirror_y = mirror == "y"
effective_mirror = mirror
# Compute pin world positions before and after the transform
pin_positions = WireDragger.compute_pin_positions_for_rotation(
sch_data, reference, float(angle), new_mirror_x, new_mirror_y, unit
)
# Build old→new map (skip pins that don't move)
old_to_new = {}
for _pin, (old_xy, new_xy) in pin_positions.items():
if old_xy == new_xy:
continue
if old_xy in old_to_new:
logger.warning(
f"rotate: pin {_pin!r} of {reference!r} shares old position "
f"{old_xy} with another pin; skipping duplicate"
)
continue
old_to_new[old_xy] = new_xy
# Drag connected wires to follow pins
anchors = set(
WireDragger.get_all_stationary_pin_positions(sch_data, reference, unit).keys()
)
anchors.update(new_xy for (_old, new_xy) in pin_positions.values())
drag_summary = WireDragger.drag_wires(
sch_data, old_to_new, anchor_points=anchors, straighten=straighten
)
# Update the symbol's rotation and mirror token in sexpdata
WireDragger.update_symbol_rotation_mirror(
sch_data, reference, float(angle), effective_mirror, unit
)
WireManager.sync_junctions(sch_data)
with open(schematic_path, "w", encoding="utf-8") as f:
f.write(kicad_dumps(sch_data))
return {
"success": True,
"reference": reference,
"unit": unit,
"angle": angle,
"mirror": effective_mirror,
"wiresMoved": drag_summary.get("endpoints_moved", 0),
"wiresRemoved": drag_summary.get("wires_removed", 0),
"labelsMoved": drag_summary.get("labels_moved", 0),
"noConnectsMoved": drag_summary.get("no_connects_moved", 0),
"wiresStraightened": drag_summary.get("wires_straightened", 0),
"wiresLeftDiagonal": drag_summary.get("wires_left_diagonal", 0),
}
except Exception as e:
logger.error(f"Error rotating schematic component: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_annotate_schematic(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Annotate unannotated components in schematic (R? -> R1, R2, ...)
utils.symbol_instances.annotate_sheet does the work. This used kicad-skip's
setAllReferences, which wrote one number into every use of a sheet used
more than once, and it took the used numbers from this file only, so a
number already taken on another sheet was handed out again (#432). It
also gave each unit of a multi-unit part its own number.
"""
logger.info("Annotating schematic")
try:
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
if not Path(schematic_path).is_file():
return {"success": False, "message": f"Schematic not found: {schematic_path}"}
annotated = annotate_sheet(Path(schematic_path))
if not annotated:
return {
"success": True,
"annotated": [],
"message": "All components already annotated",
}
return {"success": True, "annotated": annotated}
except Exception as e:
logger.error(f"Error annotating schematic: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_delete_schematic_wire(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Delete a wire from the schematic matching start/end points"""
logger.info("Deleting schematic wire")
try:
schematic_path = params.get("schematicPath")
start = params.get("start", {})
end = params.get("end", {})
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
from pathlib import Path
from commands.wire_manager import WireManager
start_point = [start.get("x", 0), start.get("y", 0)]
end_point = [end.get("x", 0), end.get("y", 0)]
deleted = WireManager.delete_wire(Path(schematic_path), start_point, end_point)
if deleted:
return {"success": True}
else:
return {"success": False, "message": "No matching wire found"}
except Exception as e:
logger.error(f"Error deleting schematic wire: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_delete_schematic_net_label(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Delete a net label from the schematic"""
logger.info("Deleting schematic net label")
try:
schematic_path = params.get("schematicPath")
net_name = params.get("netName")
position = params.get("position")
if not schematic_path or not net_name:
return {
"success": False,
"message": "schematicPath and netName are required",
}
from pathlib import Path
from commands.wire_manager import WireManager
pos_list = None
if position:
pos_list = [position.get("x", 0), position.get("y", 0)]
deleted = WireManager.delete_label(Path(schematic_path), net_name, pos_list)
if deleted:
return {"success": True}
else:
return {"success": False, "message": f"Label '{net_name}' not found"}
except Exception as e:
logger.error(f"Error deleting schematic net label: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_move_schematic_net_label(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Move a net label to a new position in the schematic."""
logger.info("Moving schematic net label")
try:
import sexpdata as _sexpdata
from sexpdata import Symbol
schematic_path = params.get("schematicPath")
net_name = params.get("netName")
new_position = params.get("newPosition", {})
new_x = new_position.get("x")
new_y = new_position.get("y")
current_position = params.get("currentPosition")
label_type = params.get("labelType")
if not schematic_path or not net_name:
return {"success": False, "message": "schematicPath and netName are required"}
if new_x is None or new_y is None:
return {"success": False, "message": "newPosition with x and y is required"}
_valid_types = {"label", "global_label", "hierarchical_label"}
if label_type is not None and label_type not in _valid_types:
return {
"success": False,
"message": f"labelType must be one of: {', '.join(sorted(_valid_types))}",
}
_SYM_AT = Symbol("at")
target_syms = (
{Symbol(label_type)}
if label_type is not None
else {Symbol(t) for t in _valid_types}
)
TOLERANCE = 0.5
with open(schematic_path, "r", encoding="utf-8") as f:
sch_data = _sexpdata.loads(f.read())
for item in sch_data:
if not (isinstance(item, list) and len(item) >= 2 and item[0] in target_syms):
continue
if item[1] != net_name:
continue
at_idx = next(
(
j
for j, p in enumerate(item)
if isinstance(p, list) and len(p) >= 3 and p[0] == _SYM_AT
),
None,
)
if at_idx is None:
continue
at_entry = item[at_idx]
old_x, old_y = float(at_entry[1]), float(at_entry[2])
if current_position is not None:
cx = current_position.get("x", 0)
cy = current_position.get("y", 0)
if not (abs(old_x - cx) < TOLERANCE and abs(old_y - cy) < TOLERANCE):
continue
rotation = at_entry[3] if len(at_entry) > 3 else 0
item[at_idx] = [_SYM_AT, float(new_x), float(new_y), rotation]
with open(schematic_path, "w", encoding="utf-8") as f:
f.write(kicad_dumps(sch_data))
return {
"success": True,
"oldPosition": {"x": old_x, "y": old_y},
"newPosition": {"x": float(new_x), "y": float(new_y)},
}
return {"success": False, "message": f"Label '{net_name}' not found"}
except Exception as e:
logger.error(f"Error moving schematic net label: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_export_schematic_svg(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Export schematic to SVG using kicad-cli"""
logger.info("Exporting schematic SVG")
import shutil
import subprocess
try:
schematic_path = params.get("schematicPath")
output_path = params.get("outputPath")
if not schematic_path or not output_path:
return {
"success": False,
"message": "schematicPath and outputPath are required",
}
if not Path(schematic_path).exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
kicad_cli = resolve_kicad_cli()
if not kicad_cli:
return {"success": False, "message": kicad_cli_not_found_message()}
# kicad-cli's --output flag for SVG export expects a directory, and
# auto-names one SVG per page after the schematic/sheet names.
# Export into a private temp directory so stale SVGs from earlier
# exports (or extra hierarchical pages) can never be picked up,
# then move only the root page to the requested outputPath.
with tempfile.TemporaryDirectory() as tmpdir:
cmd = [
kicad_cli,
"sch",
"export",
"svg",
schematic_path,
"-o",
tmpdir,
]
if params.get("blackAndWhite"):
cmd.append("--black-and-white")
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
if result.returncode != 0:
return {
"success": False,
"message": f"kicad-cli failed: {result.stderr}",
}
pages = sorted(
os.path.basename(f) for f in glob.glob(os.path.join(tmpdir, "*.svg"))
)
if not pages:
return {
"success": False,
"message": "No SVG file produced by kicad-cli",
}
root_svg = _pick_root_svg(tmpdir, schematic_path)
if root_svg is None:
return {
"success": False,
"message": "Could not identify root SVG page",
"files": pages,
}
output_dir = os.path.dirname(output_path)
if output_dir:
os.makedirs(output_dir, exist_ok=True)
shutil.move(root_svg, output_path)
return {"success": True, "file": {"path": output_path}, "pages": pages}
except FileNotFoundError:
return {"success": False, "message": kicad_cli_not_found_message()}
except Exception as e:
logger.error(f"Error exporting schematic SVG: {e}")
return {"success": False, "message": str(e)}
def _handle_get_wire_connections(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Find net name and all component pins reachable from a point or component pin."""
logger.info("Getting wire connections")
try:
from pathlib import Path
from commands.pin_locator import PinLocator
from commands.wire_connectivity import get_wire_connections
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "Missing required parameter: schematicPath"}
reference = params.get("reference")
pin = params.get("pin")
x = params.get("x")
y = params.get("y")
has_ref_pin = reference is not None and pin is not None
has_coords = x is not None and y is not None
if has_ref_pin and has_coords:
return {
"success": False,
"message": "Supply either {reference, pin} or {x, y}, not both",
}
if not has_ref_pin and not has_coords:
if reference is not None or pin is not None:
return {
"success": False,
"message": "Both reference and pin are required together",
}
return {
"success": False,
"message": "Must supply either {reference, pin} or {x, y}",
}
if has_ref_pin:
location = PinLocator().get_pin_location(Path(schematic_path), reference, str(pin))
if location is None:
return {
"success": False,
"message": f"Pin {pin} not found on {reference}",
}
x, y = location[0], location[1]
else:
try:
x, y = float(x), float(y)
except (TypeError, ValueError):
return {"success": False, "message": "Parameters x and y must be numeric"}
try:
schematic = SchematicManager.load_schematic(schematic_path)
except SchematicLoadError as e:
return e.to_response()
if not hasattr(schematic, "wire"):
return {"success": False, "message": "Schematic has no wires"}
result = get_wire_connections(schematic, schematic_path, x, y)
if result is None:
return {
"success": False,
"message": f"No wire found at ({x},{y}) — point may not be connected",
}
return {"success": True, **result}
except Exception as e:
logger.error(f"Error getting wire connections: {str(e)}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_list_schematic_texts(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""List all free-form text annotations (SCH_TEXT) in a schematic."""
logger.info("Listing schematic text annotations")
try:
from commands.wire_manager import WireManager
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
sch_file = Path(schematic_path)
if not sch_file.exists():
return {"success": False, "message": f"Schematic not found: {schematic_path}"}
texts = WireManager.list_texts(sch_file)
if texts is None:
return {"success": False, "message": "Failed to parse schematic"}
# Optional text filter
filter_text = params.get("text")
if filter_text is not None:
texts = [t for t in texts if filter_text.lower() in t["text"].lower()]
return {"success": True, "texts": texts, "count": len(texts)}
except Exception as e:
logger.error(f"Error listing schematic texts: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_add_schematic_text(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Add a free-form text annotation (SCH_TEXT) to a schematic."""
logger.info("Adding text annotation to schematic")
try:
from commands.wire_manager import WireManager
schematic_path = params.get("schematicPath")
text = params.get("text")
position = params.get("position")
angle = params.get("angle", 0)
font_size = params.get("fontSize", 1.27)
bold = params.get("bold", False)
italic = params.get("italic", False)
justify = params.get("justify", "left")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
if not text:
return {"success": False, "message": "text is required"}
if not position or len(position) != 2:
return {"success": False, "message": "position [x, y] is required"}
if justify not in ("left", "center", "right"):
return {"success": False, "message": "justify must be left, center, or right"}
if font_size <= 0:
return {"success": False, "message": "fontSize must be positive"}
sch_file = Path(schematic_path)
if not sch_file.exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
success = WireManager.add_text(
sch_file,
text,
position,
angle=angle,
font_size=font_size,
bold=bold,
italic=italic,
justify=justify,
)
if success:
return {
"success": True,
"message": f"Added text '{text}' at ({position[0]}, {position[1]})",
"position": {"x": position[0], "y": position[1]},
"angle": angle,
}
return {"success": False, "message": "Failed to add text annotation"}
except Exception as e:
logger.error(f"Error adding schematic text: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_add_schematic_hierarchical_label(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Add a hierarchical label to a sub-sheet schematic."""
logger.info("Adding hierarchical label to schematic")
try:
from commands.wire_manager import WireManager
schematic_path = params.get("schematicPath")
text = params.get("text")
position = params.get("position")
shape = params.get("shape", "bidirectional")
orientation = params.get("orientation", 0)
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
if not text:
return {"success": False, "message": "text is required"}
if not position or len(position) != 2:
return {"success": False, "message": "position [x, y] is required"}
if shape not in ("input", "output", "bidirectional"):
return {
"success": False,
"message": "shape must be input, output, or bidirectional",
}
sch_file = Path(schematic_path)
if not sch_file.exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
success = WireManager.add_hierarchical_label(
sch_file, text, position, shape=shape, orientation=orientation
)
if success:
return {
"success": True,
"message": (
f"Added hierarchical_label '{text}' " f"at {position} shape={shape}"
),
}
return {"success": False, "message": "Failed to add hierarchical label"}
except Exception as e:
logger.error(f"Error adding hierarchical label: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_add_sheet_pin(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Add a sheet pin to a sheet block on the parent schematic."""
logger.info("Adding sheet pin to schematic")
try:
from commands.wire_manager import WireManager
schematic_path = params.get("schematicPath")
sheet_name = params.get("sheetName")
pin_name = params.get("pinName")
pin_type = params.get("pinType", "bidirectional")
position = params.get("position")
orientation = params.get("orientation", 0)
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
if not sheet_name:
return {"success": False, "message": "sheetName is required"}
if not pin_name:
return {"success": False, "message": "pinName is required"}
if not position or len(position) != 2:
return {"success": False, "message": "position [x, y] is required"}
if pin_type not in ("input", "output", "bidirectional"):
return {
"success": False,
"message": "pinType must be input, output, or bidirectional",
}
sch_file = Path(schematic_path)
if not sch_file.exists():
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
with open(sch_file, "r", encoding="utf-8") as f:
content = f.read()
modified, success = WireManager.add_sheet_pin(
content,
sheet_name,
pin_name,
pin_type,
position,
orientation=orientation,
)
if not success:
return {
"success": False,
"message": f"Sheet '{sheet_name}' not found in {schematic_path}",
}
with open(sch_file, "w", encoding="utf-8") as f:
f.write(modified)
return {
"success": True,
"message": (
f"Added sheet pin '{pin_name}' ({pin_type}) " f"to sheet '{sheet_name}'"
),
}
except Exception as e:
logger.error(f"Error adding sheet pin: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_run_erc(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Run Electrical Rules Check on a schematic via kicad-cli"""
logger.info("Running ERC on schematic")
import subprocess
import tempfile
try:
schematic_path = params.get("schematicPath")
if not schematic_path or not Path(schematic_path).exists():
return {
"success": False,
"message": "Schematic file not found",
"errorDetails": f"Path does not exist: {schematic_path}",
}
kicad_cli = self.design_rule_commands._find_kicad_cli()
if not kicad_cli:
return {
"success": False,
"message": "kicad-cli not found",
"errorDetails": "Install KiCAD 8.0+ or add kicad-cli to PATH.",
}
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as tmp:
json_output = tmp.name
try:
cmd = [
kicad_cli,
"sch",
"erc",
"--format",
"json",
"--output",
json_output,
schematic_path,
]
logger.info(f"Running ERC command: {' '.join(cmd)}")
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
# kicad-cli returns non-zero when ERC violations are found —
# this is normal, not an error. Only fail when no JSON was
# produced (genuine CLI failure).
json_output_p = Path(json_output)
if not json_output_p.exists() or json_output_p.stat().st_size == 0:
logger.error(f"ERC command produced no output: {result.stderr}")
return {
"success": False,
"message": "ERC command failed - no output produced",
"errorDetails": result.stderr,
}
with open(json_output, "r", encoding="utf-8") as f:
erc_data = json.load(f)
violations = []
severity_counts = {"error": 0, "warning": 0, "info": 0}
# KiCad 9 nests violations under sheets[].violations
# instead of (or in addition to) the top-level violations
# array used by KiCad 8.
all_violations = erc_data.get("violations", [])
for sheet in erc_data.get("sheets", []):
all_violations.extend(sheet.get("violations", []))
for v in all_violations:
vseverity = v.get("severity", "error")
items = v.get("items", [])
loc = {}
if items and "pos" in items[0]:
loc = {
"x": items[0]["pos"].get("x", 0),
"y": items[0]["pos"].get("y", 0),
}
violations.append(
{
"type": v.get("type", "unknown"),
"severity": vseverity,
"message": v.get("description", ""),
"location": loc,
}
)
if vseverity in severity_counts:
severity_counts[vseverity] += 1
return {
"success": True,
"message": f"ERC complete: {len(violations)} violation(s)",
"summary": {
"total": len(violations),
"by_severity": severity_counts,
},
"violations": violations,
}
finally:
if Path(json_output).exists():
Path(json_output).unlink()
except subprocess.TimeoutExpired:
return {"success": False, "message": "ERC timed out after 120 seconds"}
except Exception as e:
logger.error(f"Error running ERC: {str(e)}")
return {"success": False, "message": str(e)}
def _build_hierarchical_pad_net_map(self, project_sch_path: str, sheets_out=None):
"""Walk the sheets reachable from the root schematic and build a
{(ref, pin_num): net_name} map.
Handles hierarchical schematics by following ``(sheet ...)`` references
from the root. Net names from global_label / hierarchical_label / local
label / power symbols are all collected, and a wired net with no label at
all gets the ``Net-(REF-PadN)`` name KiCad would give it. Wire
connectivity is traced via BFS so labels not placed directly on a pin
endpoint still reach through wire segments.
``sheets_out``, when given, receives the path of every sheet read.
Returns: (pad_net_map, net_names_set)
"""
from collections import defaultdict
from pathlib import Path
from commands.pin_locator import PinLocator
from utils.sheet_tree import sheet_tree
TOLERANCE = 0.5 # mm; schematic grid is 1.27 mm so 0.5 is safe
def snap(x, y):
"""Round to 2 dp to use exact dict lookup instead of O(n²) scan."""
return (round(float(x), 2), round(float(y), 2))
def nearby_net(pt, point_net, tol=TOLERANCE):
"""Return net name for the nearest occupied grid point, or None."""
x, y = pt
# Try exact snap first (fast path)
key = snap(x, y)
if key in point_net:
return point_net[key]
# Slow fallback for off-grid placements
for (lx, ly), name in point_net.items():
if abs(x - lx) < tol and abs(y - ly) < tol:
return name
return None
pad_net_map: dict = {}
all_net_names: set = set()
pin_locator = PinLocator()
# Only the sheets reachable from the root are part of the design. A
# recursive glob of the project directory also read KiCad's .history/
# snapshots, this server's .mcp-backups/ copies and hand-made backup
# folders as live sheets, and whichever copy sorted last silently
# overwrote the live sheet's nets (#400).
sch_files = sheet_tree(Path(project_sch_path))
if sheets_out is not None:
sheets_out.extend(str(p) for p in sch_files)
logger.info(
f"_build_hierarchical_pad_net_map: scanning {len(sch_files)} sheet(s): "
+ ", ".join(p.name for p in sch_files)
)
for sch_path in sch_files:
# A broken sheet must fail the sync loudly: silently skipping it
# would produce an incomplete pad->net map and wrong-looking
# success. SchematicLoadError carries the flat-symbol diagnosis.
sch = SchematicManager.load_schematic(str(sch_path))
# ── 1. Collect explicit label positions → net name ──────────────
point_net: dict = {} # snap(x,y) -> net_name
for attr in ("label", "global_label", "hierarchical_label"):
for lbl in getattr(sch, attr, None) or []:
try:
pos = lbl.at.value
name = lbl.value
if name:
k = snap(pos[0], pos[1])
point_net[k] = name
all_net_names.add(name)
except Exception:
pass
# Power symbols (#PWR): the Value property IS the net name; seed
# every pin position with it.
#
# #FLG (PWR_FLAG) is deliberately excluded. Its Value is the literal
# string "PWR_FLAG", an ERC marker rather than a net name. Treating
# it as one invented a bogus "PWR_FLAG" net and, worse, stamped that
# name onto the flag's pin, from where the BFS below propagated it
# over the real net. A PWR_FLAG takes whatever net it is wired to;
# it never names one.
for sym in getattr(sch, "symbol", None) or []:
try:
ref = sym.property.Reference.value
if not ref.startswith("#PWR"):
continue
net_name = sym.property.Value.value
if not net_name:
continue
all_pins = pin_locator.get_all_symbol_pins(sch_path, ref)
for _pin_num, (px, py) in all_pins.items():
k = snap(px, py)
point_net[k] = net_name
all_net_names.add(net_name)
except Exception:
pass
# ── 2. Build wire adjacency and BFS-propagate net names ──────────
wire_segments = []
for wire in getattr(sch, "wire", None) or []:
try:
pts = []
for pt in wire.pts.xy:
pts.append(snap(pt.value[0], pt.value[1]))
if len(pts) >= 2:
wire_segments.append(pts)
except Exception:
pass
# Adjacency: connect endpoints of different segments that share a grid point
point_adj: dict = defaultdict(set)
for seg in wire_segments:
# Connect consecutive points within the segment
for i in range(len(seg) - 1):
point_adj[seg[i]].add(seg[i + 1])
point_adj[seg[i + 1]].add(seg[i])
# All unique wire points
all_wire_pts = set()
for seg in wire_segments:
all_wire_pts.update(seg)
# BFS: propagate known net names through wire connections
queue = [pt for pt in all_wire_pts if pt in point_net]
visited = set(queue)
while queue:
pt = queue.pop()
net = point_net[pt]
for neighbor in point_adj[pt]:
if neighbor not in point_net:
point_net[neighbor] = net
all_net_names.add(net)
if neighbor not in visited:
visited.add(neighbor)
queue.append(neighbor)
# ── 3. Collect real (non-power) symbol pins with their positions ──
sym_pins = [] # (ref, pin_num, (px, py))
for sym in getattr(sch, "symbol", None) or []:
try:
ref = sym.property.Reference.value
if ref.startswith("#"):
continue
except Exception:
continue
pin_positions = pin_locator.get_all_symbol_pins(sch_path, ref)
for pin_num, (px, py) in pin_positions.items():
sym_pins.append((ref, pin_num, (px, py)))
# ── 3b. Name wire clusters that carry no label ────────────────────
# A net formed only by a wire between component pins -- no label, no
# power symbol -- was named by nothing above, so its pads reached the
# board with no net at all and the connection silently vanished from
# the layout (#402). KiCad names such a net itself; the rule, checked
# against kicad-cli 10.0.0 on synthetic schematics, is
# "Net-(REF-PadN)" built from the pad NUMBER (the pin's name is not
# used even when it has one), choosing the candidate that sorts
# lowest as a plain string, so R10 beats R2. Step 2's BFS floods a
# name across a whole connected cluster, so a cluster is either
# wholly named or wholly unnamed, and testing one point suffices.
def in_cluster(pt, cluster):
if snap(*pt) in cluster:
return True
x, y = pt
return any(
abs(x - cx) < TOLERANCE and abs(y - cy) < TOLERANCE for cx, cy in cluster
)
clustered: set = set()
for seed in sorted(all_wire_pts):
if seed in point_net or seed in clustered:
continue
cluster = set()
stack = [seed]
clustered.add(seed)
while stack:
cur = stack.pop()
cluster.add(cur)
for neighbor in point_adj[cur]:
if neighbor not in clustered:
clustered.add(neighbor)
stack.append(neighbor)
candidates = [
f"Net-({ref}-Pad{pin_num})"
for ref, pin_num, pt in sym_pins
if in_cluster(pt, cluster)
]
# A single pin on a wire stub is a dangling wire, not a net.
if len(candidates) < 2:
continue
auto_name = min(candidates)
for pt in cluster:
point_net[pt] = auto_name
all_net_names.add(auto_name)
# ── 3c. Match component pin positions to net names ───────────────
for ref, pin_num, (px, py) in sym_pins:
net = nearby_net((px, py), point_net)
if net:
pad_net_map[(ref, pin_num)] = net
logger.info(
f"_build_hierarchical_pad_net_map: {len(pad_net_map)} pin→net assignments, "
f"{len(all_net_names)} unique nets"
)
return pad_net_map, all_net_names
def _handle_sync_schematic_to_board(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Sync schematic netlist to PCB board (equivalent to KiCAD F8 'Update PCB from Schematic').
Reads net connections from the schematic and assigns them to the matching pads in the PCB.
"""
logger.info("Syncing schematic to board")
try:
from pathlib import Path
schematic_path = params.get("schematicPath")
board_path = params.get("boardPath")
# Determine board to work with
board = None
if board_path:
board = self._safe_load_board(board_path)
if board is None:
return {
"success": False,
"message": f"Could not load board from {board_path}",
"errorDetails": (
"pcbnew.LoadBoard failed or returned a dehydrated "
"SWIG proxy that could not be recovered"
),
}
elif self.board:
board = self.board
board_path = board.GetFileName() if not board_path else board_path
else:
return {
"success": False,
"message": "No board loaded. Use open_project first or provide boardPath.",
}
if not board_path:
board_path = board.GetFileName()
# Determine schematic path if not provided
if not schematic_path:
sch = Path(board_path).with_suffix(".kicad_sch")
if sch.exists():
schematic_path = str(sch)
else:
project_dir = Path(board_path).parent
sch_files = list(project_dir.glob("*.kicad_sch"))
if sch_files:
schematic_path = str(sch_files[0])
if not schematic_path or not Path(schematic_path).exists():
return {
"success": False,
"message": f"Schematic not found. Provide schematicPath. Tried: {schematic_path}",
}
# Build the pad->net map from the sheets reachable from the root
sheets_scanned: list = []
pad_net_map, net_names = self._build_hierarchical_pad_net_map(
schematic_path, sheets_out=sheets_scanned
)
# Add missing footprints from the schematic to the board *before*
# we add nets and assign pads — F8 in KiCad does this implicitly
# ("Update PCB from Schematic"), but our previous implementation
# only mutated nets, leaving newly-added schematic symbols with no
# PCB footprint at all.
added_footprints, skipped_footprints = self._add_missing_footprints_from_schematic(
board, schematic_path
)
# Add all nets to board
netinfo = board.GetNetInfo()
nets_by_name = netinfo.NetsByName()
added_nets = []
for net_name in net_names:
if not nets_by_name.has_key(net_name):
net_item = pcbnew.NETINFO_ITEM(board, net_name)
board.Add(net_item)
added_nets.append(net_name)
# Refresh nets map after additions
netinfo = board.GetNetInfo()
nets_by_name = netinfo.NetsByName()
# Assign nets to pads (now also covers any footprints we just added)
assigned_pads = 0
unmatched = []
for fp in board.GetFootprints():
ref = fp.GetReference()
for pad in fp.Pads():
pad_num = pad.GetNumber()
key = (ref, str(pad_num))
if key in pad_net_map:
net_name = pad_net_map[key]
if nets_by_name.has_key(net_name):
pad.SetNet(nets_by_name[net_name])
assigned_pads += 1
else:
unmatched.append(f"{ref}/{pad_num}")
with preserve_project_settings(board_path):
board.Save(board_path)
# If board was loaded fresh, update internal reference
if params.get("boardPath"):
self.board = board
self._update_command_handlers()
logger.info(
f"sync_schematic_to_board: {len(added_nets)} nets added, "
f"{len(added_footprints)} footprints added, {assigned_pads} pads assigned"
)
# Every wired pin now carries a net (an unlabeled wire gets a
# Net-(REF-PadN) name), so an unmatched pad is one the schematic
# does not wire at all, or a pad the symbol does not have. Report
# all of them: a ten-entry sample buried a 25-pad failure (#400).
warnings: List[str] = []
if unmatched:
warnings.append(
f"{len(unmatched)} pad(s) have no net in the schematic and were left "
"unchanged: " + ", ".join(unmatched)
)
logger.warning("sync_schematic_to_board: " + warnings[-1])
root_dir = Path(schematic_path).parent
sheet_names = []
for p in sheets_scanned:
try:
sheet_names.append(Path(p).relative_to(root_dir).as_posix())
except ValueError:
sheet_names.append(Path(p).name)
return {
"success": True,
"message": (
f"PCB updated from schematic: {len(added_footprints)} footprints added, "
f"{len(added_nets)} nets added, {assigned_pads} pads assigned"
),
"sheets_scanned": sheet_names,
"nets_added": added_nets,
"nets_total": len(net_names),
"pads_assigned": assigned_pads,
"unmatched_pad_count": len(unmatched),
"unmatched_pads": unmatched,
"unmatched_pads_sample": unmatched[:10],
"warnings": warnings,
"footprints_added": added_footprints,
"footprints_skipped": skipped_footprints,
}
except SchematicLoadError as e:
return e.to_response()
except Exception as e:
logger.error(f"Error in sync_schematic_to_board: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _extract_components_from_schematic(self, schematic_path: str) -> List[Dict[str, str]]:
"""Run kicad-cli netlist export and return the flat list of components.
Each entry: {"reference": str, "value": str, "footprint": str}
Empty list on any failure (kicad-cli missing, parse error, etc.) — the
caller treats that as "no missing footprints to add".
"""
import subprocess
import tempfile
import xml.etree.ElementTree as ET
kicad_cli = self._find_kicad_cli_static()
if not kicad_cli:
logger.warning("kicad-cli not found — sync will not add new footprints")
return []
with tempfile.NamedTemporaryFile(suffix=".xml", delete=False) as tmp:
tmp_path = tmp.name
try:
cmd = [
kicad_cli,
"sch",
"export",
"netlist",
"--format",
"kicadxml",
"--output",
tmp_path,
schematic_path,
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
if result.returncode != 0:
logger.warning(
f"kicad-cli netlist export failed (exit {result.returncode}): "
f"{result.stderr.strip()}"
)
return []
tree = ET.parse(tmp_path)
root = tree.getroot()
components = []
for comp in root.findall("./components/comp"):
# <tstamps> carries the schematic symbol UUID -- verified
# against kicad-cli 10.0: a bare UUID with no sheet path
# (instances of a reused sheet share it). Legacy exports
# could comma-join several; keep the first.
tstamps = (comp.findtext("tstamps") or "").replace(",", " ").split()
components.append(
{
"reference": comp.get("ref", ""),
"value": comp.findtext("value", ""),
"footprint": comp.findtext("footprint", ""),
"uuid": tstamps[0] if tstamps else "",
}
)
return components
except Exception as e:
logger.warning(f"Failed to extract components from schematic: {e}")
return []
finally:
try:
Path(tmp_path).unlink()
except OSError:
pass
@staticmethod
def _fp_lib_tables_state(project_dir: "Path", manager: Any) -> Tuple[Any, Any]:
"""Freshness key for a cached ``LibraryManager``: fp-lib-table mtimes.
A cached manager must not outlive edits to the tables it parsed —
``register_footprint_library`` rewrites the project or global
fp-lib-table mid-session, and the KiCad GUI can too (Preferences >
Manage Footprint Libraries). Two ``stat`` calls per sync are noise
next to the parse they let us skip. The global table path is
re-probed each time (rather than remembered) so a global table
*created* after the first build is also detected.
"""
def _mtime_ns(path: Any) -> Any:
if path is None:
return None
try:
return path.stat().st_mtime_ns
except OSError:
return None
# Tolerant getattr: tests substitute lightweight fakes for
# LibraryManager that don't carry the probe helper.
probe = getattr(manager, "_get_global_fp_lib_table", lambda: None)
return (_mtime_ns(project_dir / "fp-lib-table"), _mtime_ns(probe()))
def _get_project_library_manager(self, project_dir: "Path") -> Any:
"""Return a footprint ``LibraryManager`` for ``project_dir``, cached on ``self``.
Building one re-parses the global fp-lib-table plus the project's
fp-lib-table (recursively following any ``Table`` references). Doing
that on every ``sync_schematic_to_board`` call is pure waste when the
tool is invoked repeatedly against the same project, e.g. an
iterative rebuild flow (#248). Mirrors the cache-on-project-change
pattern already used for ``place_component`` above, with one addition:
the cache key includes the fp-lib-table mtimes (see
``_fp_lib_tables_state``), so a table edit mid-session triggers a
rebuild instead of serving stale library data.
"""
from commands.library import LibraryManager
cached = getattr(self, "_sync_library_manager", None)
cached_dir = getattr(self, "_sync_library_manager_project", None)
cached_state = getattr(self, "_sync_library_manager_state", None)
if (
cached is None
or cached_dir != project_dir
or cached_state != self._fp_lib_tables_state(project_dir, cached)
):
cached = LibraryManager(project_path=project_dir)
self._sync_library_manager = cached
self._sync_library_manager_project = project_dir
# Stamp freshness after the parse: if a table changes between the
# parse and the stat, the next call sees a newer mtime and rebuilds.
self._sync_library_manager_state = self._fp_lib_tables_state(project_dir, cached)
return cached
@staticmethod
def _board_symbol_uuid(fp: Any) -> str:
"""The schematic symbol UUID a board footprint is bound to, or "".
``GetPath().AsString()`` is ``/sheet-uuid/.../symbol-uuid`` (verified
against real KiCad 10); the last segment is the symbol UUID that
kicad-cli's netlist reports in ``<tstamps>``. Footprints placed by
hand have an empty path. Tolerant of test doubles without GetPath.
"""
try:
path = fp.GetPath().AsString()
except Exception:
return ""
if not isinstance(path, str) or not path:
return ""
return path.rstrip("/").rsplit("/", 1)[-1]
def _add_missing_footprints_from_schematic(
self, board: Any, schematic_path: str
) -> Tuple[List[Dict[str, str]], List[Dict[str, str]]]:
"""Add footprints to ``board`` for any schematic component not yet present.
New footprints are placed at the board origin so the user can move them
into position. Power/flag references (``#PWR``, ``#FLG``) are skipped —
they have no PCB representation.
Returns ``(added, skipped)``: each entry is
``{"reference": str, "footprint": str, "reason": str?}``.
"""
from pathlib import Path
added: List[Dict[str, str]] = []
skipped: List[Dict[str, str]] = []
components = self._extract_components_from_schematic(schematic_path)
if not components:
return added, skipped
existing_refs = {fp.GetReference() for fp in board.GetFootprints()}
# Board footprints bound (by symbol UUID) to a schematic symbol whose
# reference no netlist comp claims: the #250 shape. A schematic that
# was re-annotated after layout matches nothing by reference, and
# adding a second copy of every such part duplicates the board. The
# UUID says "same symbol, different name" -- skip the add and say so.
# Reused-sheet instances share a symbol UUID, hence the list: each
# unmatched comp claims one unclaimed footprint.
sch_refs = {c["reference"] for c in components if c.get("reference")}
unclaimed_by_uuid: Dict[str, List[Any]] = {}
for fp in board.GetFootprints():
fp_uuid = self._board_symbol_uuid(fp)
if fp_uuid and fp.GetReference() not in sch_refs:
unclaimed_by_uuid.setdefault(fp_uuid, []).append(fp)
project_dir = Path(schematic_path).parent
library_manager = self._get_project_library_manager(project_dir)
# One FootprintLoad per distinct (library, footprint), cloned per use.
# Measured on a 40-component board (#248): FootprintLoad costs the same
# every time -- nothing downstream memoizes it -- so a board with 13
# identical resistors paid for 13 identical disk reads. Scoped to this
# call rather than process-wide so an edited library is never served
# stale.
proto_cache: Dict[Tuple[str, str], Any] = {}
for comp in components:
ref = comp["reference"]
fp_str = comp["footprint"]
if not ref or ref.startswith("#"):
# Power flags / global indicators — no PCB footprint expected.
continue
if ref in existing_refs:
continue
claimed = unclaimed_by_uuid.get(comp.get("uuid", ""))
if claimed:
board_fp = claimed.pop(0)
board_ref = board_fp.GetReference()
logger.warning(
f"sync: schematic '{ref}' is already on the board as "
f"'{board_ref}' (same symbol UUID, references diverged) "
"-- not adding a duplicate footprint (#250)"
)
skipped.append(
{
"reference": ref,
"footprint": fp_str,
"reason": (
f"already on board as '{board_ref}' (same schematic "
"symbol, references diverged -- re-annotate or rename "
"to reconcile; no duplicate added)"
),
}
)
continue
if not fp_str or ":" not in fp_str:
skipped.append(
{
"reference": ref,
"footprint": fp_str,
"reason": "no Library:Name footprint set on schematic symbol",
}
)
continue
lib_name, fp_name = fp_str.split(":", 1)
library_path = library_manager.libraries.get(lib_name)
if not library_path:
skipped.append(
{
"reference": ref,
"footprint": fp_str,
"reason": f"library '{lib_name}' not in fp-lib-table",
}
)
continue
cache_key = (library_path, fp_name)
try:
if cache_key not in proto_cache:
proto_cache[cache_key] = pcbnew.FootprintLoad(library_path, fp_name)
prototype = proto_cache[cache_key]
# Always clone, including the first use: the prototype must
# never be handed to board.Add(), or the cache would be left
# holding a board-owned object that later gets mutated and
# invalidated on save/reload. Cloning costs ~0.01 ms against
# ~15 ms for a FootprintLoad.
module = clone_footprint(prototype) if prototype is not None else None
except Exception as e:
skipped.append(
{"reference": ref, "footprint": fp_str, "reason": f"FootprintLoad failed: {e}"}
)
continue
if not module:
skipped.append(
{
"reference": ref,
"footprint": fp_str,
"reason": f"footprint '{fp_name}' not in '{lib_name}'",
}
)
continue
module.SetReference(ref)
if comp["value"]:
module.SetValue(comp["value"])
module.SetFPID(pcbnew.LIB_ID(lib_name, fp_name))
# Place at board origin; user / autoplacer can position from there.
module.SetPosition(pcbnew.VECTOR2I(0, 0))
board.Add(module)
existing_refs.add(ref)
added.append({"reference": ref, "footprint": fp_str})
if added:
logger.info(f"_add_missing_footprints_from_schematic: added {len(added)} footprints")
return added, skipped
def _handle_get_schematic_view_region(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Export a cropped region of the schematic as an image"""
logger.info("Exporting schematic view region")
import base64
import subprocess
import tempfile
try:
schematic_path = params.get("schematicPath")
if not schematic_path or not Path(schematic_path).exists():
return {"success": False, "message": "Schematic file not found"}
x1 = float(params.get("x1", 0))
y1 = float(params.get("y1", 0))
x2 = float(params.get("x2", 297))
y2 = float(params.get("y2", 210))
x1, x2 = min(x1, x2), max(x1, x2)
y1, y2 = min(y1, y2), max(y1, y2)
out_format = params.get("format", "png")
width = int(params.get("width", 800))
height = int(params.get("height", 600))
kicad_cli = self.design_rule_commands._find_kicad_cli()
if not kicad_cli:
return {"success": False, "message": "kicad-cli not found"}
tmp_dir = tempfile.mkdtemp()
svg_output = None
try:
cmd = [
kicad_cli,
"sch",
"export",
"svg",
"--output",
tmp_dir,
schematic_path,
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
if result.returncode != 0:
return {
"success": False,
"message": f"SVG export failed: {result.stderr}",
}
# kicad-cli names one file per page after the schematic/sheet
# names — pick the root page deterministically.
svg_files = [f for f in os.listdir(tmp_dir) if f.endswith(".svg")]
if not svg_files:
return {
"success": False,
"message": "kicad-cli produced no SVG output",
}
svg_output = _pick_root_svg(tmp_dir, schematic_path)
if svg_output is None:
svg_output = os.path.join(tmp_dir, sorted(svg_files)[0])
logger.warning(
"Could not identify root SVG page for %s; " "falling back to %s",
schematic_path,
svg_output,
)
import xml.etree.ElementTree as ET
tree = ET.parse(svg_output)
root = tree.getroot()
# KiCad schematic SVGs use mm as viewBox units directly
vb = root.get("viewBox", "")
if vb:
parts = vb.split()
if len(parts) == 4:
orig_vb_x = float(parts[0])
orig_vb_y = float(parts[1])
new_x = orig_vb_x + x1
new_y = orig_vb_y + y1
new_w = x2 - x1
new_h = y2 - y1
root.set("viewBox", f"{new_x} {new_y} {new_w} {new_h}")
root.set("width", str(width))
root.set("height", str(height))
# Write modified SVG
cropped_svg_path = os.path.join(tmp_dir, "cropped.svg")
tree.write(cropped_svg_path, xml_declaration=True, encoding="utf-8")
if out_format == "svg":
with open(cropped_svg_path, "r", encoding="utf-8") as f:
svg_data = f.read()
return {"success": True, "imageData": svg_data, "format": "svg"}
else:
png_data = _svg_to_png(cropped_svg_path, width, height)
if png_data is None:
return {
"success": False,
"message": "No PNG converter available. Install pymupdf, inkscape, or imagemagick.",
}
return {
"success": True,
"imageData": base64.b64encode(png_data).decode("utf-8"),
"format": "png",
}
finally:
import shutil
shutil.rmtree(tmp_dir, ignore_errors=True)
except Exception as e:
logger.error(f"Error in get_schematic_view_region: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_find_wires_crossing_symbols(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Find wires that cross over component symbol bodies"""
logger.info("Finding wires crossing symbols in schematic")
try:
from pathlib import Path
from commands.schematic_analysis import find_wires_crossing_symbols
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
result = find_wires_crossing_symbols(Path(schematic_path))
return {
"success": True,
"collisions": result,
"count": len(result),
"message": f"Found {len(result)} wire(s) crossing symbols",
}
except Exception as e:
logger.error(f"Error checking wire collisions: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_find_orphaned_wires(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Find wire segments with at least one dangling (unconnected) endpoint"""
logger.info("Finding orphaned wires in schematic")
try:
from pathlib import Path
from commands.schematic_analysis import find_orphaned_wires
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
result = find_orphaned_wires(Path(schematic_path))
return {
"success": True,
**result,
"message": f"Found {result['count']} orphaned wire(s)",
}
except SchematicLoadError as e:
return e.to_response()
except Exception as e:
logger.error(f"Error finding orphaned wires: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_list_floating_labels(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""List net labels that are not connected to any component pin"""
logger.info("Listing floating net labels in schematic")
try:
from commands.wire_connectivity import list_floating_labels
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
try:
schematic = SchematicManager.load_schematic(schematic_path)
except SchematicLoadError as e:
return e.to_response()
labels = list_floating_labels(schematic, schematic_path)
return {
"success": True,
"floating_labels": labels,
"count": len(labels),
"message": f"Found {len(labels)} floating label(s)",
}
except Exception as e:
logger.error(f"Error listing floating labels: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_snap_to_grid(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Snap schematic element coordinates to the nearest grid point"""
logger.info("Snapping schematic elements to grid")
try:
from pathlib import Path
from commands.schematic_snap import snap_to_grid
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
grid_size = float(params.get("gridSize", 1.27))
elements = params.get("elements") # None → defaults inside snap_to_grid
result = snap_to_grid(Path(schematic_path), grid_size=grid_size, elements=elements)
total = result["snapped"] + result["already_on_grid"]
return {
"success": True,
**result,
"message": (
f"Snapped {result['snapped']} element(s) to {grid_size} mm grid "
f"({result['already_on_grid']} of {total} were already on grid)"
),
}
except Exception as e:
logger.error(f"Error snapping to grid: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}
def _handle_lint_offgrid(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Report (and optionally fix) off-grid connection geometry"""
logger.info("Linting schematic for off-grid geometry")
try:
from commands.schematic_lint_grid import lint_offgrid
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
if not os.path.isfile(schematic_path):
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
fix = bool(params.get("fix", False))
grid_size = float(params.get("gridSize", 1.27))
if grid_size <= 0:
return {"success": False, "message": "gridSize must be > 0"}
result = lint_offgrid(schematic_path, grid_mm=grid_size, fix=fix)
offender_count = len(result["offenders"])
if offender_count == 0:
message = f"No off-grid geometry (grid {grid_size} mm)"
elif fix:
message = (
f"Snapped {result['fixed']} of {offender_count} off-grid "
f"item(s) to the {grid_size} mm grid"
+ (
f"; {result['needsHuman']} offender(s) >0.5mm off-grid "
f"left untouched (needsHuman)"
if result["needsHuman"]
else ""
)
)
else:
message = (
f"Found {offender_count} off-grid item(s) on the "
f"{grid_size} mm grid ({result['needsHuman']} needing "
f"human review); run with fix=true to snap them"
)
return {
"success": True,
"gridSize": grid_size,
**result,
"message": message,
}
except Exception as e:
logger.error(f"Error linting off-grid geometry: {e}")
import traceback
logger.error(traceback.format_exc())
return {"success": False, "message": str(e)}