diff --git a/src/mcltspice/analysis_tools.py b/src/mcltspice/analysis_tools.py new file mode 100644 index 0000000..698771c --- /dev/null +++ b/src/mcltspice/analysis_tools.py @@ -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 + + diff --git a/src/mcltspice/batch_tools.py b/src/mcltspice/batch_tools.py new file mode 100644 index 0000000..77b54f1 --- /dev/null +++ b/src/mcltspice/batch_tools.py @@ -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], + } + + diff --git a/src/mcltspice/netlist_tools.py b/src/mcltspice/netlist_tools.py new file mode 100644 index 0000000..7655893 --- /dev/null +++ b/src/mcltspice/netlist_tools.py @@ -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), + } + + diff --git a/src/mcltspice/optimize_tools.py b/src/mcltspice/optimize_tools.py new file mode 100644 index 0000000..dac01cb --- /dev/null +++ b/src/mcltspice/optimize_tools.py @@ -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"], + } + + diff --git a/src/mcltspice/schematic_tools.py b/src/mcltspice/schematic_tools.py new file mode 100644 index 0000000..b3a0781 --- /dev/null +++ b/src/mcltspice/schematic_tools.py @@ -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() + + diff --git a/src/mcltspice/server.py b/src/mcltspice/server.py index 7ca8893..8e63e7c 100644 --- a/src/mcltspice/server.py +++ b/src/mcltspice/server.py @@ -1,1562 +1,25 @@ """FastMCP server for LTspice circuit simulation automation. -This server provides tools for: -- Running SPICE simulations on schematics and netlists -- Extracting and analyzing waveform data -- Creating circuits programmatically -- Modifying schematic components -- Browsing component libraries, models, and examples -- Design rule checks and circuit comparison +This is the assembly point. The FastMCP instance lives in _app.py; each domain +module below defines its tools/resources/prompts with `@mcp.tool()` etc. and is +imported here purely so those decorators run and register onto the shared mcp +instance before main() serves it. """ -import csv -import io -import math -import tempfile -from pathlib import Path - -import numpy as np - -# Domain modules register their tools/resources/prompts on the shared mcp -# instance at import time. Imported for that side effect only. -from . import library_tools, prompts, resources, spicebook_tools # noqa: F401,E402 +# Imported for registration side effects only (hence noqa: F401). +from . import ( # noqa: F401 + analysis_tools, + batch_tools, + library_tools, + netlist_tools, + optimize_tools, + prompts, + resources, + schematic_tools, + simulation_tools, + spicebook_tools, +) from ._app import mcp -from .batch import ( - run_monte_carlo, - run_parameter_sweep, - run_temperature_sweep, -) -from .diff import diff_schematics as _diff_schematics -from .drc import run_drc as _run_drc -from .log_parser import parse_log -from .netlist import Netlist -from .noise_analysis import ( - compute_noise_metrics, - compute_spot_noise, - compute_total_noise, -) -from .optimizer import ( - ComponentRange, - OptimizationTarget, - format_engineering, - optimize_component_values, -) -from .output_format import compact_list -from .plotting import build_waveform_svg -from .power_analysis import compute_efficiency, compute_power_metrics -from .raw_parser import parse_raw_file -from .runner import run_netlist, run_simulation -from .schematic import modify_component_value, parse_schematic -from .stability import compute_stability_metrics -from .templates import ( - ASC_TEMPLATES, - NETLIST_TEMPLATES, - all_template_names, - resolve_template, -) -from .touchstone import parse_touchstone, s_param_to_db -from .tuning import ( - UnknownParamError, - build_effective_params, - evaluate_targets, - extract_metrics, - make_suggestions, -) -from .waveform_expr import WaveformCalculator -from .waveform_math import ( - compute_bandwidth, -) -from .waveform_query import analyze_signal, extract_waveform - -# ============================================================================ -# 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 - - -# ============================================================================ -# 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 - - -# ============================================================================ -# 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"], - } - - -# ============================================================================ -# 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], - } - - -# ============================================================================ -# 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() - - -# ============================================================================ -# 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), - } - - -# ============================================================================ -# ENTRY POINT -# ============================================================================ def main(): diff --git a/src/mcltspice/simulation_tools.py b/src/mcltspice/simulation_tools.py new file mode 100644 index 0000000..1be6647 --- /dev/null +++ b/src/mcltspice/simulation_tools.py @@ -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 + +