Reference Manual

Triggers

Collection of tools for concurrency and synchronization in cocotb.

class coconext.triggers.Notify

Bases: object

Notify all waiters of an event.

cocotb.triggers.Event has state. It can be in the state where Event.is_set() is False and calling Event.wait() will return a Trigger that only fires after Event.set() is called. Or it can be in a state where Event.is_set() is True and calling Event.wait() will return a Trigger that fires immediately.

This object does not have state. It behaves like Event if it were only ever in the Event.is_set() is False state. All calls to wait() will block until the next call to notify() occurs.

Added in version 0.1.

__init__()

notify()

Wake up all waiters.

Return type:: None

async wait()

Wait until the notify() method is called.

Return type:: None

class coconext.triggers.TaskManager

Bases: object

An asynchronous context manager which enables the user to run coroutine functions or await awaitables concurrently and wait for them all to finish.

awaitables can be added to the TaskManager using start_soon(). Likewise, fork() can be used to add coroutine functions. Both methods return the Task which is running the concurrent behavior.

When control reaches the end of the context block the TaskManager blocks the Task using it until all children Tasks of the TaskManager complete.

async with TaskManager() as tm:

    @tm.fork
    async def drive(): ...

    rising_edge = cocotb.start_soon(RisingEdge(dut.clk))

    await drive  # wait for drive() to end
    rising_edge.cancel()  # cancel a child Task

# Control returns here when all child Tasks have completed

__init__()

fork(coro, *args, **kwargs)

Decorate a coroutine function to run it concurrently.

Parameters:

coro (Callable[[ParamSpec(P, bound= None)], Coroutine[Trigger, None, TypeVar(T)]]) – A coroutine function to run concurrently.
args (ParamSpecArgs) – Positional args to pass to the call of coro.
kwargs (ParamSpecKwargs) – Keyword args to pass to the call of coro.

Return type:

Task[TypeVar(T)]

Returns:

A Task which is running coro concurrently.

async with TaskManager() as tm:

    @tm.fork
    async def my_func():
        # Do stuff
        ...

    @tb.fork
    async def other_func():
        # Do other stuff in parallel to my_func
        ...

start_soon(aw, *, name=None)

Awaits its argument concurrently to other calls to this function.

Parameters:

aw (Awaitable[TypeVar(T)]) – A Awaitable to await concurrently.
name (str | None) – A name to associate with the Task awaiting aw.

Return type:

Task[TypeVar(T)]

Returns:

A Task which is awaiting aw concurrently.

async coconext.triggers.gather(*awaitables, return_exceptions=False)

Awaits on all given awaitables concurrently and returns their results once all have completed.

After the return conditions, based on return_exceptions is met, remaining waiter tasks are cancelled. This does not cancel Tasks passed as arguments, only the internal tasks awaiting upon their completion.

Parameters:

awaitables (Awaitable[Any]) – The Awaitables to concurrently await upon.
return_exceptions (bool) – If False (default), after the first awaitable results in an exception, cancels the remaining awaitables and re-raises the exception. If True, returns the exception rather than the result value when an awaitable results in an exception.

Return type:

tuple[Any, ...]

Returns:

A tuple of the results of awaiting each awaitable in the same order they were given. The order of the return tuple corresponds to the order of the input.

async coconext.triggers.select(*awaitables, return_exception=False)

Awaits on all given awaitables concurrently and returns the index and result of the first to complete.

After the first awaitable completes, remaining waiter tasks are cancelled. This does not cancel Tasks passed as arguments, only the internal tasks awaiting upon their completion.

Parameters:

awaitables (Awaitable[TypeVar(T)]) – The Awaitables to concurrently await upon.
return_exception (bool) – If False (default), re-raises the exception when an awaitable results in an exception. If True, returns the exception rather than re-raising when an awaitable results in an exception.

Raises:

ValueError – if missing at least one awaitable.

Return type:

tuple[int, Union[TypeVar(T), BaseException]]

Returns:

A tuple of the index into the argument list (0-based) of the first awaitable to complete and its result.

async coconext.triggers.wait(*awaitables, return_when)

Awaits on all given awaitables concurrently and blocks until the return_when condition is met.

Every Awaitable given to the function is awaited concurrently in its own Task. When the return conditions specified by return_when are met, this function returns those Tasks. Once the return conditions are met, any Tasks which are still running are cancel()ed.

The return_when conditions must be one of the following:

"FIRST_COMPLETED": Returns after the first awaitable completes, regardless if that was due to an exception or not.
"FIRST_EXCEPTION": Returns after all awaitables complete or after the first awaitable that completes due to an exception.
"ALL_COMPLETED": Returns after all awaitables complete.

Parameters:

awaitables (Awaitable[Any]) – The Awaitables to concurrently await upon.
return_when (Literal['FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED']) – The condition that must be met before returning. One of "FIRST_COMPLETED", "FIRST_EXCEPTION", or "ALL_COMPLETED".

Return type:

tuple[Task[Any], ...]

Returns:

A tuple of waiter Tasks. The order of the return tuple corresponds to the order of the input.

Simulation Time Utils

Collections of tools for dealing with simulation time in cocotb.

class coconext.simtime.SimTime

Bases: object

Type for dealing with simulated time.

SimTime is time quantized to simulator steps. If you can create a SimTime, you have a time that can be represented by the simulator. The type also includes a collection of functionality commonly needed when dealing with time, such as comparison, scaling, and arithmetic.

__init__(time, unit, *, round_mode='error')

Construct a SimTime of time value in unit.

Parameters:

time (float) – The time value.
unit (Literal['fs', 'ps', 'ns', 'us', 'ms', 'sec', 'step']) – Unit for time argument.
round_mode (Literal['error', 'round', 'ceil', 'floor']) – How to round the time if it can’t be accurately represented. One of "error", "round", "ceil", and "floor". See cocotb.utils.get_sim_steps() for details on these values.

in_unit(unit)

Return time in the given unit.

Return type:: float

property fs: float: Time in femtoseconds.

property ps: float: Time in picoseconds.

property ns: float: Time in nanoseconds.

property us: float: Time in microseconds.

property ms: float: Time in milliseconds.

property sec: float: Time in seconds.

property step: float: Time in simulator steps.

self == other

Return self==value.

Return type:: bool

self < other

Return self<value.

Return type:: bool

self <= other

Return self<=value.

Return type:: bool

self > other

Return self>value.

Return type:: bool

self >= other

Return self>=value.

Return type:: bool

hash(self)

Return hash(self).

Return type:: int

repr(self)

Return repr(self).

Return type:: str

mul(other, *, round_mode='error')

Scale the time via multiplication and supply a round_mode.

Return type:: SimTime

div(other, *, round_mode='error')

Scale the time via division and supply a round_mode.

Return type:: SimTime

wait()

Return a Timer Trigger of the time.

Return type:: Timer

Queue

Collection of asynchronous queues for cocotb.

class coconext.queue.AbstractQueue

Bases: Generic[T]

An asynchronous queue, useful for coordinating producer and consumer tasks.

Queues can have various semantics with respect to the values put() and get() from them. This class does not enforce any particular semantics, but leaves the implementation and semantics abstract. A concrete queue class can be created by subclassing AbstractQueue and filling in the _size(), _put(), _get(), and _peek() methods.

Additionally, there are common implementations included in this module: Queue (FIFO semantics), LifoQueue (LIFO semantics), and PriorityQueue (min-heap semantics).

Added in version 0.2.

class Lock

Bases: object

A mutex lock specialized for use by asynchronous Queue.

This class is not intended to be instantiated by the user, but rather be acquired via AbstractQueue.write_lock or AbstractQueue.read_lock.

In addition to all the functionality available on Lock, these locks are aware of the associated queue’s availability, i.e. whether the read or write side has data or space available to either get or put, respectively.

__init__()

locked()

Return True if the lock is locked.

Return type:: bool

available()

Return True if the lock is available for acquiring.

Return type:: bool

async acquire()

Acquire the lock.

Waits until the lock is available and unlocked, sets it to locked and returns. Multiple Tasks may attempt to acquire the Lock, but only one will proceed. Tasks are guaranteed to acquire the Lock in the order each attempts to acquire it.

Return type:: None

release()

Release the lock.

Sets the lock to unlocked.

Raises:: RuntimeError – If called when the lock is unlocked.
Return type:: None

__init__(maxsize=0)

Construct a queue with the given maxsize.

If maxsize is less than or equal to 0, the queue size is infinite. If it is an integer greater than 0, then put() will block when the queue reaches maxsize, until an item is removed by get().

property write_lock: Lock

Lock for exclusive write access.

After acquiring this lock, the user has exclusive access to the write-side of the queue until the lock is released. Calling put() while the write lock is acquired will result in a RuntimeError to prevent deadlock.

property read_lock: Lock

Lock for exclusive read access.

After acquiring this lock, the user has exclusive access to the read-side of the queue until the lock is released. Calling get() while the read lock is acquired will result in a RuntimeError to prevent deadlock.

qsize()

Number of items in the queue.

Return type:: int

property maxsize: int: Number of items allowed in the queue.

empty()

Return True if the queue is empty, False otherwise.

Return type:: bool

full()

Return True if there are maxsize() items in the queue.

Note

If the Queue was initialized with maxsize=0 (the default), then full() is never True.

Return type:: bool

async put(item)

Put an item into the queue.

If the queue is full, wait until a free slot is available before adding the item.

Return type:: None

put_nowait(item)

Put an item into the queue without blocking.

If no free slot is immediately available, raise QueueFull.

Return type:: None

async get()

Remove and return an item from the queue.

If the queue is empty, wait until an item is available.

Return type:: TypeVar(T)

get_nowait()

Remove and return an item from the queue.

Returns an item if one is immediately available, otherwise raises QueueEmpty.

Return type:: TypeVar(T)

async peek()

Return the next item from the queue without removing it.

If the queue is empty, wait until an item is available.

Return type:: TypeVar(T)

peek_nowait()

Return the next item from the queue without removing it.

Returns an item if one is immediately available, otherwise raises QueueEmpty.

Return type:: TypeVar(T)

class coconext.queue.Queue

Bases: AbstractQueue[T]

A subclass of AbstractQueue; retrieves oldest entries first (FIFO).

Added in version 2.0.

__init__(maxsize=0)

Construct a queue with the given maxsize.

If maxsize is less than or equal to 0, the queue size is infinite. If it is an integer greater than 0, then put() will block when the queue reaches maxsize, until an item is removed by get().

class coconext.queue.PriorityQueue

Bases: AbstractQueue[SupportsRichComparisonT]

A subclass of AbstractQueue; retrieves entries in priority order (smallest item first).

Entries are typically tuples of the form (priority number, data).

Added in version 2.0.

__init__(maxsize=0)

Construct a queue with the given maxsize.

If maxsize is less than or equal to 0, the queue size is infinite. If it is an integer greater than 0, then put() will block when the queue reaches maxsize, until an item is removed by get().

class coconext.queue.LifoQueue

Bases: AbstractQueue[T]

A subclass of AbstractQueue; retrieves most recently added entries first (LIFO).

Added in version 2.0.

__init__(maxsize=0)

Construct a queue with the given maxsize.

If maxsize is less than or equal to 0, the queue size is infinite. If it is an integer greater than 0, then put() will block when the queue reaches maxsize, until an item is removed by get().

Types

Collection of types modelled after those common in HDLs.

class coconext.types.BitArray

Bases: AbstractMutableArray[Bit]

An Array of Bits.

self == other

Return self==value.

Return type:: bool

__hash__ = None

__init__(value, range=None)

self | other

Return self|value.

Return type:: BitArray | int | LogicArray

repr(self)

Return repr(self).

Return type:: str

__ror__(other)

Return value|self.

Return type:: BitArray | int | LogicArray

str(self)

Return str(self).

Return type:: str

property range: Range: Range of the indexes of the array.