amaranth-lang__amaranth-923

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

{
  "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": "FileWriteError",
  "error_message": "Path traversal detected: '/tmp/repro.py' escapes workspace root."
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 2, in <module>\nImportError: cannot import name 'TestCaseWithSimulator' from 'amaranth.sim' (/workspace/amaranth/sim/__init__.py)\n",
  "details": {
    "stdout": "",
    "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 2, in <module>\nImportError: cannot import name 'TestCaseWithSimulator' from 'amaranth.sim' (/workspace/amaranth/sim/__init__.py)\n",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "filepath": "amaranth/sim/__init__.py",
  "content": "from .core import *\n\n\n__all__ = [\"Settle\", \"Delay\", \"Tick\", \"Passive\", \"Active\", \"Simulator\"]",
  "start_line": 1,
  "end_line": 4,
  "total_lines": 4,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "tests/test_hdl_mem.py",
  "content": "# amaranth: UnusedElaboratable=no\n\nfrom amaranth.hdl.ast import *\nfrom amaranth.hdl.mem import *\n\nfrom .utils import *\n\n\nclass MemoryTestCase(FHDLTestCase):\n    def test_name(self):\n        m1 = Memory(width=8, depth=4)\n        self.assertEqual(m1.name, \"m1\")\n        m2 = [Memory(width=8, depth=4)][0]\n        self.assertEqual(m2.name, \"$memory\")\n        m3 = Memory(width=8, depth=4, name=\"foo\")\n        self.assertEqual(m3.name, \"foo\")\n\n    def test_geometry(self):\n        m = Memory(width=8, depth=4)\n        self.assertEqual(m.width, 8)\n        self.assertEqual(m.depth, 4)\n\n    def test_geometry_wrong(self):\n        with self.assertRaisesRegex(TypeError,\n                r\"^Memory width must be a non-negative integer, not -1$\"):\n            m = Memory(width=-1, depth=4)\n        with self.assertRaisesRegex(TypeError,\n                r\"^Memory depth must be a non-negative integer, not -1$\"):\n            m = Memory(width=8, depth=-1)\n\n    def test_init(self):\n        m = Memory(width=8, depth=4, init=range(4))\n        self.assertEqual(m.init, [0, 1, 2, 3])\n\n    def test_init_wrong_count(self):\n        with self.assertRaisesRegex(ValueError,\n                r\"^Memory initialization value count exceed memory depth \\(8 > 4\\)$\"):\n            m = Memory(width=8, depth=4, init=range(8))\n\n    def test_init_wrong_type(self):\n        with self.assertRaisesRegex(TypeError,\n                (r\"^Memory initialization value at address 1: \"\n                    r\"'str' object cannot be interpreted as an integer$\")):\n            m = Memory(width=8, depth=4, init=[1, \"0\"])\n\n    def test_attrs(self):\n        m1 = Memory(width=8, depth=4)\n        self.assertEqual(m1.attrs, {})\n        m2 = Memory(width=8, depth=4, attrs={\"ram_block\": True})\n        self.assertEqual(m2.attrs, {\"ram_block\": True})\n\n    def test_read_port_transparent(self):\n        mem    = Memory(width=8, depth=4)\n        rdport = mem.read_port()\n        self.assertEqual(rdport.memory, mem)\n        self.assertEqual(rdport.domain, \"sync\")\n        self.assertEqual(rdport.transparent, True)\n        self.assertEqual(len(rdport.addr), 2)\n        self.assertEqual(len(rdport.data), 8)\n        self.assertEqual(len(rdport.en), 1)\n        self.assertIsInstance(rdport.en, Signal)\n        self.assertEqual(rdport.en.reset, 1)\n\n    def test_read_port_non_transparent(self):\n        mem    = Memory(width=8, depth=4)\n        rdport = mem.read_port(transparent=False)\n        self.assertEqual(rdport.memory, mem)\n        self.assertEqual(rdport.domain, \"sync\")\n        self.assertEqual(rdport.transparent, False)\n        self.assertEqual(len(rdport.en), 1)\n        self.assertIsInstance(rdport.en, Signal)\n        self.assertEqual(rdport.en.reset, 1)\n\n    def test_read_port_asynchronous(self):\n        mem    = Memory(width=8, depth=4)\n        rdport = mem.read_port(domain=\"comb\")\n        self.assertEqual(rdport.memory, mem)\n        self.assertEqual(rdport.domain, \"comb\")\n        self.assertEqual(rdport.transparent, True)\n        self.assertEqual(len(rdport.en), 1)\n        self.assertIsInstance(rdport.en, Const)\n        self.assertEqual(rdport.en.value, 1)\n\n    def test_read_port_wrong(self):\n        mem = Memory(width=8, depth=4)\n        with self.assertRaisesRegex(ValueError,\n                r\"^Read port cannot be simultaneously asynchronous and non-transparent$\"):\n            mem.read_port(domain=\"comb\", transparent=False)\n\n    def test_write_port(self):\n        mem    = Memory(width=8, depth=4)\n        wrport = mem.write_port()\n        self.assertEqual(wrport.memory, mem)\n        self.assertEqual(wrport.domain, \"sync\")\n        self.assertEqual(wrport.granularity, 8)\n        self.assertEqual(len(wrport.addr), 2)\n        self.assertEqual(len(wrport.data), 8)\n        self.assertEqual(len(wrport.en), 1)\n\n    def test_write_port_granularity(self):\n        mem    = Memory(width=8, depth=4)\n        wrport = mem.write_port(granularity=2)\n        self.assertEqual(wrport.memory, mem)\n        self.assertEqual(wrport.domain, \"sync\")\n        self.assertEqual(wrport.granularity, 2)\n        self.assertEqual(len(wrport.addr), 2)\n        self.assertEqual(len(wrport.data), 8)\n        self.assertEqual(len(wrport.en), 4)\n\n    def test_write_port_granularity_wrong(self):\n        mem = Memory(width=8, depth=4)\n        with self.assertRaisesRegex(TypeError,\n                r\"^Write port granularity must be a non-negative integer, not -1$\"):\n            mem.write_port(granularity=-1)\n        with self.assertRaisesRegex(ValueError,\n                r\"^Write port granularity must not be greater than memory width \\(10 > 8\\)$\"):\n            mem.write_port(granularity=10)\n        with self.assertRaisesRegex(ValueError,\n                r\"^Write port granularity must divide memory width evenly$\"):\n            mem.write_port(granularity=3)\n\n\nclass DummyPortTestCase(FHDLTestCase):\n    def test_name(self):\n        p1 = DummyPort(data_width=8, addr_width=2)\n        self.assertEqual(p1.addr.name, \"p1_addr\")\n        p2 = [DummyPort(data_width=8, addr_width=2)][0]\n        self.assertEqual(p2.addr.name, \"dummy_addr\")\n        p3 = DummyPort(data_width=8, addr_width=2, name=\"foo\")\n        self.assertEqual(p3.addr.name, \"foo_addr\")\n\n    def test_sizes(self):\n        p1 = DummyPort(data_width=8, addr_width=2)\n        self.assertEqual(p1.addr.width, 2)\n        self.assertEqual(p1.data.width, 8)\n        self.assertEqual(p1.en.width, 1)\n        p2 = DummyPort(data_width=8, addr_width=2, granularity=2)\n        self.assertEqual(p2.en.width, 4)",
  "start_line": 1,
  "end_line": 138,
  "total_lines": 138,
  "is_truncated": false
}

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

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

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 49, in <module>\n  File \"<stdin>\", line 46, in test_minimal\nTypeError: Simulator.run() got an unexpected keyword argument 'steps'\n",
  "details": {
    "stdout": "",
    "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 49, in <module>\n  File \"<stdin>\", line 46, in test_minimal\nTypeError: Simulator.run() got an unexpected keyword argument 'steps'\n",
    "exit_code": 1
  }
}