Library Reference

Modules

Bus

class cocotb_bus.bus.Bus(entity, name, signals, optional_signals=[], bus_separator='_', case_insensitive=True, array_idx=None)[source]

Wraps up a collection of signals.

Assumes we have a set of signals/nets named entity.<bus_name><bus_separator><signal>.

For example a bus stream_in with signals valid and data is assumed to be named dut.stream_in_valid and dut.stream_in_data (with the default separator ‘_’).

Parameters:

entity (SimHandle) – SimHandle instance to the entity containing the bus.
name (str) – Name of the bus. None for a nameless bus, e.g. bus-signals in an interface or a modport (untested on struct/record, but could work here as well).
signals (list or dict) – In the case of an object (passed to drive()/capture()) that has the same attribute names as the signal names of the bus, the signals argument can be a list of those names. When the object has different attribute names, the signals argument should be a dict that maps bus attribute names to object signal names.
optional_signals (list or dict, optional) – Signals that don’t have to be present on the interface. See the signals argument above for details.
bus_separator (str, optional) – Character(s) to use as separator between bus name and signal name. Defaults to ‘_’.
case_insensitive (bool, optional) – Perform case-insensitive match on signal names. Defaults to True.
array_idx (int or None, optional) – Optional index when signal is an array.

drive(obj, strict=False)[source]

Drives values onto the bus.

Parameters:

obj – Object with attribute names that match the bus signals.
strict (bool, optional) – Check that all signals are being assigned.

Raises:

AttributeError – If not all signals have been assigned when strict=True.

capture()[source]

Capture the values from the bus, returning an object representing the capture.

Returns:: A dictionary that supports access by attribute, where each attribute corresponds to each signal’s value.
Return type:: dict
Raises:: RuntimeError – If signal not present in bus, or attempt to modify a bus capture.

sample(obj, strict=False)[source]

Sample the values from the bus, assigning them to obj.

Parameters:

obj – Object with attribute names that match the bus signals.
strict (bool, optional) – Check that all signals being sampled are present in obj.

Raises:

AttributeError – If attribute is missing in obj when strict=True.

Driver

class cocotb_bus.drivers.Driver[source]

Class defining the standard interface for a driver within a testbench.

The driver is responsible for serializing transactions onto the physical pins of the interface. This may consume simulation time.

Constructor for a driver instance.

kill()[source]: Kill the coroutine sending stuff.

append(transaction, callback=None, event=None, **kwargs)[source]

Queue up a transaction to be sent over the bus.

Mechanisms are provided to permit the caller to know when the transaction is processed.

Parameters:

transaction (Any) – The transaction to be sent.
callback (Callable[[Any], Any]) – Optional function to be called when the transaction has been sent.
event (Event) – Event to be set when the transaction has been sent.
**kwargs (Any) – Any additional arguments used in child class’ _driver_send method.

Return type:

None

clear()[source]: Clear any queued transactions without sending them onto the bus.

send(transaction, sync=True, **kwargs)[source]

Blocking send call (hence must be “awaited” rather than called).

Sends the transaction over the bus.

Parameters:

transaction (Any) – The transaction to be sent.
sync (bool) – Synchronize the transfer by waiting for a rising edge.
**kwargs (Any) – Additional arguments used in child class’ _driver_send method.

Return type:

None

async _driver_send(transaction, sync=True, **kwargs)[source]

Actual implementation of the send.

Sub-classes should override this method to implement the actual send() routine.

Parameters:

transaction (Any) – The transaction to be sent.
sync (bool) – Synchronize the transfer by waiting for a rising edge.
**kwargs (Any) – Additional arguments if required for protocol implemented in a sub-class.

Return type:

None

async _send(transaction, callback, event, sync=True, **kwargs)[source]

Send coroutine.

Parameters:

transaction (Any) – The transaction to be sent.
callback (Callable[[Any], Any]) – Optional function to be called when the transaction has been sent.
event (Event) – event to be set when the transaction has been sent.
sync (bool) – Synchronize the transfer by waiting for a rising edge.
**kwargs – Any additional arguments used in child class’ _driver_send method.

Return type:

None

class cocotb_bus.drivers.BitDriver(signal, clk, generator=None)[source]

Bases: object

Drives a signal onto a single bit.

Useful for exercising ready/valid flags.

start(generator=None)[source]

Start generating data.

Parameters:

generator (Iterable[Tuple[int, int]]) –

Generator yielding data. The generator should yield tuples (on, off) with the number of cycles to be on, followed by the number of cycles to be off. Typically the generator should go on forever.

Example:

bit_driver.start((1, i % 5) for i in itertools.count())

Return type:

None

stop()[source]: Stop generating data.

class cocotb_bus.drivers.BusDriver(entity, name, clock, **kwargs)[source]

Bases: Driver

Wrapper around common functionality for buses which have:

a list of _signals (class attribute)

a list of _optional_signals (class attribute)

a clock

a name

an entity

Parameters:

entity (SimHandleBase) – A handle to the simulator entity.
name (str | None) – Name of this bus. None for a nameless bus, e.g. bus-signals in an interface or a modport. (untested on struct/record, but could work here as well).
clock (SimHandleBase) – A handle to the clock associated with this bus.
**kwargs (Any) – Keyword arguments forwarded to cocotb.Bus, see docs for that class for more information.

Constructor for a driver instance.

async _driver_send(transaction, sync=True)[source]

Implementation for BusDriver.

Parameters:

transaction – The transaction to send.
sync (bool) – Synchronize the transfer by waiting for a rising edge.

Return type:

None

_wait_for_signal(signal)[source]: This method will return when the specified signal has hit logic 1. The state will be in the ReadOnly phase so sim will need to move to NextTimeStep before registering more callbacks can occur.

_wait_for_nsignal(signal)[source]: This method will return when the specified signal has hit logic 0. The state will be in the ReadOnly phase so sim will need to move to NextTimeStep before registering more callbacks can occur.

class cocotb_bus.drivers.ValidatedBusDriver(entity, name, clock, *, valid_generator=None, **kwargs)[source]

Bases: BusDriver

Same as a BusDriver except we support an optional generator to control which cycles are valid.

Parameters:

entity (SimHandle) – A handle to the simulator entity.
name (str) – Name of this bus.
clock (SimHandle) – A handle to the clock associated with this bus.
valid_generator (generator, optional) – a generator that yields tuples of (valid, invalid) cycles to insert.
kwargs (Any) –

Constructor for a driver instance.

_next_valids()[source]

Optionally insert invalid cycles every N cycles.

The generator should yield tuples with the number of cycles to be on followed by the number of cycles to be off. The on cycles should be non-zero, we skip invalid generator entries.

set_valid_generator(valid_generator=None)[source]: Set a new valid generator for this bus.

Monitor

class cocotb_bus.monitors.Monitor(callback=None, event=None)[source]

Base class for Monitor objects.

Monitors are passive ‘listening’ objects that monitor pins going in or out of a DUT. This class should not be used directly, but should be sub-classed and the internal _monitor_recv() method should be overridden. This _monitor_recv() method should capture some behavior of the pins, form a transaction, and pass this transaction to the internal _recv() method. The _monitor_recv() method is added to the cocotb scheduler during the __init__ phase, so it should not be awaited anywhere.

The primary use of a Monitor is as an interface for a Scoreboard.

Parameters:

callback (callable) – Callback to be called with each recovered transaction as the argument. If the callback isn’t used, received transactions will be placed on a queue and the event used to notify any consumers.
event (cocotb.triggers.Event) – Event that will be called when a transaction is received through the internal _recv() method. Event.data is set to the received transaction.

kill()[source]: Kill the monitor coroutine.

add_callback(callback)[source]

Add function as a callback.

Parameters:: callback (callable) – The function to call back.

wait_for_recv(timeout=None)[source]

With timeout, wait() for transaction to arrive on monitor and return its data.

Parameters:: timeout – The timeout value for Timer. Defaults to None.
Returns:: Data of received transaction.

_monitor_recv()[source]

Actual implementation of the receiver.

Sub-classes should override this method to implement the actual receive routine and call _recv() with the recovered transaction.

_recv(transaction)[source]: Common handling of a received transaction.

class cocotb_bus.monitors.BusMonitor(entity, name, clock, reset=None, reset_n=None, callback=None, event=None, **kwargs)[source]

Bases: Monitor

Wrapper providing common functionality for monitoring buses.

property in_reset: Boolean flag showing whether the bus is in reset state or not.

Scoreboard

Common scoreboarding capability.

class cocotb_bus.scoreboard.Scoreboard(dut, reorder_depth=0, fail_immediately=True)[source]

Bases: object

Generic scoreboarding class.

We can add interfaces by providing a monitor and an expected output queue.

The expected output can either be a function which provides a transaction or a simple list containing the expected output.

Parameters:

dut (SimHandle) – Handle to the DUT.
reorder_depth (int, optional) – Consider up to reorder_depth elements of the expected result list as passing matches. Default is 0, meaning only the first element in the expected result list is considered for a passing match.
fail_immediately (bool, optional) – Raise AssertionError immediately when something is wrong instead of just recording an error. Default is True.

property result

Determine the test result, do we have any pending data remaining?

Raises:: AssertionError – If not all expected output was received or error were recorded during the test.

compare(got, exp, log, strict_type=True)[source]

Common function for comparing two transactions.

Can be re-implemented by a sub-class.

Parameters:

got – The received transaction.
exp – The expected transaction.
log – The logger for reporting messages.
strict_type (bool, optional) – Require transaction type to match exactly if True, otherwise compare its string representation.

Raises:

AssertionError – If received transaction differed from expected transaction when fail_immediately is True. If strict_type is True, also the transaction type must match.

add_interface(monitor, expected_output, compare_fn=None, reorder_depth=0, strict_type=True)[source]

Add an interface to be scoreboarded.

Provides a function which the monitor will callback with received transactions.

Simply check against the expected output.

Parameters:

monitor – The monitor object.
expected_output – Queue of expected outputs.
compare_fn (callable, optional) – Function doing the actual comparison.
reorder_depth (int, optional) – Consider up to reorder_depth elements of the expected result list as passing matches. Default is 0, meaning only the first element in the expected result list is considered for a passing match.
strict_type (bool, optional) – Require transaction type to match exactly if True, otherwise compare its string representation.

Raises:

TypeError – If no monitor is on the interface or compare_fn is not a callable function.

Implemented Testbench Structures

Drivers

AMBA

Advanced Microcontroller Bus Architecture.

class cocotb_bus.drivers.amba.AXI4Master(entity, name, clock, **kwargs)[source]

AXI4 Master

TODO: Kill all pending transactions if reset is asserted.

Constructor for a driver instance.

Parameters:

entity (SimHandleBase) –
name (str) –
clock (SimHandleBase) –
kwargs (Any) –

write(address, value, *, size=None, burst=AXIBurst.INCR, byte_enable=None, address_latency=0, data_latency=0, sync=True)[source]

Write a value to an address.

With unaligned writes (when address is not a multiple of size), only the low size - address % size Bytes are written for: * the last element of value for INCR or WRAP bursts, or * every element of value for FIXED bursts.

Parameters:

address (int) – The address to write to.
value (int | Sequence[int]) – The data value(s) to write.
size (int | None) – The size (in bytes) of each beat. Defaults to None (width of the data bus).
burst (AXIBurst) – The burst type, either FIXED, INCR or WRAP. Defaults to INCR.
byte_enable (int | None | Sequence[int | None]) – Which bytes in value to actually write. Defaults to None (write all bytes).
address_latency (int) – Delay before setting the address (in clock cycles). Default is no delay.
data_latency (int) – Delay before setting the data value (in clock cycles). Default is no delay.
sync (bool) – Wait for rising edge on clock initially. Defaults to True.

Raises:

ValueError – If any of the input parameters is invalid.
AXIProtocolError – If write response from AXI is not OKAY.

Return type:

None

read(address, length=1, *, size=None, burst=AXIBurst.INCR, return_rresp=False, sync=True)[source]

Read from an address.

With unaligned reads (when address is not a multiple of size) with INCR or WRAP bursts, the last element of the returned read data will be only the low-order size - address % size Bytes of the last word. With unaligned reads with FIXED bursts, every element of the returned read data will be only the low-order size - address % size Bytes.

Parameters:

address (int) – The address to read from.
length (int) – Number of words to transfer. Defaults to 1.
size (int | None) – The size (in bytes) of each beat. Defaults to None (width of the data bus).
burst (AXIBurst) – The burst type, either FIXED, INCR or WRAP. Defaults to INCR.
return_rresp (bool) – Return the list of RRESP values, instead of raising an AXIProtocolError in case of not OKAY. Defaults to False.
sync (bool) – Wait for rising edge on clock initially. Defaults to True.

Returns:

The read data values or, if return_rresp is True, a list of pairs each containing the data and RRESP values.-

Raises:

ValueError – If any of the input parameters is invalid.
AXIProtocolError – If read response from AXI is not OKAY and return_rresp is False
AXIReadBurstLengthMismatch – If the received number of words does not match the requested one.

Return type:

List[BinaryValue] | List[Tuple[BinaryValue, AXIxRESP]]

class cocotb_bus.drivers.amba.AXI4LiteMaster(entity, name, clock, **kwargs)[source]

AXI4-Lite Master

Constructor for a driver instance.

Parameters:

entity (SimHandleBase) –
name (str) –
clock (SimHandleBase) –
kwargs (Any) –

write(address, value, byte_enable=None, address_latency=0, data_latency=0, sync=True)[source]

Write a value to an address.

Parameters:

address (int) – The address to write to.
value (int) – The data value to write.
byte_enable (int | None) – Which bytes in value to actually write. Defaults to None (write all bytes).
address_latency (int) – Delay before setting the address (in clock cycles). Default is no delay.
data_latency (int) – Delay before setting the data value (in clock cycles). Default is no delay.
sync (bool) – Wait for rising edge on clock initially. Defaults to True.
Returns – The write response value.
Raises – ValueError: If any of the input parameters is invalid. AXIProtocolError: If write response from AXI is not OKAY.

Return type:

BinaryValue

read(address, sync=True)[source]

Read from an address.

Parameters:

address (int) – The address to read from.
length – Number of words to transfer
sync (bool) – Wait for rising edge on clock initially. Defaults to True.

Returns:

The read data value.

Raises:

AXIProtocolError – If read response from AXI is not OKAY.

Return type:

BinaryValue

class cocotb_bus.drivers.amba.AXI4Slave(entity, name, clock, memory, callback=None, event=None, big_endian=False, **kwargs)[source]

AXI4 Slave

Monitors an internal memory and handles read and write requests.

Constructor for a driver instance.

Avalon

class cocotb_bus.drivers.avalon.AvalonMM(entity, name, clock, **kwargs)[source]

Bases: BusDriver

Avalon Memory Mapped Interface (Avalon-MM) Driver.

Currently we only support the mode required to communicate with SF avalon_mapper which is a limited subset of all the signals.

Blocking operation is all that is supported at the moment, and for the near future as well. Posted responses from a slave are not supported.

Constructor for a driver instance.

class cocotb_bus.drivers.avalon.AvalonMaster(entity, name, clock, **kwargs)[source]

Bases: AvalonMM

Avalon Memory Mapped Interface (Avalon-MM) Master.

Constructor for a driver instance.

read(address, sync=True)[source]

Issue a request to the bus and block until this comes back.

Simulation time still progresses but syntactically it blocks.

Parameters:

address (int) – The address to read from.
sync (bool) – Wait for rising edge on clock initially. Defaults to True.

Returns:

The read data value.

Raises:

TestError – If master is write-only.

Return type:

BinaryValue

write(address, value)[source]

Issue a write to the given address with the specified value.

Parameters:

address (int) – The address to write to.
value (int) – The data value to write.

Raises:

TestError – If master is read-only.

Return type:

None

class cocotb_bus.drivers.avalon.AvalonMemory(entity, name, clock, readlatency_min=1, readlatency_max=1, memory=None, avl_properties={}, **kwargs)[source]

Bases: BusDriver

Emulate a memory, with back-door access.

Constructor for a driver instance.

class cocotb_bus.drivers.avalon.AvalonST(entity, name, clock, *, config={}, **kwargs)[source]

Bases: ValidatedBusDriver

Avalon Streaming Interface (Avalon-ST) Driver

Constructor for a driver instance.

class cocotb_bus.drivers.avalon.AvalonSTPkts(entity, name, clock, *, config={}, **kwargs)[source]

Bases: ValidatedBusDriver

Avalon Streaming Interface (Avalon-ST) Driver, packetized.

Constructor for a driver instance.

OPB

class cocotb_bus.drivers.opb.OPBMaster(entity, name, clock, **kwargs)[source]

Bases: BusDriver

On-chip peripheral bus master.

Constructor for a driver instance.

read(address, sync=True)[source]

Issue a request to the bus and block until this comes back.

Simulation time still progresses but syntactically it blocks.

Parameters:

address (int) – The address to read from.
sync (bool) – Wait for rising edge on clock initially. Defaults to True.

Returns:

The read data value.

Raises:

OPBException – If read took longer than 16 cycles.

Return type:

BinaryValue

write(address, value, sync=True)[source]

Issue a write to the given address with the specified value.

Parameters:

address (int) – The address to read from.
value (int) – The data value to write.
sync (bool) – Wait for rising edge on clock initially. Defaults to True.

Raises:

OPBException – If write took longer than 16 cycles.

Return type:

None

XGMII

class cocotb_bus.drivers.xgmii.XGMII(signal, clock, interleaved=True)[source]

Bases: Driver

XGMII (10 Gigabit Media Independent Interface) driver.

Parameters:

signal (SimHandleBase) – The XGMII data bus.
clock (SimHandleBase) – The associated clock (assumed to be driven by another coroutine).
interleaved (bool) – Whether control bits are interleaved with the data bytes or not.

If interleaved the bus is: byte0, byte0_control, byte1, byte1_control, …
Otherwise expect: byte0, byte1, …, byte0_control, byte1_control, …

static layer1(packet)[source]

Take an Ethernet packet (as a string) and format as a layer 1 packet.

Pad to 64 bytes, prepend preamble and append 4-byte CRC on the end.

Parameters:: packet (bytes) – The Ethernet packet to format.
Returns:: The formatted layer 1 packet.
Return type:: bytes

idle()[source]: Helper function to set bus to IDLE state.

terminate(index)[source]

Helper function to terminate from a provided lane index.

Parameters:: index (int) – The index to terminate.
Return type:: None

Monitors

Avalon

class cocotb_bus.monitors.avalon.AvalonST(entity, name, clock, *, config={}, **kwargs)[source]

Bases: BusMonitor

Avalon-ST bus.

Non-packetized so each valid word is a separate transaction.

class cocotb_bus.monitors.avalon.AvalonSTPkts(entity, name, clock, *, config={}, report_channel=False, **kwargs)[source]

Bases: BusMonitor

Packetized Avalon-ST bus.

Parameters:

entity – see BusMonitor
name – see BusMonitor
clock – see BusMonitor
config (dict) – bus configuration options
report_channel (bool) – report channel with data, default is False Setting to True on bus without channel signal will give an error

XGMII

class cocotb_bus.monitors.xgmii.XGMII(signal, clock, interleaved=True, callback=None, event=None)[source]

Bases: Monitor

XGMII (10 Gigabit Media Independent Interface) Monitor.

Assumes a single vector, either 4 or 8 bytes plus control bit for each byte.

If interleaved is True then the control bits are adjacent to the bytes.

Changed in version 1.4.0: This now emits packets of type bytes rather than str, which matches the behavior of cocotb.drivers.xgmii.XGMII.

Parameters:

signal (SimHandle) – The XGMII data bus.
clock (SimHandle) – The associated clock (assumed to be driven by another coroutine).
interleaved (bool, optional) – Whether control bits are interleaved with the data bytes or not.

If interleaved the bus is: byte0, byte0_control, byte1, byte1_control, …
Otherwise expect: byte0, byte1, …, byte0_control, byte1_control, …