Ryan Malloy
50953a4dea
Wine batch-mode runner, binary .raw parser (UTF-16 LE + mixed precision float64/float32), .asc schematic parser/editor, and 9 FastMCP tools for simulation automation on Linux.
333 lines
10 KiB
Python
333 lines
10 KiB
Python
"""Run LTspice simulations via Wine batch mode."""
|
|
|
|
import asyncio
|
|
import shutil
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from .config import (
|
|
DEFAULT_TIMEOUT,
|
|
LTSPICE_EXE,
|
|
MAX_RAW_FILE_SIZE,
|
|
get_wine_env,
|
|
validate_installation,
|
|
)
|
|
from .raw_parser import RawFile, parse_raw_file
|
|
|
|
|
|
@dataclass
|
|
class SimulationResult:
|
|
"""Result of a simulation run."""
|
|
success: bool
|
|
raw_file: Path | None
|
|
log_file: Path | None
|
|
raw_data: RawFile | None
|
|
error: str | None
|
|
stdout: str
|
|
stderr: str
|
|
elapsed_seconds: float
|
|
|
|
|
|
async def run_simulation(
|
|
schematic_path: Path | str,
|
|
timeout: float = DEFAULT_TIMEOUT,
|
|
work_dir: Path | str | None = None,
|
|
parse_results: bool = True,
|
|
) -> SimulationResult:
|
|
"""Run an LTspice simulation in batch mode.
|
|
|
|
Args:
|
|
schematic_path: Path to .asc schematic file
|
|
timeout: Maximum simulation time in seconds
|
|
work_dir: Working directory for output files (temp dir if None)
|
|
parse_results: Whether to parse the .raw file
|
|
|
|
Returns:
|
|
SimulationResult with status and data
|
|
"""
|
|
import time
|
|
start_time = time.monotonic()
|
|
|
|
# Validate installation
|
|
ok, msg = validate_installation()
|
|
if not ok:
|
|
return SimulationResult(
|
|
success=False,
|
|
raw_file=None,
|
|
log_file=None,
|
|
raw_data=None,
|
|
error=msg,
|
|
stdout="",
|
|
stderr="",
|
|
elapsed_seconds=0,
|
|
)
|
|
|
|
schematic_path = Path(schematic_path).resolve()
|
|
if not schematic_path.exists():
|
|
return SimulationResult(
|
|
success=False,
|
|
raw_file=None,
|
|
log_file=None,
|
|
raw_data=None,
|
|
error=f"Schematic not found: {schematic_path}",
|
|
stdout="",
|
|
stderr="",
|
|
elapsed_seconds=0,
|
|
)
|
|
|
|
# Set up working directory
|
|
temp_dir = None
|
|
if work_dir:
|
|
work_dir = Path(work_dir)
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
else:
|
|
temp_dir = tempfile.mkdtemp(prefix="ltspice_")
|
|
work_dir = Path(temp_dir)
|
|
|
|
try:
|
|
# Copy schematic to work directory
|
|
work_schematic = work_dir / schematic_path.name
|
|
shutil.copy2(schematic_path, work_schematic)
|
|
|
|
# Copy any .lib, .sub files from the same directory
|
|
src_dir = schematic_path.parent
|
|
for ext in [".lib", ".sub", ".inc", ".model"]:
|
|
for f in src_dir.glob(f"*{ext}"):
|
|
shutil.copy2(f, work_dir / f.name)
|
|
|
|
# Convert path to Windows format for Wine
|
|
# Wine maps Z: to root filesystem
|
|
win_path = "Z:" + str(work_schematic).replace("/", "\\")
|
|
|
|
# Build command
|
|
cmd = [
|
|
"wine",
|
|
str(LTSPICE_EXE),
|
|
"-b", # Batch mode
|
|
win_path,
|
|
]
|
|
|
|
# Run simulation
|
|
env = get_wine_env()
|
|
|
|
process = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
env=env,
|
|
cwd=str(work_dir),
|
|
)
|
|
|
|
try:
|
|
stdout_bytes, stderr_bytes = await asyncio.wait_for(
|
|
process.communicate(),
|
|
timeout=timeout
|
|
)
|
|
except asyncio.TimeoutError:
|
|
process.kill()
|
|
await process.wait()
|
|
return SimulationResult(
|
|
success=False,
|
|
raw_file=None,
|
|
log_file=None,
|
|
raw_data=None,
|
|
error=f"Simulation timed out after {timeout} seconds",
|
|
stdout="",
|
|
stderr="",
|
|
elapsed_seconds=time.monotonic() - start_time,
|
|
)
|
|
|
|
stdout = stdout_bytes.decode("utf-8", errors="replace")
|
|
stderr = stderr_bytes.decode("utf-8", errors="replace")
|
|
|
|
elapsed = time.monotonic() - start_time
|
|
|
|
# Look for output files
|
|
raw_file = work_schematic.with_suffix(".raw")
|
|
log_file = work_schematic.with_suffix(".log")
|
|
|
|
if not raw_file.exists():
|
|
# Check for error in log
|
|
error_msg = "Simulation failed - no .raw file produced"
|
|
if log_file.exists():
|
|
log_content = log_file.read_text(errors="replace")
|
|
if "Error" in log_content or "error" in log_content:
|
|
error_msg = f"Simulation error: {log_content[:500]}"
|
|
|
|
return SimulationResult(
|
|
success=False,
|
|
raw_file=None,
|
|
log_file=log_file if log_file.exists() else None,
|
|
raw_data=None,
|
|
error=error_msg,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
elapsed_seconds=elapsed,
|
|
)
|
|
|
|
# Check file size
|
|
if raw_file.stat().st_size > MAX_RAW_FILE_SIZE:
|
|
return SimulationResult(
|
|
success=True,
|
|
raw_file=raw_file,
|
|
log_file=log_file if log_file.exists() else None,
|
|
raw_data=None,
|
|
error=f"Raw file too large to parse ({raw_file.stat().st_size} bytes)",
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
elapsed_seconds=elapsed,
|
|
)
|
|
|
|
# Parse results
|
|
raw_data = None
|
|
parse_error = None
|
|
if parse_results:
|
|
try:
|
|
raw_data = parse_raw_file(raw_file)
|
|
except Exception as e:
|
|
parse_error = f"Failed to parse .raw file: {e}"
|
|
|
|
return SimulationResult(
|
|
success=True,
|
|
raw_file=raw_file,
|
|
log_file=log_file if log_file.exists() else None,
|
|
raw_data=raw_data,
|
|
error=parse_error,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
elapsed_seconds=elapsed,
|
|
)
|
|
|
|
finally:
|
|
# Clean up temp directory (but not user-specified work_dir)
|
|
if temp_dir:
|
|
# Keep files for debugging if simulation failed
|
|
pass # Let user specify cleanup behavior
|
|
|
|
|
|
async def run_netlist(
|
|
netlist_path: Path | str,
|
|
timeout: float = DEFAULT_TIMEOUT,
|
|
work_dir: Path | str | None = None,
|
|
parse_results: bool = True,
|
|
) -> SimulationResult:
|
|
"""Run an LTspice simulation from a netlist (.cir/.net) file.
|
|
|
|
This is similar to run_simulation but for netlist files.
|
|
|
|
Args:
|
|
netlist_path: Path to .cir or .net netlist file
|
|
timeout: Maximum simulation time in seconds
|
|
work_dir: Working directory for output files
|
|
parse_results: Whether to parse the .raw file
|
|
|
|
Returns:
|
|
SimulationResult with status and data
|
|
"""
|
|
import time
|
|
start_time = time.monotonic()
|
|
|
|
ok, msg = validate_installation()
|
|
if not ok:
|
|
return SimulationResult(
|
|
success=False, raw_file=None, log_file=None, raw_data=None,
|
|
error=msg, stdout="", stderr="", elapsed_seconds=0,
|
|
)
|
|
|
|
netlist_path = Path(netlist_path).resolve()
|
|
if not netlist_path.exists():
|
|
return SimulationResult(
|
|
success=False, raw_file=None, log_file=None, raw_data=None,
|
|
error=f"Netlist not found: {netlist_path}",
|
|
stdout="", stderr="", elapsed_seconds=0,
|
|
)
|
|
|
|
temp_dir = None
|
|
if work_dir:
|
|
work_dir = Path(work_dir)
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
else:
|
|
temp_dir = tempfile.mkdtemp(prefix="ltspice_")
|
|
work_dir = Path(temp_dir)
|
|
|
|
try:
|
|
# Copy netlist to work directory
|
|
work_netlist = work_dir / netlist_path.name
|
|
shutil.copy2(netlist_path, work_netlist)
|
|
|
|
# Copy included files
|
|
src_dir = netlist_path.parent
|
|
for ext in [".lib", ".sub", ".inc", ".model"]:
|
|
for f in src_dir.glob(f"*{ext}"):
|
|
shutil.copy2(f, work_dir / f.name)
|
|
|
|
win_path = "Z:" + str(work_netlist).replace("/", "\\")
|
|
|
|
cmd = ["wine", str(LTSPICE_EXE), "-b", win_path]
|
|
env = get_wine_env()
|
|
|
|
process = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
env=env,
|
|
cwd=str(work_dir),
|
|
)
|
|
|
|
try:
|
|
stdout_bytes, stderr_bytes = await asyncio.wait_for(
|
|
process.communicate(),
|
|
timeout=timeout
|
|
)
|
|
except asyncio.TimeoutError:
|
|
process.kill()
|
|
await process.wait()
|
|
return SimulationResult(
|
|
success=False, raw_file=None, log_file=None, raw_data=None,
|
|
error=f"Simulation timed out after {timeout} seconds",
|
|
stdout="", stderr="",
|
|
elapsed_seconds=time.monotonic() - start_time,
|
|
)
|
|
|
|
stdout = stdout_bytes.decode("utf-8", errors="replace")
|
|
stderr = stderr_bytes.decode("utf-8", errors="replace")
|
|
elapsed = time.monotonic() - start_time
|
|
|
|
raw_file = work_netlist.with_suffix(".raw")
|
|
log_file = work_netlist.with_suffix(".log")
|
|
|
|
if not raw_file.exists():
|
|
error_msg = "Simulation failed - no .raw file produced"
|
|
if log_file.exists():
|
|
log_content = log_file.read_text(errors="replace")
|
|
if "error" in log_content.lower():
|
|
error_msg = f"Simulation error: {log_content[:500]}"
|
|
|
|
return SimulationResult(
|
|
success=False,
|
|
raw_file=None,
|
|
log_file=log_file if log_file.exists() else None,
|
|
raw_data=None, error=error_msg,
|
|
stdout=stdout, stderr=stderr, elapsed_seconds=elapsed,
|
|
)
|
|
|
|
raw_data = None
|
|
parse_error = None
|
|
if parse_results and raw_file.stat().st_size <= MAX_RAW_FILE_SIZE:
|
|
try:
|
|
raw_data = parse_raw_file(raw_file)
|
|
except Exception as e:
|
|
parse_error = f"Failed to parse .raw file: {e}"
|
|
|
|
return SimulationResult(
|
|
success=True,
|
|
raw_file=raw_file,
|
|
log_file=log_file if log_file.exists() else None,
|
|
raw_data=raw_data,
|
|
error=parse_error,
|
|
stdout=stdout, stderr=stderr, elapsed_seconds=elapsed,
|
|
)
|
|
finally:
|
|
pass # Keep temp files for debugging
|