amaranth-lang__amaranth-923

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "amaranth/sim:\n__init__.py\n_base.py\n_pyclock.py\n_pycoro.py\n_pyrtl.py\ncore.py\npysim.py\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "amaranth/hdl/mem.py:12:class Memory(Elaboratable):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "filepath": "amaranth/hdl/mem.py",
  "content": "import operator\nfrom collections import OrderedDict\n\nfrom .. import tracer\nfrom .ast import *\nfrom .ir import Elaboratable, Instance, Fragment\n\n\n__all__ = [\"Memory\", \"ReadPort\", \"WritePort\", \"DummyPort\"]\n\n\nclass Memory(Elaboratable):\n    \"\"\"A word addressable storage.\n\n    Parameters\n    ----------\n    width : int\n        Access granularity. Each storage element of this memory is ``width`` bits in size.\n    depth : int\n        Word count. This memory contains ``depth`` storage elements.\n    init : list of int\n        Initial values. At power on, each storage element in this memory is initialized to\n        the corresponding element of ``init``, if any, or to zero otherwise.\n        Uninitialized memories are not currently supported.\n    name : str\n        Name hint for this memory. If ``None`` (default) the name is inferred from the variable\n        name this ``Signal`` is assigned to.\n    attrs : dict\n        Dictionary of synthesis attributes.\n\n    Attributes\n    ----------\n    width : int\n    depth : int\n    init : list of int\n    attrs : dict\n    \"\"\"\n    def __init__(self, *, width, depth, init=None, name=None, attrs=None, simulate=True):\n        if not isinstance(width, int) or width < 0:\n            raise TypeError(\"Memory width must be a non-negative integer, not {!r}\"\n                            .format(width))\n        if not isinstance(depth, int) or depth < 0:\n            raise TypeError(\"Memory depth must be a non-negative integer, not {!r}\"\n                            .format(depth))\n\n        self.name    = name or tracer.get_var_name(depth=2, default=\"$memory\")\n        self.src_loc = tracer.get_src_loc()\n\n        self.width = width\n        self.depth = depth\n        self.attrs = OrderedDict(() if attrs is None else attrs)\n\n        # Array of signals for simulation.\n        self._array = Array()\n        if simulate:\n            for addr in range(self.depth):\n                self._array.append(Signal(self.width, name=\"{}({})\"\n                                          .format(name or \"memory\", addr)))\n\n        self.init = init\n        self._read_ports = []\n        self._write_ports = []\n\n    @property\n    def init(self):\n        return self._init\n\n    @init.setter\n    def init(self, new_init):\n        self._init = [] if new_init is None else list(new_init)\n        if len(self.init) > self.depth:\n            raise ValueError(\"Memory initialization value count exceed memory depth ({} > {})\"\n                             .format(len(self.init), self.depth))\n\n        try:\n            for addr in range(len(self._array)):\n                if addr < len(self._init):\n                    self._array[addr].reset = operator.index(self._init[addr])\n                else:\n                    self._array[addr].reset = 0\n        except TypeError as e:\n            raise TypeError(\"Memory initialization value at address {:x}: {}\"\n                            .format(addr, e)) from None\n\n    def read_port(self, *, src_loc_at=0, **kwargs):\n        \"\"\"Get a read port.\n\n        See :class:`ReadPort` for details.\n\n        Arguments\n        ---------\n        domain : str\n        transparent : bool\n\n        Returns\n        -------\n        An instance of :class:`ReadPort` associated with this memory.\n        \"\"\"\n        return ReadPort(self, src_loc_at=1 + src_loc_at, **kwargs)\n\n    def write_port(self, *, src_loc_at=0, **kwargs):\n        \"\"\"Get a write port.\n\n        See :class:`WritePort` for details.\n\n        Arguments\n        ---------\n        domain : str\n        granularity : int\n\n        Returns\n        -------\n        An instance of :class:`WritePort` associated with this memory.\n        \"\"\"\n        return WritePort(self, src_loc_at=1 + src_loc_at, **kwargs)\n\n    def __getitem__(self, index):\n        \"\"\"Simulation only.\"\"\"\n        return self._array[index]\n\n    def elaborate(self, platform):\n        init = \"\".join(format(Const(elem, unsigned(self.width)).value, f\"0{self.width}b\") for elem in reversed(self.init))\n        init = Const(int(init or \"0\", 2), self.depth * self.width)\n        rd_clk = []\n        rd_clk_enable = 0\n        rd_transparency_mask = 0\n        for index, port in enumerate(self._read_ports):\n            if port.domain != \"comb\":\n                rd_clk.append(ClockSignal(port.domain))\n                rd_clk_enable |= 1 << index\n                if port.transparent:\n                    for write_index, write_port in enumerate(self._write_ports):\n                        if port.domain == write_port.domain:\n                            rd_transparency_mask |= 1 << (index * len(self._write_ports) + write_index)\n            else:\n                rd_clk.append(Const(0, 1))\n        f = Instance(\"$mem_v2\",\n            *((\"a\", attr, value) for attr, value in self.attrs.items()),\n            p_SIZE=self.depth,\n            p_OFFSET=0,\n            p_ABITS=Shape.cast(range(self.depth)).width,\n            p_WIDTH=self.width,\n            p_INIT=init,\n            p_RD_PORTS=len(self._read_ports),\n            p_RD_CLK_ENABLE=Const(rd_clk_enable, len(self._read_ports)) if self._read_ports else Const(0, 1),\n            p_RD_CLK_POLARITY=Const(-1, unsigned(len(self._read_ports))) if self._read_ports else Const(0, 1),\n            p_RD_TRANSPARENCY_MASK=Const(rd_transparency_mask, max(1, len(self._read_ports) * len(self._write_ports))),\n            p_RD_COLLISION_X_MASK=Const(0, max(1, len(self._read_ports) * len(self._write_ports))),\n            p_RD_WIDE_CONTINUATION=Const(0, len(self._read_ports)) if self._read_ports else Const(0, 1),\n            p_RD_CE_OVER_SRST=Const(0, len(self._read_ports)) if self._read_ports else Const(0, 1),",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 354,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "amaranth/hdl/mem.py",
  "content": "            p_RD_ARST_VALUE=Const(0, len(self._read_ports) * self.width),\n            p_RD_SRST_VALUE=Const(0, len(self._read_ports) * self.width),\n            p_RD_INIT_VALUE=Const(0, len(self._read_ports) * self.width),\n            p_WR_PORTS=len(self._write_ports),\n            p_WR_CLK_ENABLE=Const(-1, unsigned(len(self._write_ports))) if self._write_ports else Const(0, 1),\n            p_WR_CLK_POLARITY=Const(-1, unsigned(len(self._write_ports))) if self._write_ports else Const(0, 1),\n            p_WR_PRIORITY_MASK=Const(0, len(self._write_ports) * len(self._write_ports)) if self._write_ports else Const(0, 1),\n            p_WR_WIDE_CONTINUATION=Const(0, len(self._write_ports)) if self._write_ports else Const(0, 1),\n            i_RD_CLK=Cat(rd_clk),\n            i_RD_EN=Cat(port.en for port in self._read_ports),\n            i_RD_ARST=Const(0, len(self._read_ports)),\n            i_RD_SRST=Const(0, len(self._read_ports)),\n            i_RD_ADDR=Cat(port.addr for port in self._read_ports),\n            o_RD_DATA=Cat(port.data for port in self._read_ports),\n            i_WR_CLK=Cat(ClockSignal(port.domain) for port in self._write_ports),\n            i_WR_EN=Cat(Cat(en_bit.replicate(port.granularity) for en_bit in port.en) for port in self._write_ports),\n            i_WR_ADDR=Cat(port.addr for port in self._write_ports),\n            i_WR_DATA=Cat(port.data for port in self._write_ports),\n        )\n        for port in self._read_ports:\n            port._MustUse__used = True\n            if port.domain == \"comb\":\n                # Asynchronous port\n                f.add_statements(port.data.eq(self._array[port.addr]))\n                f.add_driver(port.data)\n            else:\n                # Synchronous port\n                data = self._array[port.addr]\n                for write_port in self._write_ports:\n                    if port.domain == write_port.domain and port.transparent:\n                        if len(write_port.en) > 1:\n                            parts = []\n                            for index, en_bit in enumerate(write_port.en):\n                                offset = index * write_port.granularity\n                                bits   = slice(offset, offset + write_port.granularity)\n                                cond = en_bit & (port.addr == write_port.addr)\n                                parts.append(Mux(cond, write_port.data[bits], data[bits]))\n                            data = Cat(parts)\n                        else:\n                            data = Mux(write_port.en, write_port.data, data)\n                f.add_statements(\n                    Switch(port.en, {\n                        1: port.data.eq(data)\n                    })\n                )\n                f.add_driver(port.data, port.domain)\n        for port in self._write_ports:\n            port._MustUse__used = True\n            if len(port.en) > 1:\n                for index, en_bit in enumerate(port.en):\n                    offset = index * port.granularity\n                    bits   = slice(offset, offset + port.granularity)\n                    write_data = self._array[port.addr][bits].eq(port.data[bits])\n                    f.add_statements(Switch(en_bit, { 1: write_data }))\n            else:\n                write_data = self._array[port.addr].eq(port.data)\n                f.add_statements(Switch(port.en, { 1: write_data }))\n            for signal in self._array:\n                f.add_driver(signal, port.domain)\n        return f\n\nclass ReadPort(Elaboratable):\n    \"\"\"A memory read port.\n\n    Parameters\n    ----------\n    memory : :class:`Memory`\n        Memory associated with the port.\n    domain : str\n        Clock domain. Defaults to ``\"sync\"``. If set to ``\"comb\"``, the port is asynchronous.\n        Otherwise, the read data becomes available on the next clock cycle.\n    transparent : bool\n        Port transparency. If set (default), a read at an address that is also being written to in\n        the same clock cycle will output the new value. Otherwise, the old value will be output\n        first. This behavior only applies to ports in the same domain.\n\n    Attributes\n    ----------\n    memory : :class:`Memory`\n    domain : str\n    transparent : bool\n    addr : Signal(range(memory.depth)), in\n        Read address.\n    data : Signal(memory.width), out\n        Read data.\n    en : Signal or Const, in\n        Read enable. If asserted, ``data`` is updated with the word stored at ``addr``.\n\n    Exceptions\n    ----------\n    Raises :exn:`ValueError` if the read port is simultaneously asynchronous and non-transparent.\n    \"\"\"\n    def __init__(self, memory, *, domain=\"sync\", transparent=True, src_loc_at=0):\n        if domain == \"comb\" and not transparent:\n            raise ValueError(\"Read port cannot be simultaneously asynchronous and non-transparent\")\n\n        self.memory      = memory\n        self.domain      = domain\n        self.transparent = transparent\n\n        self.addr = Signal(range(memory.depth),\n                           name=\"{}_r_addr\".format(memory.name), src_loc_at=1 + src_loc_at)\n        self.data = Signal(memory.width,\n                           name=\"{}_r_data\".format(memory.name), src_loc_at=1 + src_loc_at)\n        if self.domain != \"comb\":\n            self.en = Signal(name=\"{}_r_en\".format(memory.name), reset=1,\n                             src_loc_at=1 + src_loc_at)\n        else:\n            self.en = Const(1)\n\n        memory._read_ports.append(self)\n\n    def elaborate(self, platform):\n        if self is self.memory._read_ports[0]:\n            return self.memory\n        else:\n            return Fragment()\n\n\nclass WritePort(Elaboratable):\n    \"\"\"A memory write port.\n\n    Parameters\n    ----------\n    memory : :class:`Memory`\n        Memory associated with the port.\n    domain : str\n        Clock domain. Defaults to ``\"sync\"``. Writes have a latency of 1 clock cycle.\n    granularity : int\n        Port granularity. Defaults to ``memory.width``. Write data is split evenly in\n        ``memory.width // granularity`` chunks, which can be updated independently.\n\n    Attributes\n    ----------\n    memory : :class:`Memory`\n    domain : str\n    granularity : int\n    addr : Signal(range(memory.depth)), in\n        Write address.\n    data : Signal(memory.width), in\n        Write data.\n    en : Signal(memory.width // granularity), in\n        Write enable. Each bit selects a non-overlapping chunk of ``granularity`` bits on the\n        ``data`` signal, which is written to memory at ``addr``. Unselected chunks are ignored.\n\n    Exceptions\n    ----------\n    Raises :exn:`ValueError` if the write port granularity is greater than memory width, or does not\n    divide memory width evenly.\n    \"\"\"",
  "start_line": 151,
  "end_line": 300,
  "total_lines": 354,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "amaranth/hdl/mem.py",
  "content": "    def __init__(self, memory, *, domain=\"sync\", granularity=None, src_loc_at=0):\n        if granularity is None:\n            granularity = memory.width\n        if not isinstance(granularity, int) or granularity < 0:\n            raise TypeError(\"Write port granularity must be a non-negative integer, not {!r}\"\n                            .format(granularity))\n        if granularity > memory.width:\n            raise ValueError(\"Write port granularity must not be greater than memory width \"\n                             \"({} > {})\"\n                             .format(granularity, memory.width))\n        if memory.width // granularity * granularity != memory.width:\n            raise ValueError(\"Write port granularity must divide memory width evenly\")\n\n        self.memory       = memory\n        self.domain       = domain\n        self.granularity  = granularity\n\n        self.addr = Signal(range(memory.depth),\n                           name=\"{}_w_addr\".format(memory.name), src_loc_at=1 + src_loc_at)\n        self.data = Signal(memory.width,\n                           name=\"{}_w_data\".format(memory.name), src_loc_at=1 + src_loc_at)\n        self.en   = Signal(memory.width // granularity,\n                           name=\"{}_w_en\".format(memory.name), src_loc_at=1 + src_loc_at)\n\n        memory._write_ports.append(self)\n\n    def elaborate(self, platform):\n        if not self.memory._read_ports and self is self.memory._write_ports[0]:\n            return self.memory\n        else:\n            return Fragment()\n\n\nclass DummyPort:\n    \"\"\"Dummy memory port.\n\n    This port can be used in place of either a read or a write port for testing and verification.\n    It does not include any read/write port specific attributes, i.e. none besides ``\"domain\"``;\n    any such attributes may be set manually.\n    \"\"\"\n    def __init__(self, *, data_width, addr_width, domain=\"sync\", name=None, granularity=None):\n        self.domain = domain\n\n        if granularity is None:\n            granularity = data_width\n        if name is None:\n            name = tracer.get_var_name(depth=2, default=\"dummy\")\n\n        self.addr = Signal(addr_width,\n                           name=\"{}_addr\".format(name), src_loc_at=1)\n        self.data = Signal(data_width,\n                           name=\"{}_data\".format(name), src_loc_at=1)\n        self.en   = Signal(data_width // granularity,\n                           name=\"{}_en\".format(name), src_loc_at=1)",
  "start_line": 301,
  "end_line": 354,
  "total_lines": 354,
  "is_truncated": false
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "filepath": "amaranth/sim/pysim.py",
  "content": "from contextlib import contextmanager\nimport itertools\nimport re\nfrom vcd import VCDWriter\nfrom vcd.gtkw import GTKWSave\n\nfrom ..hdl import *\nfrom ..hdl.ast import SignalDict\nfrom ._base import *\nfrom ._pyrtl import _FragmentCompiler\nfrom ._pycoro import PyCoroProcess\nfrom ._pyclock import PyClockProcess\n\n\n__all__ = [\"PySimEngine\"]\n\n\nclass _NameExtractor:\n    def __init__(self):\n        self.names = SignalDict()\n\n    def __call__(self, fragment, *, hierarchy=(\"bench\", \"top\",)):\n        def add_signal_name(signal):\n            hierarchical_signal_name = (*hierarchy, signal.name)\n            if signal not in self.names:\n                self.names[signal] = {hierarchical_signal_name}\n            else:\n                self.names[signal].add(hierarchical_signal_name)\n\n        for domain_name, domain_signals in fragment.drivers.items():\n            if domain_name is not None:\n                domain = fragment.domains[domain_name]\n                add_signal_name(domain.clk)\n                if domain.rst is not None:\n                    add_signal_name(domain.rst)\n\n        for statement in fragment.statements:\n            for signal in statement._lhs_signals() | statement._rhs_signals():\n                if not isinstance(signal, (ClockSignal, ResetSignal)):\n                    add_signal_name(signal)\n\n        for subfragment_index, (subfragment, subfragment_name) in enumerate(fragment.subfragments):\n            if subfragment_name is None:\n                subfragment_name = \"U${}\".format(subfragment_index)\n            self(subfragment, hierarchy=(*hierarchy, subfragment_name))\n\n        return self.names\n\n\nclass _VCDWriter:\n    @staticmethod\n    def decode_to_vcd(signal, value):\n        return signal.decoder(value).expandtabs().replace(\" \", \"_\")\n\n    def __init__(self, fragment, *, vcd_file, gtkw_file=None, traces=()):\n        if isinstance(vcd_file, str):\n            vcd_file = open(vcd_file, \"wt\")\n        if isinstance(gtkw_file, str):\n            gtkw_file = open(gtkw_file, \"wt\")\n\n        self.vcd_vars = SignalDict()\n        self.vcd_file = vcd_file\n        self.vcd_writer = vcd_file and VCDWriter(self.vcd_file,\n            timescale=\"1 ps\", comment=\"Generated by Amaranth\")\n\n        self.gtkw_names = SignalDict()\n        self.gtkw_file = gtkw_file\n        self.gtkw_save = gtkw_file and GTKWSave(self.gtkw_file)\n\n        self.traces = []\n\n        signal_names = _NameExtractor()(fragment)\n\n        trace_names = SignalDict()\n        for trace in traces:\n            if trace not in signal_names:\n                trace_names[trace] = {('bench', trace.name)}\n            self.traces.append(trace)\n\n        if self.vcd_writer is None:\n            return\n\n        for signal, names in itertools.chain(signal_names.items(), trace_names.items()):\n            if signal.decoder:\n                var_type = \"string\"\n                var_size = 1\n                var_init = self.decode_to_vcd(signal, signal.reset)\n            else:\n                var_type = \"wire\"\n                var_size = signal.width\n                var_init = signal.reset\n\n            for (*var_scope, var_name) in names:\n                if re.search(r\"[ \\t\\r\\n]\", var_name):\n                    raise NameError(\"Signal '{}.{}' contains a whitespace character\"\n                                    .format(\".\".join(var_scope), var_name))\n\n                suffix = None\n                while True:\n                    try:\n                        if suffix is None:\n                            var_name_suffix = var_name\n                        else:\n                            var_name_suffix = \"{}${}\".format(var_name, suffix)\n                        if signal not in self.vcd_vars:\n                            vcd_var = self.vcd_writer.register_var(\n                                scope=var_scope, name=var_name_suffix,\n                                var_type=var_type, size=var_size, init=var_init)\n                            self.vcd_vars[signal] = vcd_var\n                        else:\n                            self.vcd_writer.register_alias(\n                                scope=var_scope, name=var_name_suffix,\n                                var=self.vcd_vars[signal])\n                        break\n                    except KeyError:\n                        suffix = (suffix or 0) + 1\n\n                if signal not in self.gtkw_names:\n                    self.gtkw_names[signal] = (*var_scope, var_name_suffix)\n\n    def update(self, timestamp, signal, value):\n        vcd_var = self.vcd_vars.get(signal)\n        if vcd_var is None:\n            return\n\n        if signal.decoder:\n            var_value = self.decode_to_vcd(signal, value)\n        else:\n            var_value = value\n        self.vcd_writer.change(vcd_var, timestamp, var_value)\n\n    def close(self, timestamp):\n        if self.vcd_writer is not None:\n            self.vcd_writer.close(timestamp)\n\n        if self.gtkw_save is not None:\n            self.gtkw_save.dumpfile(self.vcd_file.name)\n            self.gtkw_save.dumpfile_size(self.vcd_file.tell())\n\n            self.gtkw_save.treeopen(\"top\")\n            for signal in self.traces:\n                if len(signal) > 1 and not signal.decoder:\n                    suffix = \"[{}:0]\".format(len(signal) - 1)\n                else:\n                    suffix = \"\"\n                self.gtkw_save.trace(\".\".join(self.gtkw_names[signal]) + suffix)\n\n        if self.vcd_file is not None:\n            self.vcd_file.close()\n        if self.gtkw_file is not None:",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 336,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "amaranth/sim/core.py",
  "content": "import inspect\nimport warnings\n\nfrom .._utils import deprecated\nfrom ..hdl.cd import *\nfrom ..hdl.ir import *\nfrom ._base import BaseEngine\n\n\n__all__ = [\"Settle\", \"Delay\", \"Tick\", \"Passive\", \"Active\", \"Simulator\"]\n\n\nclass Command:\n    pass\n\n\nclass Settle(Command):\n    def __repr__(self):\n        return \"(settle)\"\n\n\nclass Delay(Command):\n    def __init__(self, interval=None):\n        self.interval = None if interval is None else float(interval)\n\n    def __repr__(self):\n        if self.interval is None:\n            return \"(delay ε)\"\n        else:\n            return \"(delay {:.3}us)\".format(self.interval * 1e6)\n\n\nclass Tick(Command):\n    def __init__(self, domain=\"sync\"):\n        if not isinstance(domain, (str, ClockDomain)):\n            raise TypeError(\"Domain must be a string or a ClockDomain instance, not {!r}\"\n                            .format(domain))\n        assert domain != \"comb\"\n        self.domain = domain\n\n    def __repr__(self):\n        return \"(tick {})\".format(self.domain)\n\n\nclass Passive(Command):\n    def __repr__(self):\n        return \"(passive)\"\n\n\nclass Active(Command):\n    def __repr__(self):\n        return \"(active)\"\n\n\nclass Simulator:\n    def __init__(self, fragment, *, engine=\"pysim\"):\n        if isinstance(engine, type) and issubclass(engine, BaseEngine):\n            pass\n        elif engine == \"pysim\":\n            from .pysim import PySimEngine\n            engine = PySimEngine\n        else:\n            raise TypeError(\"Value '{!r}' is not a simulation engine class or \"\n                            \"a simulation engine name\"\n                            .format(engine))\n\n        self._fragment = Fragment.get(fragment, platform=None).prepare()\n        self._engine   = engine(self._fragment)\n        self._clocked  = set()\n\n    def _check_process(self, process):\n        if not (inspect.isgeneratorfunction(process) or inspect.iscoroutinefunction(process)):\n            raise TypeError(\"Cannot add a process {!r} because it is not a generator function\"\n                            .format(process))\n        return process\n\n    def add_process(self, process):\n        process = self._check_process(process)\n        def wrapper():\n            # Only start a bench process after comb settling, so that the reset values are correct.\n            yield Settle()\n            yield from process()\n        self._engine.add_coroutine_process(wrapper, default_cmd=None)\n\n    def add_sync_process(self, process, *, domain=\"sync\"):\n        process = self._check_process(process)\n        def wrapper():\n            # Only start a sync process after the first clock edge (or reset edge, if the domain\n            # uses an asynchronous reset). This matches the behavior of synchronous FFs.\n            yield Tick(domain)\n            yield from process()\n        self._engine.add_coroutine_process(wrapper, default_cmd=Tick(domain))\n\n    def add_clock(self, period, *, phase=None, domain=\"sync\", if_exists=False):\n        \"\"\"Add a clock process.\n\n        Adds a process that drives the clock signal of ``domain`` at a 50% duty cycle.\n\n        Arguments\n        ---------\n        period : float\n            Clock period. The process will toggle the ``domain`` clock signal every ``period / 2``\n            seconds.\n        phase : None or float\n            Clock phase. The process will wait ``phase`` seconds before the first clock transition.\n            If not specified, defaults to ``period / 2``.\n        domain : str or ClockDomain\n            Driven clock domain. If specified as a string, the domain with that name is looked up\n            in the root fragment of the simulation.\n        if_exists : bool\n            If ``False`` (the default), raise an error if the driven domain is specified as\n            a string and the root fragment does not have such a domain. If ``True``, do nothing\n            in this case.\n        \"\"\"\n        if isinstance(domain, ClockDomain):\n            if (domain.name in self._fragment.domains and\n                    domain is not self._fragment.domains[domain.name]):\n                warnings.warn(\"Adding a clock process that drives a clock domain object \"\n                              \"named {!r}, which is distinct from an identically named domain \"\n                              \"in the simulated design\"\n                              .format(domain.name),\n                              UserWarning, stacklevel=2)\n        elif domain in self._fragment.domains:\n            domain = self._fragment.domains[domain]\n        elif if_exists:\n            return\n        else:\n            raise ValueError(\"Domain {!r} is not present in simulation\"\n                             .format(domain))\n        if domain in self._clocked:\n            raise ValueError(\"Domain {!r} already has a clock driving it\"\n                             .format(domain.name))\n        \n        # We represent times internally in 1 ps units, but users supply float quantities of seconds\n        period = int(period * 1e12)\n\n        if phase is None:\n            # By default, delay the first edge by half period. This causes any synchronous activity\n            # to happen at a non-zero time, distinguishing it from the reset values in the waveform\n            # viewer.\n            phase = period // 2\n        else:\n            phase = int(phase * 1e12) + period // 2\n        self._engine.add_clock_process(domain.clk, phase=phase, period=period)\n        self._clocked.add(domain)\n\n    def reset(self):\n        \"\"\"Reset the simulation.\n\n        Assign the reset value to every signal in the simulation, and restart every user process.",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 215,
  "is_truncated": true
}

{
  "status": "error",
  "error_type": "TimeoutExceeded",
  "error_message": "Command timed out after 90 seconds"
}

{
  "status": "ok",
  "filepath": "amaranth/sim/core.py",
  "content": "        \"\"\"\n        self._engine.reset()\n\n    def advance(self):\n        \"\"\"Advance the simulation.\n\n        Run every process and commit changes until a fixed point is reached, then advance time\n        to the closest deadline (if any). If there is an unstable combinatorial loop,\n        this function will never return.\n\n        Returns ``True`` if there are any active processes, ``False`` otherwise.\n        \"\"\"\n        return self._engine.advance()\n\n    def run(self):\n        \"\"\"Run the simulation while any processes are active.\n\n        Processes added with :meth:`add_process` and :meth:`add_sync_process` are initially active,\n        and may change their status using the ``yield Passive()`` and ``yield Active()`` commands.\n        Processes compiled from HDL and added with :meth:`add_clock` are always passive.\n        \"\"\"\n        while self.advance():\n            pass\n\n    def run_until(self, deadline, *, run_passive=False):\n        \"\"\"Run the simulation until it advances to ``deadline``.\n\n        If ``run_passive`` is ``False``, the simulation also stops when there are no active\n        processes, similar to :meth:`run`. Otherwise, the simulation will stop only after it\n        advances to or past ``deadline``.\n\n        If the simulation stops advancing, this function will never return.\n        \"\"\"\n        # Convert deadline in seconds into internal 1 ps units\n        deadline = deadline * 1e12\n        assert self._engine.now <= deadline\n        while (self.advance() or run_passive) and self._engine.now < deadline:\n            pass\n\n    def write_vcd(self, vcd_file, gtkw_file=None, *, traces=()):\n        \"\"\"Write waveforms to a Value Change Dump file, optionally populating a GTKWave save file.\n\n        This method returns a context manager. It can be used as: ::\n\n            sim = Simulator(frag)\n            sim.add_clock(1e-6)\n            with sim.write_vcd(\"dump.vcd\", \"dump.gtkw\"):\n                sim.run_until(1e-3)\n\n        Arguments\n        ---------\n        vcd_file : str or file-like object\n            Verilog Value Change Dump file or filename.\n        gtkw_file : str or file-like object\n            GTKWave save file or filename.\n        traces : iterable of Signal\n            Signals to display traces for.\n        \"\"\"\n        if self._engine.now != 0:\n            for file in (vcd_file, gtkw_file):\n                if hasattr(file, \"close\"):\n                    file.close()\n            raise ValueError(\"Cannot start writing waveforms after advancing simulation time\")\n\n        return self._engine.write_vcd(vcd_file=vcd_file, gtkw_file=gtkw_file, traces=traces)",
  "start_line": 151,
  "end_line": 215,
  "total_lines": 215,
  "is_truncated": false
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 51, in <module>\n  File \"/workspace/amaranth/sim/core.py\", line 86, in add_sync_process\n    process = self._check_process(process)\n  File \"/workspace/amaranth/sim/core.py\", line 73, in _check_process\n    raise TypeError(\"Cannot add a process {!r} because it is not a generator function\"\nTypeError: Cannot add a process <function <lambda> at 0x77a1e0325e10> because it is not a generator function\n",
  "details": {
    "stdout": "",
    "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 51, in <module>\n  File \"/workspace/amaranth/sim/core.py\", line 86, in add_sync_process\n    process = self._check_process(process)\n  File \"/workspace/amaranth/sim/core.py\", line 73, in _check_process\n    raise TypeError(\"Cannot add a process {!r} because it is not a generator function\"\nTypeError: Cannot add a process <function <lambda> at 0x77a1e0325e10> because it is not a generator function\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "Start test\nC2 read address 1: (sig mem_r_data)\nC4 read address 1 while writing address 2: (sig mem_r_data)\nC6 read address 1: (sig mem_r_data)\nC8 read address 1: (sig mem_r_data)\n",
  "stderr": "",
  "exit_code": 0
}