Clément SAILLANT
524 lines
18 KiB
Python
524 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
SQLite-backed KiCad v10 symbol catalog.
|
|
|
|
This module indexes the exported KiCad v10 symbol libraries once and exposes
|
|
search/query helpers for lightweight MCP servers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
from typing import Any, Dict, List, Optional, Sequence, Tuple
|
|
|
|
try:
|
|
from .kicad_text import (
|
|
KiCadComponent,
|
|
container_library_cache_root,
|
|
container_symbols_dir,
|
|
ensure_container_library_cache,
|
|
parse_inline_library_symbols,
|
|
)
|
|
except ImportError:
|
|
from kicad_text import ( # type: ignore
|
|
KiCadComponent,
|
|
container_library_cache_root,
|
|
container_symbols_dir,
|
|
ensure_container_library_cache,
|
|
parse_inline_library_symbols,
|
|
)
|
|
|
|
|
|
INDEX_VERSION = 1
|
|
SYMBOL_SQLITE_FILENAME = ".symbol_index.sqlite3"
|
|
|
|
|
|
def symbol_sqlite_index_path(cache_root: Optional[Path] = None) -> Path:
|
|
root = cache_root or container_library_cache_root()
|
|
return root / SYMBOL_SQLITE_FILENAME
|
|
|
|
|
|
def _normalize(value: Any) -> str:
|
|
return str(value or "").strip().lower()
|
|
|
|
|
|
def _fts_term(value: str) -> str:
|
|
escaped = value.replace('"', '""').strip()
|
|
return f'"{escaped}"'
|
|
|
|
|
|
def _connect(index_path: Path) -> sqlite3.Connection:
|
|
connection = sqlite3.connect(index_path)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute("PRAGMA synchronous=NORMAL")
|
|
connection.execute("PRAGMA temp_store=MEMORY")
|
|
return connection
|
|
|
|
|
|
def _index_meta(cache_root: Path) -> Dict[str, Any]:
|
|
symbols_dir = container_symbols_dir(cache_root)
|
|
library_files = sorted(symbols_dir.glob("*.kicad_sym"))
|
|
return {
|
|
"version": INDEX_VERSION,
|
|
"symbols_dir": str(symbols_dir),
|
|
"symbols_dir_mtime_ns": symbols_dir.stat().st_mtime_ns if symbols_dir.exists() else None,
|
|
"library_count": len(library_files),
|
|
"library_names": [path.name for path in library_files],
|
|
}
|
|
|
|
|
|
def _metadata_matches(connection: sqlite3.Connection, current_meta: Dict[str, Any]) -> bool:
|
|
row = connection.execute("SELECT value FROM metadata WHERE key = 'meta_json'").fetchone()
|
|
if row is None:
|
|
return False
|
|
try:
|
|
cached_meta = json.loads(row["value"])
|
|
except json.JSONDecodeError:
|
|
return False
|
|
return cached_meta == current_meta
|
|
|
|
|
|
def _create_schema(connection: sqlite3.Connection) -> None:
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE metadata (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE sources (
|
|
path TEXT PRIMARY KEY
|
|
);
|
|
|
|
CREATE TABLE symbols (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
lib_id TEXT NOT NULL,
|
|
reference TEXT NOT NULL,
|
|
value TEXT NOT NULL,
|
|
footprint TEXT NOT NULL,
|
|
datasheet TEXT NOT NULL,
|
|
description TEXT NOT NULL,
|
|
manufacturer TEXT NOT NULL,
|
|
part_number TEXT NOT NULL,
|
|
lcsc_id TEXT NOT NULL,
|
|
source_path TEXT NOT NULL,
|
|
category TEXT NOT NULL,
|
|
component_type TEXT NOT NULL,
|
|
package TEXT NOT NULL,
|
|
unit_price REAL,
|
|
stock INTEGER,
|
|
lib_id_lc TEXT NOT NULL,
|
|
reference_lc TEXT NOT NULL,
|
|
value_lc TEXT NOT NULL,
|
|
footprint_lc TEXT NOT NULL,
|
|
datasheet_lc TEXT NOT NULL,
|
|
description_lc TEXT NOT NULL,
|
|
manufacturer_lc TEXT NOT NULL,
|
|
part_number_lc TEXT NOT NULL,
|
|
lcsc_id_lc TEXT NOT NULL,
|
|
source_path_lc TEXT NOT NULL,
|
|
category_lc TEXT NOT NULL,
|
|
component_type_lc TEXT NOT NULL,
|
|
package_lc TEXT NOT NULL
|
|
);
|
|
|
|
CREATE INDEX idx_symbols_component_type ON symbols(component_type_lc);
|
|
CREATE INDEX idx_symbols_manufacturer ON symbols(manufacturer_lc);
|
|
CREATE INDEX idx_symbols_part_number ON symbols(part_number_lc);
|
|
CREATE INDEX idx_symbols_lib_id ON symbols(lib_id_lc);
|
|
CREATE INDEX idx_symbols_value ON symbols(value_lc);
|
|
CREATE INDEX idx_symbols_footprint ON symbols(footprint_lc);
|
|
CREATE INDEX idx_symbols_category ON symbols(category_lc);
|
|
CREATE INDEX idx_symbols_reference ON symbols(reference_lc);
|
|
CREATE INDEX idx_symbols_lcsc ON symbols(lcsc_id_lc);
|
|
|
|
CREATE VIRTUAL TABLE symbols_fts USING fts5(
|
|
lib_id,
|
|
reference,
|
|
value,
|
|
footprint,
|
|
description,
|
|
manufacturer,
|
|
part_number,
|
|
lcsc_id,
|
|
category,
|
|
component_type,
|
|
package
|
|
);
|
|
"""
|
|
)
|
|
|
|
|
|
def _insert_component(connection: sqlite3.Connection, component: KiCadComponent) -> None:
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO symbols (
|
|
lib_id, reference, value, footprint, datasheet, description,
|
|
manufacturer, part_number, lcsc_id, source_path, category,
|
|
component_type, package, unit_price, stock,
|
|
lib_id_lc, reference_lc, value_lc, footprint_lc, datasheet_lc,
|
|
description_lc, manufacturer_lc, part_number_lc, lcsc_id_lc,
|
|
source_path_lc, category_lc, component_type_lc, package_lc
|
|
) VALUES (
|
|
?, ?, ?, ?, ?, ?,
|
|
?, ?, ?, ?, ?,
|
|
?, ?, ?, ?,
|
|
?, ?, ?, ?, ?,
|
|
?, ?, ?, ?,
|
|
?, ?, ?, ?
|
|
)
|
|
""",
|
|
(
|
|
component.lib_id,
|
|
component.reference,
|
|
component.value,
|
|
component.footprint,
|
|
component.datasheet,
|
|
component.description,
|
|
component.manufacturer,
|
|
component.part_number,
|
|
component.lcsc_id,
|
|
component.source_path,
|
|
component.category,
|
|
component.component_type,
|
|
component.package,
|
|
component.unit_price,
|
|
component.stock,
|
|
_normalize(component.lib_id),
|
|
_normalize(component.reference),
|
|
_normalize(component.value),
|
|
_normalize(component.footprint),
|
|
_normalize(component.datasheet),
|
|
_normalize(component.description),
|
|
_normalize(component.manufacturer),
|
|
_normalize(component.part_number),
|
|
_normalize(component.lcsc_id),
|
|
_normalize(component.source_path),
|
|
_normalize(component.category),
|
|
_normalize(component.component_type),
|
|
_normalize(component.package),
|
|
),
|
|
)
|
|
rowid = cursor.lastrowid
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO symbols_fts (
|
|
rowid, lib_id, reference, value, footprint, description,
|
|
manufacturer, part_number, lcsc_id, category, component_type, package
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
rowid,
|
|
component.lib_id,
|
|
component.reference,
|
|
component.value,
|
|
component.footprint,
|
|
component.description,
|
|
component.manufacturer,
|
|
component.part_number,
|
|
component.lcsc_id,
|
|
component.category,
|
|
component.component_type,
|
|
component.package,
|
|
),
|
|
)
|
|
|
|
|
|
def _row_to_component(row: sqlite3.Row) -> KiCadComponent:
|
|
return KiCadComponent(
|
|
reference=row["reference"],
|
|
value=row["value"],
|
|
footprint=row["footprint"],
|
|
datasheet=row["datasheet"],
|
|
description=row["description"],
|
|
manufacturer=row["manufacturer"],
|
|
part_number=row["part_number"],
|
|
lcsc_id=row["lcsc_id"],
|
|
lib_id=row["lib_id"],
|
|
source_path=row["source_path"],
|
|
category=row["category"],
|
|
component_type=row["component_type"],
|
|
unit_price=row["unit_price"],
|
|
stock=row["stock"],
|
|
)
|
|
|
|
|
|
def build_symbol_sqlite_index(
|
|
cache_root: Optional[Path] = None,
|
|
*,
|
|
refresh: bool = False,
|
|
) -> Dict[str, Any]:
|
|
resolved_root = cache_root or ensure_container_library_cache(refresh=refresh)
|
|
if resolved_root is None:
|
|
return {
|
|
"index_path": None,
|
|
"symbol_libraries": 0,
|
|
"symbol_components": 0,
|
|
"source_count": 0,
|
|
}
|
|
|
|
resolved_root = resolved_root.expanduser()
|
|
index_path = symbol_sqlite_index_path(resolved_root)
|
|
current_meta = _index_meta(resolved_root)
|
|
|
|
if index_path.exists() and not refresh:
|
|
with _connect(index_path) as connection:
|
|
if _metadata_matches(connection, current_meta):
|
|
return symbol_sqlite_summary(resolved_root)
|
|
|
|
fd, tmp_name = tempfile.mkstemp(
|
|
prefix=".symbol-index-",
|
|
suffix=".sqlite3",
|
|
dir=str(resolved_root),
|
|
)
|
|
Path(tmp_name).unlink(missing_ok=True)
|
|
tmp_path = Path(tmp_name)
|
|
|
|
try:
|
|
with _connect(tmp_path) as connection:
|
|
_create_schema(connection)
|
|
connection.execute(
|
|
"INSERT INTO metadata(key, value) VALUES('meta_json', ?)",
|
|
(json.dumps(current_meta, ensure_ascii=True),),
|
|
)
|
|
|
|
sources = sorted(container_symbols_dir(resolved_root).glob("*.kicad_sym"))
|
|
for source in sources:
|
|
connection.execute("INSERT INTO sources(path) VALUES (?)", (str(source),))
|
|
for component in parse_inline_library_symbols(source):
|
|
_insert_component(connection, component)
|
|
connection.commit()
|
|
|
|
tmp_path.replace(index_path)
|
|
return symbol_sqlite_summary(resolved_root)
|
|
finally:
|
|
tmp_path.unlink(missing_ok=True)
|
|
|
|
|
|
def symbol_sqlite_summary(cache_root: Optional[Path] = None) -> Dict[str, Any]:
|
|
resolved_root = cache_root or ensure_container_library_cache()
|
|
if resolved_root is None:
|
|
return {
|
|
"index_path": None,
|
|
"symbol_libraries": 0,
|
|
"symbol_components": 0,
|
|
"source_count": 0,
|
|
}
|
|
|
|
index_path = symbol_sqlite_index_path(resolved_root)
|
|
if not index_path.exists():
|
|
return build_symbol_sqlite_index(resolved_root, refresh=False)
|
|
|
|
with _connect(index_path) as connection:
|
|
symbol_components = connection.execute("SELECT COUNT(*) AS count FROM symbols").fetchone()["count"]
|
|
symbol_libraries = connection.execute("SELECT COUNT(*) AS count FROM sources").fetchone()["count"]
|
|
return {
|
|
"index_path": str(index_path),
|
|
"symbol_libraries": symbol_libraries,
|
|
"symbol_components": symbol_components,
|
|
"source_count": symbol_libraries,
|
|
}
|
|
|
|
|
|
def load_all_symbol_components(
|
|
cache_root: Optional[Path] = None,
|
|
*,
|
|
refresh: bool = False,
|
|
) -> Tuple[List[KiCadComponent], List[Path]]:
|
|
resolved_root = cache_root or ensure_container_library_cache(refresh=refresh)
|
|
if resolved_root is None:
|
|
return [], []
|
|
|
|
build_symbol_sqlite_index(resolved_root, refresh=refresh)
|
|
index_path = symbol_sqlite_index_path(resolved_root)
|
|
with _connect(index_path) as connection:
|
|
rows = connection.execute("SELECT * FROM symbols ORDER BY lib_id").fetchall()
|
|
sources = [Path(row["path"]) for row in connection.execute("SELECT path FROM sources ORDER BY path")]
|
|
return [_row_to_component(row) for row in rows], sources
|
|
|
|
|
|
def source_sample(cache_root: Optional[Path] = None, limit: int = 10) -> List[str]:
|
|
resolved_root = cache_root or ensure_container_library_cache()
|
|
if resolved_root is None:
|
|
return []
|
|
build_symbol_sqlite_index(resolved_root, refresh=False)
|
|
with _connect(symbol_sqlite_index_path(resolved_root)) as connection:
|
|
rows = connection.execute(
|
|
"SELECT path FROM sources ORDER BY path LIMIT ?",
|
|
(limit,),
|
|
).fetchall()
|
|
return [row["path"] for row in rows]
|
|
|
|
|
|
def source_count(cache_root: Optional[Path] = None) -> int:
|
|
resolved_root = cache_root or ensure_container_library_cache()
|
|
if resolved_root is None:
|
|
return 0
|
|
build_symbol_sqlite_index(resolved_root, refresh=False)
|
|
with _connect(symbol_sqlite_index_path(resolved_root)) as connection:
|
|
return connection.execute("SELECT COUNT(*) AS count FROM sources").fetchone()["count"]
|
|
|
|
|
|
def catalog_size(cache_root: Optional[Path] = None) -> int:
|
|
resolved_root = cache_root or ensure_container_library_cache()
|
|
if resolved_root is None:
|
|
return 0
|
|
build_symbol_sqlite_index(resolved_root, refresh=False)
|
|
with _connect(symbol_sqlite_index_path(resolved_root)) as connection:
|
|
return connection.execute("SELECT COUNT(*) AS count FROM symbols").fetchone()["count"]
|
|
|
|
|
|
def resolve_symbol_component(identifier: str, cache_root: Optional[Path] = None) -> Optional[KiCadComponent]:
|
|
needle = _normalize(identifier)
|
|
if not needle:
|
|
return None
|
|
|
|
resolved_root = cache_root or ensure_container_library_cache()
|
|
if resolved_root is None:
|
|
return None
|
|
build_symbol_sqlite_index(resolved_root, refresh=False)
|
|
|
|
query = """
|
|
SELECT *
|
|
FROM symbols
|
|
WHERE lib_id_lc = ?
|
|
OR reference_lc = ?
|
|
OR value_lc = ?
|
|
OR part_number_lc = ?
|
|
OR lcsc_id_lc = ?
|
|
OR (lib_id_lc || ':' || reference_lc) = ?
|
|
LIMIT 1
|
|
"""
|
|
with _connect(symbol_sqlite_index_path(resolved_root)) as connection:
|
|
row = connection.execute(query, (needle, needle, needle, needle, needle, needle)).fetchone()
|
|
return _row_to_component(row) if row is not None else None
|
|
|
|
|
|
def _apply_spec_filter(
|
|
where_clauses: List[str],
|
|
params: List[Any],
|
|
score_terms: List[str],
|
|
score_params: List[Any],
|
|
key: str,
|
|
value: str,
|
|
) -> None:
|
|
normalized = _normalize(value)
|
|
if not normalized:
|
|
return
|
|
|
|
if key in {"value", "resistance", "capacitance"}:
|
|
where_clauses.append("instr(value_lc, ?) > 0")
|
|
params.append(normalized)
|
|
score_terms.append("CASE WHEN value_lc = ? THEN 60 WHEN instr(value_lc, ?) > 0 THEN 30 ELSE 0 END")
|
|
score_params.extend([normalized, normalized])
|
|
return
|
|
|
|
if key in {"package"}:
|
|
where_clauses.append("instr(package_lc, ?) > 0")
|
|
params.append(normalized)
|
|
score_terms.append("CASE WHEN package_lc = ? THEN 35 WHEN instr(package_lc, ?) > 0 THEN 20 ELSE 0 END")
|
|
score_params.extend([normalized, normalized])
|
|
return
|
|
|
|
if key in {"footprint"}:
|
|
where_clauses.append("(instr(package_lc, ?) > 0 OR instr(footprint_lc, ?) > 0)")
|
|
params.extend([normalized, normalized])
|
|
score_terms.append(
|
|
"CASE WHEN footprint_lc = ? THEN 35 WHEN instr(footprint_lc, ?) > 0 THEN 25 "
|
|
"WHEN package_lc = ? THEN 25 WHEN instr(package_lc, ?) > 0 THEN 15 ELSE 0 END"
|
|
)
|
|
score_params.extend([normalized, normalized, normalized, normalized])
|
|
return
|
|
|
|
if key in {"manufacturer", "brand"}:
|
|
where_clauses.append("instr(manufacturer_lc, ?) > 0")
|
|
params.append(normalized)
|
|
score_terms.append(
|
|
"CASE WHEN manufacturer_lc = ? THEN 55 WHEN instr(manufacturer_lc, ?) > 0 THEN 25 ELSE 0 END"
|
|
)
|
|
score_params.extend([normalized, normalized])
|
|
return
|
|
|
|
if key in {"part", "part_number", "mpn"}:
|
|
where_clauses.append("instr(part_number_lc, ?) > 0")
|
|
params.append(normalized)
|
|
score_terms.append(
|
|
"CASE WHEN part_number_lc = ? THEN 90 WHEN instr(part_number_lc, ?) > 0 THEN 45 ELSE 0 END"
|
|
)
|
|
score_params.extend([normalized, normalized])
|
|
return
|
|
|
|
if key in {"lcsc", "lcsc_id"}:
|
|
where_clauses.append("instr(lcsc_id_lc, ?) > 0")
|
|
params.append(normalized)
|
|
score_terms.append(
|
|
"CASE WHEN lcsc_id_lc = ? THEN 90 WHEN instr(lcsc_id_lc, ?) > 0 THEN 45 ELSE 0 END"
|
|
)
|
|
score_params.extend([normalized, normalized])
|
|
return
|
|
|
|
if key in {"type", "category"}:
|
|
where_clauses.append("(instr(component_type_lc, ?) > 0 OR instr(category_lc, ?) > 0)")
|
|
params.extend([normalized, normalized])
|
|
score_terms.append(
|
|
"CASE WHEN component_type_lc = ? THEN 70 WHEN instr(component_type_lc, ?) > 0 THEN 35 "
|
|
"WHEN instr(category_lc, ?) > 0 THEN 20 ELSE 0 END"
|
|
)
|
|
score_params.extend([normalized, normalized, normalized])
|
|
return
|
|
|
|
where_clauses.append(
|
|
"id IN (SELECT rowid FROM symbols_fts WHERE symbols_fts MATCH ?)"
|
|
)
|
|
params.append(_fts_term(normalized))
|
|
score_terms.append("20")
|
|
|
|
|
|
def search_symbol_components(
|
|
component_type: str,
|
|
specs: Dict[str, Any],
|
|
*,
|
|
limit: int = 20,
|
|
cache_root: Optional[Path] = None,
|
|
) -> List[KiCadComponent]:
|
|
resolved_root = cache_root or ensure_container_library_cache()
|
|
if resolved_root is None:
|
|
return []
|
|
build_symbol_sqlite_index(resolved_root, refresh=False)
|
|
|
|
where_clauses = ["1 = 1"]
|
|
params: List[Any] = []
|
|
score_terms: List[str] = []
|
|
score_params: List[Any] = []
|
|
|
|
wanted_type = _normalize(component_type)
|
|
if wanted_type:
|
|
where_clauses.append("(component_type_lc = ? OR instr(lib_id_lc, ?) > 0)")
|
|
params.extend([wanted_type, wanted_type])
|
|
score_terms.append(
|
|
"CASE WHEN component_type_lc = ? THEN 80 WHEN instr(lib_id_lc, ?) > 0 THEN 40 ELSE 0 END"
|
|
)
|
|
score_params.extend([wanted_type, wanted_type])
|
|
|
|
for key, value in specs.items():
|
|
_apply_spec_filter(where_clauses, params, score_terms, score_params, key.lower(), str(value))
|
|
|
|
score_sql = " + ".join(score_terms) if score_terms else "0"
|
|
sql = f"""
|
|
SELECT *,
|
|
({score_sql}) AS score
|
|
FROM symbols
|
|
WHERE {' AND '.join(where_clauses)}
|
|
ORDER BY score DESC, lib_id ASC
|
|
LIMIT ?
|
|
"""
|
|
|
|
with _connect(symbol_sqlite_index_path(resolved_root)) as connection:
|
|
rows = connection.execute(sql, tuple(score_params + params + [max(limit * 10, limit)])).fetchall()
|
|
|
|
return [_row_to_component(row) for row in rows[:limit]]
|