From 351b89ada5005e6d8c33806a3af3f0f126e03c0f Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Fri, 19 Jun 2026 22:53:14 -0600 Subject: [PATCH 01/11] Sync uv.lock to pyproject version 2026.03.05.1 --- uv.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/uv.lock b/uv.lock index 8f57302..7c7a6d9 100644 --- a/uv.lock +++ b/uv.lock @@ -1028,7 +1028,7 @@ wheels = [ [[package]] name = "mcltspice" -version = "2026.2.14.1" +version = "2026.3.5.1" source = { editable = "." } dependencies = [ { name = "fastmcp" }, From d177a8a3163a4fdd90b74457876e61548fc30f0d Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Fri, 19 Jun 2026 22:53:14 -0600 Subject: [PATCH 02/11] Add RawFile.resolve_index, dedup signal-index lookup in two tools tune_circuit and plot_waveform each hand-rolled exact case-insensitive name->row-index resolution -- a manual 3-way loop and a next(enumerate) generator respectively. Centralize as RawFile.resolve_index(), next to the existing get_variable() (which does partial match, so it can't be reused here without changing semantics). No behavior change: exact case-insensitive match and the not-found error paths (with available_signals) are preserved. Verified against the live LTspice path -- tune_circuit bandwidth and plot_waveform resolution unchanged. --- src/mcltspice/raw_parser.py | 13 +++++++++++++ src/mcltspice/server.py | 24 ++++++++---------------- 2 files changed, 21 insertions(+), 16 deletions(-) diff --git a/src/mcltspice/raw_parser.py b/src/mcltspice/raw_parser.py index f9b8a27..13742b2 100644 --- a/src/mcltspice/raw_parser.py +++ b/src/mcltspice/raw_parser.py @@ -52,6 +52,19 @@ class RawFile: return arr return None + def resolve_index(self, name: str) -> int | None: + """Return a variable's data-row index by exact name (case-insensitive). + + Unlike get_variable (partial match, returns the data array), this does an + exact case-insensitive name match and returns the integer row index into + self.data, or None if there's no exact match. + """ + name_lower = name.lower() + for var in self.variables: + if var.name.lower() == name_lower: + return var.index + return None + def get_time(self, run: int | None = None) -> np.ndarray | None: """Get the time axis (for transient analysis).""" return self.get_variable("time", run=run) diff --git a/src/mcltspice/server.py b/src/mcltspice/server.py index acfb7c3..f2aef06 100644 --- a/src/mcltspice/server.py +++ b/src/mcltspice/server.py @@ -1272,17 +1272,10 @@ async def tune_circuit( if raw is None: return {"error": "No raw data from simulation", "params_used": effective_params} - # Find the signal variable - sig_idx = None - time_idx = None - freq_idx = None - for var in raw.variables: - if var.name.lower() == signal.lower(): - sig_idx = var.index - if var.name.lower() == "time": - time_idx = var.index - if var.name.lower() == "frequency": - freq_idx = var.index + # Find the signal variable (exact, case-insensitive) + sig_idx = raw.resolve_index(signal) + time_idx = raw.resolve_index("time") + freq_idx = raw.resolve_index("frequency") if sig_idx is None: available = [v.name for v in raw.variables] @@ -1513,19 +1506,18 @@ async def plot_waveform( signal_list = signals if signals else [signal] is_multi = len(signal_list) > 1 - # Resolve all signal names (case-insensitive) + # Resolve all signal names (exact, case-insensitive) sig_names = [v.name for v in raw_data.variables] sig_lower = {s.lower(): s for s in sig_names} resolved: list[tuple[str, int]] = [] # (actual_name, index) for s in signal_list: - actual = sig_lower.get(s.lower()) - if actual is None: + idx = raw_data.resolve_index(s) + if idx is None: return { "error": f"Signal '{s}' not found", "available_signals": sig_names, } - idx = next(i for i, v in enumerate(raw_data.variables) if v.name == actual) - resolved.append((actual, idx)) + resolved.append((sig_lower[s.lower()], idx)) # Get x-axis data (time or frequency) x_var = raw_data.variables[0] From 7c4f09cbb30e74c89a9edce994c3fc0a9f27517d Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Fri, 19 Jun 2026 22:54:05 -0600 Subject: [PATCH 03/11] Test RawFile.resolve_index -- exact match, case-insensitivity, no partial Locks in the exact-match semantics that distinguish resolve_index from get_variable, so the lookup used by tune_circuit and plot_waveform is now covered without needing the live LTspice path. --- tests/test_raw_parser.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/test_raw_parser.py b/tests/test_raw_parser.py index 9a42bef..db8d45b 100644 --- a/tests/test_raw_parser.py +++ b/tests/test_raw_parser.py @@ -69,6 +69,22 @@ class TestRawFileGetVariable: assert len(result) == mock_rawfile.points +class TestRawFileResolveIndex: + def test_exact_match_returns_index(self, mock_rawfile): + assert mock_rawfile.resolve_index("V(out)") == 1 + assert mock_rawfile.resolve_index("time") == 0 + + def test_case_insensitive(self, mock_rawfile): + assert mock_rawfile.resolve_index("v(out)") == 1 + + def test_partial_does_not_match(self, mock_rawfile): + """Unlike get_variable, resolve_index requires an exact name match.""" + assert mock_rawfile.resolve_index("out") is None + + def test_missing_returns_none(self, mock_rawfile): + assert mock_rawfile.resolve_index("V(nonexistent)") is None + + class TestRawFileRunData: def test_get_run_data_slicing(self, mock_rawfile_stepped): """Extracting a single run produces correct point count.""" From c71b3bbb4c2fd9f320a17e5539ca6b1a4b584e54 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sat, 20 Jun 2026 00:45:24 -0600 Subject: [PATCH 04/11] Extract template registries into templates.py The two inline registries (_TEMPLATES, _ASC_TEMPLATES) plus their ~30 generator-function imports were pure data living in server.py. Move them to a dedicated templates.py as NETLIST_TEMPLATES / ASC_TEMPLATES, and add resolve_template() (netlist-first precedence) + all_template_names() to replace the '.get() or .get()' dual-lookup dance that tune_circuit used. server.py: 3257 -> 2949 lines. Registries verified byte-identical to the originals (keys, params, descriptions, func identities); netlist-first precedence for names in both registries preserved; live tune_circuit / create_from_template / generate_schematic unchanged. --- src/mcltspice/server.py | 356 +++---------------------------------- src/mcltspice/templates.py | 315 ++++++++++++++++++++++++++++++++ 2 files changed, 339 insertions(+), 332 deletions(-) create mode 100644 src/mcltspice/templates.py diff --git a/src/mcltspice/server.py b/src/mcltspice/server.py index f2aef06..0637859 100644 --- a/src/mcltspice/server.py +++ b/src/mcltspice/server.py @@ -21,51 +21,6 @@ import numpy as np from fastmcp import FastMCP from fastmcp.prompts import Message -from .asc_generator import ( - generate_boost_converter as generate_boost_converter_asc, -) -from .asc_generator import ( - generate_buck_converter as generate_buck_converter_asc, -) -from .asc_generator import ( - generate_colpitts_oscillator as generate_colpitts_asc, -) -from .asc_generator import ( - generate_common_emitter_amp as generate_ce_amp_asc, -) -from .asc_generator import ( - generate_current_mirror as generate_current_mirror_asc, -) -from .asc_generator import ( - generate_differential_amp as generate_diff_amp_asc, -) -from .asc_generator import ( - generate_h_bridge as generate_h_bridge_asc, -) -from .asc_generator import ( - generate_instrumentation_amp as generate_inamp_asc, -) -from .asc_generator import ( - generate_inverting_amp, -) -from .asc_generator import ( - generate_ldo_regulator as generate_ldo_asc, -) -from .asc_generator import ( - generate_non_inverting_amp as generate_noninv_amp_asc, -) -from .asc_generator import ( - generate_rc_lowpass as generate_rc_lowpass_asc, -) -from .asc_generator import ( - generate_sallen_key_lowpass as generate_sallen_key_asc, -) -from .asc_generator import ( - generate_transimpedance_amp as generate_tia_asc, -) -from .asc_generator import ( - generate_voltage_divider as generate_voltage_divider_asc, -) from .batch import ( run_monte_carlo, run_parameter_sweep, @@ -82,24 +37,7 @@ from .drc import run_drc as _run_drc from .log_parser import parse_log from .models import search_models as _search_models from .models import search_subcircuits as _search_subcircuits -from .netlist import ( - Netlist, - boost_converter, - buck_converter, - colpitts_oscillator, - common_emitter_amplifier, - current_mirror, - differential_amplifier, - h_bridge, - instrumentation_amplifier, - inverting_amplifier, - ldo_regulator, - non_inverting_amplifier, - rc_lowpass, - sallen_key_lowpass, - transimpedance_amplifier, - voltage_divider, -) +from .netlist import Netlist from .noise_analysis import ( compute_noise_metrics, compute_spot_noise, @@ -122,6 +60,12 @@ from .spicebook import ( ) from .stability import compute_stability_metrics from .svg_plot import plot_bode, plot_spectrum, plot_timeseries, plot_timeseries_multi +from .templates import ( + ASC_TEMPLATES, + NETLIST_TEMPLATES, + all_template_names, + resolve_template, +) from .touchstone import parse_touchstone, s_param_to_db from .waveform_expr import WaveformCalculator from .waveform_math import ( @@ -1200,7 +1144,7 @@ async def tune_circuit( Args: template: Template name (from list_templates). Works with both - netlist templates (_TEMPLATES) and schematic templates (_ASC_TEMPLATES). + 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, @@ -1213,14 +1157,12 @@ async def tune_circuit( Returns: Dict with metrics, target comparison, and tuning suggestions. """ - # --- 1. Look up template --- - tmpl = _TEMPLATES.get(template) - is_netlist = tmpl is not None - if tmpl is None: - tmpl = _ASC_TEMPLATES.get(template) - if tmpl is None: - names = sorted(set(list(_TEMPLATES.keys()) + list(_ASC_TEMPLATES.keys()))) + # --- 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 --- effective_params: dict[str, str] = dict(tmpl["params"]) @@ -1773,103 +1715,6 @@ async def monte_carlo( # ============================================================================ -_ASC_TEMPLATES: dict[str, dict] = { - "rc_lowpass": { - "func": generate_rc_lowpass_asc, - "description": "RC lowpass filter with AC analysis", - "params": {"r": "1k", "c": "100n"}, - }, - "voltage_divider": { - "func": generate_voltage_divider_asc, - "description": "Resistive voltage divider with .op analysis", - "params": {"r1": "10k", "r2": "10k", "vin": "5"}, - }, - "inverting_amp": { - "func": generate_inverting_amp, - "description": "Inverting op-amp amplifier (gain = -Rf/Rin)", - "params": {"rin": "10k", "rf": "100k"}, - }, - "non_inverting_amp": { - "func": generate_noninv_amp_asc, - "description": "Non-inverting op-amp amplifier (gain = 1 + Rf/Rin)", - "params": {"rin": "10k", "rf": "100k"}, - }, - "common_emitter_amp": { - "func": generate_ce_amp_asc, - "description": "Common-emitter BJT amplifier with voltage divider bias", - "params": { - "rc": "2.2k", "rb1": "56k", "rb2": "12k", "re": "1k", - "cc_in": "10u", "cc_out": "10u", "ce": "47u", - "vcc": "12", "bjt_model": "2N2222", - }, - }, - "colpitts_oscillator": { - "func": generate_colpitts_asc, - "description": "Colpitts oscillator with LC tank and BJT", - "params": { - "ind": "1u", "c1": "100p", "c2": "100p", - "rb": "47k", "rc": "1k", "re": "470", - "vcc": "12", "bjt_model": "2N2222", - }, - }, - "differential_amp": { - "func": generate_diff_amp_asc, - "description": "Differential amplifier with op-amp", - "params": {"r1": "10k", "r2": "10k", "r3": "10k", "r4": "10k"}, - }, - "buck_converter": { - "func": generate_buck_converter_asc, - "description": "Buck (step-down) converter with NMOS switch", - "params": { - "ind": "10u", "c_out": "100u", "r_load": "10", - "v_in": "12", "duty_cycle": "0.5", "freq": "100k", - "mosfet_model": "IRF540N", "diode_model": "1N5819", - }, - }, - "ldo_regulator": { - "func": generate_ldo_asc, - "description": "LDO voltage regulator with PMOS pass transistor", - "params": { - "r1": "10k", "r2": "10k", "pass_transistor": "IRF9540N", - "v_in": "8", "v_ref": "2.5", - }, - }, - "h_bridge": { - "func": generate_h_bridge_asc, - "description": "H-bridge motor driver with 4 NMOS transistors", - "params": {"v_supply": "12", "r_load": "10", "mosfet_model": "IRF540N"}, - }, - "sallen_key_lowpass": { - "func": generate_sallen_key_asc, - "description": "Sallen-Key lowpass filter (unity gain, 2nd order)", - "params": {"r1": "10k", "r2": "10k", "c1": "10n", "c2": "10n"}, - }, - "boost_converter": { - "func": generate_boost_converter_asc, - "description": "Boost (step-up) converter with NMOS switch", - "params": { - "ind": "10u", "c_out": "100u", "r_load": "50", - "v_in": "5", "duty_cycle": "0.5", "freq": "100k", - }, - }, - "instrumentation_amp": { - "func": generate_inamp_asc, - "description": "3-opamp instrumentation amplifier", - "params": {"r1": "10k", "r2": "10k", "r3": "10k", "r_gain": "10k"}, - }, - "current_mirror": { - "func": generate_current_mirror_asc, - "description": "BJT current mirror with reference and load", - "params": {"r_ref": "10k", "r_load": "1k", "vcc": "12"}, - }, - "transimpedance_amp": { - "func": generate_tia_asc, - "description": "Transimpedance amplifier (current to voltage)", - "params": {"rf": "100k", "cf": "1p", "i_source": "1u"}, - }, -} - - @mcp.tool() def generate_schematic( template: str, @@ -1894,11 +1739,11 @@ def generate_schematic( 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)) + if template not in ASC_TEMPLATES: + names = ", ".join(sorted(ASC_TEMPLATES)) return {"error": f"Unknown template '{template}'. Available: {names}"} - entry = _ASC_TEMPLATES[template] + entry = ASC_TEMPLATES[template] call_params: dict[str, str | float] = {} if params: @@ -2143,159 +1988,6 @@ def create_netlist( # CIRCUIT TEMPLATE TOOLS # ============================================================================ -# Registry of netlist templates with parameter metadata -_TEMPLATES: dict[str, dict] = { - "voltage_divider": { - "func": voltage_divider, - "description": "Resistive voltage divider with .op or custom analysis", - "params": {"v_in": "5", "r1": "10k", "r2": "10k", "sim_type": "op"}, - }, - "rc_lowpass": { - "func": rc_lowpass, - "description": "RC lowpass filter with AC sweep", - "params": {"r": "1k", "c": "100n", "f_start": "1", "f_stop": "1meg"}, - }, - "inverting_amplifier": { - "func": inverting_amplifier, - "description": "Inverting op-amp (gain = -Rf/Rin), +/-15V supply", - "params": {"r_in": "10k", "r_f": "100k", "opamp_model": "LT1001"}, - }, - "non_inverting_amplifier": { - "func": non_inverting_amplifier, - "description": "Non-inverting op-amp (gain = 1 + Rf/Rin), +/-15V supply", - "params": {"r_in": "10k", "r_f": "100k", "opamp_model": "LT1001"}, - }, - "differential_amplifier": { - "func": differential_amplifier, - "description": "Diff amp: Vout = (R2/R1)*(V2-V1), +/-15V supply", - "params": { - "r1": "10k", - "r2": "10k", - "r3": "10k", - "r4": "10k", - "opamp_model": "LT1001", - }, - }, - "common_emitter_amplifier": { - "func": common_emitter_amplifier, - "description": "BJT common-emitter with voltage divider bias", - "params": { - "rc": "2.2k", - "rb1": "56k", - "rb2": "12k", - "re": "1k", - "cc1": "10u", - "cc2": "10u", - "ce": "47u", - "vcc": "12", - "bjt_model": "2N2222", - }, - }, - "buck_converter": { - "func": buck_converter, - "description": "Step-down DC-DC converter with MOSFET switch", - "params": { - "ind": "10u", - "c_out": "100u", - "r_load": "10", - "v_in": "12", - "duty_cycle": "0.5", - "freq": "100k", - "mosfet_model": "IRF540N", - "diode_model": "1N5819", - }, - }, - "ldo_regulator": { - "func": ldo_regulator, - "description": "LDO regulator: Vout = Vref * (1 + R1/R2)", - "params": { - "opamp_model": "LT1001", - "r1": "10k", - "r2": "10k", - "pass_transistor": "IRF9540N", - "v_in": "8", - "v_ref": "2.5", - }, - }, - "colpitts_oscillator": { - "func": colpitts_oscillator, - "description": "LC oscillator: f ~ 1/(2pi*sqrt(L*Cseries))", - "params": { - "ind": "1u", - "c1": "100p", - "c2": "100p", - "rb": "47k", - "rc": "1k", - "re": "470", - "vcc": "12", - "bjt_model": "2N2222", - }, - }, - "h_bridge": { - "func": h_bridge, - "description": "4-MOSFET H-bridge motor driver with dead time", - "params": { - "v_supply": "12", - "r_load": "10", - "mosfet_model": "IRF540N", - }, - }, - "sallen_key_lowpass": { - "func": sallen_key_lowpass, - "description": "Sallen-Key lowpass filter (unity gain, 2nd order)", - "params": { - "r1": "10k", - "r2": "10k", - "c1": "10n", - "c2": "10n", - "opamp_model": "LT1001", - }, - }, - "boost_converter": { - "func": boost_converter, - "description": "Step-up DC-DC converter with MOSFET switch", - "params": { - "ind": "10u", - "c_out": "100u", - "r_load": "50", - "v_in": "5", - "duty_cycle": "0.5", - "freq": "100k", - "mosfet_model": "IRF540N", - "diode_model": "1N5819", - }, - }, - "instrumentation_amplifier": { - "func": instrumentation_amplifier, - "description": "3-opamp instrumentation amp: gain = 1 + 2*R1/R_gain", - "params": { - "r1": "10k", - "r2": "10k", - "r3": "10k", - "r_gain": "10k", - }, - }, - "current_mirror": { - "func": current_mirror, - "description": "BJT current mirror with reference resistor", - "params": { - "r_ref": "10k", - "r_load": "1k", - "vcc": "12", - "bjt_model": "2N2222", - }, - }, - "transimpedance_amplifier": { - "func": transimpedance_amplifier, - "description": "TIA: converts input current to output voltage (Vout = -Iph * Rf)", - "params": { - "rf": "100k", - "cf": "1p", - "i_source": "1u", - }, - }, -} - @mcp.tool() def create_from_template( @@ -2329,13 +2021,13 @@ def create_from_template( params: Optional dict of parameter overrides (all values as strings) output_path: Where to save .cir file (None = auto in /tmp) """ - template = _TEMPLATES.get(template_name) + 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 _TEMPLATES.items() + for k, v in NETLIST_TEMPLATES.items() ], } @@ -2386,9 +2078,9 @@ def list_templates() -> dict: "description": info["description"], "params": info["params"], } - for name, info in _TEMPLATES.items() + for name, info in NETLIST_TEMPLATES.items() ], - "total_count": len(_TEMPLATES), + "total_count": len(NETLIST_TEMPLATES), } @@ -2639,11 +2331,11 @@ def resource_templates() -> str: templates = { "netlist_templates": [ {"name": name, "description": info["description"], "params": info["params"]} - for name, info in _TEMPLATES.items() + for name, info in NETLIST_TEMPLATES.items() ], "schematic_templates": [ {"name": name, "description": info["description"], "params": info["params"]} - for name, info in _ASC_TEMPLATES.items() + for name, info in ASC_TEMPLATES.items() ], } return json.dumps(templates, indent=2) @@ -2653,8 +2345,8 @@ def resource_templates() -> str: def resource_template_detail(name: str) -> str: """Detailed information about a specific circuit template.""" # Check both registries - netlist_info = _TEMPLATES.get(name) - asc_info = _ASC_TEMPLATES.get(name) + netlist_info = NETLIST_TEMPLATES.get(name) + asc_info = ASC_TEMPLATES.get(name) if not netlist_info and not asc_info: return json.dumps({"error": f"Template '{name}' not found"}) diff --git a/src/mcltspice/templates.py b/src/mcltspice/templates.py new file mode 100644 index 0000000..e86fc71 --- /dev/null +++ b/src/mcltspice/templates.py @@ -0,0 +1,315 @@ +"""Circuit template registries. + +Two registries map template names to builder functions and default params: + +- NETLIST_TEMPLATES -> .netlist builders, producing SPICE .cir netlists +- ASC_TEMPLATES -> .asc_generator builders, producing graphical schematics + +Some names appear in both (e.g. rc_lowpass). resolve_template() applies a +netlist-first precedence so a name available in both resolves to its netlist +form, matching the historical lookup order in tune_circuit. +""" + +from .asc_generator import ( + generate_boost_converter, + generate_buck_converter, + generate_colpitts_oscillator, + generate_common_emitter_amp, + generate_current_mirror, + generate_differential_amp, + generate_h_bridge, + generate_instrumentation_amp, + generate_inverting_amp, + generate_ldo_regulator, + generate_non_inverting_amp, + generate_rc_lowpass, + generate_sallen_key_lowpass, + generate_transimpedance_amp, + generate_voltage_divider, +) +from .netlist import ( + boost_converter, + buck_converter, + colpitts_oscillator, + common_emitter_amplifier, + current_mirror, + differential_amplifier, + h_bridge, + instrumentation_amplifier, + inverting_amplifier, + ldo_regulator, + non_inverting_amplifier, + rc_lowpass, + sallen_key_lowpass, + transimpedance_amplifier, + voltage_divider, +) + +# Graphical .asc schematic templates (generate_schematic, tune_circuit fallback) +ASC_TEMPLATES: dict[str, dict] = { + "rc_lowpass": { + "func": generate_rc_lowpass, + "description": "RC lowpass filter with AC analysis", + "params": {"r": "1k", "c": "100n"}, + }, + "voltage_divider": { + "func": generate_voltage_divider, + "description": "Resistive voltage divider with .op analysis", + "params": {"r1": "10k", "r2": "10k", "vin": "5"}, + }, + "inverting_amp": { + "func": generate_inverting_amp, + "description": "Inverting op-amp amplifier (gain = -Rf/Rin)", + "params": {"rin": "10k", "rf": "100k"}, + }, + "non_inverting_amp": { + "func": generate_non_inverting_amp, + "description": "Non-inverting op-amp amplifier (gain = 1 + Rf/Rin)", + "params": {"rin": "10k", "rf": "100k"}, + }, + "common_emitter_amp": { + "func": generate_common_emitter_amp, + "description": "Common-emitter BJT amplifier with voltage divider bias", + "params": { + "rc": "2.2k", "rb1": "56k", "rb2": "12k", "re": "1k", + "cc_in": "10u", "cc_out": "10u", "ce": "47u", + "vcc": "12", "bjt_model": "2N2222", + }, + }, + "colpitts_oscillator": { + "func": generate_colpitts_oscillator, + "description": "Colpitts oscillator with LC tank and BJT", + "params": { + "ind": "1u", "c1": "100p", "c2": "100p", + "rb": "47k", "rc": "1k", "re": "470", + "vcc": "12", "bjt_model": "2N2222", + }, + }, + "differential_amp": { + "func": generate_differential_amp, + "description": "Differential amplifier with op-amp", + "params": {"r1": "10k", "r2": "10k", "r3": "10k", "r4": "10k"}, + }, + "buck_converter": { + "func": generate_buck_converter, + "description": "Buck (step-down) converter with NMOS switch", + "params": { + "ind": "10u", "c_out": "100u", "r_load": "10", + "v_in": "12", "duty_cycle": "0.5", "freq": "100k", + "mosfet_model": "IRF540N", "diode_model": "1N5819", + }, + }, + "ldo_regulator": { + "func": generate_ldo_regulator, + "description": "LDO voltage regulator with PMOS pass transistor", + "params": { + "r1": "10k", "r2": "10k", "pass_transistor": "IRF9540N", + "v_in": "8", "v_ref": "2.5", + }, + }, + "h_bridge": { + "func": generate_h_bridge, + "description": "H-bridge motor driver with 4 NMOS transistors", + "params": {"v_supply": "12", "r_load": "10", "mosfet_model": "IRF540N"}, + }, + "sallen_key_lowpass": { + "func": generate_sallen_key_lowpass, + "description": "Sallen-Key lowpass filter (unity gain, 2nd order)", + "params": {"r1": "10k", "r2": "10k", "c1": "10n", "c2": "10n"}, + }, + "boost_converter": { + "func": generate_boost_converter, + "description": "Boost (step-up) converter with NMOS switch", + "params": { + "ind": "10u", "c_out": "100u", "r_load": "50", + "v_in": "5", "duty_cycle": "0.5", "freq": "100k", + }, + }, + "instrumentation_amp": { + "func": generate_instrumentation_amp, + "description": "3-opamp instrumentation amplifier", + "params": {"r1": "10k", "r2": "10k", "r3": "10k", "r_gain": "10k"}, + }, + "current_mirror": { + "func": generate_current_mirror, + "description": "BJT current mirror with reference and load", + "params": {"r_ref": "10k", "r_load": "1k", "vcc": "12"}, + }, + "transimpedance_amp": { + "func": generate_transimpedance_amp, + "description": "Transimpedance amplifier (current to voltage)", + "params": {"rf": "100k", "cf": "1p", "i_source": "1u"}, + }, +} + +# SPICE .cir netlist templates (create_from_template, list_templates, tune_circuit) +NETLIST_TEMPLATES: dict[str, dict] = { + "voltage_divider": { + "func": voltage_divider, + "description": "Resistive voltage divider with .op or custom analysis", + "params": {"v_in": "5", "r1": "10k", "r2": "10k", "sim_type": "op"}, + }, + "rc_lowpass": { + "func": rc_lowpass, + "description": "RC lowpass filter with AC sweep", + "params": {"r": "1k", "c": "100n", "f_start": "1", "f_stop": "1meg"}, + }, + "inverting_amplifier": { + "func": inverting_amplifier, + "description": "Inverting op-amp (gain = -Rf/Rin), +/-15V supply", + "params": {"r_in": "10k", "r_f": "100k", "opamp_model": "LT1001"}, + }, + "non_inverting_amplifier": { + "func": non_inverting_amplifier, + "description": "Non-inverting op-amp (gain = 1 + Rf/Rin), +/-15V supply", + "params": {"r_in": "10k", "r_f": "100k", "opamp_model": "LT1001"}, + }, + "differential_amplifier": { + "func": differential_amplifier, + "description": "Diff amp: Vout = (R2/R1)*(V2-V1), +/-15V supply", + "params": { + "r1": "10k", + "r2": "10k", + "r3": "10k", + "r4": "10k", + "opamp_model": "LT1001", + }, + }, + "common_emitter_amplifier": { + "func": common_emitter_amplifier, + "description": "BJT common-emitter with voltage divider bias", + "params": { + "rc": "2.2k", + "rb1": "56k", + "rb2": "12k", + "re": "1k", + "cc1": "10u", + "cc2": "10u", + "ce": "47u", + "vcc": "12", + "bjt_model": "2N2222", + }, + }, + "buck_converter": { + "func": buck_converter, + "description": "Step-down DC-DC converter with MOSFET switch", + "params": { + "ind": "10u", + "c_out": "100u", + "r_load": "10", + "v_in": "12", + "duty_cycle": "0.5", + "freq": "100k", + "mosfet_model": "IRF540N", + "diode_model": "1N5819", + }, + }, + "ldo_regulator": { + "func": ldo_regulator, + "description": "LDO regulator: Vout = Vref * (1 + R1/R2)", + "params": { + "opamp_model": "LT1001", + "r1": "10k", + "r2": "10k", + "pass_transistor": "IRF9540N", + "v_in": "8", + "v_ref": "2.5", + }, + }, + "colpitts_oscillator": { + "func": colpitts_oscillator, + "description": "LC oscillator: f ~ 1/(2pi*sqrt(L*Cseries))", + "params": { + "ind": "1u", + "c1": "100p", + "c2": "100p", + "rb": "47k", + "rc": "1k", + "re": "470", + "vcc": "12", + "bjt_model": "2N2222", + }, + }, + "h_bridge": { + "func": h_bridge, + "description": "4-MOSFET H-bridge motor driver with dead time", + "params": { + "v_supply": "12", + "r_load": "10", + "mosfet_model": "IRF540N", + }, + }, + "sallen_key_lowpass": { + "func": sallen_key_lowpass, + "description": "Sallen-Key lowpass filter (unity gain, 2nd order)", + "params": { + "r1": "10k", + "r2": "10k", + "c1": "10n", + "c2": "10n", + "opamp_model": "LT1001", + }, + }, + "boost_converter": { + "func": boost_converter, + "description": "Step-up DC-DC converter with MOSFET switch", + "params": { + "ind": "10u", + "c_out": "100u", + "r_load": "50", + "v_in": "5", + "duty_cycle": "0.5", + "freq": "100k", + "mosfet_model": "IRF540N", + "diode_model": "1N5819", + }, + }, + "instrumentation_amplifier": { + "func": instrumentation_amplifier, + "description": "3-opamp instrumentation amp: gain = 1 + 2*R1/R_gain", + "params": { + "r1": "10k", + "r2": "10k", + "r3": "10k", + "r_gain": "10k", + }, + }, + "current_mirror": { + "func": current_mirror, + "description": "BJT current mirror with reference resistor", + "params": { + "r_ref": "10k", + "r_load": "1k", + "vcc": "12", + "bjt_model": "2N2222", + }, + }, + "transimpedance_amplifier": { + "func": transimpedance_amplifier, + "description": "TIA: converts input current to output voltage (Vout = -Iph * Rf)", + "params": { + "rf": "100k", + "cf": "1p", + "i_source": "1u", + }, + }, +} + + +def resolve_template(name: str) -> tuple[dict, bool] | None: + """Resolve a template name to (entry, is_netlist), netlist registry first. + + Returns None if the name is in neither registry. Netlist-first precedence + preserves the historical lookup order: a name present in both registries + (e.g. "rc_lowpass") resolves to its netlist form. + """ + if name in NETLIST_TEMPLATES: + return NETLIST_TEMPLATES[name], True + if name in ASC_TEMPLATES: + return ASC_TEMPLATES[name], False + return None + + +def all_template_names() -> list[str]: + """Sorted union of every template name across both registries.""" + return sorted(set(NETLIST_TEMPLATES) | set(ASC_TEMPLATES)) From cf1502cee9409f957de8fdc0516e4287ba2c4c0c Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sat, 20 Jun 2026 00:45:47 -0600 Subject: [PATCH 05/11] Test templates registries and resolve_template precedence Covers the netlist-first precedence contract (names in both registries), asc-only/netlist-only resolution, unknown -> None, and the deduped union in all_template_names -- the lookup logic tune_circuit depends on. --- tests/test_templates.py | 58 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) create mode 100644 tests/test_templates.py diff --git a/tests/test_templates.py b/tests/test_templates.py new file mode 100644 index 0000000..3c091f3 --- /dev/null +++ b/tests/test_templates.py @@ -0,0 +1,58 @@ +"""Tests for the circuit template registries and lookup helpers.""" + +from mcltspice.templates import ( + ASC_TEMPLATES, + NETLIST_TEMPLATES, + all_template_names, + resolve_template, +) + + +class TestRegistries: + def test_both_registries_populated(self): + assert len(NETLIST_TEMPLATES) == 15 + assert len(ASC_TEMPLATES) == 15 + + def test_every_entry_has_required_keys(self): + for reg in (NETLIST_TEMPLATES, ASC_TEMPLATES): + for name, entry in reg.items(): + assert callable(entry["func"]), name + assert isinstance(entry["description"], str) and entry["description"], name + assert isinstance(entry["params"], dict), name + + +class TestResolveTemplate: + def test_netlist_first_precedence(self): + """A name in both registries resolves to its netlist form.""" + assert "rc_lowpass" in NETLIST_TEMPLATES + assert "rc_lowpass" in ASC_TEMPLATES + entry, is_netlist = resolve_template("rc_lowpass") + assert is_netlist is True + assert entry is NETLIST_TEMPLATES["rc_lowpass"] + + def test_asc_only_name(self): + """A name only in the asc registry resolves there, is_netlist False.""" + assert "inverting_amp" in ASC_TEMPLATES + assert "inverting_amp" not in NETLIST_TEMPLATES + entry, is_netlist = resolve_template("inverting_amp") + assert is_netlist is False + assert entry is ASC_TEMPLATES["inverting_amp"] + + def test_netlist_only_name(self): + assert "inverting_amplifier" in NETLIST_TEMPLATES + entry, is_netlist = resolve_template("inverting_amplifier") + assert is_netlist is True + + def test_unknown_returns_none(self): + assert resolve_template("does_not_exist") is None + + +class TestAllTemplateNames: + def test_is_sorted_union(self): + names = all_template_names() + expected = sorted(set(NETLIST_TEMPLATES) | set(ASC_TEMPLATES)) + assert names == expected + + def test_dedupes_overlapping_names(self): + # 15 + 15 with 9 shared names -> 21 unique + assert len(all_template_names()) == 21 From dd6db95021e1e142af5b10f81a58590599a85e6f Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sat, 20 Jun 2026 01:43:09 -0600 Subject: [PATCH 06/11] Extract tune_circuit pure logic into tuning.py tune_circuit interleaved ~260 lines of pure computation (param validation, metric extraction, target evaluation, suggestion generation) with the async Wine orchestration, so that logic could only be tested through a full simulation. Split the four pure pieces into tuning.py: build_effective_params, extract_metrics, evaluate_targets, make_suggestions (plus UnknownParamError). The tool keeps the async generate-and-simulate flow and error-dict shaping. server.py: 2949 -> 2799 lines. Output verified byte-identical to the prior version across AC, transient, and all error paths. --- src/mcltspice/server.py | 194 ++++------------------------------ src/mcltspice/tuning.py | 228 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 250 insertions(+), 172 deletions(-) create mode 100644 src/mcltspice/tuning.py diff --git a/src/mcltspice/server.py b/src/mcltspice/server.py index 0637859..9e5cc7d 100644 --- a/src/mcltspice/server.py +++ b/src/mcltspice/server.py @@ -67,6 +67,13 @@ from .templates import ( 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, @@ -1165,15 +1172,13 @@ async def tune_circuit( tmpl, is_netlist = found # --- 2. Build effective params --- - effective_params: dict[str, str] = dict(tmpl["params"]) - if params: - for k, v in params.items(): - if k not in tmpl["params"]: - return { - "error": f"Unknown param '{k}' for {template}", - "valid_params": list(tmpl["params"].keys()), - } - effective_params[k] = v + 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 @@ -1209,178 +1214,23 @@ async def tune_circuit( except Exception as e: return {"error": f"Generation/simulation failed: {e}", "params_used": effective_params} - # --- 4. Extract waveform and compute metrics --- + # --- 4. Extract metrics --- raw = result.raw_data if raw is None: return {"error": "No raw data from simulation", "params_used": effective_params} - # Find the signal variable (exact, case-insensitive) - sig_idx = raw.resolve_index(signal) - time_idx = raw.resolve_index("time") - freq_idx = raw.resolve_index("frequency") - - if sig_idx is None: - available = [v.name for v in raw.variables] + extracted = extract_metrics(raw, signal) + if extracted is None: return { "error": f"Signal '{signal}' not found", - "available_signals": available, + "available_signals": [v.name for v in raw.variables], "params_used": effective_params, } + metrics, is_ac = extracted - sig_data = raw.data[sig_idx] - metrics: dict[str, float] = {} - - is_ac = freq_idx is not None - - if is_ac: - # Frequency-domain metrics - freq = np.abs(raw.data[freq_idx]).real - mag_complex = sig_data - mag_db = 20.0 * np.log10(np.maximum(np.abs(mag_complex), 1e-30)) - - # Bandwidth - try: - bw_result = compute_bandwidth(freq, mag_db) - metrics["bandwidth_hz"] = bw_result["bandwidth_hz"] - if bw_result.get("f_low"): - metrics["f_low_hz"] = bw_result["f_low"] - if bw_result.get("f_high"): - metrics["f_high_hz"] = bw_result["f_high"] - except Exception: - pass - - # DC gain (magnitude at lowest frequency) - metrics["gain_db"] = float(mag_db[0]) - metrics["dc_value"] = float(np.abs(mag_complex[0])) - - else: - # Time-domain metrics - time_data = raw.data[time_idx] if time_idx is not None else None - sig_real = np.real(sig_data) - - # RMS - metrics["rms"] = float(compute_rms(sig_real)) - - # Peak-to-peak - pp = compute_peak_to_peak(sig_real) - metrics["peak_to_peak"] = pp["peak_to_peak"] - metrics["dc_value"] = pp["mean"] - - # Settling time (if signal looks like a step response) - if time_data is not None: - time_real = np.real(time_data) - try: - settle = compute_settling_time(time_real, sig_real) - if settle["settled"]: - metrics["settling_time_s"] = settle["settling_time"] - except Exception: - pass - - # FFT for fundamental frequency - try: - fft = compute_fft(time_real, sig_real) - if fft["fundamental_freq"] > 0: - metrics["fundamental_freq_hz"] = fft["fundamental_freq"] - except Exception: - pass - - # --- 5. Compare against targets --- - targets_met = True - target_results: dict[str, dict] = {} - if targets: - for metric_name, target_str in targets.items(): - if metric_name not in metrics: - target_results[metric_name] = { - "status": "unmeasurable", - "reason": f"Metric '{metric_name}' not available from this simulation", - } - targets_met = False - continue - - actual = metrics[metric_name] - met = False - target_val = 0.0 - - if target_str.startswith(">"): - target_val = float(target_str[1:]) - met = actual > target_val - elif target_str.startswith("<"): - target_val = float(target_str[1:]) - met = actual < target_val - elif target_str.startswith("~"): - target_val = float(target_str[1:]) - tolerance = target_val * 0.1 # 10% tolerance - met = abs(actual - target_val) <= tolerance - else: - target_val = float(target_str) - met = abs(actual - target_val) <= target_val * 0.1 - - target_results[metric_name] = { - "target": target_str, - "actual": actual, - "met": met, - } - if not met: - targets_met = False - - # --- 6. Generate suggestions --- - suggestions: list[str] = [] - if targets and not targets_met: - for metric_name, result_info in target_results.items(): - if isinstance(result_info, dict) and not result_info.get("met", True): - actual = result_info.get("actual") - target_str_val = result_info.get("target", "") - if actual is None: - continue - - if metric_name == "bandwidth_hz": - if str(target_str_val).startswith(">") and actual < float( - str(target_str_val)[1:] - ): - suggestions.append( - "To increase bandwidth: decrease R or C values " - "(f_c = 1/(2*pi*R*C) for RC filters)" - ) - elif str(target_str_val).startswith("<"): - suggestions.append( - "To decrease bandwidth: increase R or C values" - ) - - elif metric_name == "gain_db": - if str(target_str_val).startswith(">") and actual < float( - str(target_str_val)[1:] - ): - suggestions.append( - "To increase gain: increase Rf/Rin ratio for op-amp circuits, " - "or increase Rc/Re ratio for CE amplifiers" - ) - elif str(target_str_val).startswith("<"): - suggestions.append("To decrease gain: decrease Rf/Rin ratio") - - elif metric_name == "peak_to_peak": - if str(target_str_val).startswith("<"): - suggestions.append( - "To reduce peak-to-peak (ripple): increase filter capacitance " - "or inductance, or increase switching frequency" - ) - - elif metric_name == "settling_time_s": - if str(target_str_val).startswith("<"): - suggestions.append( - "To reduce settling time: increase bandwidth (decrease R*C), " - "or add damping to reduce ringing" - ) - - elif metric_name == "rms": - suggestions.append( - f"RMS is {actual:.4g}, target was {target_str_val}. " - "Adjust source amplitude or gain." - ) - - if not suggestions and not targets_met: - suggestions.append( - "Adjust component values toward the target. Use smaller steps for fine-tuning." - ) + # --- 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, diff --git a/src/mcltspice/tuning.py b/src/mcltspice/tuning.py new file mode 100644 index 0000000..fbffc45 --- /dev/null +++ b/src/mcltspice/tuning.py @@ -0,0 +1,228 @@ +"""Pure circuit-tuning logic, separated from simulation I/O. + +These functions take already-parsed data (a RawFile, metric dicts) and return +plain values, so they can be unit-tested without running LTspice. The async +generate-and-simulate orchestration stays in the tune_circuit MCP tool. +""" + +import numpy as np + +from .raw_parser import RawFile +from .waveform_math import ( + compute_bandwidth, + compute_fft, + compute_peak_to_peak, + compute_rms, + compute_settling_time, +) + + +class UnknownParamError(ValueError): + """Raised when a tuning override names a param the template doesn't have.""" + + def __init__(self, key: str, valid_params: list[str]): + super().__init__(f"Unknown param '{key}'") + self.key = key + self.valid_params = valid_params + + +def build_effective_params( + default_params: dict[str, str], overrides: dict[str, str] | None +) -> dict[str, str]: + """Merge user overrides onto a template's default params. + + Raises UnknownParamError if an override names a key not in the template. + """ + effective = dict(default_params) + if overrides: + for k, v in overrides.items(): + if k not in default_params: + raise UnknownParamError(k, list(default_params.keys())) + effective[k] = v + return effective + + +def extract_metrics(raw: RawFile, signal: str) -> tuple[dict[str, float], bool] | None: + """Compute performance metrics for one signal from a parsed RawFile. + + Returns (metrics, is_ac), or None if the signal name isn't present. AC + (frequency-domain) runs yield bandwidth/gain; transient runs yield + RMS/peak-to-peak/settling-time/fundamental-frequency. + """ + sig_idx = raw.resolve_index(signal) + time_idx = raw.resolve_index("time") + freq_idx = raw.resolve_index("frequency") + + if sig_idx is None: + return None + + sig_data = raw.data[sig_idx] + metrics: dict[str, float] = {} + is_ac = freq_idx is not None + + if is_ac: + # Frequency-domain metrics + freq = np.abs(raw.data[freq_idx]).real + mag_complex = sig_data + mag_db = 20.0 * np.log10(np.maximum(np.abs(mag_complex), 1e-30)) + + try: + bw_result = compute_bandwidth(freq, mag_db) + metrics["bandwidth_hz"] = bw_result["bandwidth_hz"] + if bw_result.get("f_low"): + metrics["f_low_hz"] = bw_result["f_low"] + if bw_result.get("f_high"): + metrics["f_high_hz"] = bw_result["f_high"] + except Exception: + pass + + # DC gain (magnitude at lowest frequency) + metrics["gain_db"] = float(mag_db[0]) + metrics["dc_value"] = float(np.abs(mag_complex[0])) + + else: + # Time-domain metrics + time_data = raw.data[time_idx] if time_idx is not None else None + sig_real = np.real(sig_data) + + metrics["rms"] = float(compute_rms(sig_real)) + + pp = compute_peak_to_peak(sig_real) + metrics["peak_to_peak"] = pp["peak_to_peak"] + metrics["dc_value"] = pp["mean"] + + if time_data is not None: + time_real = np.real(time_data) + try: + settle = compute_settling_time(time_real, sig_real) + if settle["settled"]: + metrics["settling_time_s"] = settle["settling_time"] + except Exception: + pass + + try: + fft = compute_fft(time_real, sig_real) + if fft["fundamental_freq"] > 0: + metrics["fundamental_freq_hz"] = fft["fundamental_freq"] + except Exception: + pass + + return metrics, is_ac + + +def evaluate_targets( + metrics: dict[str, float], targets: dict[str, str] | None +) -> tuple[bool, dict[str, dict]]: + """Compare measured metrics against target comparison strings. + + Target strings: ">N" (greater), ""): + target_val = float(target_str[1:]) + met = actual > target_val + elif target_str.startswith("<"): + target_val = float(target_str[1:]) + met = actual < target_val + elif target_str.startswith("~"): + target_val = float(target_str[1:]) + tolerance = target_val * 0.1 # 10% tolerance + met = abs(actual - target_val) <= tolerance + else: + target_val = float(target_str) + met = abs(actual - target_val) <= target_val * 0.1 + + target_results[metric_name] = { + "target": target_str, + "actual": actual, + "met": met, + } + if not met: + targets_met = False + + return targets_met, target_results + + +def make_suggestions(target_results: dict[str, dict], targets_met: bool) -> list[str]: + """Suggest parameter adjustments for any unmet targets. + + Returns an empty list when all targets are met (or there are none); falls + back to a generic hint if no metric-specific suggestion applies. + """ + suggestions: list[str] = [] + if targets_met: + return suggestions + + for metric_name, result_info in target_results.items(): + if isinstance(result_info, dict) and not result_info.get("met", True): + actual = result_info.get("actual") + target_str_val = result_info.get("target", "") + if actual is None: + continue + + if metric_name == "bandwidth_hz": + if str(target_str_val).startswith(">") and actual < float( + str(target_str_val)[1:] + ): + suggestions.append( + "To increase bandwidth: decrease R or C values " + "(f_c = 1/(2*pi*R*C) for RC filters)" + ) + elif str(target_str_val).startswith("<"): + suggestions.append("To decrease bandwidth: increase R or C values") + + elif metric_name == "gain_db": + if str(target_str_val).startswith(">") and actual < float( + str(target_str_val)[1:] + ): + suggestions.append( + "To increase gain: increase Rf/Rin ratio for op-amp circuits, " + "or increase Rc/Re ratio for CE amplifiers" + ) + elif str(target_str_val).startswith("<"): + suggestions.append("To decrease gain: decrease Rf/Rin ratio") + + elif metric_name == "peak_to_peak": + if str(target_str_val).startswith("<"): + suggestions.append( + "To reduce peak-to-peak (ripple): increase filter capacitance " + "or inductance, or increase switching frequency" + ) + + elif metric_name == "settling_time_s": + if str(target_str_val).startswith("<"): + suggestions.append( + "To reduce settling time: increase bandwidth (decrease R*C), " + "or add damping to reduce ringing" + ) + + elif metric_name == "rms": + suggestions.append( + f"RMS is {actual:.4g}, target was {target_str_val}. " + "Adjust source amplitude or gain." + ) + + if not suggestions: + suggestions.append( + "Adjust component values toward the target. Use smaller steps for fine-tuning." + ) + + return suggestions From 181d1cb925cf8aead6b986e3f6349233490cb6ba Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sat, 20 Jun 2026 01:43:44 -0600 Subject: [PATCH 07/11] Test tuning.py pure logic without a simulator 15 tests over build_effective_params, extract_metrics (AC + transient via hand-built RawFiles), evaluate_targets (all comparison operators + unmeasurable), and make_suggestions. This logic previously required a full LTspice run to exercise. --- tests/test_tuning.py | 122 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 122 insertions(+) create mode 100644 tests/test_tuning.py diff --git a/tests/test_tuning.py b/tests/test_tuning.py new file mode 100644 index 0000000..8480ce0 --- /dev/null +++ b/tests/test_tuning.py @@ -0,0 +1,122 @@ +"""Tests for the pure tuning logic extracted from tune_circuit. + +These run without LTspice -- they feed hand-built RawFiles and metric dicts +directly into the pure functions. +""" + +import numpy as np +import pytest + +from mcltspice.raw_parser import RawFile, Variable +from mcltspice.tuning import ( + UnknownParamError, + build_effective_params, + evaluate_targets, + extract_metrics, + make_suggestions, +) + + +def _ac_rawfile(corner_hz: float = 1000.0) -> RawFile: + """A single-pole RC lowpass AC response, frequency + complex V(out).""" + freq = np.logspace(0, 6, 601) + h = 1.0 / (1.0 + 1j * (freq / corner_hz)) + data = np.array([freq.astype(complex), h]) + return RawFile( + title="ac", date="", plotname="AC Analysis", flags=["complex"], + variables=[Variable(0, "frequency", "frequency"), Variable(1, "V(out)", "voltage")], + points=len(freq), data=data, + ) + + +def _tran_rawfile() -> RawFile: + """A transient run: time + a 1 kHz sine on V(out).""" + t = np.linspace(0, 0.01, 1000) + v = 2.0 * np.sin(2 * np.pi * 1000 * t) + data = np.array([t, v]) + return RawFile( + title="tran", date="", plotname="Transient Analysis", flags=["real"], + variables=[Variable(0, "time", "time"), Variable(1, "V(out)", "voltage")], + points=len(t), data=data, + ) + + +class TestBuildEffectiveParams: + def test_merges_overrides(self): + result = build_effective_params({"r": "1k", "c": "100n"}, {"r": "2k"}) + assert result == {"r": "2k", "c": "100n"} + + def test_none_overrides_returns_defaults_copy(self): + defaults = {"r": "1k"} + result = build_effective_params(defaults, None) + assert result == {"r": "1k"} + assert result is not defaults # must be a copy + + def test_unknown_key_raises(self): + with pytest.raises(UnknownParamError) as exc: + build_effective_params({"r": "1k"}, {"bogus": "5"}) + assert exc.value.key == "bogus" + assert exc.value.valid_params == ["r"] + + +class TestExtractMetrics: + def test_missing_signal_returns_none(self): + assert extract_metrics(_ac_rawfile(), "V(ghost)") is None + + def test_ac_metrics(self): + metrics, is_ac = extract_metrics(_ac_rawfile(corner_hz=1000.0), "V(out)") + assert is_ac is True + assert "bandwidth_hz" in metrics + assert metrics["bandwidth_hz"] == pytest.approx(1000.0, rel=0.05) + assert metrics["gain_db"] == pytest.approx(0.0, abs=0.1) # 0 dB DC + + def test_transient_metrics(self): + metrics, is_ac = extract_metrics(_tran_rawfile(), "V(out)") + assert is_ac is False + assert metrics["rms"] == pytest.approx(2.0 / np.sqrt(2), rel=0.02) + assert metrics["peak_to_peak"] == pytest.approx(4.0, rel=0.02) + + +class TestEvaluateTargets: + def test_no_targets(self): + assert evaluate_targets({"bandwidth_hz": 1000.0}, None) == (True, {}) + + def test_greater_than_met_and_unmet(self): + met, results = evaluate_targets({"bw": 1000.0}, {"bw": ">500"}) + assert met is True and results["bw"]["met"] is True + met2, results2 = evaluate_targets({"bw": 1000.0}, {"bw": ">5000"}) + assert met2 is False and results2["bw"]["met"] is False + + def test_less_than(self): + met, _ = evaluate_targets({"ripple": 0.05}, {"ripple": "<0.1"}) + assert met is True + + def test_approx_within_ten_percent(self): + assert evaluate_targets({"f": 1050.0}, {"f": "~1000"})[0] is True # 5% off -> met + assert evaluate_targets({"f": 1200.0}, {"f": "~1000"})[0] is False # 20% off -> unmet + + def test_bare_number_is_approx(self): + assert evaluate_targets({"f": 1000.0}, {"f": "1000"})[0] is True + + def test_unmeasurable_metric(self): + met, results = evaluate_targets({"rms": 1.0}, {"bandwidth_hz": ">500"}) + assert met is False + assert results["bandwidth_hz"]["status"] == "unmeasurable" + + +class TestMakeSuggestions: + def test_met_yields_no_suggestions(self): + _, results = evaluate_targets({"bw": 1000.0}, {"bw": ">500"}) + assert make_suggestions(results, True) == [] + + def test_bandwidth_increase_hint(self): + _, results = evaluate_targets({"bandwidth_hz": 100.0}, {"bandwidth_hz": ">5000"}) + out = make_suggestions(results, False) + assert any("increase bandwidth" in s for s in out) + + def test_unmeasurable_falls_back_to_generic(self): + _, results = evaluate_targets({"rms": 1.0}, {"bandwidth_hz": ">500"}) + out = make_suggestions(results, False) + assert out == [ + "Adjust component values toward the target. Use smaller steps for fine-tuning." + ] From b634eb9cd17786d96cb05235de05f30b8ad0909d Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sat, 20 Jun 2026 21:06:02 -0600 Subject: [PATCH 08/11] Extract plot_waveform core into plotting.py plot_waveform was the largest tool at 207 lines, interleaving signal resolution, auto plot-type detection, X-range clipping, downsampling, and svg_plot dispatch with the .raw read and SVG write. Move the RawFile -> SVG transform into plotting.build_waveform_svg (returns the SVG string + metadata, no filesystem); the tool keeps only the parse and write I/O. server.py: 2799 -> 2677 lines. Output verified byte-identical (SVG sha256 + metadata) to the prior version across bode/spectrum/time, x-range+downsample, custom dimensions, and both error paths. --- src/mcltspice/plotting.py | 176 ++++++++++++++++++++++++++++++++++++++ src/mcltspice/server.py | 164 +++++------------------------------ 2 files changed, 197 insertions(+), 143 deletions(-) create mode 100644 src/mcltspice/plotting.py diff --git a/src/mcltspice/plotting.py b/src/mcltspice/plotting.py new file mode 100644 index 0000000..f1c9869 --- /dev/null +++ b/src/mcltspice/plotting.py @@ -0,0 +1,176 @@ +"""Waveform-plot construction, separated from file I/O. + +build_waveform_svg() turns a parsed RawFile into an SVG string plus metadata. +It does no filesystem access, so it can be unit-tested with a hand-built +RawFile -- the plot_waveform MCP tool wraps it with the .raw read and SVG write. +""" + +import numpy as np + +from .raw_parser import RawFile +from .svg_plot import plot_bode, plot_spectrum, plot_timeseries, plot_timeseries_multi + + +def build_waveform_svg( + raw_data: RawFile, + signal_list: list[str], + plot_type: str = "auto", + 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: + """Render one or more signals from a RawFile to an SVG string. + + Returns {"svg", "plot_type", "signal", "points", "dimensions"} on success, + or {"error", ...} if a signal is missing or a multi-signal overlay is + requested for a non-time-domain plot. plot_type "auto" picks bode (complex + frequency data), spectrum (real frequency data), or time (everything else). + """ + is_multi = len(signal_list) > 1 + + # Resolve all signal names (exact, case-insensitive) + sig_names = [v.name for v in raw_data.variables] + sig_lower = {s.lower(): s for s in sig_names} + resolved: list[tuple[str, int]] = [] # (actual_name, index) + for s in signal_list: + idx = raw_data.resolve_index(s) + if idx is None: + return { + "error": f"Signal '{s}' not found", + "available_signals": sig_names, + } + resolved.append((sig_lower[s.lower()], idx)) + + # Get x-axis data (time or frequency) + x_var = raw_data.variables[0] + x_data = raw_data.data[0] + is_freq = x_var.name.lower() == "frequency" + + # For single-signal: check complex data (AC analysis) + first_values = raw_data.data[resolved[0][1]] + is_complex = bool(np.iscomplexobj(first_values)) + + # Determine plot type + if plot_type == "auto": + if is_freq and is_complex: + plot_type = "bode" + elif is_freq: + plot_type = "spectrum" + else: + plot_type = "time" + + # Multi-signal only supported for time-domain + if is_multi and plot_type != "time": + return { + "error": "Multi-signal overlay is only supported for time-domain plots.", + "plot_type": plot_type, + "hint": "Use single signal for bode/spectrum, or set plot_type='time'.", + } + + # Clip to X range (before downsampling for efficiency) + real_x = np.real(x_data) + mask = None + if x_min is not None or x_max is not None: + mask = np.ones(len(real_x), dtype=bool) + if x_min is not None: + mask &= real_x >= x_min + if x_max is not None: + mask &= real_x <= x_max + x_data = x_data[mask] + real_x = real_x[mask] + + # Downsample (stride-based) + step = None + if max_points and len(x_data) > max_points: + step = max(1, len(x_data) // max_points) + x_data = x_data[::step] + real_x = real_x[::step] + + # Height default + if height is None: + height = 500 if plot_type == "bode" else 400 + + if is_multi: + # Multi-signal time-domain overlay + traces: list[tuple[str, np.ndarray]] = [] + for name, idx in resolved: + v = raw_data.data[idx] + if mask is not None: + v = v[mask] + if step is not None: + v = v[::step] + traces.append((name, np.real(v))) + + if title is None: + names_str = ", ".join(n for n, _ in resolved) + title = f"Time Domain — {names_str}" + + svg = plot_timeseries_multi( + time=real_x, traces=traces, + title=title, width=width, height=height, + x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, + ) + actual_name = ", ".join(n for n, _ in resolved) + else: + # Single-signal path + actual_name, sig_idx = resolved[0] + values = raw_data.data[sig_idx] + if mask is not None: + values = values[mask] + if step is not None: + values = values[::step] + + if np.iscomplexobj(values): + mag_db = 20.0 * np.log10(np.maximum(np.abs(values), 1e-30)) + phase_deg = np.degrees(np.angle(values)) + else: + mag_db = None + phase_deg = None + + # Title default + if title is None: + if plot_type == "bode": + title = f"Bode Plot — {actual_name}" + elif plot_type == "spectrum": + title = f"Spectrum — {actual_name}" + else: + title = f"Time Domain — {actual_name}" + + # Generate SVG + if plot_type == "bode": + if mag_db is None: + mag_db = np.real(values) + phase_deg = None + svg = plot_bode( + freq=real_x, mag_db=mag_db, phase_deg=phase_deg, + title=title, width=width, height=height, + x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, + ) + elif plot_type == "spectrum": + if mag_db is None: + mag_db = 20.0 * np.log10(np.maximum(np.abs(np.real(values)), 1e-30)) + svg = plot_spectrum( + freq=real_x, mag_db=mag_db, + title=title, width=width, height=height, + x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, + ) + else: # time + svg = plot_timeseries( + time=real_x, values=np.real(values), + title=title, ylabel=actual_name, + width=width, height=height, + x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, + ) + + return { + "svg": svg, + "plot_type": plot_type, + "signal": actual_name, + "points": len(real_x), + "dimensions": f"{width}x{height}", + } diff --git a/src/mcltspice/server.py b/src/mcltspice/server.py index 9e5cc7d..426998d 100644 --- a/src/mcltspice/server.py +++ b/src/mcltspice/server.py @@ -49,6 +49,7 @@ from .optimizer import ( format_engineering, optimize_component_values, ) +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 @@ -59,7 +60,6 @@ from .spicebook import ( ltspice_to_ngspice, ) from .stability import compute_stability_metrics -from .svg_plot import plot_bode, plot_spectrum, plot_timeseries, plot_timeseries_multi from .templates import ( ASC_TEMPLATES, NETLIST_TEMPLATES, @@ -1296,158 +1296,36 @@ async def plot_waveform( # Build signal list (signals overrides signal when set) signal_list = signals if signals else [signal] - is_multi = len(signal_list) > 1 - # Resolve all signal names (exact, case-insensitive) - sig_names = [v.name for v in raw_data.variables] - sig_lower = {s.lower(): s for s in sig_names} - resolved: list[tuple[str, int]] = [] # (actual_name, index) - for s in signal_list: - idx = raw_data.resolve_index(s) - if idx is None: - return { - "error": f"Signal '{s}' not found", - "available_signals": sig_names, - } - resolved.append((sig_lower[s.lower()], idx)) - - # Get x-axis data (time or frequency) - x_var = raw_data.variables[0] - x_data = raw_data.data[0] - is_freq = x_var.name.lower() == "frequency" - - # For single-signal: check complex data (AC analysis) - first_values = raw_data.data[resolved[0][1]] - if np.iscomplexobj(first_values): - is_complex = True - else: - is_complex = False - - # Determine plot type - if plot_type == "auto": - if is_freq and is_complex: - plot_type = "bode" - elif is_freq: - plot_type = "spectrum" - else: - plot_type = "time" - - # Multi-signal only supported for time-domain - if is_multi and plot_type != "time": - return { - "error": "Multi-signal overlay is only supported for time-domain plots.", - "plot_type": plot_type, - "hint": "Use single signal for bode/spectrum, or set plot_type='time'.", - } - - # Clip to X range (before downsampling for efficiency) - real_x = np.real(x_data) - mask = None - if x_min is not None or x_max is not None: - mask = np.ones(len(real_x), dtype=bool) - if x_min is not None: - mask &= real_x >= x_min - if x_max is not None: - mask &= real_x <= x_max - x_data = x_data[mask] - real_x = real_x[mask] - - # Downsample (stride-based, same pattern as get_waveform) - step = None - if max_points and len(x_data) > max_points: - step = max(1, len(x_data) // max_points) - x_data = x_data[::step] - real_x = real_x[::step] - - # Height default - if height is None: - height = 500 if plot_type == "bode" else 400 - - if is_multi: - # Multi-signal time-domain overlay - traces: list[tuple[str, np.ndarray]] = [] - for name, idx in resolved: - v = raw_data.data[idx] - if mask is not None: - v = v[mask] - if step is not None: - v = v[::step] - traces.append((name, np.real(v))) - - if title is None: - names_str = ", ".join(n for n, _ in resolved) - title = f"Time Domain \u2014 {names_str}" - - svg = plot_timeseries_multi( - time=real_x, traces=traces, - title=title, width=width, height=height, - x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, - ) - actual_name = ", ".join(n for n, _ in resolved) - else: - # Single-signal path (original logic) - actual_name, sig_idx = resolved[0] - values = raw_data.data[sig_idx] - if mask is not None: - values = values[mask] - if step is not None: - values = values[::step] - - if np.iscomplexobj(values): - mag_db = 20.0 * np.log10(np.maximum(np.abs(values), 1e-30)) - phase_deg = np.degrees(np.angle(values)) - else: - mag_db = None - phase_deg = None - - # Title default - if title is None: - if plot_type == "bode": - title = f"Bode Plot \u2014 {actual_name}" - elif plot_type == "spectrum": - title = f"Spectrum \u2014 {actual_name}" - else: - title = f"Time Domain \u2014 {actual_name}" - - # Generate SVG - if plot_type == "bode": - if mag_db is None: - mag_db = np.real(values) - phase_deg = None - svg = plot_bode( - freq=real_x, mag_db=mag_db, phase_deg=phase_deg, - title=title, width=width, height=height, - x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, - ) - elif plot_type == "spectrum": - if mag_db is None: - mag_db = 20.0 * np.log10(np.maximum(np.abs(np.real(values)), 1e-30)) - svg = plot_spectrum( - freq=real_x, mag_db=mag_db, - title=title, width=width, height=height, - x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, - ) - else: # time - svg = plot_timeseries( - time=real_x, values=np.real(values), - title=title, ylabel=actual_name, - width=width, height=height, - x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, - ) + 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(svg) + out.write_text(result["svg"]) return { "svg_path": str(out), - "plot_type": plot_type, - "signal": actual_name, - "points": len(real_x), - "dimensions": f"{width}x{height}", + "plot_type": result["plot_type"], + "signal": result["signal"], + "points": result["points"], + "dimensions": result["dimensions"], } From 35b12e61c47790fc7fb19561d0c2834ad0a73c17 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sat, 20 Jun 2026 21:06:35 -0600 Subject: [PATCH 09/11] Test build_waveform_svg without a simulator or filesystem 11 tests over auto plot-type detection (bode/spectrum/time), missing-signal and multi-signal-overlay errors, dimension/height defaults, downsampling, and x-range clipping -- all with hand-built RawFiles, no .raw read or SVG write. --- tests/test_plotting.py | 103 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 103 insertions(+) create mode 100644 tests/test_plotting.py diff --git a/tests/test_plotting.py b/tests/test_plotting.py new file mode 100644 index 0000000..774739f --- /dev/null +++ b/tests/test_plotting.py @@ -0,0 +1,103 @@ +"""Tests for build_waveform_svg -- the plot_waveform core, sans file I/O. + +Hand-built RawFiles drive every branch (auto plot-type detection, clipping, +multi-signal, errors) without LTspice or writing an SVG to disk. +""" + +import numpy as np + +from mcltspice.plotting import build_waveform_svg +from mcltspice.raw_parser import RawFile, Variable + + +def _ac_raw() -> RawFile: + """Frequency axis + complex V(out), V(in) -> AC analysis.""" + freq = np.logspace(0, 6, 400) + hout = 1.0 / (1.0 + 1j * (freq / 1000.0)) + data = np.array([freq.astype(complex), hout, np.ones_like(freq, dtype=complex)]) + return RawFile( + title="ac", date="", plotname="AC Analysis", flags=["complex"], + variables=[ + Variable(0, "frequency", "frequency"), + Variable(1, "V(out)", "voltage"), + Variable(2, "V(in)", "voltage"), + ], + points=len(freq), data=data, + ) + + +def _tran_raw() -> RawFile: + """Time axis + two real signals -> transient.""" + t = np.linspace(0, 0.01, 500) + data = np.array([t, np.sin(2 * np.pi * 1000 * t), np.cos(2 * np.pi * 1000 * t)]) + return RawFile( + title="tran", date="", plotname="Transient Analysis", flags=["real"], + variables=[ + Variable(0, "time", "time"), + Variable(1, "V(out)", "voltage"), + Variable(2, "V(ref)", "voltage"), + ], + points=len(t), data=data, + ) + + +class TestAutoPlotType: + def test_complex_frequency_is_bode(self): + r = build_waveform_svg(_ac_raw(), ["V(out)"]) + assert r["plot_type"] == "bode" + assert r["svg"].lstrip().startswith("<") # real SVG markup + + def test_real_frequency_is_spectrum(self): + # frequency axis but real signal data -> spectrum + freq = np.logspace(0, 6, 200) + data = np.array([freq, np.abs(1.0 / (1.0 + 1j * freq / 1000.0))]) + raw = RawFile( + title="", date="", plotname="AC Analysis", flags=["real"], + variables=[Variable(0, "frequency", "frequency"), Variable(1, "V(out)", "voltage")], + points=len(freq), data=data, + ) + assert build_waveform_svg(raw, ["V(out)"])["plot_type"] == "spectrum" + + def test_time_axis_is_time(self): + assert build_waveform_svg(_tran_raw(), ["V(out)"])["plot_type"] == "time" + + def test_explicit_overrides_auto(self): + assert build_waveform_svg(_tran_raw(), ["V(out)"], plot_type="time")["plot_type"] == "time" + + +class TestErrorsAndMetadata: + def test_missing_signal(self): + r = build_waveform_svg(_ac_raw(), ["V(ghost)"]) + assert "error" in r and "V(ghost)" in r["error"] + assert r["available_signals"] == ["frequency", "V(out)", "V(in)"] + + def test_multi_signal_non_time_is_error(self): + r = build_waveform_svg(_ac_raw(), ["V(out)", "V(in)"], plot_type="bode") + assert "error" in r and "time-domain" in r["error"] + + def test_dimensions_and_signal_metadata(self): + r = build_waveform_svg(_tran_raw(), ["V(out)"], width=1000, height=600) + assert r["dimensions"] == "1000x600" + assert r["signal"] == "V(out)" + + def test_default_height_depends_on_plot_type(self): + assert build_waveform_svg(_ac_raw(), ["V(out)"])["dimensions"] == "800x500" # bode + assert build_waveform_svg(_tran_raw(), ["V(out)"])["dimensions"] == "800x400" # time + + +class TestSamplingAndMulti: + def test_max_points_downsamples(self): + r = build_waveform_svg(_tran_raw(), ["V(out)"], max_points=100) + assert r["points"] <= 100 + + def test_x_range_clip_reduces_points(self): + full = build_waveform_svg(_tran_raw(), ["V(out)"], max_points=10000)["points"] + clipped = build_waveform_svg( + _tran_raw(), ["V(out)"], x_min=0.002, x_max=0.004, max_points=10000 + )["points"] + assert clipped < full + + def test_multi_signal_time_overlay(self): + r = build_waveform_svg(_tran_raw(), ["V(out)", "V(ref)"]) + assert r["plot_type"] == "time" + assert r["signal"] == "V(out), V(ref)" From 1118ac2be12214ff6f9a53d98cdb4e7c3a0fc3f9 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sat, 20 Jun 2026 22:25:00 -0600 Subject: [PATCH 10/11] Extract get_waveform/analyze_waveform cores; relocate output helpers Move the shared JSON-compaction helpers (round_sig, compact_list, downsample_indices) out of server.py into output_format.py, and extract the get_waveform and analyze_waveform cores into waveform_query.py (extract_waveform, analyze_signal) -- both take a parsed RawFile and return a JSON-ready dict, so they test without a simulator. The tools keep only the .raw read. server.py: 2677 -> 2472 lines. Output verified byte-identical across AC/transient/downsample/error/all-analysis cases. --- src/mcltspice/output_format.py | 81 +++++++++++ src/mcltspice/server.py | 249 +++----------------------------- src/mcltspice/waveform_query.py | 197 +++++++++++++++++++++++++ 3 files changed, 300 insertions(+), 227 deletions(-) create mode 100644 src/mcltspice/output_format.py create mode 100644 src/mcltspice/waveform_query.py diff --git a/src/mcltspice/output_format.py b/src/mcltspice/output_format.py new file mode 100644 index 0000000..314d772 --- /dev/null +++ b/src/mcltspice/output_format.py @@ -0,0 +1,81 @@ +"""Output-formatting helpers for compact JSON responses. + +Full float64 precision (~17 digits) bloats JSON output; these round to an +appropriate precision and downsample long arrays before serialization. +""" + +import math + +import numpy as np + + +def round_sig(x: float, sig: int = 6) -> float: + """Round to N significant figures for compact JSON output.""" + if x == 0 or not math.isfinite(x): + return float(x) + digits = sig - 1 - int(math.floor(math.log10(abs(x)))) + return round(x, digits) + + +def compact_list( + values, decimal_places: int | None = None, sig_figs: int | None = None +) -> list[float]: + """Round a list of floats for compact JSON output. + + Full float64 precision (~17 digits) causes massive JSON bloat. + Rounding to appropriate precision cuts output size 50-60%. + """ + if decimal_places is not None: + return [round(float(v), decimal_places) for v in values] + if sig_figs is not None: + return [round_sig(float(v), sig_figs) for v in values] + return [float(v) for v in values] + + +def downsample_indices( + n_total: int, + max_points: int, + signals: list[np.ndarray] | None = None, +) -> np.ndarray: + """Compute indices for downsampling, preserving peaks in AC data. + + For AC data (when complex signal arrays are provided), uses + peak-preserving bucket selection: divides the data into max_points + buckets and keeps the point in each bucket whose magnitude deviates + most from the bucket mean. This ensures narrow resonance peaks + (e.g. high-Q bandpass filters) aren't lost by blind stride sampling. + + For transient data (no signals provided), uses simple stride. + """ + if n_total <= max_points: + return np.arange(n_total) + + if signals is None or len(signals) == 0: + step = max(1, n_total // max_points) + return np.arange(0, n_total, step) + + # Pre-compute magnitude_dB for each signal + mag_db_list = [] + for sig in signals: + mag = np.abs(sig) + with np.errstate(divide="ignore"): + db = np.where(mag > 0, 20 * np.log10(mag), -200.0) + mag_db_list.append(db) + + # Bucket-based peak-preserving selection + indices = [] + bucket_size = n_total / max_points + + for i in range(max_points): + start = int(i * bucket_size) + end = min(int((i + 1) * bucket_size), n_total) + if start >= end: + continue + # Combined importance: sum of |deviation from bucket mean| across signals + combined = np.zeros(end - start) + for db in mag_db_list: + bucket = db[start:end] + combined += np.abs(bucket - np.mean(bucket)) + indices.append(start + int(np.argmax(combined))) + + return np.array(indices) diff --git a/src/mcltspice/server.py b/src/mcltspice/server.py index 426998d..616142e 100644 --- a/src/mcltspice/server.py +++ b/src/mcltspice/server.py @@ -49,6 +49,7 @@ from .optimizer import ( 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 @@ -77,13 +78,8 @@ from .tuning import ( from .waveform_expr import WaveformCalculator from .waveform_math import ( compute_bandwidth, - compute_fft, - compute_peak_to_peak, - compute_rise_time, - compute_rms, - compute_settling_time, - compute_thd, ) +from .waveform_query import analyze_signal, extract_waveform mcp = FastMCP( name="mcltspice", @@ -125,83 +121,6 @@ mcp = FastMCP( ) -# ============================================================================ -# OUTPUT HELPERS -# ============================================================================ - - -def _round_sig(x: float, sig: int = 6) -> float: - """Round to N significant figures for compact JSON output.""" - if x == 0 or not math.isfinite(x): - return float(x) - digits = sig - 1 - int(math.floor(math.log10(abs(x)))) - return round(x, digits) - - -def _compact_list( - values, decimal_places: int | None = None, sig_figs: int | None = None -) -> list[float]: - """Round a list of floats for compact JSON output. - - Full float64 precision (~17 digits) causes massive JSON bloat. - Rounding to appropriate precision cuts output size 50-60%. - """ - if decimal_places is not None: - return [round(float(v), decimal_places) for v in values] - if sig_figs is not None: - return [_round_sig(float(v), sig_figs) for v in values] - return [float(v) for v in values] - - -def _downsample_indices( - n_total: int, - max_points: int, - signals: list[np.ndarray] | None = None, -) -> np.ndarray: - """Compute indices for downsampling, preserving peaks in AC data. - - For AC data (when complex signal arrays are provided), uses - peak-preserving bucket selection: divides the data into max_points - buckets and keeps the point in each bucket whose magnitude deviates - most from the bucket mean. This ensures narrow resonance peaks - (e.g. high-Q bandpass filters) aren't lost by blind stride sampling. - - For transient data (no signals provided), uses simple stride. - """ - if n_total <= max_points: - return np.arange(n_total) - - if signals is None or len(signals) == 0: - step = max(1, n_total // max_points) - return np.arange(0, n_total, step) - - # Pre-compute magnitude_dB for each signal - mag_db_list = [] - for sig in signals: - mag = np.abs(sig) - with np.errstate(divide="ignore"): - db = np.where(mag > 0, 20 * np.log10(mag), -200.0) - mag_db_list.append(db) - - # Bucket-based peak-preserving selection - indices = [] - bucket_size = n_total / max_points - - for i in range(max_points): - start = int(i * bucket_size) - end = min(int((i + 1) * bucket_size), n_total) - if start >= end: - continue - # Combined importance: sum of |deviation from bucket mean| across signals - combined = np.zeros(end - start) - for db in mag_db_list: - bucket = db[start:end] - combined += np.abs(bucket - np.mean(bucket)) - indices.append(start + int(np.argmax(combined))) - - return np.array(indices) - - # ============================================================================ # SIMULATION TOOLS # ============================================================================ @@ -329,89 +248,14 @@ def get_waveform( x_max: Maximum x-axis value (frequency Hz or time s) to include """ raw = parse_raw_file(raw_file_path) - - # Extract specific run if requested - if run is not None: - if not raw.is_stepped: - return {"error": "Not a stepped simulation - no multiple runs available"} - if run < 1 or run > raw.n_runs: - return {"error": f"Run {run} out of range (1..{raw.n_runs})"} - raw = raw.get_run_data(run) - - x_axis = raw.get_time() - x_name = "time" - if x_axis is None: - x_axis = raw.get_frequency() - x_name = "frequency" - - if x_axis is None: - return {"error": "No time or frequency axis found in raw file"} - - x_real = x_axis.real if np.iscomplexobj(x_axis) else x_axis - is_complex = np.iscomplexobj(raw.data) - - # Apply x-axis range filter - 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_indices = np.where(mask)[0] - - if len(filtered_indices) == 0: - return { - "error": f"No data points in {x_name} range [{x_min}, {x_max}]", - "total_points": len(x_real), - f"{x_name}_range": [float(x_real[0]), float(x_real[-1])], - } - - # Collect signal data for peak-preserving downsampling (AC only) - ac_signals = [] - if is_complex: - for name in signal_names: - data = raw.get_variable(name) - if data is not None: - ac_signals.append(data[filtered_indices]) - - # Downsample: peak-preserving for AC, stride for transient - ds_indices = _downsample_indices( - len(filtered_indices), - max_points, - signals=ac_signals if ac_signals else None, + return extract_waveform( + raw, + signal_names, + max_points=max_points, + run=run, + x_min=x_min, + x_max=x_max, ) - sample_indices = filtered_indices[ds_indices] - - result = { - "x_axis_name": x_name, - "x_axis_data": _compact_list(x_real[sample_indices], sig_figs=6), - "signals": {}, - "total_points": len(x_real), - "returned_points": len(sample_indices), - "is_stepped": raw.is_stepped, - "n_runs": raw.n_runs, - } - - for name in signal_names: - data = raw.get_variable(name) - if data is not None: - sampled = data[sample_indices] - if np.iscomplexobj(sampled): - result["signals"][name] = { - "magnitude_db": _compact_list( - [20 * math.log10(abs(x)) if abs(x) > 0 else -200 for x in sampled], - decimal_places=2, - ), - "phase_degrees": _compact_list( - [math.degrees(math.atan2(x.imag, x.real)) for x in sampled], - decimal_places=2, - ), - } - else: - result["signals"][name] = { - "values": _compact_list(sampled, sig_figs=6) - } - - return result @mcp.tool() @@ -484,66 +328,17 @@ def analyze_waveform( thd_n_harmonics: Number of harmonics for THD calculation """ raw = parse_raw_file(raw_file_path) - - time = raw.get_time() - signal = raw.get_variable(signal_name) - - if signal is None: - return { - "error": f"Signal '{signal_name}' not found. Available: " - f"{[v.name for v in raw.variables]}" - } - - # Use real parts for time-domain analysis - if np.iscomplexobj(time): - time = time.real - if np.iscomplexobj(signal): - signal = np.abs(signal) - - results = {"signal": signal_name} - - for analysis in analyses: - if analysis == "rms": - results["rms"] = compute_rms(signal) - - elif analysis == "peak_to_peak": - results["peak_to_peak"] = compute_peak_to_peak(signal) - - elif analysis == "settling_time": - if time is not None: - results["settling_time"] = compute_settling_time( - time, - signal, - final_value=settling_final_value, - tolerance_percent=settling_tolerance_pct, - ) - - elif analysis == "rise_time": - if time is not None: - results["rise_time"] = compute_rise_time( - time, - signal, - low_pct=rise_low_pct, - high_pct=rise_high_pct, - ) - - elif analysis == "fft": - if time is not None: - results["fft"] = compute_fft( - time, - signal, - max_harmonics=fft_max_harmonics, - ) - - elif analysis == "thd": - if time is not None: - results["thd"] = compute_thd( - time, - signal, - n_harmonics=thd_n_harmonics, - ) - - return results + 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() @@ -1052,9 +847,9 @@ def evaluate_waveform_expression( 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["x_axis_data"] = compact_list(x_real[sample_indices], sig_figs=6) - response["values"] = _compact_list(result[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/waveform_query.py b/src/mcltspice/waveform_query.py new file mode 100644 index 0000000..aab81c7 --- /dev/null +++ b/src/mcltspice/waveform_query.py @@ -0,0 +1,197 @@ +"""Read-side queries over a parsed RawFile: data extraction and analysis. + +Both functions take an already-parsed RawFile and return a JSON-ready dict, so +they can be unit-tested without an LTspice run. The get_waveform / analyze_waveform +MCP tools wrap these with the .raw file read. +""" + +import math + +import numpy as np + +from .output_format import compact_list, downsample_indices +from .raw_parser import RawFile +from .waveform_math import ( + compute_fft, + compute_peak_to_peak, + compute_rise_time, + compute_rms, + compute_settling_time, + compute_thd, +) + + +def extract_waveform( + raw: RawFile, + signal_names: list[str], + max_points: int = 1000, + run: int | None = None, + x_min: float | None = None, + x_max: float | None = None, +) -> dict: + """Extract sampled signal data from a RawFile. + + Returns the x-axis plus per-signal values (magnitude_db/phase for AC, + values for transient), or an {"error": ...} dict for a bad run number or an + empty x-range. AC data is downsampled peak-preservingly; transient by stride. + """ + # Extract specific run if requested + if run is not None: + if not raw.is_stepped: + return {"error": "Not a stepped simulation - no multiple runs available"} + if run < 1 or run > raw.n_runs: + return {"error": f"Run {run} out of range (1..{raw.n_runs})"} + raw = raw.get_run_data(run) + + x_axis = raw.get_time() + x_name = "time" + if x_axis is None: + x_axis = raw.get_frequency() + x_name = "frequency" + + if x_axis is None: + return {"error": "No time or frequency axis found in raw file"} + + x_real = x_axis.real if np.iscomplexobj(x_axis) else x_axis + is_complex = np.iscomplexobj(raw.data) + + # Apply x-axis range filter + 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_indices = np.where(mask)[0] + + if len(filtered_indices) == 0: + return { + "error": f"No data points in {x_name} range [{x_min}, {x_max}]", + "total_points": len(x_real), + f"{x_name}_range": [float(x_real[0]), float(x_real[-1])], + } + + # Collect signal data for peak-preserving downsampling (AC only) + ac_signals = [] + if is_complex: + for name in signal_names: + data = raw.get_variable(name) + if data is not None: + ac_signals.append(data[filtered_indices]) + + # Downsample: peak-preserving for AC, stride for transient + ds_indices = downsample_indices( + len(filtered_indices), + max_points, + signals=ac_signals if ac_signals else None, + ) + sample_indices = filtered_indices[ds_indices] + + result = { + "x_axis_name": x_name, + "x_axis_data": compact_list(x_real[sample_indices], sig_figs=6), + "signals": {}, + "total_points": len(x_real), + "returned_points": len(sample_indices), + "is_stepped": raw.is_stepped, + "n_runs": raw.n_runs, + } + + for name in signal_names: + data = raw.get_variable(name) + if data is not None: + sampled = data[sample_indices] + if np.iscomplexobj(sampled): + result["signals"][name] = { + "magnitude_db": compact_list( + [20 * math.log10(abs(x)) if abs(x) > 0 else -200 for x in sampled], + decimal_places=2, + ), + "phase_degrees": compact_list( + [math.degrees(math.atan2(x.imag, x.real)) for x in sampled], + decimal_places=2, + ), + } + else: + result["signals"][name] = { + "values": compact_list(sampled, sig_figs=6) + } + + return result + + +def analyze_signal( + raw: RawFile, + 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: + """Run one or more analyses on a signal from a RawFile. + + Supported analyses: rms, peak_to_peak, settling_time, rise_time, fft, thd. + Returns a dict keyed by analysis name, or {"error": ...} if the signal is + absent. Frequency-domain analyses are skipped when there's no time axis. + """ + time = raw.get_time() + signal = raw.get_variable(signal_name) + + if signal is None: + return { + "error": f"Signal '{signal_name}' not found. Available: " + f"{[v.name for v in raw.variables]}" + } + + # Use real parts for time-domain analysis + if np.iscomplexobj(time): + time = time.real + if np.iscomplexobj(signal): + signal = np.abs(signal) + + results = {"signal": signal_name} + + for analysis in analyses: + if analysis == "rms": + results["rms"] = compute_rms(signal) + + elif analysis == "peak_to_peak": + results["peak_to_peak"] = compute_peak_to_peak(signal) + + elif analysis == "settling_time": + if time is not None: + results["settling_time"] = compute_settling_time( + time, + signal, + final_value=settling_final_value, + tolerance_percent=settling_tolerance_pct, + ) + + elif analysis == "rise_time": + if time is not None: + results["rise_time"] = compute_rise_time( + time, + signal, + low_pct=rise_low_pct, + high_pct=rise_high_pct, + ) + + elif analysis == "fft": + if time is not None: + results["fft"] = compute_fft( + time, + signal, + max_harmonics=fft_max_harmonics, + ) + + elif analysis == "thd": + if time is not None: + results["thd"] = compute_thd( + time, + signal, + n_harmonics=thd_n_harmonics, + ) + + return results From 90cd07d9bc89a97c65870ceb716fd27da435cc33 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sat, 20 Jun 2026 22:25:54 -0600 Subject: [PATCH 11/11] Test output_format and waveform_query without a simulator 17 tests: round_sig/compact_list rounding, downsample_indices stride vs peak-preserving (a narrow spike must survive), and extract_waveform/ analyze_signal across AC/transient/x-range/error/all-analysis paths. --- tests/test_output_format.py | 52 ++++++++++++++++++++++++ tests/test_waveform_query.py | 77 ++++++++++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+) create mode 100644 tests/test_output_format.py create mode 100644 tests/test_waveform_query.py diff --git a/tests/test_output_format.py b/tests/test_output_format.py new file mode 100644 index 0000000..56cab11 --- /dev/null +++ b/tests/test_output_format.py @@ -0,0 +1,52 @@ +"""Tests for the JSON output-formatting helpers.""" + +import math + +import numpy as np + +from mcltspice.output_format import compact_list, downsample_indices, round_sig + + +class TestRoundSig: + def test_significant_figures(self): + assert round_sig(123456.789, 3) == 123000.0 + assert round_sig(0.00123456, 3) == 0.00123 + + def test_zero_and_nonfinite_pass_through(self): + assert round_sig(0.0) == 0.0 + assert math.isinf(round_sig(float("inf"))) + assert math.isnan(round_sig(float("nan"))) + + +class TestCompactList: + def test_decimal_places(self): + assert compact_list([1.23456, 2.98765], decimal_places=2) == [1.23, 2.99] + + def test_sig_figs(self): + assert compact_list([123456.0], sig_figs=3) == [123000.0] + + def test_no_rounding_returns_floats(self): + out = compact_list([1, 2, 3]) + assert out == [1.0, 2.0, 3.0] + assert all(isinstance(v, float) for v in out) + + +class TestDownsampleIndices: + def test_no_downsample_when_under_limit(self): + idx = downsample_indices(50, 100) + assert np.array_equal(idx, np.arange(50)) + + def test_stride_for_transient(self): + idx = downsample_indices(1000, 100) + assert len(idx) <= 100 + # plain stride: evenly spaced + assert idx[0] == 0 + assert idx[1] - idx[0] == idx[2] - idx[1] + + def test_peak_preserving_keeps_a_narrow_spike(self): + # A flat magnitude with one sharp spike: peak-preserving must keep it. + n = 1000 + mag = np.ones(n, dtype=complex) + mag[497] = 1000.0 # narrow resonance peak + idx = downsample_indices(n, 50, signals=[mag]) + assert 497 in idx # spike survived downsampling diff --git a/tests/test_waveform_query.py b/tests/test_waveform_query.py new file mode 100644 index 0000000..7ecc0e7 --- /dev/null +++ b/tests/test_waveform_query.py @@ -0,0 +1,77 @@ +"""Tests for extract_waveform / analyze_signal -- RawFile queries sans simulator.""" + +import numpy as np + +from mcltspice.raw_parser import RawFile, Variable +from mcltspice.waveform_query import analyze_signal, extract_waveform + + +def _ac_raw() -> RawFile: + freq = np.logspace(0, 6, 300) + hout = 1.0 / (1.0 + 1j * (freq / 1000.0)) + data = np.array([freq.astype(complex), hout]) + return RawFile( + title="", date="", plotname="AC Analysis", flags=["complex"], + variables=[Variable(0, "frequency", "frequency"), Variable(1, "V(out)", "voltage")], + points=len(freq), data=data, + ) + + +def _tran_raw() -> RawFile: + t = np.linspace(0, 0.01, 800) + v = 2.0 * np.sin(2 * np.pi * 1000 * t) + return RawFile( + title="", date="", plotname="Transient Analysis", flags=["real"], + variables=[Variable(0, "time", "time"), Variable(1, "V(out)", "voltage")], + points=len(t), data=np.array([t, v]), + ) + + +class TestExtractWaveform: + def test_transient_values(self): + r = extract_waveform(_tran_raw(), ["V(out)"], max_points=100) + assert r["x_axis_name"] == "time" + assert r["returned_points"] <= 100 + assert "values" in r["signals"]["V(out)"] + + def test_ac_gives_magnitude_and_phase(self): + r = extract_waveform(_ac_raw(), ["V(out)"]) + assert r["x_axis_name"] == "frequency" + sig = r["signals"]["V(out)"] + assert "magnitude_db" in sig and "phase_degrees" in sig + + def test_x_range_filter(self): + full = extract_waveform(_tran_raw(), ["V(out)"], max_points=10000)["returned_points"] + clipped = extract_waveform( + _tran_raw(), ["V(out)"], max_points=10000, x_min=0.002, x_max=0.004 + )["returned_points"] + assert clipped < full + + def test_empty_range_error(self): + r = extract_waveform(_tran_raw(), ["V(out)"], x_min=1e12) + assert "error" in r and "range" in r["error"] + + def test_run_on_non_stepped_error(self): + r = extract_waveform(_tran_raw(), ["V(out)"], run=2) + assert "error" in r and "stepped" in r["error"] + + +class TestAnalyzeSignal: + def test_rms_and_peak_to_peak(self): + r = analyze_signal(_tran_raw(), "V(out)", ["rms", "peak_to_peak"]) + assert r["signal"] == "V(out)" + assert abs(r["rms"] - 2.0 / np.sqrt(2)) < 0.05 # RMS of a 2 V amplitude sine + assert abs(r["peak_to_peak"]["peak_to_peak"] - 4.0) < 0.05 + + def test_fft_and_thd_present(self): + r = analyze_signal(_tran_raw(), "V(out)", ["fft", "thd"]) + assert "fft" in r and "thd" in r + + def test_missing_signal_error(self): + r = analyze_signal(_tran_raw(), "V(ghost)", ["rms"]) + assert "error" in r and "V(ghost)" in r["error"] + + def test_complex_signal_uses_magnitude(self): + # AC (complex) signal: analysis should run on |signal| without raising + r = analyze_signal(_ac_raw(), "V(out)", ["rms"]) + assert "rms" in r and r["rms"] >= 0