"""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 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 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