Fix compatibility with upcoming cocotb 2.0

.github/workflows/tests.yml

@@ -26,7 +26,7 @@ jobs:
       matrix:
         python-version: ["3.6", "3.12"]
         # NOTE: align with versions in noxfile.py:
-        cocotb-version: ["1.6.0", "1.9.0"]
+        cocotb-version: ["1.6.0", "1.9.0", "github-b9dd5ee1"]
         include:
         - sim: icarus
           sim-version: apt

examples/dff/tests/dff_cocotb.py

@@ -31,9 +31,8 @@
 import cocotb
 from cocotb.clock import Clock
 from cocotb.triggers import ReadOnly, RisingEdge
-from cocotb.binary import BinaryValue
 
-from cocotb_bus.compat import TestFactory
+from cocotb_bus.compat import create_binary, TestFactory
 from cocotb_bus.monitors import Monitor
 from cocotb_bus.drivers import BitDriver
 from cocotb_bus.scoreboard import Scoreboard
@@ -138,7 +137,7 @@ async def run_test(dut):
 
     cocotb.start_soon(Clock(dut.c, 10, 'us').start(start_high=False))
 
-    tb = DFF_TB(dut, init_val=BinaryValue("0"))
+    tb = DFF_TB(dut, init_val=create_binary("0", 1, big_endian=True))
 
     clkedge = RisingEdge(dut.c)
 
@@ -159,5 +158,6 @@ async def run_test(dut):
 
 
 # Register the test.
+
 factory = TestFactory(run_test)
 factory.generate_tests()

examples/endian_swapper/tests/test_endian_swapper.py

@@ -35,7 +35,7 @@
 
 from cocotb.clock import Clock
 from cocotb.triggers import Timer, RisingEdge, ReadOnly
-from cocotb_bus.compat import TestFactory
+from cocotb_bus.compat import TestFactory, convert_binary_to_unsigned
 from cocotb_bus.drivers import BitDriver
 from cocotb_bus.drivers.avalon import AvalonSTPkts as AvalonSTDriver
 from cocotb_bus.drivers.avalon import AvalonMaster
@@ -223,8 +223,12 @@ async def run_test(dut, data_in=None, config_coroutine=None, idle_inserter=None,
 
     pkt_count = await tb.csr.read(1)
 
-    assert pkt_count.integer == tb.pkts_sent, "DUT recorded %d packets but tb counted %d" % (pkt_count.integer, tb.pkts_sent)
-    dut._log.info("DUT correctly counted %d packets" % pkt_count.integer)
+    assert convert_binary_to_unsigned(pkt_count) == tb.pkts_sent, (
+        "DUT recorded %d packets but tb counted %d" % (
+            convert_binary_to_unsigned(pkt_count), tb.pkts_sent
+        )
+    )
+    dut._log.info("DUT correctly counted %d packets" % convert_binary_to_unsigned(pkt_count))
 
     raise tb.scoreboard.result
 

noxfile.py

@@ -2,9 +2,13 @@
 
 
 @nox.session
-@nox.parametrize("cocotb", ["1.6.0", "1.9.0"])
+@nox.parametrize("cocotb", ["1.6.0", "1.9.0", "github-b9dd5ee1"])
 def tests(session, cocotb):
-    session.install("pytest", "coverage", f"cocotb=={cocotb}")
+    if cocotb.startswith("github-"):
+        cocotb_req = "git+https://github.com/cocotb/cocotb@" + cocotb[len("github-"):]
+    else:
+        cocotb_req = f"cocotb=={cocotb}"
+    session.install("pytest", "coverage", cocotb_req)
     session.install(".")
     session.run("make", external=True)
 

setup.py

@@ -30,8 +30,9 @@ def get_version(version_file):
         packages=find_packages("src"),
         package_dir={"": "src"},
         install_requires=[
-            "cocotb>=1.6.0,<2.0",
+            "cocotb>=1.6.0",
             "scapy",
+            "packaging",
         ],
         python_requires='>=3.6'
     )

src/cocotb_bus/compat.py

@@ -4,6 +4,7 @@
 
 
 from packaging.version import parse as parse_version
+from typing import Optional, Union
 import warnings
 
 import cocotb
@@ -15,17 +16,104 @@ def TestFactory(*args, **kwargs):
         return cocotb.regression.TestFactory(*args, **kwargs)
 
 
-if parse_version(cocotb.__version__) > parse_version("2.dev0"):
+cocotb_2x_or_newer = parse_version(cocotb.__version__) > parse_version("2.dev0")
+if cocotb_2x_or_newer:
+    from cocotb.types import LogicArray, Range
+    BinaryType = LogicArray
+
+
     def set_event(event, data):
         event.data = data
         event.set()
 
     def coroutine(f):
         return f
 
+
+    def create_binary(binstr: Union[int, str, bytes, LogicArray], bit_count: int, big_endian: bool):
+        if big_endian:
+            range=Range(0, 'to', bit_count - 1)
+        else:
+            range=Range(bit_count - 1, 'downto', 0)
+
+        if isinstance(binstr, bytes):
+            if not big_endian:
+                binstr = reversed(binstr)
+            binstr = "".join([format(char, "08b") for char in binstr])
+
+            if big_endian:
+                binstr = binstr + "0" * (bit_count - len(binstr))
+            else:
+                binstr = "0" * (bit_count - len(binstr)) + binstr
+
+        return LogicArray(binstr, range=range)
+
+
+    def create_binary_from_other(value: BinaryType, binstr: Union[int, str, bytes]):
+        return LogicArray(value=binstr, range=value.range)
+
+
+    def convert_binary_to_bytes(value: BinaryType, big_endian: bool):
+        return value.to_bytes('big' if big_endian else 'little')
+
+
+    def convert_binary_to_unsigned(value: Union[int, BinaryType]):
+        if isinstance(value, int):
+            return value
+        return value.to_unsigned()
+
+
+    def binary_slice(value: BinaryType, start: int, end: int):
+        # On 2.x the default slice direction is downto, whereas on 1.9.x it's to
+        # Additionally, in the case of BinaryValue slices always operate within 0..len(v) index
+        # range regardless of how BinaryValue was acquired. In the case of LogicArray slices
+        # operate on its range which doesn't necessarily start at zero. E.g. the following
+        # returns top 8 bits: LogicArray(..., Range(31, 'downto', 0))[31:15][31:20][31:24]
+        offset = min(value.range.left, value.range.right)
+        start = len(value) - start - 1 + offset
+        end = len(value) - end - 1 + offset
+        return value[start:end]
+
 else:
+    from cocotb.binary import BinaryValue
+    from cocotb.binary import _RESOLVE_TO_CHOICE
+
+    coroutine = cocotb.coroutine
+    BinaryType = BinaryValue
+
+
     def set_event(event, data):
         event.set(data)
 
-    def coroutine(f):
-        return cocotb.coroutine(f)
+
+    def create_binary(binstr: Union[int, str, bytes, BinaryValue], bit_count: int,
+                      big_endian: bool):
+        return BinaryValue(value=binstr, n_bits=bit_count, bigEndian=big_endian)
+
+
+    def create_binary_from_other(value: BinaryType, binstr: Union[int, str, bytes]):
+        return BinaryValue(value=binstr, n_bits=value.n_bits, bigEndian=value.big_endian)
+
+
+    def convert_binary_to_bytes(value: BinaryType, big_endian: bool):
+        # Setting bigEndian does not affect initialization because value.binstr already is adjusted
+        # to contain n_bits number of bits. Only access via buff is affected.
+        value = BinaryValue(value=value.binstr, n_bits=value.n_bits, bigEndian=big_endian)
+        return value.buff
+
+
+    def convert_binary_to_unsigned(value: Union[int, BinaryType]):
+        # In 1.9.x code has more automatic conversions, therefore code does not consistently
+        # apply .integer
+        if isinstance(value, int):
+            return value
+        return value.integer
+
+
+    def binary_slice(value: BinaryType, start: int, end: int):
+        # On 2.x the default slice direction is downto, whereas on 1.9.x it's to
+        return value[start:end]
+
+
+def binary_is_resolvable(value: BinaryType):
+    return value.is_resolvable

src/cocotb_bus/drivers/__init__.py

@@ -16,7 +16,7 @@
 from cocotb.handle import SimHandleBase
 
 from cocotb_bus.bus import Bus
-from cocotb_bus.compat import coroutine
+from cocotb_bus.compat import coroutine, convert_binary_to_unsigned
 
 
 class BitDriver:
@@ -260,7 +260,7 @@ async def _wait_for_signal(self, signal):
         registering more callbacks can occur.
         """
         await ReadOnly()
-        while signal.value.integer != 1:
+        while convert_binary_to_unsigned(signal.value) != 1:
             await RisingEdge(signal)
             await ReadOnly()
         await NextTimeStep()
@@ -274,7 +274,7 @@ async def _wait_for_nsignal(self, signal):
         registering more callbacks can occur.
         """
         await ReadOnly()
-        while signal.value.integer != 0:
+        while convert_binary_to_unsigned(signal.value) != 0:
             await Edge(signal)
             await ReadOnly()
         await NextTimeStep()