Merge pull request #61 from Mehanik/fix/tool-schema-descriptions
feat: add_schematic_wire & add_schematic_junction with pin snapping and polyline routing
This commit is contained in:
mixelpixx
@@ -30,172 +30,6 @@ class ConnectionManager:
|
||||
cls._pin_locator = PinLocator()
|
||||
return cls._pin_locator
|
||||
|
||||
@staticmethod
|
||||
def add_wire(
|
||||
schematic_path: Path,
|
||||
start_point: list,
|
||||
end_point: list,
|
||||
properties: dict = None,
|
||||
):
|
||||
"""
|
||||
Add a wire between two points using WireManager
|
||||
|
||||
Args:
|
||||
schematic_path: Path to .kicad_sch file
|
||||
start_point: [x, y] coordinates for wire start
|
||||
end_point: [x, y] coordinates for wire end
|
||||
properties: Optional wire properties (stroke_width, stroke_type)
|
||||
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
if not WIRE_MANAGER_AVAILABLE:
|
||||
logger.error("WireManager not available")
|
||||
return False
|
||||
|
||||
stroke_width = properties.get("stroke_width", 0) if properties else 0
|
||||
stroke_type = (
|
||||
properties.get("stroke_type", "default") if properties else "default"
|
||||
)
|
||||
|
||||
success = WireManager.add_wire(
|
||||
schematic_path,
|
||||
start_point,
|
||||
end_point,
|
||||
stroke_width=stroke_width,
|
||||
stroke_type=stroke_type,
|
||||
)
|
||||
return success
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding wire: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def get_pin_location(symbol, pin_name: str):
|
||||
"""
|
||||
Get the absolute location of a pin on a symbol
|
||||
|
||||
Args:
|
||||
symbol: Symbol object
|
||||
pin_name: Name or number of the pin (e.g., "1", "GND", "VCC")
|
||||
|
||||
Returns:
|
||||
[x, y] coordinates or None if pin not found
|
||||
"""
|
||||
try:
|
||||
if not hasattr(symbol, "pin"):
|
||||
logger.warning(f"Symbol {symbol.property.Reference.value} has no pins")
|
||||
return None
|
||||
|
||||
# Find the pin by name
|
||||
target_pin = None
|
||||
for pin in symbol.pin:
|
||||
if pin.name == pin_name:
|
||||
target_pin = pin
|
||||
break
|
||||
|
||||
if not target_pin:
|
||||
logger.warning(
|
||||
f"Pin '{pin_name}' not found on {symbol.property.Reference.value}"
|
||||
)
|
||||
return None
|
||||
|
||||
# Get pin location relative to symbol
|
||||
pin_loc = target_pin.location
|
||||
# Get symbol location
|
||||
symbol_at = symbol.at.value
|
||||
|
||||
# Calculate absolute position
|
||||
# pin_loc is relative to symbol origin, need to add symbol position
|
||||
abs_x = symbol_at[0] + pin_loc[0]
|
||||
abs_y = symbol_at[1] + pin_loc[1]
|
||||
|
||||
return [abs_x, abs_y]
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting pin location: {e}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def add_connection(
|
||||
schematic_path: Path,
|
||||
source_ref: str,
|
||||
source_pin: str,
|
||||
target_ref: str,
|
||||
target_pin: str,
|
||||
routing: str = "direct",
|
||||
):
|
||||
"""
|
||||
Add a wire connection between two component pins
|
||||
|
||||
Args:
|
||||
schematic_path: Path to .kicad_sch file
|
||||
source_ref: Reference designator of source component (e.g., "R1", "R1_")
|
||||
source_pin: Pin name/number on source component
|
||||
target_ref: Reference designator of target component (e.g., "C1", "C1_")
|
||||
target_pin: Pin name/number on target component
|
||||
routing: Routing style ('direct', 'orthogonal_h', 'orthogonal_v')
|
||||
|
||||
Returns:
|
||||
True if connection was successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
if not WIRE_MANAGER_AVAILABLE:
|
||||
logger.error("WireManager/PinLocator not available")
|
||||
return False
|
||||
|
||||
locator = ConnectionManager.get_pin_locator()
|
||||
if not locator:
|
||||
logger.error("Pin locator unavailable")
|
||||
return False
|
||||
|
||||
# Get pin locations
|
||||
source_loc = locator.get_pin_location(
|
||||
schematic_path, source_ref, source_pin
|
||||
)
|
||||
target_loc = locator.get_pin_location(
|
||||
schematic_path, target_ref, target_pin
|
||||
)
|
||||
|
||||
if not source_loc or not target_loc:
|
||||
logger.error("Could not determine pin locations")
|
||||
return False
|
||||
|
||||
# Create wire based on routing style
|
||||
if routing == "direct":
|
||||
# Simple direct wire
|
||||
success = WireManager.add_wire(schematic_path, source_loc, target_loc)
|
||||
elif routing == "orthogonal_h":
|
||||
# Orthogonal routing (horizontal first)
|
||||
path = WireManager.create_orthogonal_path(
|
||||
source_loc, target_loc, prefer_horizontal_first=True
|
||||
)
|
||||
success = WireManager.add_polyline_wire(schematic_path, path)
|
||||
elif routing == "orthogonal_v":
|
||||
# Orthogonal routing (vertical first)
|
||||
path = WireManager.create_orthogonal_path(
|
||||
source_loc, target_loc, prefer_horizontal_first=False
|
||||
)
|
||||
success = WireManager.add_polyline_wire(schematic_path, path)
|
||||
else:
|
||||
logger.error(f"Unknown routing style: {routing}")
|
||||
return False
|
||||
|
||||
if success:
|
||||
logger.info(
|
||||
f"Connected {source_ref}/{source_pin} to {target_ref}/{target_pin} (routing: {routing})"
|
||||
)
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding connection: {e}")
|
||||
import traceback
|
||||
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def add_net_label(schematic: Schematic, net_name: str, position: list):
|
||||
"""
|
||||
@@ -257,15 +91,20 @@ class ConnectionManager:
|
||||
|
||||
# Add a small wire stub from the pin (2.54mm = 0.1 inch, standard grid spacing)
|
||||
# Stub direction follows the pin's outward angle from the PinLocator
|
||||
pin_angle_deg = getattr(locator, '_last_pin_angle', 0)
|
||||
pin_angle_deg = getattr(locator, "_last_pin_angle", 0)
|
||||
try:
|
||||
pin_angle_deg = locator.get_pin_angle(schematic_path, component_ref, pin_name) or 0
|
||||
pin_angle_deg = (
|
||||
locator.get_pin_angle(schematic_path, component_ref, pin_name) or 0
|
||||
)
|
||||
except Exception:
|
||||
pin_angle_deg = 0
|
||||
import math as _math
|
||||
|
||||
angle_rad = _math.radians(pin_angle_deg)
|
||||
stub_end = [round(pin_loc[0] + 2.54 * _math.cos(angle_rad), 4),
|
||||
round(pin_loc[1] - 2.54 * _math.sin(angle_rad), 4)]
|
||||
stub_end = [
|
||||
round(pin_loc[0] + 2.54 * _math.cos(angle_rad), 4),
|
||||
round(pin_loc[1] - 2.54 * _math.sin(angle_rad), 4),
|
||||
]
|
||||
|
||||
# Create wire stub using WireManager
|
||||
wire_success = WireManager.add_wire(schematic_path, pin_loc, stub_end)
|
||||
@@ -333,9 +172,15 @@ class ConnectionManager:
|
||||
connected = []
|
||||
failed = []
|
||||
|
||||
for pin_num in sorted(src_pins.keys(), key=lambda x: int(x) if x.isdigit() else 0):
|
||||
for pin_num in sorted(
|
||||
src_pins.keys(), key=lambda x: int(x) if x.isdigit() else 0
|
||||
):
|
||||
try:
|
||||
net_name = f"{net_prefix}_{int(pin_num) + pin_offset}" if pin_num.isdigit() else f"{net_prefix}_{pin_num}"
|
||||
net_name = (
|
||||
f"{net_prefix}_{int(pin_num) + pin_offset}"
|
||||
if pin_num.isdigit()
|
||||
else f"{net_prefix}_{pin_num}"
|
||||
)
|
||||
|
||||
ok_src = ConnectionManager.connect_to_net(
|
||||
schematic_path, source_ref, pin_num, net_name
|
||||
@@ -355,11 +200,15 @@ class ConnectionManager:
|
||||
failed.append(f"{target_ref}/{pin_num} (pin not found)")
|
||||
continue
|
||||
|
||||
connected.append(f"{source_ref}/{pin_num} <-> {target_ref}/{pin_num} [{net_name}]")
|
||||
connected.append(
|
||||
f"{source_ref}/{pin_num} <-> {target_ref}/{pin_num} [{net_name}]"
|
||||
)
|
||||
except Exception as e:
|
||||
failed.append(f"{source_ref}/{pin_num}: {e}")
|
||||
|
||||
logger.info(f"connect_passthrough: {len(connected)} connected, {len(failed)} failed")
|
||||
logger.info(
|
||||
f"connect_passthrough: {len(connected)} connected, {len(failed)} failed"
|
||||
)
|
||||
return {"connected": connected, "failed": failed}
|
||||
|
||||
@staticmethod
|
||||
@@ -607,35 +456,3 @@ class ConnectionManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating netlist: {e}")
|
||||
return {"nets": [], "components": []}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Example Usage (for testing)
|
||||
from schematic import (
|
||||
SchematicManager,
|
||||
) # Assuming schematic.py is in the same directory
|
||||
|
||||
# Create a new schematic
|
||||
test_sch = SchematicManager.create_schematic("ConnectionTestSchematic")
|
||||
|
||||
# Add some wires
|
||||
wire1 = ConnectionManager.add_wire(test_sch, [100, 100], [200, 100])
|
||||
wire2 = ConnectionManager.add_wire(test_sch, [200, 100], [200, 200])
|
||||
|
||||
# Note: add_connection, remove_connection, get_net_connections are placeholders
|
||||
# and require more complex implementation based on kicad-skip's structure.
|
||||
|
||||
# Example of how you might add a net label (requires finding a point on a wire)
|
||||
# from skip import Label
|
||||
# if wire1:
|
||||
# net_label_pos = wire1.start # Or calculate a point on the wire
|
||||
# net_label = test_sch.add_label(text="Net_01", at=net_label_pos)
|
||||
# print(f"Added net label 'Net_01' at {net_label_pos}")
|
||||
|
||||
# Save the schematic (optional)
|
||||
# SchematicManager.save_schematic(test_sch, "connection_test.kicad_sch")
|
||||
|
||||
# Clean up (if saved)
|
||||
# if os.path.exists("connection_test.kicad_sch"):
|
||||
# os.remove("connection_test.kicad_sch")
|
||||
# print("Cleaned up connection_test.kicad_sch")
|
||||
|
||||
@@ -11,24 +11,29 @@ import logging
|
||||
import math
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple, Optional, Dict
|
||||
from typing import List, Tuple, Optional
|
||||
import sexpdata
|
||||
from sexpdata import Symbol
|
||||
|
||||
logger = logging.getLogger("kicad_interface")
|
||||
logger = logging.getLogger('kicad_interface')
|
||||
|
||||
# Module-level Symbol constants — avoids repeated allocation on every call
|
||||
_SYM_WIRE = Symbol('wire')
|
||||
_SYM_PTS = Symbol('pts')
|
||||
_SYM_XY = Symbol('xy')
|
||||
_SYM_STROKE = Symbol('stroke')
|
||||
_SYM_WIDTH = Symbol('width')
|
||||
_SYM_TYPE = Symbol('type')
|
||||
_SYM_UUID = Symbol('uuid')
|
||||
_SYM_SHEET_INSTANCES = Symbol('sheet_instances')
|
||||
|
||||
|
||||
class WireManager:
|
||||
"""Manage wires in KiCad schematics using S-expression manipulation"""
|
||||
|
||||
@staticmethod
|
||||
def add_wire(
|
||||
schematic_path: Path,
|
||||
start_point: List[float],
|
||||
end_point: List[float],
|
||||
stroke_width: float = 0,
|
||||
stroke_type: str = "default",
|
||||
) -> bool:
|
||||
def add_wire(schematic_path: Path, start_point: List[float], end_point: List[float],
|
||||
stroke_width: float = 0, stroke_type: str = 'default') -> bool:
|
||||
"""
|
||||
Add a wire to the schematic using S-expression manipulation
|
||||
|
||||
@@ -44,36 +49,25 @@ class WireManager:
|
||||
"""
|
||||
try:
|
||||
# Read schematic
|
||||
with open(schematic_path, "r", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'r', encoding='utf-8') as f:
|
||||
sch_content = f.read()
|
||||
|
||||
sch_data = sexpdata.loads(sch_content)
|
||||
|
||||
# Break any existing wire that passes through a new endpoint (T-junction support)
|
||||
for pt in (start_point, end_point):
|
||||
splits = WireManager._break_wires_at_point(sch_data, pt)
|
||||
if splits:
|
||||
logger.info(f"Broke {splits} wire(s) at new wire endpoint {pt}")
|
||||
|
||||
# Create wire S-expression
|
||||
# Format: (wire (pts (xy x1 y1) (xy x2 y2)) (stroke (width N) (type default)) (uuid ...))
|
||||
wire_sexp = [
|
||||
Symbol("wire"),
|
||||
[
|
||||
Symbol("pts"),
|
||||
[Symbol("xy"), start_point[0], start_point[1]],
|
||||
[Symbol("xy"), end_point[0], end_point[1]],
|
||||
],
|
||||
[
|
||||
Symbol("stroke"),
|
||||
[Symbol("width"), stroke_width],
|
||||
[Symbol("type"), Symbol(stroke_type)],
|
||||
],
|
||||
[Symbol("uuid"), str(uuid.uuid4())],
|
||||
]
|
||||
wire_sexp = WireManager._make_wire_sexp(start_point, end_point, stroke_width, stroke_type)
|
||||
|
||||
# Find insertion point (before sheet_instances)
|
||||
sheet_instances_index = None
|
||||
for i, item in enumerate(sch_data):
|
||||
if (
|
||||
isinstance(item, list)
|
||||
and len(item) > 0
|
||||
and item[0] == Symbol("sheet_instances")
|
||||
):
|
||||
if isinstance(item, list) and len(item) > 0 and item[0] == _SYM_SHEET_INSTANCES:
|
||||
sheet_instances_index = i
|
||||
break
|
||||
|
||||
@@ -86,7 +80,7 @@ class WireManager:
|
||||
logger.info(f"Injected wire from {start_point} to {end_point}")
|
||||
|
||||
# Write back
|
||||
with open(schematic_path, "w", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'w', encoding='utf-8') as f:
|
||||
output = sexpdata.dumps(sch_data)
|
||||
f.write(output)
|
||||
|
||||
@@ -96,17 +90,12 @@ class WireManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding wire: {e}")
|
||||
import traceback
|
||||
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def add_polyline_wire(
|
||||
schematic_path: Path,
|
||||
points: List[List[float]],
|
||||
stroke_width: float = 0,
|
||||
stroke_type: str = "default",
|
||||
) -> bool:
|
||||
def add_polyline_wire(schematic_path: Path, points: List[List[float]],
|
||||
stroke_width: float = 0, stroke_type: str = 'default') -> bool:
|
||||
"""
|
||||
Add a multi-segment wire (polyline) to the schematic
|
||||
|
||||
@@ -125,36 +114,28 @@ class WireManager:
|
||||
return False
|
||||
|
||||
# Read schematic
|
||||
with open(schematic_path, "r", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'r', encoding='utf-8') as f:
|
||||
sch_content = f.read()
|
||||
|
||||
sch_data = sexpdata.loads(sch_content)
|
||||
|
||||
# Create pts list
|
||||
pts_list = [Symbol("pts")]
|
||||
for point in points:
|
||||
pts_list.append([Symbol("xy"), point[0], point[1]])
|
||||
# Break any existing wire at the outer endpoints of the new path
|
||||
for pt in (points[0], points[-1]):
|
||||
splits = WireManager._break_wires_at_point(sch_data, pt)
|
||||
if splits:
|
||||
logger.info(f"Broke {splits} wire(s) at new polyline endpoint {pt}")
|
||||
|
||||
# Create wire S-expression with multiple points
|
||||
wire_sexp = [
|
||||
Symbol("wire"),
|
||||
pts_list,
|
||||
[
|
||||
Symbol("stroke"),
|
||||
[Symbol("width"), stroke_width],
|
||||
[Symbol("type"), Symbol(stroke_type)],
|
||||
],
|
||||
[Symbol("uuid"), str(uuid.uuid4())],
|
||||
# KiCAD wire elements only support exactly 2 pts each.
|
||||
# Split N waypoints into N-1 individual wire segments.
|
||||
wire_sexps = [
|
||||
WireManager._make_wire_sexp(points[i], points[i + 1], stroke_width, stroke_type)
|
||||
for i in range(len(points) - 1)
|
||||
]
|
||||
|
||||
# Find insertion point
|
||||
sheet_instances_index = None
|
||||
for i, item in enumerate(sch_data):
|
||||
if (
|
||||
isinstance(item, list)
|
||||
and len(item) > 0
|
||||
and item[0] == Symbol("sheet_instances")
|
||||
):
|
||||
if isinstance(item, list) and len(item) > 0 and item[0] == _SYM_SHEET_INSTANCES:
|
||||
sheet_instances_index = i
|
||||
break
|
||||
|
||||
@@ -162,12 +143,13 @@ class WireManager:
|
||||
logger.error("No sheet_instances section found in schematic")
|
||||
return False
|
||||
|
||||
# Insert wire
|
||||
sch_data.insert(sheet_instances_index, wire_sexp)
|
||||
logger.info(f"Injected polyline wire with {len(points)} points")
|
||||
# Insert all segments (in reverse so order is preserved after inserts)
|
||||
for wire_sexp in reversed(wire_sexps):
|
||||
sch_data.insert(sheet_instances_index, wire_sexp)
|
||||
logger.info(f"Injected {len(wire_sexps)} wire segments for {len(points)}-point polyline")
|
||||
|
||||
# Write back
|
||||
with open(schematic_path, "w", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'w', encoding='utf-8') as f:
|
||||
output = sexpdata.dumps(sch_data)
|
||||
f.write(output)
|
||||
|
||||
@@ -177,18 +159,12 @@ class WireManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding polyline wire: {e}")
|
||||
import traceback
|
||||
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def add_label(
|
||||
schematic_path: Path,
|
||||
text: str,
|
||||
position: List[float],
|
||||
label_type: str = "label",
|
||||
orientation: int = 0,
|
||||
) -> bool:
|
||||
def add_label(schematic_path: Path, text: str, position: List[float],
|
||||
label_type: str = 'label', orientation: int = 0) -> bool:
|
||||
"""
|
||||
Add a net label to the schematic
|
||||
|
||||
@@ -204,7 +180,7 @@ class WireManager:
|
||||
"""
|
||||
try:
|
||||
# Read schematic
|
||||
with open(schematic_path, "r", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'r', encoding='utf-8') as f:
|
||||
sch_content = f.read()
|
||||
|
||||
sch_data = sexpdata.loads(sch_content)
|
||||
@@ -214,24 +190,19 @@ class WireManager:
|
||||
label_sexp = [
|
||||
Symbol(label_type),
|
||||
text,
|
||||
[Symbol("at"), position[0], position[1], orientation],
|
||||
[Symbol("fields_autoplaced"), Symbol("yes")],
|
||||
[
|
||||
Symbol("effects"),
|
||||
[Symbol("font"), [Symbol("size"), 1.27, 1.27]],
|
||||
[Symbol("justify"), Symbol("left"), Symbol("bottom")],
|
||||
[Symbol('at'), position[0], position[1], orientation],
|
||||
[Symbol('fields_autoplaced'), Symbol('yes')],
|
||||
[Symbol('effects'),
|
||||
[Symbol('font'), [Symbol('size'), 1.27, 1.27]],
|
||||
[Symbol('justify'), Symbol('left'), Symbol('bottom')]
|
||||
],
|
||||
[Symbol("uuid"), str(uuid.uuid4())],
|
||||
[Symbol('uuid'), str(uuid.uuid4())]
|
||||
]
|
||||
|
||||
# Find insertion point
|
||||
sheet_instances_index = None
|
||||
for i, item in enumerate(sch_data):
|
||||
if (
|
||||
isinstance(item, list)
|
||||
and len(item) > 0
|
||||
and item[0] == Symbol("sheet_instances")
|
||||
):
|
||||
if isinstance(item, list) and len(item) > 0 and item[0] == _SYM_SHEET_INSTANCES:
|
||||
sheet_instances_index = i
|
||||
break
|
||||
|
||||
@@ -244,7 +215,7 @@ class WireManager:
|
||||
logger.info(f"Injected label '{text}' at {position}")
|
||||
|
||||
# Write back
|
||||
with open(schematic_path, "w", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'w', encoding='utf-8') as f:
|
||||
output = sexpdata.dumps(sch_data)
|
||||
f.write(output)
|
||||
|
||||
@@ -254,16 +225,115 @@ class WireManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding label: {e}")
|
||||
import traceback
|
||||
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def add_junction(
|
||||
schematic_path: Path, position: List[float], diameter: float = 0
|
||||
) -> bool:
|
||||
def _parse_wire(wire_item) -> Optional[Tuple[Tuple[float, float], Tuple[float, float], float, str]]:
|
||||
"""
|
||||
Add a junction (connection dot) to the schematic
|
||||
Parse a wire S-expression item in a single pass.
|
||||
Returns ((x1,y1), (x2,y2), stroke_width, stroke_type), or None if not a valid wire.
|
||||
"""
|
||||
if not (isinstance(wire_item, list) and len(wire_item) >= 2
|
||||
and wire_item[0] == _SYM_WIRE):
|
||||
return None
|
||||
start = end = None
|
||||
stroke_width: float = 0
|
||||
stroke_type: str = 'default'
|
||||
for part in wire_item[1:]:
|
||||
if not isinstance(part, list) or not part:
|
||||
continue
|
||||
tag = part[0]
|
||||
if tag == _SYM_PTS:
|
||||
found: List[Tuple[float, float]] = []
|
||||
for p in part[1:]:
|
||||
if isinstance(p, list) and len(p) >= 3 and p[0] == _SYM_XY:
|
||||
found.append((float(p[1]), float(p[2])))
|
||||
if len(found) == 2:
|
||||
break
|
||||
if len(found) == 2:
|
||||
start, end = found[0], found[1]
|
||||
elif tag == _SYM_STROKE:
|
||||
for sp in part[1:]:
|
||||
if isinstance(sp, list) and len(sp) >= 2:
|
||||
if sp[0] == _SYM_WIDTH:
|
||||
stroke_width = sp[1]
|
||||
elif sp[0] == _SYM_TYPE:
|
||||
stroke_type = str(sp[1])
|
||||
if start is not None and end is not None:
|
||||
return start, end, stroke_width, stroke_type
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _point_strictly_on_wire(px: float, py: float,
|
||||
x1: float, y1: float,
|
||||
x2: float, y2: float,
|
||||
eps: float = 1e-6) -> bool:
|
||||
"""
|
||||
Return True if (px, py) lies strictly between (x1,y1) and (x2,y2)
|
||||
on a horizontal or vertical wire segment (not at either endpoint).
|
||||
"""
|
||||
if abs(y1 - y2) < eps: # horizontal wire
|
||||
if abs(py - y1) > eps:
|
||||
return False
|
||||
lo, hi = min(x1, x2), max(x1, x2)
|
||||
return lo + eps < px < hi - eps
|
||||
if abs(x1 - x2) < eps: # vertical wire
|
||||
if abs(px - x1) > eps:
|
||||
return False
|
||||
lo, hi = min(y1, y2), max(y1, y2)
|
||||
return lo + eps < py < hi - eps
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _make_wire_sexp(start: List[float], end: List[float],
|
||||
stroke_width: float = 0, stroke_type: str = 'default') -> list:
|
||||
return [
|
||||
_SYM_WIRE,
|
||||
[_SYM_PTS,
|
||||
[_SYM_XY, start[0], start[1]],
|
||||
[_SYM_XY, end[0], end[1]]],
|
||||
[_SYM_STROKE,
|
||||
[_SYM_WIDTH, stroke_width],
|
||||
[_SYM_TYPE, Symbol(stroke_type)]],
|
||||
[_SYM_UUID, str(uuid.uuid4())]
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _break_wires_at_point(sch_data: list, position: List[float]) -> int:
|
||||
"""
|
||||
Split any wire segment that passes through *position* as a strict
|
||||
midpoint (i.e. position is not an existing endpoint). Mirrors
|
||||
KiCAD's SCH_LINE_WIRE_BUS_TOOL::BreakSegments behaviour.
|
||||
|
||||
Returns the number of wires split.
|
||||
"""
|
||||
px, py = float(position[0]), float(position[1])
|
||||
splits = 0
|
||||
i = 0
|
||||
while i < len(sch_data):
|
||||
parsed = WireManager._parse_wire(sch_data[i])
|
||||
if parsed is not None:
|
||||
(x1, y1), (x2, y2), stroke_width, stroke_type = parsed
|
||||
if WireManager._point_strictly_on_wire(px, py, x1, y1, x2, y2):
|
||||
seg_a = WireManager._make_wire_sexp([x1, y1], [px, py], stroke_width, stroke_type)
|
||||
seg_b = WireManager._make_wire_sexp([px, py], [x2, y2], stroke_width, stroke_type)
|
||||
sch_data[i:i + 1] = [seg_a, seg_b]
|
||||
logger.info(f"Split wire ({x1},{y1})->({x2},{y2}) at ({px},{py})")
|
||||
splits += 1
|
||||
i += 2 # skip the two new segments
|
||||
continue
|
||||
i += 1
|
||||
return splits
|
||||
|
||||
@staticmethod
|
||||
def add_junction(schematic_path: Path, position: List[float], diameter: float = 0) -> bool:
|
||||
"""
|
||||
Add a junction (connection dot) to the schematic.
|
||||
|
||||
Mirrors KiCAD's AddJunction behaviour: any wire whose interior passes
|
||||
through *position* is split into two segments at that point so that
|
||||
the BFS-based get_wire_connections tool can traverse the T/X branch.
|
||||
|
||||
Args:
|
||||
schematic_path: Path to .kicad_sch file
|
||||
@@ -275,29 +345,31 @@ class WireManager:
|
||||
"""
|
||||
try:
|
||||
# Read schematic
|
||||
with open(schematic_path, "r", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'r', encoding='utf-8') as f:
|
||||
sch_content = f.read()
|
||||
|
||||
sch_data = sexpdata.loads(sch_content)
|
||||
|
||||
# Split any wire that passes through the junction as a midpoint
|
||||
# (mirrors KiCAD's AddJunction / BreakSegments behaviour)
|
||||
splits = WireManager._break_wires_at_point(sch_data, position)
|
||||
if splits:
|
||||
logger.info(f"Broke {splits} wire(s) at junction position {position}")
|
||||
|
||||
# Create junction S-expression
|
||||
# Format: (junction (at x y) (diameter 0) (color 0 0 0 0) (uuid ...))
|
||||
junction_sexp = [
|
||||
Symbol("junction"),
|
||||
[Symbol("at"), position[0], position[1]],
|
||||
[Symbol("diameter"), diameter],
|
||||
[Symbol("color"), 0, 0, 0, 0],
|
||||
[Symbol("uuid"), str(uuid.uuid4())],
|
||||
Symbol('junction'),
|
||||
[Symbol('at'), position[0], position[1]],
|
||||
[Symbol('diameter'), diameter],
|
||||
[Symbol('color'), 0, 0, 0, 0],
|
||||
[Symbol('uuid'), str(uuid.uuid4())]
|
||||
]
|
||||
|
||||
# Find insertion point
|
||||
sheet_instances_index = None
|
||||
for i, item in enumerate(sch_data):
|
||||
if (
|
||||
isinstance(item, list)
|
||||
and len(item) > 0
|
||||
and item[0] == Symbol("sheet_instances")
|
||||
):
|
||||
if isinstance(item, list) and len(item) > 0 and item[0] == _SYM_SHEET_INSTANCES:
|
||||
sheet_instances_index = i
|
||||
break
|
||||
|
||||
@@ -310,7 +382,7 @@ class WireManager:
|
||||
logger.info(f"Injected junction at {position}")
|
||||
|
||||
# Write back
|
||||
with open(schematic_path, "w", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'w', encoding='utf-8') as f:
|
||||
output = sexpdata.dumps(sch_data)
|
||||
f.write(output)
|
||||
|
||||
@@ -320,7 +392,6 @@ class WireManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding junction: {e}")
|
||||
import traceback
|
||||
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
@@ -338,7 +409,7 @@ class WireManager:
|
||||
"""
|
||||
try:
|
||||
# Read schematic
|
||||
with open(schematic_path, "r", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'r', encoding='utf-8') as f:
|
||||
sch_content = f.read()
|
||||
|
||||
sch_data = sexpdata.loads(sch_content)
|
||||
@@ -346,19 +417,15 @@ class WireManager:
|
||||
# Create no_connect S-expression
|
||||
# Format: (no_connect (at x y) (uuid ...))
|
||||
no_connect_sexp = [
|
||||
Symbol("no_connect"),
|
||||
[Symbol("at"), position[0], position[1]],
|
||||
[Symbol("uuid"), str(uuid.uuid4())],
|
||||
Symbol('no_connect'),
|
||||
[Symbol('at'), position[0], position[1]],
|
||||
[Symbol('uuid'), str(uuid.uuid4())]
|
||||
]
|
||||
|
||||
# Find insertion point
|
||||
sheet_instances_index = None
|
||||
for i, item in enumerate(sch_data):
|
||||
if (
|
||||
isinstance(item, list)
|
||||
and len(item) > 0
|
||||
and item[0] == Symbol("sheet_instances")
|
||||
):
|
||||
if isinstance(item, list) and len(item) > 0 and item[0] == _SYM_SHEET_INSTANCES:
|
||||
sheet_instances_index = i
|
||||
break
|
||||
|
||||
@@ -371,7 +438,7 @@ class WireManager:
|
||||
logger.info(f"Injected no-connect at {position}")
|
||||
|
||||
# Write back
|
||||
with open(schematic_path, "w", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'w', encoding='utf-8') as f:
|
||||
output = sexpdata.dumps(sch_data)
|
||||
f.write(output)
|
||||
|
||||
@@ -381,17 +448,12 @@ class WireManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding no-connect: {e}")
|
||||
import traceback
|
||||
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def delete_wire(
|
||||
schematic_path: Path,
|
||||
start_point: List[float],
|
||||
end_point: List[float],
|
||||
tolerance: float = 0.5,
|
||||
) -> bool:
|
||||
def delete_wire(schematic_path: Path, start_point: List[float], end_point: List[float],
|
||||
tolerance: float = 0.5) -> bool:
|
||||
"""
|
||||
Delete a wire from the schematic matching given start/end coordinates.
|
||||
|
||||
@@ -405,7 +467,7 @@ class WireManager:
|
||||
True if a wire was found and removed, False otherwise
|
||||
"""
|
||||
try:
|
||||
with open(schematic_path, "r", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'r', encoding='utf-8') as f:
|
||||
sch_content = f.read()
|
||||
|
||||
sch_data = sexpdata.loads(sch_content)
|
||||
@@ -414,54 +476,34 @@ class WireManager:
|
||||
ex, ey = end_point
|
||||
|
||||
for i, item in enumerate(sch_data):
|
||||
if not (
|
||||
isinstance(item, list)
|
||||
and len(item) > 0
|
||||
and item[0] == Symbol("wire")
|
||||
):
|
||||
if not (isinstance(item, list) and len(item) > 0 and item[0] == Symbol('wire')):
|
||||
continue
|
||||
|
||||
# Extract pts from the wire s-expression
|
||||
pts_list = None
|
||||
for part in item[1:]:
|
||||
if (
|
||||
isinstance(part, list)
|
||||
and len(part) > 0
|
||||
and part[0] == Symbol("pts")
|
||||
):
|
||||
if isinstance(part, list) and len(part) > 0 and part[0] == Symbol('pts'):
|
||||
pts_list = part
|
||||
break
|
||||
|
||||
if pts_list is None:
|
||||
continue
|
||||
|
||||
xy_points = [
|
||||
p
|
||||
for p in pts_list[1:]
|
||||
if isinstance(p, list) and len(p) >= 3 and p[0] == Symbol("xy")
|
||||
]
|
||||
xy_points = [p for p in pts_list[1:] if isinstance(p, list) and len(p) >= 3 and p[0] == Symbol('xy')]
|
||||
if len(xy_points) < 2:
|
||||
continue
|
||||
|
||||
x1, y1 = float(xy_points[0][1]), float(xy_points[0][2])
|
||||
x2, y2 = float(xy_points[-1][1]), float(xy_points[-1][2])
|
||||
|
||||
match_fwd = (
|
||||
abs(x1 - sx) < tolerance
|
||||
and abs(y1 - sy) < tolerance
|
||||
and abs(x2 - ex) < tolerance
|
||||
and abs(y2 - ey) < tolerance
|
||||
)
|
||||
match_rev = (
|
||||
abs(x1 - ex) < tolerance
|
||||
and abs(y1 - ey) < tolerance
|
||||
and abs(x2 - sx) < tolerance
|
||||
and abs(y2 - sy) < tolerance
|
||||
)
|
||||
match_fwd = (abs(x1 - sx) < tolerance and abs(y1 - sy) < tolerance and
|
||||
abs(x2 - ex) < tolerance and abs(y2 - ey) < tolerance)
|
||||
match_rev = (abs(x1 - ex) < tolerance and abs(y1 - ey) < tolerance and
|
||||
abs(x2 - sx) < tolerance and abs(y2 - sy) < tolerance)
|
||||
|
||||
if match_fwd or match_rev:
|
||||
del sch_data[i]
|
||||
with open(schematic_path, "w", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'w', encoding='utf-8') as f:
|
||||
f.write(sexpdata.dumps(sch_data))
|
||||
logger.info(f"Deleted wire from {start_point} to {end_point}")
|
||||
return True
|
||||
@@ -472,17 +514,13 @@ class WireManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting wire: {e}")
|
||||
import traceback
|
||||
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def delete_label(
|
||||
schematic_path: Path,
|
||||
net_name: str,
|
||||
position: Optional[List[float]] = None,
|
||||
tolerance: float = 0.5,
|
||||
) -> bool:
|
||||
def delete_label(schematic_path: Path, net_name: str,
|
||||
position: Optional[List[float]] = None,
|
||||
tolerance: float = 0.5) -> bool:
|
||||
"""
|
||||
Delete a net label from the schematic by name (and optionally position).
|
||||
|
||||
@@ -496,17 +534,14 @@ class WireManager:
|
||||
True if a label was found and removed, False otherwise
|
||||
"""
|
||||
try:
|
||||
with open(schematic_path, "r", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'r', encoding='utf-8') as f:
|
||||
sch_content = f.read()
|
||||
|
||||
sch_data = sexpdata.loads(sch_content)
|
||||
|
||||
for i, item in enumerate(sch_data):
|
||||
if not (
|
||||
isinstance(item, list)
|
||||
and len(item) > 0
|
||||
and item[0] == Symbol("label")
|
||||
):
|
||||
if not (isinstance(item, list) and len(item) > 0
|
||||
and item[0] == Symbol('label')):
|
||||
continue
|
||||
|
||||
# Second element is the label text
|
||||
@@ -516,26 +551,19 @@ class WireManager:
|
||||
if position is not None:
|
||||
# Find (at x y ...) sub-expression and check coordinates
|
||||
at_entry = next(
|
||||
(
|
||||
p
|
||||
for p in item[1:]
|
||||
if isinstance(p, list)
|
||||
and len(p) >= 3
|
||||
and p[0] == Symbol("at")
|
||||
),
|
||||
(p for p in item[1:] if isinstance(p, list)
|
||||
and len(p) >= 3 and p[0] == Symbol('at')),
|
||||
None,
|
||||
)
|
||||
if at_entry is None:
|
||||
continue
|
||||
lx, ly = float(at_entry[1]), float(at_entry[2])
|
||||
if not (
|
||||
abs(lx - position[0]) < tolerance
|
||||
and abs(ly - position[1]) < tolerance
|
||||
):
|
||||
if not (abs(lx - position[0]) < tolerance
|
||||
and abs(ly - position[1]) < tolerance):
|
||||
continue
|
||||
|
||||
del sch_data[i]
|
||||
with open(schematic_path, "w", encoding="utf-8") as f:
|
||||
with open(schematic_path, 'w', encoding='utf-8') as f:
|
||||
f.write(sexpdata.dumps(sch_data))
|
||||
logger.info(f"Deleted label '{net_name}'")
|
||||
return True
|
||||
@@ -546,14 +574,12 @@ class WireManager:
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting label: {e}")
|
||||
import traceback
|
||||
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def create_orthogonal_path(
|
||||
start: List[float], end: List[float], prefer_horizontal_first: bool = True
|
||||
) -> List[List[float]]:
|
||||
def create_orthogonal_path(start: List[float], end: List[float],
|
||||
prefer_horizontal_first: bool = True) -> List[List[float]]:
|
||||
"""
|
||||
Create an orthogonal (right-angle) path between two points
|
||||
|
||||
@@ -582,11 +608,10 @@ class WireManager:
|
||||
return [start, corner, end]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if __name__ == '__main__':
|
||||
# Test wire creation
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, "/home/chris/MCP/KiCAD-MCP-Server/python")
|
||||
sys.path.insert(0, '/home/chris/MCP/KiCAD-MCP-Server/python')
|
||||
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
@@ -596,10 +621,8 @@ if __name__ == "__main__":
|
||||
print("=" * 80)
|
||||
|
||||
# Create test schematic (cross-platform temp directory)
|
||||
test_path = Path(tempfile.gettempdir()) / "test_wire_manager.kicad_sch"
|
||||
template_path = Path(
|
||||
"/home/chris/MCP/KiCAD-MCP-Server/python/templates/empty.kicad_sch"
|
||||
)
|
||||
test_path = Path(tempfile.gettempdir()) / 'test_wire_manager.kicad_sch'
|
||||
template_path = Path('/home/chris/MCP/KiCAD-MCP-Server/python/templates/empty.kicad_sch')
|
||||
|
||||
shutil.copy(template_path, test_path)
|
||||
print(f"\n✓ Created test schematic: {test_path}")
|
||||
@@ -635,9 +658,8 @@ if __name__ == "__main__":
|
||||
print("\n[Verification] Loading with kicad-skip...")
|
||||
try:
|
||||
from skip import Schematic
|
||||
|
||||
sch = Schematic(str(test_path))
|
||||
wire_count = len(list(sch.wire)) if hasattr(sch, "wire") else 0
|
||||
wire_count = len(list(sch.wire)) if hasattr(sch, 'wire') else 0
|
||||
print(f" ✓ Loaded successfully")
|
||||
print(f" ✓ Wire count: {wire_count}")
|
||||
except Exception as e:
|
||||
|
||||
Reference in New Issue
Block a user