Clément SAILLANT
483 lines
18 KiB
Python
483 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
KiCad component database MCP server.
|
|
|
|
The server stays line-delimited JSON-RPC because the existing KIC-AI plugin
|
|
clients read one JSON object per line from stdout.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Sequence
|
|
|
|
try:
|
|
from .kicad_text import (
|
|
KiCadComponent,
|
|
container_symbols_dir,
|
|
discover_symbol_library_files,
|
|
discover_schematic_files,
|
|
discover_search_roots,
|
|
parse_inline_library_symbols,
|
|
parse_schematic_components,
|
|
)
|
|
from .kicad_symbol_sqlite import (
|
|
catalog_size as sqlite_catalog_size,
|
|
resolve_symbol_component,
|
|
search_symbol_components,
|
|
source_count as sqlite_source_count,
|
|
source_sample as sqlite_source_sample,
|
|
)
|
|
except ImportError:
|
|
from kicad_text import ( # type: ignore
|
|
KiCadComponent,
|
|
container_symbols_dir,
|
|
discover_symbol_library_files,
|
|
discover_schematic_files,
|
|
discover_search_roots,
|
|
parse_inline_library_symbols,
|
|
parse_schematic_components,
|
|
)
|
|
from kicad_symbol_sqlite import ( # type: ignore
|
|
catalog_size as sqlite_catalog_size,
|
|
resolve_symbol_component,
|
|
search_symbol_components,
|
|
source_count as sqlite_source_count,
|
|
source_sample as sqlite_source_sample,
|
|
)
|
|
|
|
|
|
class ComponentDatabaseServer:
|
|
"""Component database backed by locally discoverable KiCad files."""
|
|
|
|
def __init__(self):
|
|
self.base_dir = Path(__file__).resolve()
|
|
self.search_roots = discover_search_roots(self.base_dir)
|
|
self.container_symbols_root = container_symbols_dir()
|
|
self._catalog: Optional[List[KiCadComponent]] = None
|
|
self._sources: Optional[List[str]] = None
|
|
|
|
def _load_catalog(self) -> List[KiCadComponent]:
|
|
if self._catalog is not None:
|
|
return self._catalog
|
|
|
|
components: List[KiCadComponent] = []
|
|
source_files: List[str] = []
|
|
seen = set()
|
|
|
|
for schematic_path in discover_schematic_files(self.search_roots):
|
|
source_files.append(str(schematic_path))
|
|
|
|
for component in parse_schematic_components(schematic_path):
|
|
key = (
|
|
component.lib_id,
|
|
component.value,
|
|
component.footprint,
|
|
component.manufacturer,
|
|
component.part_number,
|
|
component.lcsc_id,
|
|
)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
components.append(component)
|
|
|
|
# Inline libraries from template and expanded schematics are useful
|
|
# when no global KiCad libraries are installed on the host.
|
|
for component in parse_inline_library_symbols(schematic_path):
|
|
key = (
|
|
"inline",
|
|
component.lib_id,
|
|
component.value,
|
|
component.footprint,
|
|
component.manufacturer,
|
|
component.part_number,
|
|
component.lcsc_id,
|
|
)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
components.append(component)
|
|
|
|
for library_path in discover_symbol_library_files(self.search_roots):
|
|
try:
|
|
if library_path.resolve().parent == self.container_symbols_root.resolve():
|
|
continue
|
|
except FileNotFoundError:
|
|
pass
|
|
source_files.append(str(library_path))
|
|
for component in parse_inline_library_symbols(library_path):
|
|
key = (
|
|
"library",
|
|
component.lib_id,
|
|
component.value,
|
|
component.footprint,
|
|
component.manufacturer,
|
|
component.part_number,
|
|
component.lcsc_id,
|
|
)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
components.append(component)
|
|
|
|
self._catalog = components
|
|
self._sources = sorted(set(source_files))
|
|
return self._catalog
|
|
|
|
def _local_catalog_sources(self) -> List[str]:
|
|
self._load_catalog()
|
|
return self._sources or []
|
|
|
|
def _catalog_sources_sample(self, limit: int = 10) -> List[str]:
|
|
sample = list(self._local_catalog_sources()[:limit])
|
|
if len(sample) >= limit:
|
|
return sample
|
|
|
|
for source in sqlite_source_sample(limit=limit):
|
|
if source in sample:
|
|
continue
|
|
sample.append(source)
|
|
if len(sample) >= limit:
|
|
break
|
|
return sample
|
|
|
|
def _catalog_source_count(self) -> int:
|
|
return len(self._local_catalog_sources()) + sqlite_source_count()
|
|
|
|
def _catalog_size(self) -> int:
|
|
return len(self._load_catalog()) + sqlite_catalog_size()
|
|
|
|
def _resolve_component(self, identifier: str) -> Optional[KiCadComponent]:
|
|
needle = identifier.strip().lower()
|
|
if not needle:
|
|
return None
|
|
|
|
for component in self._load_catalog():
|
|
candidates = [
|
|
component.lib_id,
|
|
component.reference,
|
|
component.value,
|
|
component.part_number,
|
|
component.lcsc_id,
|
|
f"{component.lib_id}:{component.reference}",
|
|
]
|
|
if any(candidate and candidate.lower() == needle for candidate in candidates):
|
|
return component
|
|
|
|
return resolve_symbol_component(identifier)
|
|
|
|
def _matches_type(self, component: KiCadComponent, component_type: str) -> bool:
|
|
if not component_type:
|
|
return True
|
|
wanted = component_type.strip().lower()
|
|
return wanted in component.component_type.lower() or wanted in component.lib_id.lower()
|
|
|
|
def _matches_specs(self, component: KiCadComponent, specs: Dict[str, Any]) -> bool:
|
|
for key, value in specs.items():
|
|
expected = str(value).strip().lower()
|
|
if not expected:
|
|
continue
|
|
haystacks: Sequence[str]
|
|
key_lower = key.lower()
|
|
if key_lower in {"value", "resistance", "capacitance"}:
|
|
haystacks = [component.value]
|
|
elif key_lower in {"package", "footprint"}:
|
|
haystacks = [component.package, component.footprint]
|
|
elif key_lower in {"manufacturer", "brand"}:
|
|
haystacks = [component.manufacturer]
|
|
elif key_lower in {"part", "part_number", "mpn"}:
|
|
haystacks = [component.part_number]
|
|
elif key_lower in {"lcsc", "lcsc_id"}:
|
|
haystacks = [component.lcsc_id]
|
|
elif key_lower in {"type", "category"}:
|
|
haystacks = [component.component_type, component.category]
|
|
else:
|
|
haystacks = [
|
|
component.reference,
|
|
component.value,
|
|
component.footprint,
|
|
component.description,
|
|
component.manufacturer,
|
|
component.part_number,
|
|
component.lcsc_id,
|
|
component.lib_id,
|
|
component.category,
|
|
]
|
|
if not any(expected in (candidate or "").lower() for candidate in haystacks):
|
|
return False
|
|
return True
|
|
|
|
def _score_component(self, component: KiCadComponent, component_type: str, specs: Dict[str, Any]) -> int:
|
|
score = 0
|
|
if component_type:
|
|
wanted = component_type.strip().lower()
|
|
if component.component_type == wanted:
|
|
score += 80
|
|
elif wanted in component.lib_id.lower():
|
|
score += 40
|
|
|
|
for key, value in specs.items():
|
|
expected = str(value).strip().lower()
|
|
if not expected:
|
|
continue
|
|
if expected == component.value.lower():
|
|
score += 60
|
|
elif expected in component.value.lower():
|
|
score += 30
|
|
if expected and expected in component.footprint.lower():
|
|
score += 25
|
|
if expected and expected in component.part_number.lower():
|
|
score += 35
|
|
if expected and expected in component.description.lower():
|
|
score += 15
|
|
|
|
if component.lcsc_id:
|
|
score += 10
|
|
if component.manufacturer:
|
|
score += 5
|
|
return score
|
|
|
|
def _alternatives(self, target: KiCadComponent, limit: int = 3) -> List[str]:
|
|
matches: List[str] = []
|
|
for component in self._load_catalog():
|
|
if component.lib_id == target.lib_id:
|
|
continue
|
|
if component.component_type != target.component_type:
|
|
continue
|
|
if target.value and component.value and component.value.lower() != target.value.lower():
|
|
continue
|
|
matches.append(component.lib_id)
|
|
if len(matches) >= limit:
|
|
return matches
|
|
|
|
extra = search_symbol_components(
|
|
target.component_type,
|
|
{"value": target.value} if target.value else {},
|
|
limit=limit + 3,
|
|
)
|
|
for component in extra:
|
|
if component.lib_id == target.lib_id or component.lib_id in matches:
|
|
continue
|
|
matches.append(component.lib_id)
|
|
if len(matches) >= limit:
|
|
break
|
|
return matches
|
|
|
|
def _search_components(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
|
|
component_type = arguments.get("type", "")
|
|
specs = arguments.get("specs", {})
|
|
limit = int(arguments.get("limit", 20))
|
|
|
|
matches = [
|
|
component
|
|
for component in self._load_catalog()
|
|
if self._matches_type(component, component_type) and self._matches_specs(component, specs)
|
|
]
|
|
matches.sort(
|
|
key=lambda component: self._score_component(component, component_type, specs),
|
|
reverse=True,
|
|
)
|
|
seen_ids = {component.lib_id for component in matches}
|
|
for component in search_symbol_components(component_type, specs, limit=max(limit * 3, limit)):
|
|
if component.lib_id in seen_ids:
|
|
continue
|
|
seen_ids.add(component.lib_id)
|
|
matches.append(component)
|
|
|
|
matches.sort(
|
|
key=lambda component: self._score_component(component, component_type, specs),
|
|
reverse=True,
|
|
)
|
|
results = []
|
|
for component in matches[:limit]:
|
|
payload = component.to_dict(include_values=True, include_footprints=True)
|
|
payload["part_id"] = component.lib_id
|
|
payload["alternatives"] = self._alternatives(component)
|
|
results.append(payload)
|
|
|
|
return {
|
|
"components": results,
|
|
"total_found": len(results),
|
|
"indexed_source_count": self._catalog_source_count(),
|
|
"indexed_sources_sample": self._catalog_sources_sample(),
|
|
"catalog_size": self._catalog_size(),
|
|
"data_mode": "file-backed-local+container-sqlite",
|
|
}
|
|
|
|
def _get_pricing(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
|
|
identifiers = list(arguments.get("part_numbers", []))
|
|
if not identifiers and arguments.get("part_number"):
|
|
identifiers.append(arguments["part_number"])
|
|
|
|
pricing: Dict[str, Dict[str, Any]] = {}
|
|
for identifier in identifiers:
|
|
component = self._resolve_component(identifier)
|
|
if component is None:
|
|
pricing[identifier] = {
|
|
"available": False,
|
|
"message": "Component not found in local KiCad catalog",
|
|
}
|
|
continue
|
|
|
|
pricing[identifier] = {
|
|
"available": component.unit_price is not None,
|
|
"unit_price": component.unit_price,
|
|
"currency": "USD" if component.unit_price is not None else None,
|
|
"stock": component.stock,
|
|
"lcsc_id": component.lcsc_id or None,
|
|
"manufacturer": component.manufacturer or None,
|
|
"part_number": component.part_number or None,
|
|
"source": component.source_path,
|
|
}
|
|
|
|
return {
|
|
"pricing": pricing,
|
|
"data_mode": "file-backed-local+container-sqlite",
|
|
"note": "Pricing is only returned when local or cached KiCad symbol metadata contains explicit price fields.",
|
|
}
|
|
|
|
def _check_availability(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
|
|
identifiers = list(arguments.get("part_numbers", []))
|
|
availability: Dict[str, Any] = {}
|
|
details: Dict[str, Dict[str, Any]] = {}
|
|
for identifier in identifiers:
|
|
component = self._resolve_component(identifier)
|
|
if component is None:
|
|
availability[identifier] = None
|
|
details[identifier] = {
|
|
"available": False,
|
|
"message": "Component not found in local KiCad catalog",
|
|
}
|
|
continue
|
|
|
|
availability[identifier] = component.stock
|
|
details[identifier] = {
|
|
"available": component.stock is not None,
|
|
"stock": component.stock,
|
|
"lcsc_id": component.lcsc_id or None,
|
|
"source": component.source_path,
|
|
}
|
|
|
|
return {
|
|
"availability": availability,
|
|
"details": details,
|
|
"data_mode": "file-backed-local+container-sqlite",
|
|
"note": "Availability is only returned when local or cached KiCad symbol metadata contains explicit stock fields.",
|
|
}
|
|
|
|
async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
|
|
method = request.get("method")
|
|
params = request.get("params", {})
|
|
request_id = request.get("id")
|
|
|
|
if method == "initialize":
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"result": {
|
|
"protocolVersion": "2025-03-26",
|
|
"capabilities": {"tools": {}},
|
|
"serverInfo": {
|
|
"name": "component-database",
|
|
"version": "2.1.0",
|
|
},
|
|
},
|
|
}
|
|
|
|
if method == "tools/list":
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"result": {
|
|
"tools": [
|
|
{
|
|
"name": "search_components",
|
|
"description": "Search components indexed from local KiCad schematics, inline symbol libraries, and cached KiCad v10 symbol libraries",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"type": "string"},
|
|
"specs": {"type": "object"},
|
|
"limit": {"type": "integer", "default": 20},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "get_pricing",
|
|
"description": "Return explicit price metadata present in local or cached KiCad symbol properties",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"part_numbers": {"type": "array", "items": {"type": "string"}},
|
|
"part_number": {"type": "string"},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"name": "check_availability",
|
|
"description": "Return explicit stock metadata present in local or cached KiCad symbol properties",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"part_numbers": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
"required": ["part_numbers"],
|
|
},
|
|
},
|
|
]
|
|
},
|
|
}
|
|
|
|
if method == "tools/call":
|
|
tool_name = params.get("name")
|
|
arguments = params.get("arguments", {})
|
|
if tool_name == "search_components":
|
|
result = self._search_components(arguments)
|
|
elif tool_name == "get_pricing":
|
|
result = self._get_pricing(arguments)
|
|
elif tool_name == "check_availability":
|
|
result = self._check_availability(arguments)
|
|
else:
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"error": {"code": -32601, "message": f"Unknown tool: {tool_name}"},
|
|
}
|
|
return {"jsonrpc": "2.0", "id": request_id, "result": result}
|
|
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": request_id,
|
|
"error": {"code": -32601, "message": "Method not found"},
|
|
}
|
|
|
|
|
|
async def main():
|
|
server = ComponentDatabaseServer()
|
|
while True:
|
|
try:
|
|
line = sys.stdin.readline()
|
|
if not line:
|
|
break
|
|
|
|
request = json.loads(line.strip())
|
|
response = await server.handle_request(request)
|
|
print(json.dumps(response))
|
|
sys.stdout.flush()
|
|
except Exception as exc:
|
|
error_response = {
|
|
"jsonrpc": "2.0",
|
|
"id": None,
|
|
"error": {"code": -32603, "message": f"Internal error: {exc}"},
|
|
}
|
|
print(json.dumps(error_response))
|
|
sys.stdout.flush()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|