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