MCP/mcltspice
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Decompose remaining tools into 6 domain modules; server.py is now an assembly file

Browse Source 
Move all ~30 remaining tools out of server.py into domain modules that register
on the shared mcp instance:
- simulation_tools  (simulate, simulate_netlist)
- analysis_tools    (waveform extraction, stability, noise, DC op, power, expressions)
- optimize_tools    (optimize_circuit, tune_circuit, plot_waveform)
- batch_tools       (parameter/temperature sweeps, monte_carlo)
- schematic_tools   (generate/read/edit/diff schematic, drc, touchstone)
- netlist_tools     (create_netlist, create_from_template, list_templates)

server.py: 1568 -> 30 lines (imports-for-registration + main). Tool bodies moved
verbatim. 42 tools / 6 resources / 8 prompts register; 473 unit + 8 integration
tests pass; MCP protocol smoke test confirms tools served over stdio.
This commit is contained in:
Ryan Malloy 2026-06-21 21:06:23 -06:00
parent dfb94bfcb9
commit 16cd0e3501
7 changed files with 1600 additions and 1554 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

670
src/mcltspice/analysis_tools.py Normal file
View File

@ -0,0 +1,670 @@
"""Waveform extraction and signal analysis tools (stability, noise, DC op, power, expressions)."""
import csv
import io
import math
import tempfile
from pathlib import Path
import numpy as np
from ._app import mcp
from .noise_analysis import (
compute_noise_metrics,
compute_spot_noise,
compute_total_noise,
)
from .output_format import compact_list
from .power_analysis import compute_efficiency, compute_power_metrics
from .raw_parser import parse_raw_file
from .stability import compute_stability_metrics
from .waveform_expr import WaveformCalculator
from .waveform_math import (
compute_bandwidth,
)
from .waveform_query import analyze_signal, extract_waveform
# ============================================================================
# WAVEFORM & ANALYSIS TOOLS
# ============================================================================
@mcp.tool()
def get_waveform(
raw_file_path: str,
signal_names: list[str],
max_points: int = 1000,
run: int | None = None,
x_min: float | None = None,
x_max: float | None = None,
) -> dict:
"""Extract waveform data from a .raw simulation results file.
For transient analysis, returns time + voltage/current values.
For AC analysis, returns frequency + magnitude(dB)/phase(degrees).
For stepped simulations (.step, .mc, .temp), specify `run` (1-based)
to extract a single run's data. Omit `run` to get all data combined.
Use x_min/x_max to zoom into a frequency or time range of interest
without needing excessive max_points for the full sweep.
AC data uses peak-preserving downsampling that keeps resonance peaks
and notches visible even at low point counts.
Args:
raw_file_path: Path to .raw file from simulation
signal_names: Signal names to extract, e.g. ["V(out)", "I(R1)"]
max_points: Maximum data points (downsampled if needed)
run: Run number (1-based) for stepped simulations (None = all data)
x_min: Minimum x-axis value (frequency Hz or time s) to include
x_max: Maximum x-axis value (frequency Hz or time s) to include
"""
raw = parse_raw_file(raw_file_path)
return extract_waveform(
raw,
signal_names,
max_points=max_points,
run=run,
x_min=x_min,
x_max=x_max,
)
@mcp.tool()
def list_simulation_runs(raw_file_path: str) -> dict:
"""List runs in a stepped simulation (.step, .mc, .temp).
Returns run count and boundary information for multi-run .raw files.
Args:
raw_file_path: Path to .raw file from simulation
"""
raw = parse_raw_file(raw_file_path)
result = {
"is_stepped": raw.is_stepped,
"n_runs": raw.n_runs,
"total_points": raw.points,
"plotname": raw.plotname,
"variables": [{"name": v.name, "type": v.type} for v in raw.variables],
}
if raw.is_stepped and raw.run_boundaries:
runs = []
for i in range(raw.n_runs):
start, end = raw._run_slice(i + 1)
runs.append(
{
"run": i + 1,
"start_index": start,
"end_index": end,
"points": end - start,
}
)
result["runs"] = runs
return result
@mcp.tool()
def analyze_waveform(
raw_file_path: str,
signal_name: str,
analyses: list[str],
settling_tolerance_pct: float = 2.0,
settling_final_value: float | None = None,
rise_low_pct: float = 10.0,
rise_high_pct: float = 90.0,
fft_max_harmonics: int = 50,
thd_n_harmonics: int = 10,
) -> dict:
"""Analyze a signal from simulation results.
Run one or more analyses on a waveform. Available analyses:
- "rms": Root mean square value
- "peak_to_peak": Min, max, peak-to-peak swing, mean
- "settling_time": Time to settle within tolerance of final value
- "rise_time": 10%-90% rise time (configurable)
- "fft": Frequency spectrum via FFT
- "thd": Total Harmonic Distortion
Args:
raw_file_path: Path to .raw file
signal_name: Signal to analyze, e.g. "V(out)"
analyses: List of analysis types to run
settling_tolerance_pct: Tolerance for settling time (default 2%)
settling_final_value: Target value (None = use last sample)
rise_low_pct: Low threshold for rise time (default 10%)
rise_high_pct: High threshold for rise time (default 90%)
fft_max_harmonics: Max harmonics to return in FFT
thd_n_harmonics: Number of harmonics for THD calculation
"""
raw = parse_raw_file(raw_file_path)
return analyze_signal(
raw,
signal_name,
analyses,
settling_tolerance_pct=settling_tolerance_pct,
settling_final_value=settling_final_value,
rise_low_pct=rise_low_pct,
rise_high_pct=rise_high_pct,
fft_max_harmonics=fft_max_harmonics,
thd_n_harmonics=thd_n_harmonics,
)
@mcp.tool()
def measure_bandwidth(
raw_file_path: str,
signal_name: str,
ref_db: float | None = None,
) -> dict:
"""Measure -3dB bandwidth from an AC analysis result.
Computes the frequency range where the signal is within 3dB
of its peak (or a specified reference level).
Args:
raw_file_path: Path to .raw file from AC simulation
signal_name: Signal to measure, e.g. "V(out)"
ref_db: Reference level in dB (None = use peak)
"""
raw = parse_raw_file(raw_file_path)
freq = raw.get_frequency()
signal = raw.get_variable(signal_name)
if freq is None:
return {"error": "Not an AC analysis - no frequency data found"}
if signal is None:
return {"error": f"Signal '{signal_name}' not found"}
# Convert complex signal to magnitude in dB
mag_db = np.array([20 * math.log10(abs(x)) if abs(x) > 0 else -200 for x in signal])
return compute_bandwidth(freq.real, mag_db, ref_db=ref_db)
@mcp.tool()
def export_csv(
raw_file_path: str,
signal_names: list[str] | None = None,
output_path: str | None = None,
max_points: int = 10000,
) -> dict:
"""Export simulation waveform data to CSV format.
Args:
raw_file_path: Path to .raw file
signal_names: Signals to export (None = all)
output_path: Where to save CSV (None = auto-generate in /tmp)
max_points: Maximum rows to export
"""
raw = parse_raw_file(raw_file_path)
# Determine x-axis
x_axis = raw.get_time()
x_name = "time"
if x_axis is None:
x_axis = raw.get_frequency()
x_name = "frequency"
# Select signals
if signal_names is None:
signal_names = [
v.name for v in raw.variables if v.name not in (x_name, "time", "frequency")
]
# Downsample
total = raw.points
step = max(1, total // max_points)
# Build CSV
buf = io.StringIO()
writer = csv.writer(buf)
# Header
if x_axis is not None and np.iscomplexobj(x_axis):
headers = [x_name]
else:
headers = [x_name]
for name in signal_names:
data = raw.get_variable(name)
if data is not None:
if np.iscomplexobj(data):
headers.extend([f"{name}_magnitude_db", f"{name}_phase_deg"])
else:
headers.append(name)
writer.writerow(headers)
# Data rows
indices = range(0, total, step)
for i in indices:
row = []
if x_axis is not None:
row.append(x_axis[i].real if np.iscomplexobj(x_axis) else x_axis[i])
for name in signal_names:
data = raw.get_variable(name)
if data is not None:
if np.iscomplexobj(data):
val = data[i]
row.append(20 * math.log10(abs(val)) if abs(val) > 0 else -200)
row.append(math.degrees(math.atan2(val.imag, val.real)))
else:
row.append(data[i])
writer.writerow(row)
csv_content = buf.getvalue()
# Save to file
if output_path is None:
raw_name = Path(raw_file_path).stem
output_path = str(Path(tempfile.gettempdir()) / f"{raw_name}.csv")
Path(output_path).write_text(csv_content)
return {
"output_path": output_path,
"rows": len(indices),
"columns": headers,
}
# ============================================================================
# STABILITY ANALYSIS TOOLS
# ============================================================================
@mcp.tool()
def analyze_stability(
raw_file_path: str,
signal_name: str,
) -> dict:
"""Measure gain margin and phase margin from AC loop gain data.
Computes Bode plot (magnitude + phase) and finds the crossover
frequencies where gain = 0 dB and phase = -180 degrees.
Args:
raw_file_path: Path to .raw file from AC simulation
signal_name: Loop gain signal, e.g. "V(out)" or "V(loop_gain)"
"""
raw = parse_raw_file(raw_file_path)
freq = raw.get_frequency()
signal = raw.get_variable(signal_name)
if freq is None:
return {"error": "Not an AC analysis - no frequency data found"}
if signal is None:
return {
"error": f"Signal '{signal_name}' not found. Available: "
f"{[v.name for v in raw.variables]}"
}
return compute_stability_metrics(freq.real, signal)
# ============================================================================
# NOISE ANALYSIS TOOLS
# ============================================================================
@mcp.tool()
def analyze_noise(
raw_file_path: str,
noise_signal: str = "onoise",
source_resistance: float = 50.0,
) -> dict:
"""Comprehensive noise analysis from a .noise simulation.
Returns spectral density, spot noise at standard frequencies (10Hz,
100Hz, 1kHz, 10kHz, 100kHz), total integrated RMS noise, noise figure,
and 1/f corner frequency estimate.
Run a simulation with .noise directive first, e.g.:
.noise V(out) V1 dec 100 1 1meg
Args:
raw_file_path: Path to .raw file from .noise simulation
noise_signal: Which noise variable to analyze ("onoise" for
output-referred or "inoise" for input-referred)
source_resistance: Source impedance in ohms for noise figure (default 50)
"""
raw = parse_raw_file(raw_file_path)
freq = raw.get_frequency()
signal = raw.get_variable(noise_signal)
if freq is None:
return {"error": "No frequency data found. Is this a .noise simulation?"}
if signal is None:
available = [v.name for v in raw.variables]
return {
"error": f"Signal '{noise_signal}' not found. Available: {available}",
}
return compute_noise_metrics(freq.real, signal, source_resistance)
@mcp.tool()
def get_spot_noise(
raw_file_path: str,
target_freq: float,
noise_signal: str = "onoise",
) -> dict:
"""Get noise spectral density at a specific frequency.
Interpolates between data points to estimate the noise density
at the requested frequency.
Args:
raw_file_path: Path to .raw file from .noise simulation
target_freq: Frequency in Hz to measure noise at
noise_signal: "onoise" (output-referred) or "inoise" (input-referred)
"""
raw = parse_raw_file(raw_file_path)
freq = raw.get_frequency()
signal = raw.get_variable(noise_signal)
if freq is None:
return {"error": "No frequency data found"}
if signal is None:
return {"error": f"Signal '{noise_signal}' not found"}
return compute_spot_noise(freq.real, signal, target_freq)
@mcp.tool()
def get_total_noise(
raw_file_path: str,
noise_signal: str = "onoise",
f_low: float | None = None,
f_high: float | None = None,
) -> dict:
"""Integrate noise over a frequency band to get total RMS noise.
Computes total_rms = sqrt(integral(|noise|^2 * df)) over the
specified frequency range.
Args:
raw_file_path: Path to .raw file from .noise simulation
noise_signal: "onoise" or "inoise"
f_low: Lower frequency bound in Hz (default: data minimum)
f_high: Upper frequency bound in Hz (default: data maximum)
"""
raw = parse_raw_file(raw_file_path)
freq = raw.get_frequency()
signal = raw.get_variable(noise_signal)
if freq is None:
return {"error": "No frequency data found"}
if signal is None:
return {"error": f"Signal '{noise_signal}' not found"}
return compute_total_noise(freq.real, signal, f_low, f_high)
# ============================================================================
# DC OPERATING POINT & TRANSFER FUNCTION TOOLS
# ============================================================================
@mcp.tool()
def get_operating_point(raw_file_path: str) -> dict:
"""Extract DC operating point results from a .raw file.
The .op analysis computes all node voltages and branch currents
at the DC bias point, stored as a single data point in the .raw file.
Run a simulation with .op directive first, then pass the .raw file.
Args:
raw_file_path: Path to .raw file from simulation
"""
raw = parse_raw_file(raw_file_path)
if raw.plotname and "operating point" not in raw.plotname.lower():
return {
"error": f"Not an operating point analysis (plotname: '{raw.plotname}'). "
"Ensure the simulation uses a .op directive."
}
# Extract single-point values, separating voltages from currents
voltages = {}
currents = {}
other = {}
for v in raw.variables:
data = raw.get_variable(v.name)
if data is None or len(data) == 0:
continue
val = float(data[0].real) if hasattr(data[0], "real") else float(data[0])
if v.name.lower().startswith("v("):
voltages[v.name] = val
elif v.name.lower().startswith("i(") or v.name.lower().startswith("ix("):
currents[v.name] = val
else:
other[v.name] = val
return {
"voltages": voltages,
"currents": currents,
"device_params": other,
"total_entries": len(voltages) + len(currents) + len(other),
}
@mcp.tool()
def get_transfer_function(raw_file_path: str) -> dict:
"""Extract .tf (transfer function) results from a .raw file.
The .tf analysis computes:
- Transfer function (gain or transresistance)
- Input impedance at the source
- Output impedance at the output node
Run a simulation with .tf directive first (e.g., ".tf V(out) V1"),
then pass the .raw file.
Args:
raw_file_path: Path to .raw file from simulation
"""
raw = parse_raw_file(raw_file_path)
if raw.plotname and "transfer function" not in raw.plotname.lower():
return {
"error": f"Not a transfer function analysis (plotname: '{raw.plotname}'). "
"Ensure the simulation uses a .tf directive, e.g., '.tf V(out) V1'."
}
result: dict = {}
for v in raw.variables:
data = raw.get_variable(v.name)
if data is None or len(data) == 0:
continue
val = float(data[0].real) if hasattr(data[0], "real") else float(data[0])
name_lower = v.name.lower()
if "transfer_function" in name_lower:
result["transfer_function"] = val
elif "output_impedance" in name_lower:
result["output_impedance_ohms"] = val
elif "input_impedance" in name_lower:
result["input_impedance_ohms"] = val
# Always include raw data with original name
result.setdefault("raw_data", {})[v.name] = val
if not result:
return {
"error": "No transfer function data found in .raw file.",
"plotname": raw.plotname,
"variables": [v.name for v in raw.variables],
}
return result
# ============================================================================
# POWER ANALYSIS TOOLS
# ============================================================================
@mcp.tool()
def analyze_power(
raw_file_path: str,
voltage_signal: str,
current_signal: str,
) -> dict:
"""Compute power metrics from voltage and current waveforms.
Returns average power, RMS power, peak power, and power factor.
Args:
raw_file_path: Path to .raw file from transient simulation
voltage_signal: Voltage signal name, e.g. "V(out)"
current_signal: Current signal name, e.g. "I(R1)"
"""
raw = parse_raw_file(raw_file_path)
time = raw.get_time()
voltage = raw.get_variable(voltage_signal)
current = raw.get_variable(current_signal)
if time is None:
return {"error": "Not a transient analysis - no time data found"}
if voltage is None:
return {"error": f"Voltage signal '{voltage_signal}' not found"}
if current is None:
return {"error": f"Current signal '{current_signal}' not found"}
return compute_power_metrics(time, voltage, current)
@mcp.tool()
def compute_efficiency_tool(
raw_file_path: str,
input_voltage_signal: str,
input_current_signal: str,
output_voltage_signal: str,
output_current_signal: str,
) -> dict:
"""Compute power conversion efficiency.
Compares input power to output power for regulators, converters, etc.
Args:
raw_file_path: Path to .raw file from transient simulation
input_voltage_signal: Input voltage, e.g. "V(vin)"
input_current_signal: Input current, e.g. "I(Vin)"
output_voltage_signal: Output voltage, e.g. "V(out)"
output_current_signal: Output current, e.g. "I(Rload)"
"""
raw = parse_raw_file(raw_file_path)
time = raw.get_time()
if time is None:
return {"error": "Not a transient analysis"}
v_in = raw.get_variable(input_voltage_signal)
i_in = raw.get_variable(input_current_signal)
v_out = raw.get_variable(output_voltage_signal)
i_out = raw.get_variable(output_current_signal)
for name, sig in [
(input_voltage_signal, v_in),
(input_current_signal, i_in),
(output_voltage_signal, v_out),
(output_current_signal, i_out),
]:
if sig is None:
return {"error": f"Signal '{name}' not found"}
return compute_efficiency(time, v_in, i_in, v_out, i_out)
# ============================================================================
# WAVEFORM EXPRESSION TOOLS
# ============================================================================
@mcp.tool()
def evaluate_waveform_expression(
raw_file_path: str,
expression: str,
max_points: int = 1000,
x_min: float | None = None,
x_max: float | None = None,
) -> dict:
"""Evaluate a math expression on simulation waveforms.
Supports: +, -, *, /, abs(), sqrt(), log10(), dB()
Signal names reference variables from the .raw file.
Examples:
"V(out) * I(R1)" - instantaneous power
"V(out) / V(in)" - voltage gain
"dB(V(out))" - magnitude in dB
Args:
raw_file_path: Path to .raw file
expression: Math expression using signal names
max_points: Maximum data points to return
x_min: Minimum x-axis value (frequency Hz or time s) to include
x_max: Maximum x-axis value (frequency Hz or time s) to include
"""
raw = parse_raw_file(raw_file_path)
calc = WaveformCalculator(raw)
try:
result = calc.calc(expression)
except ValueError as e:
return {"error": str(e), "available_signals": calc.available_signals()}
# Get x-axis
x_axis = raw.get_time()
x_name = "time"
if x_axis is None:
x_axis = raw.get_frequency()
x_name = "frequency"
total = len(result)
# Apply x-axis range filter
if x_axis is not None and (x_min is not None or x_max is not None):
x_real = x_axis.real if np.iscomplexobj(x_axis) else x_axis
mask = np.ones(len(x_real), dtype=bool)
if x_min is not None:
mask &= x_real >= x_min
if x_max is not None:
mask &= x_real <= x_max
filtered = np.where(mask)[0]
if len(filtered) == 0:
return {"error": f"No data points in {x_name} range [{x_min}, {x_max}]"}
else:
filtered = np.arange(total)
step = max(1, len(filtered) // max_points)
sample_indices = filtered[::step]
response = {
"expression": expression,
"total_points": total,
"returned_points": len(sample_indices),
}
if x_axis is not None:
x_real = x_axis.real if np.iscomplexobj(x_axis) else x_axis
response["x_axis_name"] = x_name
response["x_axis_data"] = compact_list(x_real[sample_indices], sig_figs=6)
response["values"] = compact_list(result[sample_indices], sig_figs=6)
response["available_signals"] = calc.available_signals()
return response

121
src/mcltspice/batch_tools.py Normal file
View File

@ -0,0 +1,121 @@
"""Batch simulation tools: parameter/temperature sweeps and Monte Carlo."""
import numpy as np
from ._app import mcp
from .batch import (
run_monte_carlo,
run_parameter_sweep,
run_temperature_sweep,
)
# ============================================================================
# BATCH SIMULATION TOOLS
# ============================================================================
@mcp.tool()
async def parameter_sweep(
netlist_text: str,
param_name: str,
start: float,
stop: float,
num_points: int = 10,
timeout_seconds: float = 300,
) -> dict:
"""Sweep a parameter across a range of values.
Runs multiple simulations, substituting the parameter value each time.
The netlist should contain a .param directive for the parameter.
Args:
netlist_text: Netlist with .param directive
param_name: Parameter to sweep (e.g., "Rval")
start: Start value
stop: Stop value
num_points: Number of sweep points
timeout_seconds: Per-simulation timeout
"""
values = np.linspace(start, stop, num_points).tolist()
result = await run_parameter_sweep(
netlist_text,
param_name,
values,
timeout=timeout_seconds,
)
return {
"success_count": result.success_count,
"failure_count": result.failure_count,
"total_elapsed": result.total_elapsed,
"parameter_values": result.parameter_values,
"raw_files": [str(r.raw_file) if r.raw_file else None for r in result.results],
}
@mcp.tool()
async def temperature_sweep(
netlist_text: str,
temperatures: list[float],
timeout_seconds: float = 300,
) -> dict:
"""Run simulations at different temperatures.
Args:
netlist_text: Netlist text
temperatures: List of temperatures in degrees C
timeout_seconds: Per-simulation timeout
"""
result = await run_temperature_sweep(
netlist_text,
temperatures,
timeout=timeout_seconds,
)
return {
"success_count": result.success_count,
"failure_count": result.failure_count,
"total_elapsed": result.total_elapsed,
"parameter_values": result.parameter_values,
"raw_files": [str(r.raw_file) if r.raw_file else None for r in result.results],
}
@mcp.tool()
async def monte_carlo(
netlist_text: str,
n_runs: int,
tolerances: dict[str, float],
timeout_seconds: float = 300,
seed: int | None = None,
) -> dict:
"""Run Monte Carlo analysis with component tolerances.
Randomly varies component values within tolerance using a normal
distribution, then runs simulations for each variant.
Args:
netlist_text: Netlist text
n_runs: Number of Monte Carlo iterations
tolerances: Component tolerances, e.g. {"R1": 0.05} for 5%
timeout_seconds: Per-simulation timeout
seed: Optional RNG seed for reproducibility
"""
result = await run_monte_carlo(
netlist_text,
n_runs,
tolerances,
timeout=timeout_seconds,
seed=seed,
)
return {
"success_count": result.success_count,
"failure_count": result.failure_count,
"total_elapsed": result.total_elapsed,
"parameter_values": result.parameter_values,
"raw_files": [str(r.raw_file) if r.raw_file else None for r in result.results],
}

175
src/mcltspice/netlist_tools.py Normal file
View File

@ -0,0 +1,175 @@
"""Netlist builder and circuit-template tools."""
import tempfile
from pathlib import Path
from ._app import mcp
from .netlist import Netlist
from .templates import (
NETLIST_TEMPLATES,
)
# ============================================================================
# NETLIST BUILDER TOOLS
# ============================================================================
@mcp.tool()
def create_netlist(
title: str,
components: list[dict],
directives: list[str],
output_path: str | None = None,
) -> dict:
"""Create a SPICE netlist programmatically and save to a .cir file.
Build circuits from scratch without needing a graphical schematic.
The created .cir file can be simulated with simulate_netlist.
Args:
title: Circuit title/description
components: List of component dicts, each with:
- name: Component name (R1, C1, V1, M1, X1, etc.)
- nodes: List of node names (use "0" for ground)
- value: Value or model name
- params: Optional extra parameters string
directives: List of SPICE directives, e.g.:
[".tran 10m", ".ac dec 100 1 1meg",
".meas tran vmax MAX V(out)"]
output_path: Where to save .cir file (None = auto in /tmp)
Example components:
[
{"name": "V1", "nodes": ["in", "0"], "value": "AC 1"},
{"name": "R1", "nodes": ["in", "out"], "value": "10k"},
{"name": "C1", "nodes": ["out", "0"], "value": "100n"}
]
"""
nl = Netlist(title=title)
for comp in components:
nl.add_component(
name=comp["name"],
nodes=comp["nodes"],
value=comp["value"],
params=comp.get("params", ""),
)
for directive in directives:
nl.add_directive(directive)
# Determine output path
if output_path is None:
safe_title = "".join(c if c.isalnum() else "_" for c in title)[:30]
output_path = str(Path(tempfile.gettempdir()) / f"{safe_title}.cir")
saved = nl.save(output_path)
return {
"success": True,
"output_path": str(saved),
"netlist_preview": nl.render(),
"component_count": len(nl.components),
}
# ============================================================================
# CIRCUIT TEMPLATE TOOLS
# ============================================================================
@mcp.tool()
def create_from_template(
template_name: str,
params: dict[str, str] | None = None,
output_path: str | None = None,
) -> dict:
"""Create a circuit netlist from a pre-built template.
Available templates:
- voltage_divider: params {v_in, r1, r2, sim_type}
- rc_lowpass: params {r, c, f_start, f_stop}
- inverting_amplifier: params {r_in, r_f, opamp_model}
- non_inverting_amplifier: params {r_in, r_f, opamp_model}
- differential_amplifier: params {r1, r2, r3, r4, opamp_model}
- common_emitter_amplifier: params {rc, rb1, rb2, re, cc1, cc2, ce, vcc, bjt_model}
- buck_converter: params {ind, c_out, r_load, v_in, duty_cycle, freq, mosfet_model, diode_model}
- ldo_regulator: params {opamp_model, r1, r2, pass_transistor, v_in, v_ref}
- colpitts_oscillator: params {ind, c1, c2, rb, rc, re, vcc, bjt_model}
- h_bridge: params {v_supply, r_load, mosfet_model}
- sallen_key_lowpass: params {r1, r2, c1, c2, opamp_model}
- boost_converter: params {ind, c_out, r_load, v_in, duty_cycle, freq, mosfet_model, diode_model}
- instrumentation_amplifier: params {r1, r2, r3, r_gain}
- current_mirror: params {r_ref, r_load, vcc, bjt_model}
- transimpedance_amplifier: params {rf, cf, i_source}
All parameter values are optional -- defaults are used if omitted.
Args:
template_name: Template name from the list above
params: Optional dict of parameter overrides (all values as strings)
output_path: Where to save .cir file (None = auto in /tmp)
"""
template = NETLIST_TEMPLATES.get(template_name)
if template is None:
return {
"error": f"Unknown template '{template_name}'",
"available_templates": [
{"name": k, "description": v["description"], "params": v["params"]}
for k, v in NETLIST_TEMPLATES.items()
],
}
# Build kwargs from params, converting duty_cycle to float for buck_converter
kwargs: dict = {}
if params:
for k, v in params.items():
if k not in template["params"]:
return {
"error": f"Unknown parameter '{k}' for template '{template_name}'",
"valid_params": template["params"],
}
# duty_cycle needs to be float, not string
if k == "duty_cycle":
kwargs[k] = float(v)
else:
kwargs[k] = v
nl = template["func"](**kwargs)
if output_path is None:
output_path = str(Path(tempfile.gettempdir()) / f"{template_name}.cir")
saved = nl.save(output_path)
return {
"success": True,
"template": template_name,
"description": template["description"],
"output_path": str(saved),
"netlist_preview": nl.render(),
"component_count": len(nl.components),
"params_used": {**template["params"], **(params or {})},
}
@mcp.tool()
def list_templates() -> dict:
"""List all available circuit templates with their parameters and defaults.
Returns template names, descriptions, and the parameters each accepts
with their default values.
"""
return {
"templates": [
{
"name": name,
"description": info["description"],
"params": info["params"],
}
for name, info in NETLIST_TEMPLATES.items()
],
"total_count": len(NETLIST_TEMPLATES),
}

296
src/mcltspice/optimize_tools.py Normal file
View File

@ -0,0 +1,296 @@
"""Optimization, tuning, and plotting tools."""
import tempfile
from pathlib import Path
from ._app import mcp
from .optimizer import (
ComponentRange,
OptimizationTarget,
format_engineering,
optimize_component_values,
)
from .plotting import build_waveform_svg
from .raw_parser import parse_raw_file
from .runner import run_netlist, run_simulation
from .templates import (
all_template_names,
resolve_template,
)
from .tuning import (
UnknownParamError,
build_effective_params,
evaluate_targets,
extract_metrics,
make_suggestions,
)
# ============================================================================
# OPTIMIZER TOOLS
# ============================================================================
@mcp.tool()
async def optimize_circuit(
netlist_template: str,
targets: list[dict],
component_ranges: list[dict],
max_iterations: int = 20,
) -> dict:
"""Automatically optimize component values to hit target specifications.
Runs real LTspice simulations in a loop, adjusting component values
using binary search (single component) or coordinate descent (multiple).
Args:
netlist_template: Netlist text with {ComponentName} placeholders
(e.g., {R1}, {C1}) that get substituted each iteration.
targets: List of target specs, each with:
- signal_name: Signal to measure (e.g., "V(out)")
- metric: One of "bandwidth_hz", "rms", "peak_to_peak",
"settling_time", "gain_db", "phase_margin_deg"
- target_value: Desired value
- weight: Importance weight (default 1.0)
component_ranges: List of tunable components, each with:
- component_name: Name matching {placeholder} (e.g., "R1")
- min_value: Minimum value in base units
- max_value: Maximum value in base units
- preferred_series: Optional "E12", "E24", or "E96" for snapping
max_iterations: Max simulation iterations (default 20)
"""
opt_targets = [
OptimizationTarget(
signal_name=t["signal_name"],
metric=t["metric"],
target_value=t["target_value"],
weight=t.get("weight", 1.0),
)
for t in targets
]
opt_ranges = [
ComponentRange(
component_name=r["component_name"],
min_value=r["min_value"],
max_value=r["max_value"],
preferred_series=r.get("preferred_series"),
)
for r in component_ranges
]
result = await optimize_component_values(
netlist_template,
opt_targets,
opt_ranges,
max_iterations,
)
return {
"best_values": {k: format_engineering(v) for k, v in result.best_values.items()},
"best_values_raw": result.best_values,
"best_cost": result.best_cost,
"iterations": result.iterations,
"targets_met": result.targets_met,
"final_metrics": result.final_metrics,
"history_length": len(result.history),
}
# ============================================================================
# CIRCUIT TUNING TOOL
# ============================================================================
@mcp.tool()
async def tune_circuit(
template: str,
params: dict[str, str] | None = None,
targets: dict[str, str] | None = None,
signal: str = "V(out)",
) -> dict:
"""Measure circuit performance and suggest parameter adjustments.
Single-shot workflow: generates a circuit from a template, simulates it,
measures key metrics, compares against targets, and suggests what to change.
Call this tool repeatedly with adjusted params until targets are met.
Args:
template: Template name (from list_templates). Works with both
netlist templates (NETLIST_TEMPLATES) and schematic templates (ASC_TEMPLATES).
params: Component value overrides (e.g., {"r": "2.2k", "c": "47n"}).
Use list_templates to see available parameters.
targets: Performance targets to check against. Each key is a metric name,
value is a comparison string like ">5000" or "<0.1" or "~1000".
Supported metrics: bandwidth_hz, gain_db, rms, peak_to_peak,
settling_time_s, dc_value, fundamental_freq_hz.
Example: {"bandwidth_hz": ">5000", "gain_db": ">20"}
signal: Signal name to measure (default: "V(out)")
Returns:
Dict with metrics, target comparison, and tuning suggestions.
"""
# --- 1. Look up template (netlist registry first) ---
found = resolve_template(template)
if found is None:
names = all_template_names()
return {"error": f"Unknown template '{template}'. Available: {', '.join(names)}"}
tmpl, is_netlist = found
# --- 2. Build effective params ---
try:
effective_params = build_effective_params(tmpl["params"], params)
except UnknownParamError as e:
return {
"error": f"Unknown param '{e.key}' for {template}",
"valid_params": e.valid_params,
}
# --- 3. Generate and simulate ---
import tempfile
from pathlib import Path
try:
# Build kwargs (handle duty_cycle float conversion)
call_kwargs: dict = {}
for k, v in effective_params.items():
if k == "duty_cycle":
call_kwargs[k] = float(v)
else:
call_kwargs[k] = v
if is_netlist:
nl = tmpl["func"](**call_kwargs)
out_path = Path(tempfile.gettempdir()) / f"tune_{template}.cir"
nl.save(out_path)
result = await run_netlist(out_path)
else:
sch = tmpl["func"](**call_kwargs)
out_path = Path(tempfile.gettempdir()) / f"tune_{template}.asc"
sch.save(out_path)
result = await run_simulation(out_path)
if not result.success:
return {
"error": "Simulation failed",
"detail": result.error or result.stderr,
"params_used": effective_params,
}
except Exception as e:
return {"error": f"Generation/simulation failed: {e}", "params_used": effective_params}
# --- 4. Extract metrics ---
raw = result.raw_data
if raw is None:
return {"error": "No raw data from simulation", "params_used": effective_params}
extracted = extract_metrics(raw, signal)
if extracted is None:
return {
"error": f"Signal '{signal}' not found",
"available_signals": [v.name for v in raw.variables],
"params_used": effective_params,
}
metrics, is_ac = extracted
# --- 5 + 6. Compare against targets and suggest adjustments ---
targets_met, target_results = evaluate_targets(metrics, targets)
suggestions = make_suggestions(target_results, targets_met)
return {
"template": template,
"params_used": effective_params,
"metrics": metrics,
"targets": target_results if targets else {},
"targets_met": targets_met,
"suggestions": suggestions,
"signal": signal,
"analysis_type": "ac" if is_ac else "transient",
}
# ============================================================================
# WAVEFORM PLOTTING TOOL
# ============================================================================
@mcp.tool()
async def plot_waveform(
raw_file: str,
signal: str = "V(out)",
signals: list[str] | None = None,
plot_type: str = "auto",
output_path: str | None = None,
max_points: int = 10000,
x_min: float | None = None,
x_max: float | None = None,
y_min: float | None = None,
y_max: float | None = None,
width: int = 800,
height: int | None = None,
title: str | None = None,
) -> dict:
"""Generate an SVG plot from simulation results.
Parses a .raw file and creates a publication-quality SVG waveform plot.
Supports time-domain, Bode (frequency response), and FFT spectrum plots.
Args:
raw_file: Path to the LTspice .raw binary file
signal: Signal name to plot (e.g. "V(out)", "I(R1)")
signals: Multiple signal names to overlay (e.g. ["V(tx)", "V(rx)"]). Overrides signal when set. Time-domain only.
plot_type: "auto" (detect from data), "time", "bode", or "spectrum"
output_path: Where to save SVG file (None = auto in /tmp)
max_points: Maximum data points to plot (stride-sampled if exceeded)
x_min: Left X-axis bound (seconds for time, Hz for frequency). None = auto
x_max: Right X-axis bound. None = auto
y_min: Lower Y-axis bound (volts for time, dB for bode/spectrum). None = auto
y_max: Upper Y-axis bound. None = auto
width: SVG width in pixels
height: SVG height in pixels (None = 500 for bode, 400 otherwise)
title: Plot title (None = auto-generated from signal name and plot type)
"""
raw_path = Path(raw_file)
if not raw_path.exists():
return {"error": f"Raw file not found: {raw_file}"}
try:
raw_data = parse_raw_file(str(raw_path))
except Exception as e:
return {"error": f"Failed to parse raw file: {e}"}
# Build signal list (signals overrides signal when set)
signal_list = signals if signals else [signal]
result = build_waveform_svg(
raw_data,
signal_list,
plot_type=plot_type,
max_points=max_points,
x_min=x_min,
x_max=x_max,
y_min=y_min,
y_max=y_max,
width=width,
height=height,
title=title,
)
if "error" in result:
return result
# Save SVG
if output_path is None:
out = Path(tempfile.mktemp(suffix=".svg", prefix="ltspice_plot_"))
else:
out = Path(output_path)
out.write_text(result["svg"])
return {
"svg_path": str(out),
"plot_type": result["plot_type"],
"signal": result["signal"],
"points": result["points"],
"dimensions": result["dimensions"],
}

224
src/mcltspice/schematic_tools.py Normal file
View File

@ -0,0 +1,224 @@
"""Schematic generation, Touchstone parsing, and schematic editing tools."""
import tempfile
from pathlib import Path
from ._app import mcp
from .diff import diff_schematics as _diff_schematics
from .drc import run_drc as _run_drc
from .schematic import modify_component_value, parse_schematic
from .templates import (
ASC_TEMPLATES,
)
from .touchstone import parse_touchstone, s_param_to_db
# ============================================================================
# SCHEMATIC GENERATION TOOLS
# ============================================================================
@mcp.tool()
def generate_schematic(
template: str,
params: dict[str, str] | None = None,
output_path: str | None = None,
) -> dict:
"""Generate an LTspice .asc graphical schematic file from a template.
Creates a ready-to-simulate .asc file with proper component placement,
wire routing, and simulation directives.
Available templates (use list_templates for full details):
- rc_lowpass, voltage_divider, inverting_amp, non_inverting_amp,
common_emitter_amp, colpitts_oscillator, differential_amp,
buck_converter, ldo_regulator, h_bridge,
sallen_key_lowpass, boost_converter, instrumentation_amp,
current_mirror, transimpedance_amp
Args:
template: Template name (see list above)
params: Override default parameters as key-value pairs.
Use list_templates to see available parameters for each template.
output_path: Where to save the .asc file (None = auto in /tmp)
"""
if template not in ASC_TEMPLATES:
names = ", ".join(sorted(ASC_TEMPLATES))
return {"error": f"Unknown template '{template}'. Available: {names}"}
entry = ASC_TEMPLATES[template]
call_params: dict[str, str | float] = {}
if params:
for k, v in params.items():
if k not in entry["params"]:
valid = ", ".join(sorted(entry["params"]))
return {
"error": f"Unknown param '{k}' for {template}. Valid: {valid}"
}
# duty_cycle needs float conversion for buck_converter
if k == "duty_cycle":
call_params[k] = float(v)
else:
call_params[k] = v
sch = entry["func"](**call_params)
if output_path is None:
output_path = str(Path(tempfile.gettempdir()) / f"{template}.asc")
saved = sch.save(output_path)
return {
"success": True,
"output_path": str(saved),
"template": template,
"schematic_preview": sch.render()[:500],
}
# ============================================================================
# TOUCHSTONE / S-PARAMETER TOOLS
# ============================================================================
@mcp.tool()
def read_touchstone(file_path: str) -> dict:
"""Parse a Touchstone (.s1p, .s2p, .snp) S-parameter file.
Returns S-parameter data, frequency points, and port information.
Args:
file_path: Path to Touchstone file
"""
try:
data = parse_touchstone(file_path)
except (ValueError, FileNotFoundError) as e:
return {"error": str(e)}
# Convert S-parameter data to a more digestible format
s_params = {}
for i in range(data.n_ports):
for j in range(data.n_ports):
key = f"S{i + 1}{j + 1}"
s_data = data.data[:, i, j]
s_params[key] = {
"magnitude_db": s_param_to_db(s_data).tolist(),
}
return {
"filename": data.filename,
"n_ports": data.n_ports,
"n_frequencies": len(data.frequencies),
"freq_range_hz": [float(data.frequencies[0]), float(data.frequencies[-1])],
"reference_impedance": data.reference_impedance,
"s_parameters": s_params,
"comments": data.comments[:5], # First 5 comment lines
}
# ============================================================================
# SCHEMATIC TOOLS
# ============================================================================
@mcp.tool()
def read_schematic(schematic_path: str) -> dict:
"""Read and parse an LTspice schematic file.
Returns component list, net names, and SPICE directives.
Args:
schematic_path: Path to .asc schematic file
"""
sch = parse_schematic(schematic_path)
return {
"version": sch.version,
"components": [
{
"name": c.name,
"symbol": c.symbol,
"value": c.value,
"x": c.x,
"y": c.y,
"attributes": c.attributes,
}
for c in sch.components
],
"nets": [f.name for f in sch.flags],
"directives": sch.get_spice_directives(),
"wire_count": len(sch.wires),
}
@mcp.tool()
def edit_component(
schematic_path: str,
component_name: str,
new_value: str,
output_path: str | None = None,
) -> dict:
"""Modify a component's value in a schematic.
Args:
schematic_path: Path to .asc schematic file
component_name: Instance name like "R1", "C2", "M1"
new_value: New value string, e.g., "10k", "100n", "2N7000"
output_path: Where to save (None = overwrite original)
"""
try:
sch = modify_component_value(
schematic_path,
component_name,
new_value,
output_path,
)
comp = sch.get_component(component_name)
return {
"success": True,
"component": component_name,
"new_value": new_value,
"output_path": output_path or schematic_path,
"symbol": comp.symbol if comp else None,
}
except ValueError as e:
return {"success": False, "error": str(e)}
@mcp.tool()
def diff_schematics(
schematic_a: str,
schematic_b: str,
) -> dict:
"""Compare two schematics and show what changed.
Reports component additions, removals, value changes,
directive changes, and wire/net topology differences.
Args:
schematic_a: Path to "before" .asc file
schematic_b: Path to "after" .asc file
"""
diff = _diff_schematics(schematic_a, schematic_b)
return diff.to_dict()
@mcp.tool()
def run_drc(schematic_path: str) -> dict:
"""Run design rule checks on a schematic.
Checks for common issues:
- Missing ground connection
- Floating nodes
- Missing simulation directive
- Voltage source loops
- Missing component values
- Duplicate component names
- Unconnected components
Args:
schematic_path: Path to .asc schematic file
"""
result = _run_drc(schematic_path)
return result.to_dict()

1571
src/mcltspice/server.py
View File

File diff suppressed because it is too large Load Diff

97
src/mcltspice/simulation_tools.py Normal file
View File

@ -0,0 +1,97 @@
"""Simulation tools: run schematics and netlists. Registers on the shared mcp."""
from ._app import mcp
from .log_parser import parse_log
from .runner import run_netlist, run_simulation
# ============================================================================
# SIMULATION TOOLS
# ============================================================================
@mcp.tool()
async def simulate(
schematic_path: str,
timeout_seconds: float = 300,
) -> dict:
"""Run an LTspice simulation on a schematic file.
Executes any simulation directives (.tran, .ac, .dc, .op, etc.)
found in the schematic. Returns available signal names and
the path to the .raw file for waveform extraction.
Args:
schematic_path: Absolute path to .asc schematic file
timeout_seconds: Maximum time to wait for simulation (default 5 min)
"""
result = await run_simulation(
schematic_path,
timeout=timeout_seconds,
parse_results=True,
)
response = {
"success": result.success,
"elapsed_seconds": result.elapsed_seconds,
"error": result.error,
}
if result.raw_data:
response["variables"] = [
{"name": v.name, "type": v.type} for v in result.raw_data.variables
]
response["points"] = result.raw_data.points
response["plotname"] = result.raw_data.plotname
response["raw_file"] = str(result.raw_file) if result.raw_file else None
if result.log_file and result.log_file.exists():
log = parse_log(result.log_file)
if log.measurements:
response["measurements"] = log.get_all_measurements()
if log.errors:
response["log_errors"] = log.errors
return response
@mcp.tool()
async def simulate_netlist(
netlist_path: str,
timeout_seconds: float = 300,
) -> dict:
"""Run an LTspice simulation on a netlist file (.cir or .net).
Args:
netlist_path: Absolute path to .cir or .net netlist file
timeout_seconds: Maximum time to wait for simulation
"""
result = await run_netlist(
netlist_path,
timeout=timeout_seconds,
parse_results=True,
)
response = {
"success": result.success,
"elapsed_seconds": result.elapsed_seconds,
"error": result.error,
}
if result.raw_data:
response["variables"] = [
{"name": v.name, "type": v.type} for v in result.raw_data.variables
]
response["points"] = result.raw_data.points
response["raw_file"] = str(result.raw_file) if result.raw_file else None
if result.log_file and result.log_file.exists():
log = parse_log(result.log_file)
if log.measurements:
response["measurements"] = log.get_all_measurements()
if log.errors:
response["log_errors"] = log.errors
return response