Add first simple test with a simple 2-dimensional array

tests/cocotb_tests/libs/AMBA3AHBLiteDriver.py

@@ -0,0 +1,140 @@
+from enum import Enum
+import math
+import cocotb
+from cocotb.triggers import Timer, RisingEdge
+
+# TODO: Does not yet implement HREADY_OUT == 0
+# TODO: Add support for HRESP (and throw error if HRESP occurs)
+
+class HTRANS(Enum):
+    IDLE = 0
+    BUSY = 1
+    NONSEQ = 2
+    SEQ = 3
+
+class AMBA3AHBLiteDriver:
+    """Wraps up a collection of functions to drive an AMBA3AHBLite Bus.
+
+    This is not an extensive set of features and merely enough to test
+    out SRDL2SV registers.
+    """
+
+    def __init__(self, dut, nbytes: int):
+        self._nbytes = nbytes
+        self._dut = dut
+
+    @cocotb.coroutine
+    async def reset(self, time: int = 10):
+        """Resets bus for a given amount of time"""
+
+        self._dut.HRESETn <= 0
+
+        await Timer(time, units='ns')
+        await RisingEdge(self._dut.clk)
+
+        self._dut.HRESETn <= 1
+
+    @cocotb.coroutine
+    async def write(self, address: int, value: int, nbytes = None, step_size = None):
+        if not nbytes:
+            nbytes = self._nbytes
+
+        if not step_size:
+            if (step_size := address % self._nbytes) == 0:
+                step_size = self._nbytes
+
+        # Dictionary to return address/value pairs
+        write_dict = {}
+
+        # Start counting bytes
+        nbytes_cnt = 0
+
+        # Initiate write
+        self._dut.HSEL <= 1
+        self._dut.HWRITE <= 1
+        self._dut.HADDR <= address
+        self._dut.HTRANS <= HTRANS.NONSEQ.value
+        self._dut.HSIZE <= int(math.log2(step_size))
+
+        await RisingEdge(self._dut.clk)
+
+        while True:
+            # Save address from previous phase
+            previous_address = hex(self._dut.HADDR.value)
+
+            # Set data for dataphase
+            self._dut.HWDATA <= (value >> (nbytes_cnt * 8))
+
+            # Check if we are done in next phase
+            if (nbytes_cnt := nbytes_cnt + step_size) >= nbytes:
+                self._dut.HTRANS <= HTRANS.IDLE.value
+            else:
+                # Update address
+                self._dut.HADDR <= self._dut.HADDR.value + step_size
+
+            # Wait for next clock cycle
+            await RisingEdge(self._dut.clk)
+
+            # Save into dictionary
+            write_dict[previous_address] = hex(self._dut.HWDATA.value)
+
+            if nbytes_cnt >= nbytes:
+                break
+
+        self._dut.HWRITE <= 0
+        self._dut.HSEL <= 0
+
+        await RisingEdge(self._dut.clk)
+
+        return write_dict
+
+    @cocotb.coroutine
+    async def read(self, address: int, nbytes = None, step_size = None):
+        if not nbytes:
+            nbytes = self._nbytes
+
+        if not step_size:
+            if (step_size := address % self._nbytes) == 0:
+                step_size = self._nbytes
+
+        # Dictionary to return address/value pairs
+        read_dict = {}
+
+        # Start counting bytes
+        nbytes_cnt = 0
+
+        # Initiate read
+        self._dut.HSEL <= 1
+        self._dut.HWRITE <= 0
+        self._dut.HADDR <= address
+        self._dut.HTRANS <= HTRANS.NONSEQ.value
+        self._dut.HSIZE <= int(math.log2(step_size))
+
+        await RisingEdge(self._dut.clk)
+
+        while True:
+            # Save address from previous phase
+            previous_address = hex(self._dut.HADDR.value)
+
+            # Check if we are done in next phase
+            if (nbytes_cnt := nbytes_cnt + step_size) >= nbytes:
+                self._dut.HTRANS <= HTRANS.IDLE.value
+            else:
+                # Update address
+                self._dut.HADDR <= self._dut.HADDR.value + step_size
+
+            # Wait for next clock cycle
+            await RisingEdge(self._dut.clk)
+
+            # Save into dictionary
+            read_dict[previous_address] = hex(self._dut.HRDATA.value)
+
+            if nbytes_cnt >= nbytes:
+                break
+
+        self._dut.HTRANS <= HTRANS.IDLE.value
+        self._dut.HSEL <= 0
+
+        await RisingEdge(self._dut.clk)
+
+        return read_dict