nearxos/kicad-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
602d76af236628abc06a3c70e5b247c591efb755
kicad-mcp-server/python/commands/schematic_handlers.py
mixelpixx 602d76af23 refactor: extract schematic handlers into SchematicHandlersMixin 
kicad_interface.py had grown to 8,596 lines. Most domains already delegate to
classes in commands/, but the schematic domain (43 _handle_* methods + the
schematic-only _svg_to_png helper, ~3,100 lines) was still inline. Move it into
commands/schematic_handlers.py as SchematicHandlersMixin and have KiCADInterface
inherit from it.

Pure mechanical relocation — no logic change:
- Methods stay bound to the same instance via the mixin, so every self.board,
  self._auto_save_board(), cross-handler call, and the command_routes table
  resolve exactly as before.
- Verified all 159 functions/methods (116 kept + 43 moved + _svg_to_png) are
  AST-identical between the original and the split.
- tests/test_sync_schematic_to_board_footprints.py patch targets updated to
  follow the moved module (commands.schematic_handlers.pcbnew).
- Full suite at the exact current main baseline (8 failed / 1044 passed; the 8
  are the pre-existing Java-not-installed freerouting/autoroute set).

Regenerated against main after the June merge wave (#226-229, #231-234, #238,
#252, #253); deliberately excludes the 19 _handle_export_* kicad-cli handlers
from #253 and the datasheet delegators.
2026-06-12 12:29:46 -04:00

3105 lines
125 KiB
Python
Raw Blame History

"""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 Any, Dict, List, Optional, Tuple
import pcbnew
import sexpdata
from commands.library_schematic import LibraryManager as SchematicLibraryManager
from commands.schematic import SchematicManager
from commands.wire_manager import WireManager
logger = logging.getLogger("kicad_interface")
def _svg_to_png(svg_path: str, width: int, height: int) -> Optional[bytes]:
"""Convert SVG to PNG. No cffi dependency.
Priority:
1. pymupdf (fitz) — bundled MuPDF renderer, pure Python, no system deps
2. Inkscape CLI — accurate KiCAD SVG rendering
3. ImageMagick convert — broad availability fallback
Returns PNG bytes or None if all converters fail.
"""
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
out_path = os.path.join(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 os.path.exists(out_path):
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}", out_path],
capture_output=True,
timeout=60,
)
if r.returncode == 0 and os.path.exists(out_path):
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."""
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
path = os.path.dirname(filename) or "."
project_name = project_name or os.path.basename(filename)
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
)
base_name = (
project_name if project_name.endswith(".kicad_sch") else f"{project_name}.kicad_sch"
)
normalized_path = path or "."
file_path = os.path.join(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"}
schematic = SchematicManager.load_schematic(filename)
success = schematic is not None
if success:
metadata = SchematicManager.get_schematic_metadata(schematic)
return {"success": success, "metadata": metadata}
else:
return {"success": False, "message": "Failed to load schematic"}
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)
# 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,
project_path=derived_project_path,
)
return {
"success": True,
"component_reference": reference,
"symbol_source": f"{library}:{comp_type}",
}
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")
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,
):
blocks_to_delete.append((pos, end))
search_start = end + 1
if not blocks_to_delete:
return {
"success": False,
"message": f"Component '{reference}' not found in schematic (note: this tool removes schematic symbols, use delete_component for PCB footprints)",
}
# 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}")
return {
"success": True,
"reference": reference,
"deleted_count": deleted_count,
"schematic": str(sch_file),
}
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 _escape_sexpr_string(value: str) -> str:
"""Escape a string for safe insertion into an S-expression double-quoted token."""
return value.replace("\\", "\\\\").replace('"', '\\"')
@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}")
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
prop_pattern = re.compile(
r'\(property\s+"([^"]*)"\s+"([^"]*)"\s+\(at\s+([\d\.\-]+)\s+([\d\.\-]+)\s+([\d\.\-]+)\s*\)'
)
fields = {}
for m in prop_pattern.finditer(block_text):
name, value, x, y, angle = (
m.group(1),
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
snapped_info = []
if snap_to_pins:
from commands.pin_locator import PinLocator
locator = PinLocator()
sch_path = Path(schematic_path)
# Load schematic to iterate all symbols
from skip import Schematic as SkipSchematic
sch = SkipSchematic(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, tolerance: Any) -> Any:
"""Find the nearest pin within tolerance of a point."""
best = None
best_dist = tolerance
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 dist <= best_dist:
best_dist = dist
best = (ref, pin_num, coords)
return best
# Snap first endpoint
match = find_nearest_pin(points[0], snap_tolerance)
if match:
ref, pin_num, coords = match
logger.info(
f"Snapped start point {points[0]} -> {coords} (pin {ref}/{pin_num})"
)
snapped_info.append(
f"start snapped to {ref}/{pin_num} at [{coords[0]}, {coords[1]}]"
)
points[0] = list(coords)
# Snap last endpoint
match = find_nearest_pin(points[-1], snap_tolerance)
if match:
ref, pin_num, coords = match
logger.info(f"Snapped end point {points[-1]} -> {coords} (pin {ref}/{pin_num})")
snapped_info.append(
f"end snapped to {ref}/{pin_num} at [{coords[0]}, {coords[1]}]"
)
points[-1] = list(coords)
# 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)
return {"success": True, "message": message}
else:
return {"success": False, "message": "Failed to add wire"}
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 os.path.exists(schematic_path):
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
import subprocess
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 in PATH"}
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 = params.get("orientation", 0)
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}"
)
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,
}
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 os.path.exists(schematic_path):
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
with tempfile.TemporaryDirectory() as tmpdir:
svg_path = os.path.join(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 may name the file after the schematic, find it
import glob
svg_files = glob.glob(os.path.join(tmpdir, "*.svg"))
if not svg_files:
return {
"success": False,
"message": "No SVG file produced by kicad-cli",
}
svg_path = svg_files[0]
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 in PATH"}
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}",
}
schematic = SchematicManager.load_schematic(schematic_path)
if not schematic:
return {"success": False, "message": "Failed to load schematic"}
# 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.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_net,
)
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
schematic = SchematicManager.load_schematic(schematic_path)
if not schematic:
return {"success": False, "message": "Failed to load schematic"}
# 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)
nets = []
for net_name in sorted(net_names):
connections = get_connections_for_net(schematic, schematic_path, 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,
)
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"}
schematic = SchematicManager.load_schematic(schematic_path)
if not schematic:
return {"success": False, "message": "Failed to load schematic"}
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"}
schematic = SchematicManager.load_schematic(schematic_path)
if not schematic:
return {"success": False, "message": "Failed to load schematic"}
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)}
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)
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)
if found is None:
return {"success": False, "message": f"Component {reference} not found"}
_, 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)
)
# 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
drag_summary = WireDragger.drag_wires(sch_data, old_to_new)
# 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
)
drag_summary["wires_synthesized"] = wires_synthesized
# Update symbol position
WireDragger.update_symbol_position(sch_data, reference, float(new_x), float(new_y))
WireManager.sync_junctions(sch_data)
with open(schematic_path, "w", encoding="utf-8") as f:
f.write(sexpdata.dumps(sch_data))
return {
"success": True,
"oldPosition": old_position,
"newPosition": {"x": new_x, "y": new_y},
"wiresMoved": drag_summary.get("endpoints_moved", 0),
"wiresRemoved": drag_summary.get("wires_removed", 0),
"wiresSynthesized": drag_summary.get("wires_synthesized", 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
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)
if found is None:
return {"success": False, "message": f"Component {reference} not found"}
# 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
)
# 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
drag_summary = WireDragger.drag_wires(sch_data, old_to_new)
# Update the symbol's rotation and mirror token in sexpdata
WireDragger.update_symbol_rotation_mirror(
sch_data, reference, float(angle), effective_mirror
)
WireManager.sync_junctions(sch_data)
with open(schematic_path, "w", encoding="utf-8") as f:
f.write(_sexpdata.dumps(sch_data))
return {
"success": True,
"reference": reference,
"angle": angle,
"mirror": effective_mirror,
"wiresMoved": drag_summary.get("endpoints_moved", 0),
"wiresRemoved": drag_summary.get("wires_removed", 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, ...)"""
logger.info("Annotating schematic")
try:
import re
schematic_path = params.get("schematicPath")
if not schematic_path:
return {"success": False, "message": "schematicPath is required"}
schematic = SchematicManager.load_schematic(schematic_path)
if not schematic:
return {"success": False, "message": "Failed to load schematic"}
# Collect existing references by prefix
existing_refs = {} # prefix -> set of numbers
unannotated = [] # (symbol, prefix)
for symbol in schematic.symbol:
if not hasattr(symbol.property, "Reference"):
continue
ref = symbol.property.Reference.value
if ref.startswith("_TEMPLATE"):
continue
# Split reference into prefix and number
match = re.match(r"^([A-Za-z_]+)(\d+)$", ref)
if match:
prefix = match.group(1)
num = int(match.group(2))
if prefix not in existing_refs:
existing_refs[prefix] = set()
existing_refs[prefix].add(num)
elif ref.endswith("?"):
prefix = ref[:-1]
unannotated.append((symbol, prefix))
if not unannotated:
return {
"success": True,
"annotated": [],
"message": "All components already annotated",
}
annotated = []
for symbol, prefix in unannotated:
if prefix not in existing_refs:
existing_refs[prefix] = set()
# Find next available number
next_num = 1
while next_num in existing_refs[prefix]:
next_num += 1
old_ref = symbol.property.Reference.value
new_ref = f"{prefix}{next_num}"
symbol.setAllReferences(new_ref)
existing_refs[prefix].add(next_num)
uuid_val = str(symbol.uuid.value) if hasattr(symbol, "uuid") else ""
annotated.append(
{
"uuid": uuid_val,
"oldReference": old_ref,
"newReference": new_ref,
}
)
SchematicManager.save_schematic(schematic, schematic_path)
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(_sexpdata.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 glob
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 os.path.exists(schematic_path):
return {
"success": False,
"message": f"Schematic not found: {schematic_path}",
}
# kicad-cli's --output flag for SVG export expects a directory, not a file path.
# The output file is auto-named based on the schematic name.
output_dir = os.path.dirname(output_path)
if not output_dir:
output_dir = "."
os.makedirs(output_dir, exist_ok=True)
cmd = [
"kicad-cli",
"sch",
"export",
"svg",
schematic_path,
"-o",
output_dir,
]
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}",
}
# kicad-cli names the file after the schematic, so find the generated SVG
svg_files = glob.glob(os.path.join(output_dir, "*.svg"))
if not svg_files:
return {
"success": False,
"message": "No SVG file produced by kicad-cli",
}
generated_svg = svg_files[0]
# Move/rename to the user-specified output path if it differs
if os.path.abspath(generated_svg) != os.path.abspath(output_path):
shutil.move(generated_svg, output_path)
return {"success": True, "file": {"path": output_path}}
except FileNotFoundError:
return {"success": False, "message": "kicad-cli not found in PATH"}
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"}
schematic = SchematicManager.load_schematic(schematic_path)
if not schematic:
return {"success": False, "message": "Failed to load schematic"}
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 os
import subprocess
import tempfile
try:
schematic_path = params.get("schematicPath")
if not schematic_path or not os.path.exists(schematic_path):
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).
if not os.path.exists(json_output) or os.path.getsize(json_output) == 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 os.path.exists(json_output):
os.unlink(json_output)
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):
"""Walk all .kicad_sch files in the project and build a {(ref, pin_num): net_name} map.
Handles hierarchical schematics by scanning every sub-sheet file. Net names
from global_label / hierarchical_label / local label / power symbols are all
collected. Wire connectivity is traced via BFS so labels not placed directly
on a pin endpoint still reach through wire segments.
Returns: (pad_net_map, net_names_set)
"""
from collections import defaultdict
from pathlib import Path
from commands.pin_locator import PinLocator
from skip import Schematic
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
project_dir = Path(project_sch_path).parent
pad_net_map: dict = {}
all_net_names: set = set()
pin_locator = PinLocator()
sch_files = sorted(project_dir.rglob("*.kicad_sch"))
logger.info(f"_build_hierarchical_pad_net_map: scanning {len(sch_files)} schematic files")
for sch_path in sch_files:
try:
sch = Schematic(str(sch_path))
except Exception as e:
logger.warning(f"Could not load {sch_path}: {e}")
continue
# ── 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 / #FLG): value property IS the net name; use pin 1 pos
for sym in getattr(sch, "symbol", None) or []:
try:
ref = sym.property.Reference.value
if not (ref.startswith("#PWR") or ref.startswith("#FLG")):
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. Match component pin positions to net names ────────────────
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():
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 hierarchical pad→net map (walks all sub-sheets)
pad_net_map, net_names = self._build_hierarchical_pad_net_map(schematic_path)
# 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}")
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"
)
return {
"success": True,
"message": (
f"PCB updated from schematic: {len(added_footprints)} footprints added, "
f"{len(added_nets)} nets added, {assigned_pads} pads assigned"
),
"nets_added": added_nets,
"nets_total": len(net_names),
"pads_assigned": assigned_pads,
"unmatched_pads_sample": unmatched[:10],
"footprints_added": added_footprints,
"footprints_skipped": skipped_footprints,
}
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"):
components.append(
{
"reference": comp.get("ref", ""),
"value": comp.findtext("value", ""),
"footprint": comp.findtext("footprint", ""),
}
)
return components
except Exception as e:
logger.warning(f"Failed to extract components from schematic: {e}")
return []
finally:
try:
os.unlink(tmp_path)
except OSError:
pass
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
from commands.library import LibraryManager
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()}
project_dir = Path(schematic_path).parent
library_manager = LibraryManager(project_path=project_dir)
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
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
try:
module = pcbnew.FootprintLoad(library_path, fp_name)
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 os
import subprocess
import tempfile
try:
schematic_path = params.get("schematicPath")
if not schematic_path or not os.path.exists(schematic_path):
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 the file after the schematic
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 = os.path.join(tmp_dir, svg_files[0])
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 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"}
schematic = SchematicManager.load_schematic(schematic_path)
if not schematic:
return {"success": False, "message": "Failed to load schematic"}
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)}
Reference in New Issue View Git Blame Copy Permalink