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.
This commit is contained in:
parent
cf1502cee9
commit
dd6db95021
@ -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,
|
||||
|
||||
228
src/mcltspice/tuning.py
Normal file
228
src/mcltspice/tuning.py
Normal file
@ -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), "<N" (less), "~N" or bare "N" (within 10%).
|
||||
Returns (all_targets_met, per_metric_results).
|
||||
"""
|
||||
targets_met = True
|
||||
target_results: dict[str, dict] = {}
|
||||
if not targets:
|
||||
return targets_met, target_results
|
||||
|
||||
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
|
||||
|
||||
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
|
||||
Loading…
x
Reference in New Issue
Block a user