Ryan Malloy
71dfdc8d94
Bridge mcltspice to SpiceBook (spicebook.warehack.ing) with 5 new MCP tools (publish, list, get, delete, simulate), a status resource, and a guided publish prompt. Includes LTspice-to-ngspice netlist conversion (strips .backanno, Rser=, Windows paths) with structured warnings. Path traversal protection on file reads (extension allowlist + resolve), notebook ID validation, narrowed exception handling, and size limits. 35 unit tests + 4 live integration tests.
232 lines
7.6 KiB
Python
232 lines
7.6 KiB
Python
"""SpiceBook integration -- publish LTspice circuits as interactive notebooks.
|
|
|
|
SpiceBook (spicebook.warehack.ing) is a web notebook service for SPICE circuits.
|
|
This module provides:
|
|
- SpiceBookClient: async HTTP client for the SpiceBook REST API
|
|
- ltspice_to_ngspice: best-effort netlist dialect conversion
|
|
- build_notebook_cells: assemble notebook cells for the compose endpoint
|
|
"""
|
|
|
|
import re
|
|
|
|
import httpx
|
|
|
|
from .config import SPICEBOOK_TIMEOUT, SPICEBOOK_URL
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Netlist conversion: LTspice -> ngspice
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Directives that are LTspice-only and should be stripped (precomputed lowercase)
|
|
_LTSPICE_ONLY_DIRECTIVES_LOWER = {
|
|
".backanno",
|
|
".options plotwinsize=0",
|
|
}
|
|
|
|
# Notebook IDs must be alphanumeric with dashes/underscores
|
|
_NOTEBOOK_ID_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]*$")
|
|
|
|
# Regex for .lib/.include with Windows-style paths (backslashes or drive letters)
|
|
_WINDOWS_PATH_RE = re.compile(
|
|
r"^\s*\.(lib|include)\s+.*([A-Za-z]:\\|\\\\)", re.IGNORECASE
|
|
)
|
|
|
|
# Regex for Rser= on capacitor or inductor lines
|
|
_RSER_RE = re.compile(r"\s+Rser=\S+", re.IGNORECASE)
|
|
|
|
|
|
def ltspice_to_ngspice(netlist_text: str) -> tuple[str, list[str]]:
|
|
"""Convert an LTspice netlist to ngspice-compatible form.
|
|
|
|
Returns (cleaned_netlist, warnings) where warnings describe any
|
|
constructs that were modified or removed.
|
|
"""
|
|
lines = netlist_text.splitlines()
|
|
out: list[str] = []
|
|
warnings: list[str] = []
|
|
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
|
|
# Strip LTspice-only directives
|
|
if stripped.lower() in _LTSPICE_ONLY_DIRECTIVES_LOWER:
|
|
continue
|
|
|
|
# Comment out .lib/.include with Windows paths
|
|
if _WINDOWS_PATH_RE.match(line):
|
|
out.append(f"* {line} ; commented out -- Windows path")
|
|
warnings.append(f"Commented out Windows path: {stripped}")
|
|
continue
|
|
|
|
# Strip Rser= from component lines (C or L)
|
|
if stripped and stripped[0].upper() in ("C", "L") and "Rser=" in line:
|
|
cleaned = _RSER_RE.sub("", line)
|
|
out.append(cleaned)
|
|
warnings.append(f"Stripped Rser= from: {stripped}")
|
|
continue
|
|
|
|
out.append(line)
|
|
|
|
return "\n".join(out), warnings
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cell builder for SpiceBook compose endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def build_notebook_cells(
|
|
title: str,
|
|
description: str | None,
|
|
netlist: str,
|
|
warnings: list[str] | None = None,
|
|
analysis: dict | None = None,
|
|
svgs: list[str] | None = None,
|
|
) -> list[dict]:
|
|
"""Build the cell list for SpiceBook's POST /api/notebooks/compose.
|
|
|
|
Args:
|
|
title: Notebook title (also used in the intro cell heading).
|
|
description: Optional circuit description.
|
|
netlist: ngspice-compatible netlist text.
|
|
warnings: Conversion warnings from ltspice_to_ngspice.
|
|
analysis: Optional dict of analysis results (rendered as a table).
|
|
svgs: Optional list of SVG strings to embed as images.
|
|
|
|
Returns:
|
|
List of cell dicts ready for the compose endpoint.
|
|
"""
|
|
cells: list[dict] = []
|
|
|
|
# 1. Intro markdown cell
|
|
intro_parts = [f"# {title}"]
|
|
if description:
|
|
intro_parts.append(description)
|
|
if warnings:
|
|
intro_parts.append("### Conversion notes")
|
|
for w in warnings:
|
|
intro_parts.append(f"- {w}")
|
|
cells.append({"type": "markdown", "source": "\n\n".join(intro_parts)})
|
|
|
|
# 2. SPICE netlist cell
|
|
cells.append({"type": "spice", "source": netlist})
|
|
|
|
# 3. Optional analysis results as markdown table
|
|
if analysis:
|
|
rows = ["| Metric | Value |", "|--------|-------|"]
|
|
for key, val in analysis.items():
|
|
rows.append(f"| {key} | {val} |")
|
|
cells.append({"type": "markdown", "source": "\n".join(rows)})
|
|
|
|
# 4. Optional SVG plot cells
|
|
if svgs:
|
|
for i, svg in enumerate(svgs):
|
|
label = f"### Plot {i + 1}" if len(svgs) > 1 else "### Plot"
|
|
cells.append({"type": "markdown", "source": f"{label}\n\n{svg}"})
|
|
|
|
return cells
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SpiceBook async HTTP client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class SpiceBookClient:
|
|
"""Thin async wrapper around the SpiceBook REST API."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str = SPICEBOOK_URL,
|
|
timeout: float = SPICEBOOK_TIMEOUT,
|
|
):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.timeout = timeout
|
|
|
|
def _client(self) -> httpx.AsyncClient:
|
|
return httpx.AsyncClient(
|
|
base_url=self.base_url,
|
|
timeout=self.timeout,
|
|
)
|
|
|
|
async def health(self) -> bool:
|
|
"""Check if SpiceBook is reachable. Returns False on any network error."""
|
|
try:
|
|
async with self._client() as client:
|
|
resp = await client.get("/health")
|
|
return resp.status_code == 200
|
|
except (httpx.HTTPError, OSError):
|
|
return False
|
|
|
|
@staticmethod
|
|
def _validate_notebook_id(notebook_id: str) -> str | None:
|
|
"""Return an error message if notebook_id is unsafe for URL interpolation."""
|
|
if not notebook_id or not _NOTEBOOK_ID_RE.match(notebook_id):
|
|
return f"Invalid notebook_id: expected alphanumeric/dash/underscore, got '{notebook_id[:80]}'"
|
|
return None
|
|
|
|
async def compose(
|
|
self,
|
|
title: str,
|
|
cells: list[dict],
|
|
tags: list[str] | None = None,
|
|
run: bool = False,
|
|
engine: str = "ngspice",
|
|
) -> dict:
|
|
"""Create a notebook with cells in one atomic call."""
|
|
payload: dict = {
|
|
"title": title,
|
|
"cells": cells,
|
|
"engine": engine,
|
|
}
|
|
if tags:
|
|
payload["tags"] = tags
|
|
if run:
|
|
payload["run"] = True
|
|
|
|
async with self._client() as client:
|
|
resp = await client.post("/api/notebooks/compose", json=payload)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def list_notebooks(self) -> list[dict]:
|
|
"""List all notebooks."""
|
|
async with self._client() as client:
|
|
resp = await client.get("/api/notebooks")
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def get_notebook(self, notebook_id: str) -> dict:
|
|
"""Get full notebook by ID."""
|
|
err = self._validate_notebook_id(notebook_id)
|
|
if err:
|
|
raise ValueError(err)
|
|
async with self._client() as client:
|
|
resp = await client.get(f"/api/notebooks/{notebook_id}")
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def delete_notebook(self, notebook_id: str) -> bool:
|
|
"""Delete a notebook. Returns True on success."""
|
|
err = self._validate_notebook_id(notebook_id)
|
|
if err:
|
|
raise ValueError(err)
|
|
async with self._client() as client:
|
|
resp = await client.delete(f"/api/notebooks/{notebook_id}")
|
|
resp.raise_for_status()
|
|
return True
|
|
|
|
async def simulate(
|
|
self,
|
|
netlist: str,
|
|
engine: str = "ngspice",
|
|
) -> dict:
|
|
"""Run a stateless simulation without creating a notebook."""
|
|
async with self._client() as client:
|
|
resp = await client.post(
|
|
"/api/simulate",
|
|
json={"netlist": netlist, "engine": engine},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|