Module circuitsim.simulators
Interface with simulators.
Expand source code
"""Interface with simulators."""
import math
import subprocess
from multiprocessing import Pool
from pathlib import Path
available_simulators = ["iverilog", "verilator", "vcs"]
compile_args = {
    "vcs": ["vcs", "-full64"],
    "iverilog": ["iverilog"],
    "verilator": ["verilator", "--exe", "tb.cpp", "--cc"],
}
post_compile_args = {"verilator": ["make", "-C", "obj_dir", "-f", "Vtb.mk"]}
simulate_args = {
    "vcs": ["./simv", "-q"],
    "iverilog": ["vvp", "a.out"],
    "verilator": ["./obj_dir/Vtb"],
}
class SimulationCompilationError(Exception):
    """Thrown when a simulation fails during compilation."""
class SimulationExecutionError(Exception):
    """Thrown when a simulation fails during execution."""
def _convert_value(v, allow_x=False):
    if v in [True, False, 0, 1]:
        return str(int(v))
    if isinstance(v, str) and v in "01":
        return v
    if allow_x and v.lower() == "x":
        return v.lower()
    raise ValueError(f"Unknown input value '{v}'")
def _convert_values(values, allow_x=False):
    return {k: _convert_value(v, allow_x=allow_x) for k, v in values.items()}
def compile_simulator(simulator, netlists, working_dir):
    """
    Compile a circuit simulation.
    Parameters
    ----------
    simulator: str
            The simulator to use.
    netlist: list of str or pathlib.Path
            The netlists to compile.
    working_dir: str or pathlib.Path
            The path to comiple the simulation in.
    """
    netlists = [str(Path(n).absolute()) for n in netlists]
    working_dir = Path(working_dir)
    with open(working_dir / "compile.log", "w+") as f:
        try:
            subprocess.run(
                compile_args[simulator] + netlists,
                cwd=working_dir,
                stdout=f,
                stderr=f,
                check=True,
            )
        except subprocess.CalledProcessError as e:
            f.seek(0)
            message = f.read()
            raise SimulationCompilationError(message) from e
    if simulator in post_compile_args:
        with open(working_dir / "post_compile.log", "w+") as f:
            try:
                subprocess.run(
                    post_compile_args[simulator],
                    cwd=working_dir,
                    stdout=f,
                    stderr=f,
                    check=True,
                )
            except subprocess.CalledProcessError as e:
                f.seek(0)
                message = f.read()
                raise SimulationCompilationError(message) from e
def _run_process(p, simulator, working_dir):
    with open(working_dir / f"simulate_{p}.log", "w") as f:
        return subprocess.run(
            simulate_args[simulator]
            + [f"+input_file=input_file_{p}.txt", f"+output_file=output_file_{p}.txt"],
            cwd=working_dir,
            stdout=f,
            stderr=f,
            check=True,
        )
def execute_simulator(
    simulator, inputs, vectors, working_dir, num_processes, allow_x=False
):
    """
    Execute a compiled simulation.
    Parameters
    ----------
    simulator; str
            The simulator to use.
    inputs: list of str
            The inputs to the module.
    vectors: list of dict of str:bool
            The vectors to simulate. Each vector is represented as a
            dictionary mapping an input to a logical value.
    working_dir: str or pathlib.Path
            The directory to compile and run the simulation in.
    num_process: int
            The number of simulation processes to run. Specifying a
            number more than 1 will cause multiple simulations to be
            executed in parallel.
    allow_x: bool
            If True, the inputs/outputs can contain "don't care" or "x"
            values in addition to 0 and 1. Specify an don't care value
            for an input by setting it to "x".
    """
    working_dir = Path(working_dir)
    vectors_per_process = math.ceil(len(vectors) / num_processes)
    for p, v in enumerate(
        [
            vectors[i : i + vectors_per_process]
            for i in range(0, len(vectors), vectors_per_process)
        ]
    ):
        with open(working_dir / f"input_file_{p}.txt", "w") as f:
            for vector in v:
                vector = _convert_values(vector, allow_x=allow_x)
                try:
                    f.write("".join(vector[i] for i in inputs) + "\n")
                except KeyError as e:
                    missing_inputs = ", ".join(set(inputs) - set(vector))
                    if len(missing_inputs) > 20:
                        missing_inputs = missing_inputs[:17] + "..."
                    raise ValueError(
                        "Value not provided for inputs '{missing_inputs}'"
                    ) from e
    if num_processes > 1:
        with Pool(num_processes) as pool:
            results = pool.starmap_async(
                _run_process,
                [(p, simulator, working_dir) for p in range(num_processes)],
            )
            sim_results = results.get()
            for sim_res in sim_results:
                if sim_res.returncode != 0:
                    with open(working_dir / "simulate.log") as f:
                        message = f.read()
                    raise SimulationExecutionError(message)
    else:
        sim_res = _run_process(0, simulator, working_dir)
def parse_simulation_output(working_dir, outputs, num_processes, allow_x=False):
    """
    Parse the output of a simulation.
    Parameters
    ----------
    working_dir: str or pathlib.Path
            The directory that the simulation was run from.
    Returns
    -------
    list of dict of str to bool
            The simulation results, as a list of dictionaries mapping
            outputs to values.
    num_process: int
            The number of simulation processes that were run.
    allow_x: bool
            If True, the outputs can contain "don't care" or "x"
            values in addition to 0 and 1. The outputs will then be
            returned as a dict of str:str, where each output is either
            "0", "1", or "x".
    Returns
    -------
    list o dict o str:bool or str:str
            The simulation outputs. Each vector is represented as a
            dictionary mapping an output to a logical value. If `allow_x`
            is True, then instead of logic values, each output will be
            mapped to either "0", "1", or "x".
    """
    working_dir = Path(working_dir)
    vectors = []
    for p in range(num_processes):
        with open(working_dir / f"output_file_{p}.txt") as f:
            for line in f:
                if allow_x:
                    if not set(line.strip().lower()) <= {"0", "1", "x"}:
                        raise ValueError(f"Unknown value in simulation output: {line}")
                    vectors.append(dict(zip(outputs, line.strip().lower())))
                else:
                    if not set(line.strip()) <= {"0", "1"}:
                        raise ValueError(f"Unknown value in simulation output: {line}")
                    vectors.append(
                        {o: bool(int(v)) for o, v in zip(outputs, line.strip())}
                    )
    return vectors
Functions
def compile_simulator(simulator, netlists, working_dir)- 
Compile a circuit simulation.
Parameters
simulator:str- The simulator to use.
 netlist:listofstrorpathlib.Path- The netlists to compile.
 working_dir:strorpathlib.Path- The path to comiple the simulation in.
 
Expand source code
def compile_simulator(simulator, netlists, working_dir): """ Compile a circuit simulation. Parameters ---------- simulator: str The simulator to use. netlist: list of str or pathlib.Path The netlists to compile. working_dir: str or pathlib.Path The path to comiple the simulation in. """ netlists = [str(Path(n).absolute()) for n in netlists] working_dir = Path(working_dir) with open(working_dir / "compile.log", "w+") as f: try: subprocess.run( compile_args[simulator] + netlists, cwd=working_dir, stdout=f, stderr=f, check=True, ) except subprocess.CalledProcessError as e: f.seek(0) message = f.read() raise SimulationCompilationError(message) from e if simulator in post_compile_args: with open(working_dir / "post_compile.log", "w+") as f: try: subprocess.run( post_compile_args[simulator], cwd=working_dir, stdout=f, stderr=f, check=True, ) except subprocess.CalledProcessError as e: f.seek(0) message = f.read() raise SimulationCompilationError(message) from e def execute_simulator(simulator, inputs, vectors, working_dir, num_processes, allow_x=False)- 
Execute a compiled simulation.
Parameters
- simulator; str
 - The simulator to use.
 inputs:listofstr- The inputs to the module.
 vectors:listofdictofstr:bool- The vectors to simulate. Each vector is represented as a dictionary mapping an input to a logical value.
 working_dir:strorpathlib.Path- The directory to compile and run the simulation in.
 num_process:int- The number of simulation processes to run. Specifying a number more than 1 will cause multiple simulations to be executed in parallel.
 allow_x:bool- If True, the inputs/outputs can contain "don't care" or "x" values in addition to 0 and 1. Specify an don't care value for an input by setting it to "x".
 
Expand source code
def execute_simulator( simulator, inputs, vectors, working_dir, num_processes, allow_x=False ): """ Execute a compiled simulation. Parameters ---------- simulator; str The simulator to use. inputs: list of str The inputs to the module. vectors: list of dict of str:bool The vectors to simulate. Each vector is represented as a dictionary mapping an input to a logical value. working_dir: str or pathlib.Path The directory to compile and run the simulation in. num_process: int The number of simulation processes to run. Specifying a number more than 1 will cause multiple simulations to be executed in parallel. allow_x: bool If True, the inputs/outputs can contain "don't care" or "x" values in addition to 0 and 1. Specify an don't care value for an input by setting it to "x". """ working_dir = Path(working_dir) vectors_per_process = math.ceil(len(vectors) / num_processes) for p, v in enumerate( [ vectors[i : i + vectors_per_process] for i in range(0, len(vectors), vectors_per_process) ] ): with open(working_dir / f"input_file_{p}.txt", "w") as f: for vector in v: vector = _convert_values(vector, allow_x=allow_x) try: f.write("".join(vector[i] for i in inputs) + "\n") except KeyError as e: missing_inputs = ", ".join(set(inputs) - set(vector)) if len(missing_inputs) > 20: missing_inputs = missing_inputs[:17] + "..." raise ValueError( "Value not provided for inputs '{missing_inputs}'" ) from e if num_processes > 1: with Pool(num_processes) as pool: results = pool.starmap_async( _run_process, [(p, simulator, working_dir) for p in range(num_processes)], ) sim_results = results.get() for sim_res in sim_results: if sim_res.returncode != 0: with open(working_dir / "simulate.log") as f: message = f.read() raise SimulationExecutionError(message) else: sim_res = _run_process(0, simulator, working_dir) def parse_simulation_output(working_dir, outputs, num_processes, allow_x=False)- 
Parse the output of a simulation.
Parameters
working_dir:strorpathlib.Path- The directory that the simulation was run from.
 
Returns
listofdictofstr to bool- The simulation results, as a list of dictionaries mapping outputs to values.
 num_process:int- The number of simulation processes that were run.
 allow_x:bool- If True, the outputs can contain "don't care" or "x" values in addition to 0 and 1. The outputs will then be returned as a dict of str:str, where each output is either "0", "1", or "x".
 
Returns
list o dict o str:boolorstr:str- The simulation outputs. Each vector is represented as a
dictionary mapping an output to a logical value. If 
allow_xis True, then instead of logic values, each output will be mapped to either "0", "1", or "x". 
Expand source code
def parse_simulation_output(working_dir, outputs, num_processes, allow_x=False): """ Parse the output of a simulation. Parameters ---------- working_dir: str or pathlib.Path The directory that the simulation was run from. Returns ------- list of dict of str to bool The simulation results, as a list of dictionaries mapping outputs to values. num_process: int The number of simulation processes that were run. allow_x: bool If True, the outputs can contain "don't care" or "x" values in addition to 0 and 1. The outputs will then be returned as a dict of str:str, where each output is either "0", "1", or "x". Returns ------- list o dict o str:bool or str:str The simulation outputs. Each vector is represented as a dictionary mapping an output to a logical value. If `allow_x` is True, then instead of logic values, each output will be mapped to either "0", "1", or "x". """ working_dir = Path(working_dir) vectors = [] for p in range(num_processes): with open(working_dir / f"output_file_{p}.txt") as f: for line in f: if allow_x: if not set(line.strip().lower()) <= {"0", "1", "x"}: raise ValueError(f"Unknown value in simulation output: {line}") vectors.append(dict(zip(outputs, line.strip().lower()))) else: if not set(line.strip()) <= {"0", "1"}: raise ValueError(f"Unknown value in simulation output: {line}") vectors.append( {o: bool(int(v)) for o, v in zip(outputs, line.strip())} ) return vectors 
Classes
class SimulationCompilationError (*args, **kwargs)- 
Thrown when a simulation fails during compilation.
Expand source code
class SimulationCompilationError(Exception): """Thrown when a simulation fails during compilation."""Ancestors
- builtins.Exception
 - builtins.BaseException
 
 class SimulationExecutionError (*args, **kwargs)- 
Thrown when a simulation fails during execution.
Expand source code
class SimulationExecutionError(Exception): """Thrown when a simulation fails during execution."""Ancestors
- builtins.Exception
 - builtins.BaseException