feat: Complete Phase 1 & 2 - Schematic workflow fix and JLCPCB integration
Phase 1: Schematic Workflow Fix (Issue #26) - Fixed broken schematic workflow using template-based symbol cloning - Updated create_project to generate both PCB and schematic files - Rewrote add_schematic_component to use kicad-skip clone() API - Added template schematics with cloneable R, C, LED symbols - All schematic tests now passing Phase 2: JLCPCB Integration Complete - Integrated JLCSearch public API (no authentication required) - Access to ~100k JLCPCB parts with real-time stock and pricing - Implemented parametric search for resistors, capacitors, components - Added package-to-footprint mapping for KiCad integration - Cost optimization with Basic vs Extended library classification - Alternative part suggestions with price comparison New Components: - python/commands/jlcsearch.py - JLCSearch API client - python/templates/ - Template schematics for symbol cloning - docs/JLCPCB_INTEGRATION.md - Comprehensive API documentation - docs/SCHEMATIC_WORKFLOW_FIX.md - Phase 1 technical details - CHANGELOG.md - Consolidated unified changelog - PHASE_2_COMPLETE.md - Phase 2 implementation summary MCP Tools Available: - download_jlcpcb_database - Download full parts catalog - search_jlcpcb_parts - Parametric search with filters - get_jlcpcb_part - Part details + footprint suggestions - get_jlcpcb_database_stats - Database statistics - suggest_jlcpcb_alternatives - Find similar/cheaper parts Technical Improvements: - SQLite database with FTS5 full-text search - HMAC-SHA256 authentication support (official JLCPCB API) - Improved .gitignore to exclude credentials and databases - Template-based schematic creation workflow Testing: - All integration tests passing - Database operations validated - Live API connectivity confirmed - Schematic workflow end-to-end verified Credits: - JLCSearch API: @tscircuit - Local JLCPCB search: @l3wi Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
KiCAD MCP Bot
@@ -1,6 +1,6 @@
|
||||
from skip import Schematic
|
||||
# Symbol class might not be directly importable in the current version
|
||||
import os
|
||||
import uuid
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -8,42 +8,78 @@ logger = logging.getLogger(__name__)
|
||||
class ComponentManager:
|
||||
"""Manage components in a schematic"""
|
||||
|
||||
# Template symbol references mapping component type to template reference
|
||||
TEMPLATE_MAP = {
|
||||
'R': '_TEMPLATE_R',
|
||||
'C': '_TEMPLATE_C',
|
||||
'D': '_TEMPLATE_D',
|
||||
'LED': '_TEMPLATE_D',
|
||||
# Add more mappings as needed
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def add_component(schematic: Schematic, component_def: dict):
|
||||
"""Add a component to the schematic"""
|
||||
"""Add a component to the schematic by cloning from template"""
|
||||
try:
|
||||
logger.info(f"Adding component: lib={component_def.get('library')}, name={component_def.get('type')}, ref={component_def.get('reference')}")
|
||||
logger.info(f"Adding component: type={component_def.get('type')}, ref={component_def.get('reference')}")
|
||||
logger.debug(f"Full component_def: {component_def}")
|
||||
|
||||
# Create a new symbol
|
||||
symbol = schematic.add_symbol(
|
||||
lib=component_def.get('library', 'Device'),
|
||||
name=component_def.get('type', 'R'), # Default to Resistor symbol 'R'
|
||||
reference=component_def.get('reference', 'R?'),
|
||||
at=[component_def.get('x', 0), component_def.get('y', 0)],
|
||||
unit=component_def.get('unit', 1),
|
||||
rotation=component_def.get('rotation', 0)
|
||||
)
|
||||
# Get component type and determine template
|
||||
comp_type = component_def.get('type', 'R')
|
||||
template_ref = ComponentManager.TEMPLATE_MAP.get(comp_type, '_TEMPLATE_R')
|
||||
|
||||
# Set properties
|
||||
# Check if schematic has template symbols
|
||||
if not hasattr(schematic.symbol, template_ref):
|
||||
logger.error(f"Template symbol {template_ref} not found in schematic. Available symbols: {[str(s.property.Reference.value) for s in schematic.symbol]}")
|
||||
raise ValueError(f"Template symbol {template_ref} not found. The schematic must be created from template_with_symbols.kicad_sch")
|
||||
|
||||
# Get template symbol and clone it
|
||||
template_symbol = getattr(schematic.symbol, template_ref)
|
||||
new_symbol = template_symbol.clone()
|
||||
logger.debug(f"Cloned template symbol {template_ref}")
|
||||
|
||||
# Set reference
|
||||
reference = component_def.get('reference', 'R?')
|
||||
new_symbol.property.Reference.value = reference
|
||||
logger.debug(f"Set reference to {reference}")
|
||||
|
||||
# Set value
|
||||
if 'value' in component_def:
|
||||
symbol.property.Value.value = component_def['value']
|
||||
new_symbol.property.Value.value = component_def['value']
|
||||
logger.debug(f"Set value to {component_def['value']}")
|
||||
|
||||
# Set footprint
|
||||
if 'footprint' in component_def:
|
||||
symbol.property.Footprint.value = component_def['footprint']
|
||||
new_symbol.property.Footprint.value = component_def['footprint']
|
||||
logger.debug(f"Set footprint to {component_def['footprint']}")
|
||||
|
||||
# Set datasheet
|
||||
if 'datasheet' in component_def:
|
||||
symbol.property.Datasheet.value = component_def['datasheet']
|
||||
new_symbol.property.Datasheet.value = component_def['datasheet']
|
||||
|
||||
# Add additional properties
|
||||
for key, value in component_def.get('properties', {}).items():
|
||||
# Avoid overwriting standard properties unless explicitly intended
|
||||
if key not in ['Reference', 'Value', 'Footprint', 'Datasheet']:
|
||||
symbol.property.append(key, value)
|
||||
# Set position
|
||||
x = component_def.get('x', 0)
|
||||
y = component_def.get('y', 0)
|
||||
rotation = component_def.get('rotation', 0)
|
||||
new_symbol.at.value = [x, y, rotation]
|
||||
logger.debug(f"Set position to ({x}, {y}, {rotation})")
|
||||
|
||||
logger.info(f"Successfully added component {symbol.reference} ({symbol.name}) to schematic.")
|
||||
return symbol
|
||||
# Set BOM and board flags
|
||||
new_symbol.in_bom.value = component_def.get('in_bom', True)
|
||||
new_symbol.on_board.value = component_def.get('on_board', True)
|
||||
new_symbol.dnp.value = component_def.get('dnp', False)
|
||||
|
||||
# Generate new UUID
|
||||
new_symbol.uuid.value = str(uuid.uuid4())
|
||||
|
||||
# Append to schematic
|
||||
schematic.symbol.append(new_symbol)
|
||||
logger.info(f"Successfully added component {reference} to schematic")
|
||||
|
||||
return new_symbol
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding component: {e}", exc_info=True)
|
||||
return None
|
||||
raise
|
||||
|
||||
@staticmethod
|
||||
def remove_component(schematic: Schematic, component_ref: str):
|
||||
|
||||
@@ -9,6 +9,12 @@ import os
|
||||
import logging
|
||||
import requests
|
||||
import time
|
||||
import hmac
|
||||
import hashlib
|
||||
import secrets
|
||||
import string
|
||||
import base64
|
||||
import json
|
||||
from typing import Optional, Dict, List, Callable
|
||||
from pathlib import Path
|
||||
|
||||
@@ -19,73 +25,100 @@ class JLCPCBClient:
|
||||
"""
|
||||
Client for JLCPCB API
|
||||
|
||||
Handles authentication and fetching the complete parts library
|
||||
from JLCPCB's external API.
|
||||
Handles HMAC-SHA256 signature-based authentication and fetching
|
||||
the complete parts library from JLCPCB's external API.
|
||||
"""
|
||||
|
||||
BASE_URL = "https://jlcpcb.com/external"
|
||||
|
||||
def __init__(self, api_key: Optional[str] = None, api_secret: Optional[str] = None):
|
||||
def __init__(self, app_id: Optional[str] = None, access_key: Optional[str] = None, secret_key: Optional[str] = None):
|
||||
"""
|
||||
Initialize JLCPCB API client
|
||||
|
||||
Args:
|
||||
api_key: JLCPCB API key (or reads from JLCPCB_API_KEY env var)
|
||||
api_secret: JLCPCB API secret (or reads from JLCPCB_API_SECRET env var)
|
||||
app_id: JLCPCB App ID (or reads from JLCPCB_APP_ID env var)
|
||||
access_key: JLCPCB Access Key (or reads from JLCPCB_API_KEY env var)
|
||||
secret_key: JLCPCB Secret Key (or reads from JLCPCB_API_SECRET env var)
|
||||
"""
|
||||
self.api_key = api_key or os.getenv('JLCPCB_API_KEY')
|
||||
self.api_secret = api_secret or os.getenv('JLCPCB_API_SECRET')
|
||||
self.token = None
|
||||
self.token_expiry = 0
|
||||
self.app_id = app_id or os.getenv('JLCPCB_APP_ID')
|
||||
self.access_key = access_key or os.getenv('JLCPCB_API_KEY')
|
||||
self.secret_key = secret_key or os.getenv('JLCPCB_API_SECRET')
|
||||
|
||||
if not self.api_key or not self.api_secret:
|
||||
logger.warning("JLCPCB API credentials not found. Set JLCPCB_API_KEY and JLCPCB_API_SECRET environment variables.")
|
||||
if not self.app_id or not self.access_key or not self.secret_key:
|
||||
logger.warning("JLCPCB API credentials not found. Set JLCPCB_APP_ID, JLCPCB_API_KEY, and JLCPCB_API_SECRET environment variables.")
|
||||
|
||||
def authenticate(self) -> str:
|
||||
@staticmethod
|
||||
def _generate_nonce() -> str:
|
||||
"""Generate a 32-character random nonce"""
|
||||
chars = string.ascii_letters + string.digits
|
||||
return ''.join(secrets.choice(chars) for _ in range(32))
|
||||
|
||||
def _build_signature_string(self, method: str, path: str, timestamp: int, nonce: str, body: str) -> str:
|
||||
"""
|
||||
Get authentication token from JLCPCB API
|
||||
Build the signature string according to JLCPCB spec
|
||||
|
||||
Format:
|
||||
<HTTP Method>\n
|
||||
<Request Path>\n
|
||||
<Timestamp>\n
|
||||
<Nonce>\n
|
||||
<Request Body>\n
|
||||
|
||||
Args:
|
||||
method: HTTP method (GET, POST, etc.)
|
||||
path: Request path with query params
|
||||
timestamp: Unix timestamp in seconds
|
||||
nonce: 32-character random string
|
||||
body: Request body (empty string for GET)
|
||||
|
||||
Returns:
|
||||
Authentication token
|
||||
|
||||
Raises:
|
||||
Exception if authentication fails
|
||||
Signature string
|
||||
"""
|
||||
if not self.api_key or not self.api_secret:
|
||||
raise Exception("JLCPCB API credentials not configured. Please set JLCPCB_API_KEY and JLCPCB_API_SECRET environment variables.")
|
||||
return f"{method}\n{path}\n{timestamp}\n{nonce}\n{body}\n"
|
||||
|
||||
# Check if we have a valid token
|
||||
if self.token and time.time() < self.token_expiry:
|
||||
return self.token
|
||||
def _sign(self, signature_string: str) -> str:
|
||||
"""
|
||||
Sign the signature string with HMAC-SHA256
|
||||
|
||||
logger.info("Authenticating with JLCPCB API...")
|
||||
Args:
|
||||
signature_string: The string to sign
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{self.BASE_URL}/genToken",
|
||||
json={
|
||||
"appKey": self.api_key,
|
||||
"appSecret": self.api_secret
|
||||
},
|
||||
timeout=30
|
||||
)
|
||||
Returns:
|
||||
Base64-encoded signature
|
||||
"""
|
||||
signature_bytes = hmac.new(
|
||||
self.secret_key.encode('utf-8'),
|
||||
signature_string.encode('utf-8'),
|
||||
hashlib.sha256
|
||||
).digest()
|
||||
return base64.b64encode(signature_bytes).decode('utf-8')
|
||||
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
def _get_auth_header(self, method: str, path: str, body: str = "") -> str:
|
||||
"""
|
||||
Generate the Authorization header for JLCPCB API requests
|
||||
|
||||
if data.get('code') != 200:
|
||||
raise Exception(f"Authentication failed: {data.get('msg', 'Unknown error')}")
|
||||
Args:
|
||||
method: HTTP method (GET, POST, etc.)
|
||||
path: Request path with query params
|
||||
body: Request body JSON string (empty for GET)
|
||||
|
||||
self.token = data['data']['token']
|
||||
# Tokens typically expire after 2 hours, we'll refresh after 1.5 hours to be safe
|
||||
self.token_expiry = time.time() + (90 * 60)
|
||||
Returns:
|
||||
Authorization header value
|
||||
"""
|
||||
if not self.app_id or not self.access_key or not self.secret_key:
|
||||
raise Exception("JLCPCB API credentials not configured. Please set JLCPCB_APP_ID, JLCPCB_API_KEY, and JLCPCB_API_SECRET environment variables.")
|
||||
|
||||
logger.info("Successfully authenticated with JLCPCB API")
|
||||
return self.token
|
||||
nonce = self._generate_nonce()
|
||||
timestamp = int(time.time())
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"Failed to authenticate with JLCPCB API: {e}")
|
||||
raise Exception(f"JLCPCB API authentication failed: {e}")
|
||||
signature_string = self._build_signature_string(method, path, timestamp, nonce, body)
|
||||
signature = self._sign(signature_string)
|
||||
|
||||
logger.debug(f"Signature string:\n{repr(signature_string)}")
|
||||
logger.debug(f"Signature: {signature}")
|
||||
logger.debug(f"Auth header: JOP appid=\"{self.app_id}\",accesskey=\"{self.access_key}\",nonce=\"{nonce}\",timestamp=\"{timestamp}\",signature=\"{signature}\"")
|
||||
|
||||
return f'JOP appid="{self.app_id}",accesskey="{self.access_key}",nonce="{nonce}",timestamp="{timestamp}",signature="{signature}"'
|
||||
|
||||
def fetch_parts_page(self, last_key: Optional[str] = None) -> Dict:
|
||||
"""
|
||||
@@ -97,29 +130,41 @@ class JLCPCBClient:
|
||||
Returns:
|
||||
Response dict with parts data and pagination info
|
||||
"""
|
||||
token = self.authenticate()
|
||||
|
||||
headers = {
|
||||
"externalApiToken": token
|
||||
}
|
||||
path = "/component/getComponentInfos"
|
||||
|
||||
payload = {}
|
||||
if last_key:
|
||||
payload["lastKey"] = last_key
|
||||
|
||||
# Convert payload to JSON string for signing
|
||||
# For POST requests, we always send JSON, even if empty dict
|
||||
body_str = json.dumps(payload, separators=(',', ':'))
|
||||
|
||||
# Generate authorization header
|
||||
auth_header = self._get_auth_header("POST", path, body_str)
|
||||
|
||||
headers = {
|
||||
"Authorization": auth_header,
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{self.BASE_URL}/component/getComponentInfos",
|
||||
f"{self.BASE_URL}{path}",
|
||||
headers=headers,
|
||||
json=payload,
|
||||
timeout=60
|
||||
)
|
||||
|
||||
logger.debug(f"Response status: {response.status_code}")
|
||||
logger.debug(f"Response headers: {response.headers}")
|
||||
logger.debug(f"Response text: {response.text}")
|
||||
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
if data.get('code') != 200:
|
||||
raise Exception(f"API request failed: {data.get('msg', 'Unknown error')}")
|
||||
raise Exception(f"API request failed (code {data.get('code')}): {data.get('msg', 'Unknown error')} - Full response: {data}")
|
||||
|
||||
return data['data']
|
||||
|
||||
@@ -200,20 +245,22 @@ class JLCPCBClient:
|
||||
return None
|
||||
|
||||
|
||||
def test_jlcpcb_connection(api_key: Optional[str] = None, api_secret: Optional[str] = None) -> bool:
|
||||
def test_jlcpcb_connection(app_id: Optional[str] = None, access_key: Optional[str] = None, secret_key: Optional[str] = None) -> bool:
|
||||
"""
|
||||
Test JLCPCB API connection
|
||||
|
||||
Args:
|
||||
api_key: Optional API key (uses env var if not provided)
|
||||
api_secret: Optional API secret (uses env var if not provided)
|
||||
app_id: Optional App ID (uses env var if not provided)
|
||||
access_key: Optional Access Key (uses env var if not provided)
|
||||
secret_key: Optional Secret Key (uses env var if not provided)
|
||||
|
||||
Returns:
|
||||
True if connection successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
client = JLCPCBClient(api_key, api_secret)
|
||||
token = client.authenticate()
|
||||
client = JLCPCBClient(app_id, access_key, secret_key)
|
||||
# Test by fetching first page
|
||||
data = client.fetch_parts_page()
|
||||
logger.info("JLCPCB API connection test successful")
|
||||
return True
|
||||
except Exception as e:
|
||||
|
||||
@@ -162,6 +162,91 @@ class JLCPCBPartsManager:
|
||||
else:
|
||||
return 'Extended' # Default to Extended
|
||||
|
||||
def import_jlcsearch_parts(self, parts: List[Dict], progress_callback=None):
|
||||
"""
|
||||
Import parts into database from JLCSearch API response
|
||||
|
||||
Args:
|
||||
parts: List of part dicts from JLCSearch API
|
||||
progress_callback: Optional callback(current, total, message)
|
||||
"""
|
||||
cursor = self.conn.cursor()
|
||||
imported = 0
|
||||
skipped = 0
|
||||
|
||||
for i, part in enumerate(parts):
|
||||
try:
|
||||
# JLCSearch format is different from official API
|
||||
# LCSC is an integer, we need to add 'C' prefix
|
||||
lcsc = part.get('lcsc')
|
||||
if isinstance(lcsc, int):
|
||||
lcsc = f"C{lcsc}"
|
||||
|
||||
# Build price JSON from jlcsearch single price
|
||||
price = part.get('price') or part.get('price1')
|
||||
price_json = json.dumps([{"qty": 1, "price": price}] if price else [])
|
||||
|
||||
# Determine library type from is_basic flag
|
||||
library_type = 'Basic' if part.get('is_basic') else 'Extended'
|
||||
if part.get('is_preferred'):
|
||||
library_type = 'Preferred'
|
||||
|
||||
# Extract description from various fields
|
||||
description_parts = []
|
||||
if 'resistance' in part:
|
||||
description_parts.append(f"{part['resistance']}Ω")
|
||||
if 'capacitance' in part:
|
||||
description_parts.append(f"{part['capacitance']}F")
|
||||
if 'tolerance_fraction' in part:
|
||||
tol = part['tolerance_fraction'] * 100
|
||||
description_parts.append(f"±{tol}%")
|
||||
if 'power_watts' in part:
|
||||
description_parts.append(f"{part['power_watts']}mW")
|
||||
if 'voltage' in part:
|
||||
description_parts.append(f"{part['voltage']}V")
|
||||
|
||||
description = part.get('description', ' '.join(description_parts))
|
||||
|
||||
cursor.execute('''
|
||||
INSERT OR REPLACE INTO components (
|
||||
lcsc, category, subcategory, mfr_part, package,
|
||||
solder_joints, manufacturer, library_type, description,
|
||||
datasheet, stock, price_json, last_updated
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
''', (
|
||||
lcsc, # lcsc with C prefix
|
||||
part.get('category', ''), # category
|
||||
part.get('subcategory', ''), # subcategory
|
||||
part.get('mfr', ''), # mfr_part
|
||||
part.get('package', ''), # package
|
||||
0, # solder_joints (not in jlcsearch)
|
||||
part.get('manufacturer', ''), # manufacturer
|
||||
library_type, # library_type
|
||||
description, # description
|
||||
'', # datasheet (not in jlcsearch)
|
||||
part.get('stock', 0), # stock
|
||||
price_json, # price_json
|
||||
int(datetime.now().timestamp()) # last_updated
|
||||
))
|
||||
|
||||
imported += 1
|
||||
|
||||
if progress_callback and (i + 1) % 1000 == 0:
|
||||
progress_callback(i + 1, len(parts), f"Imported {imported} parts...")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error importing part {part.get('lcsc')}: {e}")
|
||||
skipped += 1
|
||||
|
||||
# Update FTS index
|
||||
cursor.execute('''
|
||||
INSERT INTO components_fts(components_fts)
|
||||
VALUES('rebuild')
|
||||
''')
|
||||
|
||||
self.conn.commit()
|
||||
logger.info(f"Import complete: {imported} parts imported, {skipped} skipped")
|
||||
|
||||
def search_parts(
|
||||
self,
|
||||
query: Optional[str] = None,
|
||||
|
||||
246
python/commands/jlcsearch.py
Normal file
246
python/commands/jlcsearch.py
Normal file
@@ -0,0 +1,246 @@
|
||||
"""
|
||||
JLCSearch API client (public, no authentication required)
|
||||
|
||||
Alternative to official JLCPCB API using the community-maintained
|
||||
jlcsearch service at https://jlcsearch.tscircuit.com/
|
||||
"""
|
||||
|
||||
import logging
|
||||
import requests
|
||||
from typing import Optional, Dict, List, Callable
|
||||
import time
|
||||
|
||||
logger = logging.getLogger('kicad_interface')
|
||||
|
||||
|
||||
class JLCSearchClient:
|
||||
"""
|
||||
Client for JLCSearch public API (tscircuit)
|
||||
|
||||
Provides access to JLCPCB parts database without authentication
|
||||
via the community-maintained jlcsearch service.
|
||||
"""
|
||||
|
||||
BASE_URL = "https://jlcsearch.tscircuit.com"
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize JLCSearch API client"""
|
||||
pass
|
||||
|
||||
def search_components(
|
||||
self,
|
||||
category: str = "components",
|
||||
limit: int = 100,
|
||||
offset: int = 0,
|
||||
**filters
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Search components in JLCSearch database
|
||||
|
||||
Args:
|
||||
category: Component category (e.g., "resistors", "capacitors", "components")
|
||||
limit: Maximum number of results
|
||||
offset: Offset for pagination
|
||||
**filters: Additional filters (e.g., package="0603", resistance=1000)
|
||||
|
||||
Returns:
|
||||
List of component dicts
|
||||
"""
|
||||
url = f"{self.BASE_URL}/{category}/list.json"
|
||||
|
||||
params = {
|
||||
"limit": limit,
|
||||
"offset": offset,
|
||||
**filters
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.get(url, params=params, timeout=30)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
# The response has the category name as key
|
||||
# e.g., {"resistors": [...]} or {"components": [...]}
|
||||
for key, value in data.items():
|
||||
if isinstance(value, list):
|
||||
return value
|
||||
|
||||
return []
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"Failed to search JLCSearch: {e}")
|
||||
raise Exception(f"JLCSearch API request failed: {e}")
|
||||
|
||||
def search_resistors(self, resistance: Optional[int] = None, package: Optional[str] = None, limit: int = 100) -> List[Dict]:
|
||||
"""
|
||||
Search for resistors
|
||||
|
||||
Args:
|
||||
resistance: Resistance value in ohms
|
||||
package: Package type (e.g., "0603", "0805")
|
||||
limit: Maximum results
|
||||
|
||||
Returns:
|
||||
List of resistor dicts with fields:
|
||||
- lcsc: LCSC number (integer)
|
||||
- mfr: Manufacturer part number
|
||||
- package: Package size
|
||||
- is_basic: True if basic library part
|
||||
- resistance: Resistance in ohms
|
||||
- tolerance_fraction: Tolerance (0.01 = 1%)
|
||||
- power_watts: Power rating in mW
|
||||
- stock: Available stock
|
||||
- price1: Price per unit
|
||||
"""
|
||||
filters = {}
|
||||
if resistance is not None:
|
||||
filters["resistance"] = resistance
|
||||
if package:
|
||||
filters["package"] = package
|
||||
|
||||
return self.search_components("resistors", limit=limit, **filters)
|
||||
|
||||
def search_capacitors(self, capacitance: Optional[float] = None, package: Optional[str] = None, limit: int = 100) -> List[Dict]:
|
||||
"""
|
||||
Search for capacitors
|
||||
|
||||
Args:
|
||||
capacitance: Capacitance value in farads
|
||||
package: Package type
|
||||
limit: Maximum results
|
||||
|
||||
Returns:
|
||||
List of capacitor dicts
|
||||
"""
|
||||
filters = {}
|
||||
if capacitance is not None:
|
||||
filters["capacitance"] = capacitance
|
||||
if package:
|
||||
filters["package"] = package
|
||||
|
||||
return self.search_components("capacitors", limit=limit, **filters)
|
||||
|
||||
def get_part_by_lcsc(self, lcsc_number: int) -> Optional[Dict]:
|
||||
"""
|
||||
Get part details by LCSC number
|
||||
|
||||
Args:
|
||||
lcsc_number: LCSC number (integer, without 'C' prefix)
|
||||
|
||||
Returns:
|
||||
Part dict or None if not found
|
||||
"""
|
||||
# Search across all components filtering by LCSC
|
||||
# Note: jlcsearch doesn't have a dedicated single-part endpoint
|
||||
# so we search and filter
|
||||
try:
|
||||
results = self.search_components("components", limit=1, lcsc=lcsc_number)
|
||||
return results[0] if results else None
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get part C{lcsc_number}: {e}")
|
||||
return None
|
||||
|
||||
def download_all_components(
|
||||
self,
|
||||
callback: Optional[Callable[[int, str], None]] = None,
|
||||
batch_size: int = 1000
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Download all components from jlcsearch database
|
||||
|
||||
Args:
|
||||
callback: Optional progress callback function(parts_count, status_msg)
|
||||
batch_size: Number of parts per batch
|
||||
|
||||
Returns:
|
||||
List of all parts
|
||||
"""
|
||||
all_parts = []
|
||||
offset = 0
|
||||
|
||||
logger.info("Starting full jlcsearch parts database download...")
|
||||
|
||||
while True:
|
||||
try:
|
||||
batch = self.search_components(
|
||||
"components",
|
||||
limit=batch_size,
|
||||
offset=offset
|
||||
)
|
||||
|
||||
if not batch:
|
||||
break
|
||||
|
||||
all_parts.extend(batch)
|
||||
offset += len(batch)
|
||||
|
||||
if callback:
|
||||
callback(len(all_parts), f"Downloaded {len(all_parts)} parts...")
|
||||
else:
|
||||
logger.info(f"Downloaded {len(all_parts)} parts so far...")
|
||||
|
||||
# If we got fewer results than requested, we've reached the end
|
||||
if len(batch) < batch_size:
|
||||
break
|
||||
|
||||
# Rate limiting - be nice to the API
|
||||
time.sleep(0.1)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error downloading parts at offset {offset}: {e}")
|
||||
if len(all_parts) > 0:
|
||||
logger.warning(f"Partial download available: {len(all_parts)} parts")
|
||||
return all_parts
|
||||
else:
|
||||
raise
|
||||
|
||||
logger.info(f"Download complete: {len(all_parts)} parts retrieved")
|
||||
return all_parts
|
||||
|
||||
|
||||
def test_jlcsearch_connection() -> bool:
|
||||
"""
|
||||
Test JLCSearch API connection
|
||||
|
||||
Returns:
|
||||
True if connection successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
client = JLCSearchClient()
|
||||
# Test by searching for 1k resistors
|
||||
results = client.search_resistors(resistance=1000, limit=5)
|
||||
logger.info(f"JLCSearch API connection test successful - found {len(results)} resistors")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"JLCSearch API connection test failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Test the JLCSearch client
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
print("Testing JLCSearch API connection...")
|
||||
if test_jlcsearch_connection():
|
||||
print("✓ Connection successful!")
|
||||
|
||||
client = JLCSearchClient()
|
||||
|
||||
print("\nSearching for 1k 0603 resistors...")
|
||||
resistors = client.search_resistors(resistance=1000, package="0603", limit=5)
|
||||
print(f"✓ Found {len(resistors)} resistors")
|
||||
|
||||
if resistors:
|
||||
print(f"\nExample resistor:")
|
||||
r = resistors[0]
|
||||
print(f" LCSC: C{r.get('lcsc')}")
|
||||
print(f" MFR: {r.get('mfr')}")
|
||||
print(f" Package: {r.get('package')}")
|
||||
print(f" Resistance: {r.get('resistance')}Ω")
|
||||
print(f" Tolerance: {r.get('tolerance_fraction', 0) * 100}%")
|
||||
print(f" Power: {r.get('power_watts')}mW")
|
||||
print(f" Stock: {r.get('stock')}")
|
||||
print(f" Price: ${r.get('price1')}")
|
||||
print(f" Basic Library: {'Yes' if r.get('is_basic') else 'No'}")
|
||||
else:
|
||||
print("✗ Connection failed")
|
||||
@@ -5,6 +5,7 @@ Project-related command implementations for KiCAD interface
|
||||
import os
|
||||
import pcbnew # type: ignore
|
||||
import logging
|
||||
import shutil
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
logger = logging.getLogger('kicad_interface')
|
||||
@@ -57,12 +58,37 @@ class ProjectCommands:
|
||||
board.SetFileName(board_path)
|
||||
pcbnew.SaveBoard(board_path, board)
|
||||
|
||||
# Create project file
|
||||
# Create schematic from template (use template_with_symbols for component cloning support)
|
||||
schematic_path = project_path.replace(".kicad_pro", ".kicad_sch")
|
||||
template_sch_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
'..', 'templates', 'template_with_symbols.kicad_sch'
|
||||
)
|
||||
|
||||
if os.path.exists(template_sch_path):
|
||||
# Copy template schematic
|
||||
shutil.copy(template_sch_path, schematic_path)
|
||||
logger.info(f"Created schematic from template: {schematic_path}")
|
||||
else:
|
||||
# Fallback: create minimal schematic
|
||||
logger.warning(f"Template not found at {template_sch_path}, creating minimal schematic")
|
||||
with open(schematic_path, 'w') as f:
|
||||
f.write(f'(kicad_sch (version 20230121) (generator "KiCAD-MCP-Server")\n\n')
|
||||
f.write(f' (uuid 00000000-0000-0000-0000-000000000000)\n\n')
|
||||
f.write(f' (paper "A4")\n\n')
|
||||
f.write(f' (lib_symbols\n )\n\n')
|
||||
f.write(f' (sheet_instances\n (path "/" (page "1"))\n )\n')
|
||||
f.write(f')\n')
|
||||
|
||||
# Create project file with schematic reference
|
||||
with open(project_path, 'w') as f:
|
||||
f.write('{\n')
|
||||
f.write(' "board": {\n')
|
||||
f.write(f' "filename": "{os.path.basename(board_path)}"\n')
|
||||
f.write(' }\n')
|
||||
f.write(' },\n')
|
||||
f.write(' "sheets": [\n')
|
||||
f.write(f' ["root", "{os.path.basename(schematic_path)}"]\n')
|
||||
f.write(' ]\n')
|
||||
f.write('}\n')
|
||||
|
||||
self.board = board
|
||||
@@ -73,7 +99,8 @@ class ProjectCommands:
|
||||
"project": {
|
||||
"name": project_name,
|
||||
"path": project_path,
|
||||
"boardPath": board_path
|
||||
"boardPath": board_path,
|
||||
"schematicPath": schematic_path
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from skip import Schematic
|
||||
import os
|
||||
import shutil
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger('kicad_interface')
|
||||
@@ -9,35 +10,40 @@ class SchematicManager:
|
||||
|
||||
@staticmethod
|
||||
def create_schematic(name, metadata=None):
|
||||
"""Create a new empty schematic"""
|
||||
# kicad-skip requires a filepath to create a schematic
|
||||
# We'll create a blank schematic file by loading an existing file
|
||||
# or we can create a template file first.
|
||||
|
||||
# Create an empty template file first
|
||||
temp_path = f"{name}_template.kicad_sch"
|
||||
with open(temp_path, 'w') as f:
|
||||
# Write minimal schematic file content
|
||||
f.write("(kicad_sch (version 20230121) (generator \"KiCAD-MCP-Server\"))\n")
|
||||
|
||||
# Now load it
|
||||
sch = Schematic(temp_path)
|
||||
sch.version = "20230121" # Set appropriate version
|
||||
sch.generator = "KiCAD-MCP-Server"
|
||||
|
||||
# Clean up the template
|
||||
os.remove(temp_path)
|
||||
# Add metadata if provided
|
||||
if metadata:
|
||||
for key, value in metadata.items():
|
||||
# kicad-skip doesn't have a direct metadata property on Schematic,
|
||||
# but we can add properties to the root sheet if needed, or
|
||||
# include it in the file path/name convention.
|
||||
# For now, we'll just create the schematic.
|
||||
pass # Placeholder for potential metadata handling
|
||||
"""Create a new empty schematic from template"""
|
||||
try:
|
||||
# Determine template path (use template_with_symbols for component cloning support)
|
||||
template_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
'..', 'templates', 'template_with_symbols.kicad_sch'
|
||||
)
|
||||
|
||||
logger.info(f"Created new schematic: {name}")
|
||||
return sch
|
||||
# Determine output path
|
||||
output_path = name if name.endswith('.kicad_sch') else f"{name}.kicad_sch"
|
||||
|
||||
if os.path.exists(template_path):
|
||||
# Copy template to target location
|
||||
shutil.copy(template_path, output_path)
|
||||
logger.info(f"Created schematic from template: {output_path}")
|
||||
else:
|
||||
# Fallback: create minimal schematic
|
||||
logger.warning(f"Template not found at {template_path}, creating minimal schematic")
|
||||
with open(output_path, 'w') as f:
|
||||
f.write('(kicad_sch (version 20230121) (generator "KiCAD-MCP-Server")\n\n')
|
||||
f.write(' (uuid 00000000-0000-0000-0000-000000000000)\n\n')
|
||||
f.write(' (paper "A4")\n\n')
|
||||
f.write(' (lib_symbols\n )\n\n')
|
||||
f.write(' (sheet_instances\n (path "/" (page "1"))\n )\n')
|
||||
f.write(')\n')
|
||||
|
||||
# Load the schematic
|
||||
sch = Schematic(output_path)
|
||||
logger.info(f"Loaded new schematic: {output_path}")
|
||||
return sch
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating schematic: {e}")
|
||||
raise
|
||||
|
||||
@staticmethod
|
||||
def load_schematic(file_path):
|
||||
|
||||
Reference in New Issue
Block a user