Library Reference

class cocotb_tools.runner.Runner[source]#

Expand for references to cocotb_tools.runner.Runner

build(hdl_library='top', verilog_sources=[], vhdl_sources=[], sources=[], includes=[], defines={}, parameters={}, build_args=[], hdl_toplevel=None, always=False, build_dir='sim_build', clean=False, verbose=False, timescale=None, waves=False, log_file=None)[source]#

Build the HDL sources.

With mixed language simulators, sources will be built, followed by vhdl_sources, then verilog_sources. With simulators that only support either VHDL or Verilog, sources will be built, followed by vhdl_sources and verilog_sources, respectively.

If your source files use an atypical file extension, use VHDL and Verilog to tag the path as a VHDL or Verilog source file, respectively. If the filepaths aren’t tagged, the extension is used to determine if they are VHDL or Verilog files.

Language	File Extensions
VHDL	`.vhd`, `.vhdl`
Verilog	`.v`, `.sv`, `.vh`, `.svh`

runner.build(
    sources=[
        VHDL("/my/file.is_actually_vhdl"),
        Verilog("/other/file.verilog"),
    ],
)

The same tagging works for build_args. Tagged build_args only supply that option to the compiler when building the source file for the tagged language. Non-tagged build_args are supplied when compiling any language.

Parameters:

hdl_library (str) – The library name to compile into.
verilog_sources (Sequence[PathLike[str] | str]) – Verilog source files to build.
vhdl_sources (Sequence[PathLike[str] | str]) – VHDL source files to build.
sources (Sequence[PathLike[str] | str | VHDL | Verilog]) – Language-agnostic list of source files to build.
includes (Sequence[PathLike[str] | str]) – Verilog include directories.
defines (Mapping[str, object]) – Defines to set.
parameters (Mapping[str, object]) – Verilog parameters or VHDL generics.
build_args (Sequence[str | VHDL | Verilog]) – Extra build arguments for the simulator.
hdl_toplevel (str | None) – The name of the HDL toplevel module.
always (bool) – Always run the build step.
build_dir (PathLike[str] | str) – Directory to run the build step in.
clean (bool) – Delete build_dir before building.
verbose (bool) – Enable verbose messages.
timescale (Tuple[str, str] | None) – Tuple containing time unit and time precision for simulation.
waves (bool) – Record signal traces.
log_file (PathLike[str] | str | None) – File to write the build log to.

Deprecated since version 2.0: Uses of the verilog_sources and vhdl_sources parameters should be replaced with the language-agnostic sources argument.

Expand for references to cocotb_tools.runner.Runner.build

test(test_module, hdl_toplevel, hdl_toplevel_library='top', hdl_toplevel_lang=None, gpi_interfaces=None, testcase=None, seed=None, elab_args=[], test_args=[], plusargs=[], extra_env={}, waves=False, gui=False, parameters=None, build_dir=None, test_dir=None, results_xml=None, pre_cmd=None, verbose=False, timescale=None, log_file=None, test_filter=None)[source]#

Run the tests.

Parameters:

test_module (str | Sequence[str]) – Name(s) of the Python module(s) containing the tests to run. Can be a comma-separated list.
hdl_toplevel (str) – Name of the HDL toplevel module.
hdl_toplevel_library (str) – The library name for HDL toplevel module.
hdl_toplevel_lang (str | None) – Language of the HDL toplevel module.
gpi_interfaces (List[str] | None) – List of GPI interfaces to use, with the first one being the entry point.
testcase (str | Sequence[str] | None) – Name(s) of a specific testcase(s) to run. If not set, run all testcases found in test_module. Can be a comma-separated list.
seed (int | str | None) – A specific random seed to use.
elab_args (Sequence[str]) – A list of elaboration arguments for the simulator.
test_args (Sequence[str]) – A list of extra arguments for the simulator.
plusargs (Sequence[str]) – ‘plusargs’ to set for the simulator.
extra_env (Mapping[str, str]) – Extra environment variables to set.
waves (bool) – Record signal traces.
gui (bool) – Run with simulator GUI.
parameters (Mapping[str, object] | None) – Verilog parameters or VHDL generics.
build_dir (PathLike[str] | str | None) – Directory the build step has been run in.
test_dir (PathLike[str] | str | None) – Directory to run the tests in.
results_xml (str | None) – Name of xUnit XML file to store test results in. If an absolute path is provided it will be used as-is, {build_dir}/results.xml otherwise. This argument should not be set when run with pytest.
verbose (bool) – Enable verbose messages.
pre_cmd (List[str] | None) – Commands to run before simulation begins. Typically Tcl commands for simulators that support them.
timescale (Tuple[str, str] | None) – Tuple containing time unit and time precision for simulation.
log_file (PathLike[str] | str | None) – File to write the test log to.
test_filter (str | None) – Regular expression which matches test names. Only matched tests are run if this argument if given.

Returns:

The absolute location of the results XML file which can be defined by the results_xml argument.

Return type:

Path

Expand for references to cocotb_tools.runner.Runner.test

Library Reference / Writing and Generating Tests

class cocotb_tools.runner.VHDL[source]#: Tags source files and build arguments to Runner.build() as VHDL-specific.

class cocotb_tools.runner.Verilog[source]#: Tags source files and build arguments to Runner.build() as Verilog-specific.

cocotb_tools.runner.MAX_PARALLEL_BUILD_JOBS: int = 4#

The maximum number of parallel build threads in calls to Runner.build().

If the number of CPU cores is less than this value, it uses the CPU core count. Set this variable to globally change the number of parallel build jobs.

Simulator Runners#

class cocotb_tools.runner.Icarus[source]#

Implementation of Runner for Icarus.

hdl_toplevel argument to build() is required.
waves argument must be given to build() if used.
timescale argument to build() must be given to support dumping the command file.
Does not support the gui argument to test().
Does not support the pre_cmd argument to test().

class cocotb_tools.runner.Verilator[source]#

Implementation of Runner for Verilator.

Does not support the pre_cmd argument to test().
Does not support the gui argument to test().

class cocotb_tools.runner.Riviera[source]#

Implementation of Runner for Riviera-PRO.

Does not support the pre_cmd argument to test().
Does not support the gui argument to test().
Does not support the timescale argument to build() or test().

class cocotb_tools.runner.Questa[source]#

Implementation of Runner for Questa.

Does not support the timescale argument to build() or test().

class cocotb_tools.runner.Xcelium[source]#

Implementation of Runner for Xcelium.

Does not support the pre_cmd argument to test().
Does not support the timescale argument to build() or test().

class cocotb_tools.runner.Ghdl[source]#

Implementation of Runner for GHDL.

Does not support the pre_cmd argument to test().
Does not support the gui argument to test().

class cocotb_tools.runner.Nvc[source]#

Implementation of Runner for NVC.

Does not support the pre_cmd argument to test().
Does not support the gui argument to test().
Does not support the timescale argument to build() or test().

class cocotb_tools.runner.Vcs[source]#

Implementation of Runner for VCS.

Does not support the pre_cmd argument to test().
Does not support VHDL.

class cocotb_tools.runner.Dsim[source]#

Implementation of Runner for DSim.

Does not support the gui argument to test().
Does not support the pre_cmd argument to test().

Results#

cocotb_tools.runner.get_results(results_xml_file)[source]#

Return number of tests and fails in results_xml_file.

Returns:: Tuple of number of tests and number of fails.
Raises:: RuntimeError – If results_xml_file is non-existent.
Return type:: Tuple[int, int]

File Utilities#

cocotb_tools.runner.get_abs_path(path)[source]#

Return path in absolute form.

cocotb_tools.runner.get_abs_paths(paths)[source]#

Return list of paths in absolute form.

cocotb_tools.runner.outdated(output, dependencies)[source]#

Return True if any source files in dependencies are newer than the output directory.

Returns:: True if any source files are newer, False otherwise.
Return type:: bool

class cocotb_tools.runner.UnknownFileExtension(source)[source]#

Raised when a source file type cannot be determined from the file extension.

See Runner.build() for details on supported standard file extensions and the use of VHDL and Verilog for specifying file type.

Writing and Generating Tests#

cocotb.test(_func: F | _Parameterized[F]) → F[source]#

cocotb.test(*, timeout_time: float | None = None, timeout_unit: TimeUnit = 'step', expect_fail: bool = False, expect_error: Type[BaseException] | Tuple[Type[BaseException], ...] = (), skip: bool = False, stage: int = 0, name: str | None = None, _expect_sim_failure: bool = False) → Callable[[F | _Parameterized[F]], F]

Decorator to register a Callable which returns a Coroutine as a test.

The test decorator provides a test timeout, and allows us to mark tests as skipped or expecting errors or failures. Tests are evaluated in the order they are defined in a test module.

Usage:

@cocotb.test(timeout_time=10, timeout_unit="ms")
async def test_thing(dut): ...

Parameters:

timeout_time –
Simulation time duration before timeout occurs.

Added in version 1.3.

Note

Test timeout is intended for protection against deadlock. Users should use with_timeout if they require a more general-purpose timeout mechanism.
timeout_unit –
Units of timeout_time, accepts any units that Timer does.

Added in version 1.3.

Changed in version 2.0: Passing None as the timeout_unit argument was removed, use 'step' instead.
expect_fail – If True and the test fails a functional check via an assert statement, pytest.raises, pytest.warns, or pytest.deprecated_call the test is considered to have passed. If True and the test passes successfully, the test is considered to have failed.
expect_error –
Mark the result as a pass only if one of the exception types is raised in the test. This is primarily for cocotb internal regression use for when a simulator error is expected.

Users are encouraged to use the following idiom instead:
```
@cocotb.test()
async def my_test(dut):
    try:
        await thing_that_should_fail()
    except ExceptionIExpect:
        pass
    else:
        assert False, "Exception did not occur"
```
Changed in version 1.3: Specific exception types can be expected

Changed in version 2.0: Passing a bool value was removed. Pass a specific Exception or a tuple of Exceptions instead.
skip – Don’t execute this test as part of the regression. Test can still be run manually by setting COCOTB_TESTCASE.
stage – Order tests logically into stages, where multiple tests can share a stage. Defaults to 0.
name –
Override the default name of the test. The default test name is the __qualname__() of the decorated test function.

Added in version 2.0.

Returns:

The test function to which the decorator is applied.

Note

To extend the test decorator, use the following template to create a new cocotb.test-like wrapper.

import functools


def test_extender(**decorator_kwargs):
    def decorator(obj):
        @cocotb.test(**decorator_kwargs)
        @functools.wraps(obj)
        async def test(dut, **test_kwargs):
            pass  # your code here

        return obj

    return decorator

Changed in version 2.0: Support using decorator on test function without supplying parameters first.

Assumes all default values for the test parameters.

@cocotb.test
async def test_thing(dut): ...

Changed in version 2.0: Decorated tests now return the decorated object.

Expand for references to cocotb.test

Coroutines and Tasks / Async generators

Coroutines and Tasks / Waiting For Multiple Events Simultaneously / Determining Which Task Finishes First

Coroutines and Tasks / Waiting For Multiple Events Simultaneously / Waiting For Multiple Events

Coroutines and Tasks / Waiting For Multiple Events Simultaneously / Waiting For One Of Multiple Events

Quickstart Guide / Creating a Test

Writing Testbenches / Reading values from signals / Cleaning up resources

Writing Testbenches / Reading values from signals / Concurrent and sequential execution

Writing Testbenches / Logging

Writing Testbenches / Reading values from signals / Passing and failing tests

Writing Testbenches / Reading values from signals / Identifying tests

cocotb.parametrize(*options_by_tuple, **options_by_name)[source]#

Decorator to generate parametrized tests from a single test function.

Decorates a test function with named test parameters. The call to parametrize should include the name of each test parameter and the possible values each parameter can hold. This will generate a test for each of the Cartesian products of the parameters and their values.

@cocotb.test(
    skip=False,
)
@cocotb.parametrize(
    arg1=[0, 1],
    arg2=["a", "b"],
)
async def my_test(arg1: int, arg2: str) -> None: ...

The above is equivalent to the following.

@cocotb.test(skip=False)
async def my_test_0_a() -> None:
    arg1, arg2 = 0, "a"
    ...


@cocotb.test(skip=False)
async def my_test_0_b() -> None:
    arg1, arg2 = 0, "b"
    ...


@cocotb.test(skip=False)
async def my_test_1_a() -> None:
    arg1, arg2 = 1, "a"
    ...


@cocotb.test(skip=False)
async def my_test_1_b() -> None:
    arg1, arg2 = 1, "b"
    ...

Options can also be specified in much the same way that TestFactory.add_option can, either by supplying tuples of the parameter name to values, or a sequence of variable names and a sequence of values.

@cocotb.parametrize(
    ("arg1", [0, 1]),
    (("arg2", "arg3"), [(1, 2), (3, 4)])
)

Parameters:

options_by_tuple (Tuple[str, Sequence[Any]] | Tuple[Sequence[str], Sequence[Sequence[Any]]]) – Tuple of parameter name to sequence of values for that parameter, or tuple of sequence of parameter names to sequence of sequences of values for that pack of parameters.
options_by_name (Sequence[Any]) – Mapping of parameter name to sequence of values for that parameter.

Added in version 2.0.

Expand for references to cocotb.parametrize

Library Reference / Writing and Generating Tests

class cocotb.regression.TestFactory(test_function, *args, **kwargs)[source]#

Factory to automatically generate tests.

Parameters:

test_function (F) – A Callable that returns the test Coroutine. Must take dut as the first argument.
*args (Any) – Remaining arguments are passed directly to the test function. Note that these arguments are not varied. An argument that varies with each test must be a keyword argument to the test function.
**kwargs (Any) – Remaining keyword arguments are passed directly to the test function. Note that these arguments are not varied. An argument that varies with each test must be a keyword argument to the test function.

Assuming we have a common test function that will run a test. This test function will take keyword arguments (for example generators for each of the input interfaces) and generate tests that call the supplied function.

This Factory allows us to generate sets of tests based on the different permutations of the possible arguments to the test function.

For example, if we have a module that takes backpressure, has two configurable features where enabling feature_b requires feature_a to be active, and need to test against data generation routines gen_a and gen_b:

>>> tf = TestFactory(test_function=run_test)
>>> tf.add_option(name="data_in", optionlist=[gen_a, gen_b])
>>> tf.add_option("backpressure", [None, random_backpressure])
>>> tf.add_option(
...     ("feature_a", "feature_b"), [(False, False), (True, False), (True, True)]
... )
>>> tf.generate_tests()

We would get the following tests:

gen_a with no backpressure and both features disabled

gen_a with no backpressure and only feature_a enabled

gen_a with no backpressure and both features enabled

gen_a with random_backpressure and both features disabled

gen_a with random_backpressure and only feature_a enabled

gen_a with random_backpressure and both features enabled

gen_b with no backpressure and both features disabled

gen_b with no backpressure and only feature_a enabled

gen_b with no backpressure and both features enabled

gen_b with random_backpressure and both features disabled

gen_b with random_backpressure and only feature_a enabled

gen_b with random_backpressure and both features enabled

The tests are appended to the calling module for auto-discovery.

Tests are simply named test_function_N. The docstring for the test (hence the test description) includes the name and description of each generator.

Changed in version 1.5: Groups of options are now supported

Changed in version 2.0: You can now pass cocotb.test() decorator arguments when generating tests.

Deprecated since version 2.0: Use cocotb.parametrize() instead.

add_option(name: str, optionlist: Sequence[Any]) → None[source]#

add_option(name: Sequence[str], optionlist: Sequence[Sequence[Any]]) → None

Add a named option to the test.

Parameters:

name – An option name, or an iterable of several option names. Passed to test as keyword arguments.
optionlist – A list of possible options for this test knob. If N names were specified, this must be a list of N-tuples or lists, where each element specifies a value for its respective option.

Changed in version 1.5: Groups of options are now supported

generate_tests(*, prefix=None, postfix=None, stacklevel=0, name=None, timeout_time=None, timeout_unit='step', expect_fail=False, expect_error=(), skip=False, stage=0, _expect_sim_failure=False)[source]#

Generate an exhaustive set of tests using the cartesian product of the possible keyword arguments.

The generated tests are appended to the namespace of the calling module.

Parameters:

prefix (str | None) –
Text string to append to start of test_function name when naming generated test cases. This allows reuse of a single test_function with multiple TestFactories without name clashes.

Deprecated since version 2.0: Use the more flexible name field instead.
postfix (str | None) –
Text string to append to end of test_function name when naming generated test cases. This allows reuse of a single test_function with multiple TestFactories without name clashes.

Deprecated since version 2.0: Use the more flexible name field instead.
stacklevel (int) – Which stack level to add the generated tests to. This can be used to make a custom TestFactory wrapper.
name (str | None) –
Passed as name argument to cocotb.test().

Added in version 2.0.
timeout_time (float | None) –
Passed as timeout_time argument to cocotb.test().

Added in version 2.0.
timeout_unit (Literal["step"] | TimeUnitWithoutStep) –
Passed as timeout_unit argument to cocotb.test().

Added in version 2.0.
expect_fail (bool) –
Passed as expect_fail argument to cocotb.test().

Added in version 2.0.
expect_error (Type[BaseException] | Tuple[Type[BaseException], ...]) –
Passed as expect_error argument to cocotb.test().

Added in version 2.0.
skip (bool) –
Passed as skip argument to cocotb.test().

Added in version 2.0.
stage (int) –
Passed as stage argument to cocotb.test().

Added in version 2.0.

Interacting with the Simulator#

Task Management#

cocotb.start_soon(coro)[source]#

Schedule a coroutine to be run concurrently in a Task.

Note that this is not an async function, and the new task will not execute until the calling task yields control.

Parameters:: coro (Task[ResultType] | Coroutine[Any, Any, ResultType]) – A task or coroutine to be run.
Returns:: The Task that is scheduled to be run.
Return type:: Task[ResultType]

Added in version 1.6.

Expand for references to cocotb.start_soon

Library Reference / Test Utilities / Clock Driver

Library Reference / Triggers / Concurrency Triggers

Library Reference / Triggers / Synchronization Triggers

Library Reference / Interacting with the Simulator / Task Management

Coroutines and Tasks / Waiting For Multiple Events Simultaneously / Determining Which Task Finishes First

Coroutines and Tasks / Waiting For Multiple Events Simultaneously / Waiting For Multiple Events

Coroutines and Tasks / Waiting For Multiple Events Simultaneously / Waiting For One Of Multiple Events

Quickstart Guide / Creating a Test

Writing Testbenches / Reading values from signals / Cleaning up resources

Writing Testbenches / Reading values from signals / Concurrent and sequential execution

Writing Testbenches / Reading values from signals / Passing and failing tests

cocotb.start(coro)[source]#

Schedule a coroutine to be run concurrently, then yield control to allow pending tasks to execute.

The calling task will resume execution before control is returned to the simulator.

When the calling task resumes, the newly scheduled task may have completed, raised an Exception, or be pending on a Trigger.

Parameters:: coro (Task[ResultType] | Coroutine[Any, Any, ResultType]) – A task or coroutine to be run.
Returns:: The Task that has been scheduled and allowed to execute.
Return type:: Task[ResultType]

Added in version 1.6.

Deprecated since version 2.0: Use cocotb.start_soon() instead. If you need the scheduled Task to start before continuing the current Task, use an Event to block the current Task until the scheduled Task starts, like so:

async def coro(started: Event) -> None:
    started.set()
    # Do stuff...


task_started = Event()
task = cocotb.start_soon(coro(task_started))
await task_started.wait()

cocotb.create_task(coro)[source]#

Construct a coroutine into a Task without scheduling the task.

The task can later be scheduled with cocotb.start() or cocotb.start_soon().

Parameters:: coro (Task[ResultType] | Coroutine[Any, Any, ResultType]) – An existing task or a coroutine to be wrapped.
Returns:: Either the provided Task or a new Task wrapping the coroutine.
Return type:: Task[ResultType]

Added in version 1.6.

class cocotb.task.ResultType#

Task result type

alias of TypeVar(‘ResultType’)

class cocotb.task.Task(inst)[source]#

Concurrently executing task.

This class is not intended for users to directly instantiate. Use cocotb.create_task() to create a Task object or cocotb.start_soon() to create a Task and schedule it to run.

Changed in version 1.8: Moved to the cocotb.task module.

Changed in version 2.0: The retval, _finished, and __bool__ methods were removed. Use result(), done(), and done() methods instead, respectively.

cancel(msg=None)[source]#

Cancel a Task’s further execution.

When a Task is cancelled, a asyncio.CancelledError is thrown into the Task.

Returns: True if the Task was cancelled; False otherwise.

cancelled()[source]#

Return True if the Task was cancelled.

property complete: TaskComplete[ResultType]#

Trigger which fires when the Task completes.

Unlike join(), this Trigger does not return the result of the Task when awaited.

async def coro_inner():
    await Timer(1, unit="ns")
    raise ValueError("Oops")


task = cocotb.start_soon(coro_inner())
await task.complete  # no exception raised here
assert task.exception() == ValueError("Oops")

done()[source]#

Return True if the Task has finished executing.

exception()[source]#

Return the exception of the Task.

If the Task ran to completion, None is returned. If the Task failed with an exception, the exception is returned. If the Task was cancelled, the asyncio.CancelledError is re-raised. If the coroutine is not yet complete, an asyncio.InvalidStateError is raised.

join()[source]#

Block until the Task completes and return the result.

Equivalent to calling Join(self).

async def coro_inner():
    await Timer(1, unit="ns")
    return "Hello world"


task = cocotb.start_soon(coro_inner())
result = await task.join()
assert result == "Hello world"

Returns:: Object that can be awaited or passed into First or Combine; the result of which will be the result of the Task.
Return type:: Join[ResultType]

Deprecated since version 2.0: Using task directly is preferred to task.join() in all situations where the latter could be used.

kill()[source]#

Kill a coroutine.

result()[source]#

Return the result of the Task.

If the Task ran to completion, the result is returned. If the Task failed with an exception, the exception is re-raised. If the Task was cancelled, the asyncio.CancelledError is re-raised. If the coroutine is not yet complete, an asyncio.InvalidStateError is raised.

Dealing with non-`async` code#

cocotb.bridge(func)[source]#

Converts a blocking function into a coroutine function.

This function converts a blocking function into a coroutine function with the expectation that the function being converted is intended to call a cocotb.resume() converted function. This creates a bridge through non-async code for code wanting to eventually await on cocotb triggers.

When a converted function call is used in an await statement, the current Task blocks until the converted function finishes.

Results of the converted function are returned from the await expression.

Note

Bridge threads must either finish or block on a cocotb.resume() converted function before control is given back to the simulator. This is done to prevent any code from executing in parallel with the simulation.

Parameters:: func (Callable[[...], Result]) – The blocking function to convert into a coroutine function.
Returns:: func as a coroutine function.
Return type:: Callable[[…], Coroutine[Any, Any, Result]]

Changed in version 2.0: Renamed from external. No longer implemented as a type. The log attribute is no longer available.

cocotb.resume(func)[source]#

Converts a coroutine function into a blocking function.

This allows a coroutine function that awaits cocotb triggers to be called from a blocking function converted by cocotb.bridge(). This completes the bridge through non-async code.

When a converted coroutine function is called the current function blocks until the converted function exits.

Results of the converted function are returned from the function call.

Parameters:: func (Callable[[...], Coroutine[Any, Any, Result]]) – The coroutine function to convert into a blocking function.
Returns:: func as a blocking function.
Raises:: RuntimeError – If the function that is returned is subsequently called from a thread that was not started with cocotb.bridge.
Return type:: Callable[[…], Result]

Changed in version 2.0: Renamed from function. No longer implemented as a type. The log attribute is no longer available.

HDL Datatypes#

These are a set of datatypes that model the behavior of common HDL datatypes.

Added in version 1.6.

class cocotb.types.Logic(value)[source]#

Model of a 9-value (U, X, 0, 1, Z, W, L, H, -) datatype commonly seen in VHDL.

This is modeled after VHDL’s std_ulogic type. (System)Verilog’s 4-value logic type only utilizes X, 0, 1, and Z values.

Logic can be converted to and from int, str, and bool. The list of values convertible to Logic includes "U", "X", "0", "1", "Z", "W", "L", "H", "-", 0, 1, True, and False.

>>> Logic("X")
Logic('X')
>>> Logic(True)
Logic('1')
>>> Logic(1)
Logic('1')

>>> str(Logic("Z"))
'Z'
>>> bool(Logic(0))
False
>>> int(Logic(1))
1

Note

The int and bool conversions will raise ValueError if the value is not 0, 1, L, or H.

Logic values are immutable and therefore hashable and can be placed in sets and used as keys in dicts.

Logic supports the common logic operations &, |, ^, and ~.

>>> def full_adder(a: Logic, b: Logic, carry: Logic) -> Tuple[Logic, Logic]:
...     res = a ^ b ^ carry
...     carry_out = (a & b) | (b & carry) | (a & carry)
...     return res, carry_out

>>> full_adder(a=Logic("0"), b=Logic("1"), carry=Logic("1"))
(Logic('0'), Logic('1'))

Parameters:

value (str | int | bool | Logic) – value to construct into a Logic.

Raises:

ValueError – If the value if of the correct type, but cannot be constructed into a Logic.
TypeError – If the value is of a type that can’t be constructed into a Logic.

class cocotb.types.Range(left: int, direction: int)[source]#

class cocotb.types.Range(left: int, direction: str, right: int)

class cocotb.types.Range(left: int, *, right: int)

Variant of range with inclusive right bound.

In Python, range and slice have a non-inclusive right bound. In both Verilog and VHDL, ranges and arrays have an inclusive right bound. This type mimics Python’s range type, but implements HDL-like inclusive right bounds, using the names left and right as replacements for start and stop to match VHDL. Range directionality can be specified using 'to' or 'downto' between the left and right bounds. Not specifying directionality will cause the directionality to be inferred.

>>> r = Range(-2, 3)
>>> r.left, r.right, len(r)
(-2, 3, 6)

>>> s = Range(8, "downto", 1)
>>> s.left, s.right, len(s)
(8, 1, 8)

from_range() and to_range() can be used to convert from and to range.

>>> r = Range(-2, 3)
>>> r.to_range()
range(-2, 4)

Range supports “null” ranges as seen in VHDL. “null” ranges occur when a left bound cannot reach a right bound with the given direction. They have a length of 0, but the left, right, and direction values remain as given.

>>> r = Range(1, "to", 0)  # no way to count from 1 'to' 0
>>> r.left, r.direction, r.right
(1, 'to', 0)
>>> len(r)
0

Note

This is only possible when specifying the direction.

Ranges also support all the features of range including, but not limited to:

value in range to see if a value is in the range,
range.index(value) to see what position in the range the value is,

The typical use case of this type is in conjunction with Array.

Parameters:

left (int) – leftmost bound of range
direction (int | str | None) – 'to' if values are ascending, 'downto' if descending
right (int | None) – rightmost bound of range (inclusive)

property direction: str#: 'to' if Range is ascending, 'downto' otherwise.

classmethod from_range(range)[source]#

Convert range to Range.

property left: int#: Leftmost value in a Range.

property right: int#: Rightmost value in a Range.

to_range()[source]#

Convert Range to range.

class cocotb.types.Array(value, range=None)[source]#

Fixed-size, arbitrarily-indexed, homogeneous collection type.

Arrays are similar to, but different from Python lists. An array can store values of any type or values of multiple types at a time, just like a list. Array constructor values can be any Iterable.

>>> Array([1, False, "example", None])
Array([1, False, 'example', None], Range(0, 'to', 3))

>>> Array("Hello!")
Array(['H', 'e', 'l', 'l', 'o', '!'], Range(0, 'to', 5))

Unlike lists, an array’s size cannot change. This means they do not support methods like append() or insert().

The indexes of an Array can start or end at any integer value, they are not limited to 0-based indexing. Indexing schemes are selected using Range objects. If range is not given, the range Range(0, "to", len(value)-1) is used. If an int is passed for range, the range Range(0, "to", range-1) is used.

>>> a = Array([0, 1, 2, 3], Range(-1, "downto", -4))
>>> a[-1]
0
>>> a[-4]
3

Arrays can also have 0 length using “null” Ranges.

>>> Array([], Range(1, "to", 0))  # 1 to 0 is a null Range
Array([], Range(1, 'to', 0))

Indexing and slicing is very similar to lists, but it uses the indexing scheme specified with the range. Unlike list, slicing uses an inclusive right bound, which is common in HDLs. But like list, if a start or stop index is not specified in the slice, it is inferred as the start or end of the array. Slicing an Array returns a new Array object, whose bounds are the slice indexes.

>>> a = Array("1234abcd")
>>> a[7]
'd'
>>> a[2:5]
Array(['3', '4', 'a', 'b'], Range(2, 'to', 5))
>>> a[2:5] = reversed(a[2:5])
>>> "".join(a)
'12ba43cd'

>>> b = Array("1234", Range(0, -3))
>>> b[-2]
'3'
>>> b[-1:]
Array(['2', '3', '4'], Range(-1, 'downto', -3))
>>> b[:] = reversed(b)
>>> b
Array(['4', '3', '2', '1'], Range(0, 'downto', -3))

Warning

Arrays behave differently in certain situations than Python’s built-in sequence types (list, tuple, etc.).

Arrays are not necessarily 0-based and slices use inclusive right bounds, so many functions that work on Python sequences by index (like bisect) may not work on arrays.
Slice indexes must be specified in the same direction as the array and do not support specifying a “step”.
When setting a slice, the new value must be an iterable of the same size as the slice.
Negative indexes are not treated as an offset from the end of the array, but are treated literally.

Arrays are equal to other arrays of the same length with the same values (structural equality). Bounds do not matter for equality.

>>> a = Array([1, 1, 2, 3, 5], Range(4, "downto", 0))
>>> b = Array([1, 1, 2, 3, 5], Range(-2, "to", 2))
>>> a == b
True

You can change the bounds of an array by setting the range to a new value. The new bounds must be the same length of the array.

>>> a = Array("1234")
>>> a.range
Range(0, 'to', 3)
>>> a.range = Range(3, "downto", 0)
>>> a.range
Range(3, 'downto', 0)

Arrays support many of the methods and semantics defined by collections.abc.Sequence.

>>> a = Array("stuff", Range(2, "downto", -2))
>>> len(a)
5
>>> "t" in a
True
>>> a.index("u")
0
>>> for c in a:
...     print(c)
s
t
u
f
f

Parameters:

value (Iterable[T]) – Initial value for the Array.
range (Range | int | None) – The indexing scheme of the Array.

Raises:

ValueError – When argument values cannot be used to construct an Array.
TypeError – When invalid argument types are used.

count(value)#

Return number of occurrences of value.

Parameters:: value (T) – Value to search for.
Returns:: Number of occurrences of value.
Return type:: int

property direction: str#: "to" if indexes are ascending, "downto" otherwise.

index(value, start=None, stop=None)#

Find first occurrence of value.

Parameters:

value (T) – Value to search for.
start (int | None) – Index to start search at.
stop (int | None) – Index to stop search at.

Returns:

Index of first occurrence of value.

Raises:

ValueError – If the value is not present.

Return type:

int

property left: int#: Leftmost index of the array.

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

property right: int#: Rightmost index of the array.

class cocotb.types.LogicArray(value, range=None)[source]#

Fixed-sized, arbitrarily-indexed, array of cocotb.types.Logic.

LogicArrays can be constructed from iterables of values constructible into Logic like bool, str, int, or it can be constructed str or int literals syntaxes, as seen below.

Like Array, if range is not given, the range Range(len(value)-1, "downto", 0) is used. If an int is passed for range, the range Range(range-1, "downto", 0) is used.

>>> LogicArray(0b0111, 4)
LogicArray('0111', Range(3, 'downto', 0))

>>> LogicArray("01XZ")
LogicArray('01XZ', Range(3, 'downto', 0))

>>> LogicArray([0, True, "X"])
LogicArray('01X', Range(2, 'downto', 0))

Note

If constructing from an unsigned int literal, range must be given.

LogicArrays can be constructed from ints using from_unsigned() or from_signed().

>>> LogicArray.from_unsigned(0xA, 4)
LogicArray('1010', Range(3, 'downto', 0))

>>> LogicArray.from_signed(-4, Range(0, "to", 3))  # will sign-extend
LogicArray('1100', Range(0, 'to', 3))

LogicArrays can be constructed from bytes or bytearray using from_bytes(). Use the byteorder argument to control endianness.

>>> LogicArray.from_bytes(b"1n", byteorder="big")
LogicArray('0011000101101110', Range(15, 'downto', 0))

>>> LogicArray.from_bytes(b"1n", byteorder="little")
LogicArray('0110111000110001', Range(15, 'downto', 0))

LogicArrays support the same list-like operations as Array; however, it enforces the condition that all elements must be a Logic.

>>> la = LogicArray("1010")
>>> la[0]  # is indexable
Logic('0')

>>> la[1:]  # is slice-able
LogicArray('10', Range(1, 'downto', 0))

>>> Logic("0") in la  # is a collection
True

>>> list(la)  # is an iterable
[Logic('1'), Logic('0'), Logic('1'), Logic('0')]

When setting an element or slice, the value is first constructed into a Logic.

>>> la = LogicArray("1010")
>>> la[3] = "Z"
>>> la[3]
Logic('Z')

>>> la[2:] = ["X", True, 0]
>>> la
LogicArray('ZX10', Range(3, 'downto', 0))

>>> la[:] = 0b0101
>>> la
LogicArray('0101', Range(3, 'downto', 0))

LogicArrays can be converted into their str or int literal values using casts.

>>> la = LogicArray("1010")
>>> str(la)
'1010'
>>> int(la)
10

Warning

The int cast assumes the value is entirely 0 or 1 and will raise an exception otherwise.

The to_unsigned(), to_signed(), and to_bytes() methods can be used to convert the value into an unsigned or signed integer, or bytes, respectively.

>>> la.to_unsigned()
10

>>> la.to_signed()
-6

>>> la.to_bytes(byteorder="big")
b'\n'

Warning

These operations assume the value is entirely 0 or 1 and will raise an exception otherwise.

You can also convert LogicArrays to hexadecimal or binary strings using the built-ins hex:() and bin(), respectively.

>>> la = LogicArray("01111010")
>>> hex(la)
'0x7a'
>>> bin(la)
'0b1111010'

Warning

Using hex() or bin() first turns the LogicArray into an int. This means the exact length of the LogicArray is lost. It also means that these expressions will raise an exception if the value is not entirely 0 or 1.

LogicArrays also support element-wise logical operations: &, |, ^, and ~.

>>> def big_mux(a: LogicArray, b: LogicArray, sel: Logic) -> LogicArray:
...     s = LogicArray([sel] * len(a))
...     return (a & ~s) | (b & s)

>>> la = LogicArray("0110")
>>> p = LogicArray("1110")
>>> sel = Logic("1")  # choose second option
>>> big_mux(la, p, sel)
LogicArray('1110', Range(3, 'downto', 0))

Parameters:

value (int | str | Iterable[str | int | bool | Logic]) – Initial value for the LogicArray.
range (Range | int | None) – The indexing scheme of the LogicArray.

Raises:

TypeError – When invalid argument types are used.
ValueError – When value will not fit in a LogicArray of the given range.

property binstr: str#: Convert the value to the str literal representation.

Deprecated since version 2.0.

property buff: bytes#

Convert the value to bytes by interpreting it as an unsigned integer in big-endian byte order.

The object is first converted to an int as in to_unsigned(). Then the object is converted to bytes by converting the resulting integer value as in int.to_bytes(). This assumes big-endian byte order and the minimal number of bytes necessary to hold any value of the current object.

Returns:: A bytes object equivalent to the value.

Deprecated since version 2.0.

count(value)#

Return number of occurrences of value.

Parameters:: value (T) – Value to search for.
Returns:: Number of occurrences of value.
Return type:: int

property direction: str#: "to" if indexes are ascending, "downto" otherwise.

classmethod from_bytes(value, range=None, *, byteorder)[source]#

Construct a LogicArray from bytes.

The bytes is first converted to an unsigned integer using byteorder-endian representation, then is converted to a LogicArray as in from_unsigned().

Parameters:

value (bytes | bytearray) – The bytes to convert.
range (Range | int | None) – Indexing scheme for the LogicArray.
byteorder (Literal['big'] | Literal['little']) – The endianness used to construct the intermediate integer, either "big" or "little".

Returns:

A LogicArray equivalent to the value.

Raises:

ValueError – When a LogicArray of the given range can’t hold the value.

Return type:

LogicArray

classmethod from_signed(value, range)[source]#

Construct a LogicArray from an int with two’s complement representation.

The int is treated as an arbitrary-length bit vector with two’s complement representation where the left-most bit is the most significant bit. This bit vector is then constructed into a LogicArray.

Parameters:

value (int) – The integer to convert.
range (Range | int) – Indexing scheme for the LogicArray.

Returns:

A LogicArray equivalent to the value.

Raises:

TypeError – When invalid argument types are used.
ValueError – When a LogicArray of the given range can’t hold the value.

Return type:

LogicArray

classmethod from_unsigned(value, range)[source]#

Construct a LogicArray from an int with unsigned representation.

The int is treated as an arbitrary-length bit vector with unsigned representation where the left-most bit is the most significant bit. This bit vector is then constructed into a LogicArray.

Parameters:

value (int) – The integer to convert.
range (Range | int) – Indexing scheme for the LogicArray.

Returns:

A LogicArray equivalent to the value.

Raises:

TypeError – When invalid argument types are used.
ValueError – When a LogicArray of the given range can’t hold the value, or value is negative.

Return type:

LogicArray

index(value, start=None, stop=None)#

Find first occurrence of value.

Parameters:

value (T) – Value to search for.
start (int | None) – Index to start search at.
stop (int | None) – Index to stop search at.

Returns:

Index of first occurrence of value.

Raises:

ValueError – If the value is not present.

Return type:

int

property integer: int#

Convert the value to an int by interpreting it using unsigned representation.

The LogicArray is treated as an arbitrary-length vector of bits with the left-most bit being the most significant bit in the integer value. The bit vector is then interpreted as an integer using unsigned representation.

Returns:: An int equivalent to the value by interpreting it using unsigned representation.

Deprecated since version 2.0.

property is_resolvable: bool#: True if all elements are 0, 1, L, H.

property left: int#: Leftmost index of the array.

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

property right: int#: Rightmost index of the array.

property signed_integer: int#

Convert the value to an int by interpreting it using two’s complement representation.

The LogicArray is treated as an arbitrary-length vector of bits with the left-most bit being the most significant bit in the integer value. The bit vector is then interpreted as an integer using two’s complement representation.

Returns:: An int equivalent to the value by interpreting it using two’s complement representation.

Deprecated since version 2.0.

to_bytes(*, byteorder)[source]#

Convert the value to bytes.

The LogicArray is converted to an unsigned integer as in to_unsigned(), then is converted to bytes using byteorder-endian representation with the minimum number of bytes which can store all the bits in the original LogicArray.

Parameters:: byteorder (Literal['big'] | Literal['little']) – The endianness used to construct the intermediate integer, either "big" or "little".
Returns:: bytes equivalent to the value.
Return type:: bytes

to_signed(resolve=None)[source]#

Convert the value to an integer by interpreting it using two’s complement representation.

The LogicArray is treated as an arbitrary-length vector of bits with the left-most bit being the most significant bit in the integer value. The bit vector is then interpreted as an integer using two’s complement representation.

Parameters:: resolve (ResolveX | Literal['error'] | Literal['zeros'] | Literal['ones'] | Literal['random'] | None) – How X, Z, U, W, and - values should be resolve to 0 or 1 to allow conversion to an integer. See ResolveX for details. Defaults to the current value of RESOLVE_X.
Returns:: An integer equivalent to the value by interpreting it using two’s complement representation.
Return type:: int

to_unsigned(resolve=None)[source]#

Convert the value to an integer by interpreting it using unsigned representation.

The LogicArray is treated as an arbitrary-length vector of bits with the left-most bit being the most significant bit in the integer value. The bit vector is then interpreted as an integer using unsigned representation.

Parameters:: resolve (ResolveX | Literal['error'] | Literal['zeros'] | Literal['ones'] | Literal['random'] | None) – How X, Z, U, W, and - values should be resolve to 0 or 1 to allow conversion to an integer. See ResolveX for details. Defaults to the current value of RESOLVE_X.
Returns:: An integer equivalent to the value by interpreting it using unsigned representation.
Return type:: int

class cocotb.types.logic_array.ResolveX(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Resolution behaviors supported when converting a LogicArray to an integer.

The values L and H are always resolved to 0 and 1 respectively. These behaviors exist to resolve the Logic values X, Z, U, W, and - to either 0 or 1.

cocotb.types.logic_array.RESOLVE_X = ResolveX.VALUE_ERROR#

Global default for resolving X, Z, U, W, and - values to 0 or 1.

Set using COCOTB_RESOLVE_X before boot, or via this variable any time thereafter. Defaults to VALUE_ERROR.

Warning

This exists for backwards-compatibility reasons. Using any value besides VALUE_ERROR is not recommended.

Triggers#

cocotb.triggers.current_gpi_trigger()[source]#

Return the last GPITrigger that fired.

Edge Triggers#

class cocotb.triggers.RisingEdge(signal)[source]#

Fires on the rising edge of signal, on a transition to 1.

Only valid for scalar logic or bit-typed signals.

Parameters:: signal (_SignalType) – The signal upon which to wait for a rising edge.
Raises:: TypeError – If signal is not a 1-bit logic or bit-typed object.

Note

Prefer await signal.rising_edge to await RisingEdge(signal).

Warning

On many simulators this will trigger on transitions from non-0/1 value to 1, not just from 0 to 1 like the rising_edge function in VHDL.

class cocotb.triggers.FallingEdge(signal)[source]#

Fires on the falling edge of signal, on a transition to 0.

Only valid for scalar logic or bit-typed signals.

Parameters:: signal (_SignalType) – The signal upon which to wait for a rising edge.
Raises:: TypeError – If signal is not a 1-bit logic or bit-typed object.

Note

Prefer await signal.falling_edge to await FallingEdge(signal).

Warning

On many simulators this will trigger on transitions from non-0/1 value to 0, not just from 1 to 0 like the falling_edge function in VHDL.

class cocotb.triggers.ClockCycles(signal: LogicObject, num_cycles: int)[source]#

class cocotb.triggers.ClockCycles(signal: LogicObject, num_cycles: int, edge_type: Type[RisingEdge] | Type[FallingEdge] | Type[ValueChange] | None = None)

class cocotb.triggers.ClockCycles(signal: LogicObject, num_cycles: int, *, rising: bool)

Finishes after num_cycles transitions of signal.

awaiting this Trigger returns the ClockCycle object.

Parameters:

signal (cocotb.handle.LogicObject) – The signal to monitor.
num_cycles (int) – The number of cycles to count.
rising (bool | None) – If True, count rising edges; if False, count falling edges.
edge – The kind of Edge Triggers to count.

Warning

On many simulators transitions occur when the signal changes value from non-0 to 0 or non-1 to 1, not just from 1 to 0 or 0 to 1.

Added in version 2.0: Passing the edge trigger type: RisingEdge, FallingEdge, or ValueChange as the third positional argument or by the keyword edge_type.

property edge_type: Type[RisingEdge] | Type[FallingEdge] | Type[ValueChange]#: The type of edge trigger used.

property num_cycles: int#: The number of cycles to wait.

property signal: LogicObject#: The signal being monitored.

class cocotb.triggers.ValueChange(signal)[source]#

Fires on any value change of signal.

Parameters:: signal (_SignalType) – The signal upon which to wait for a value change.
Raises:: TypeError – If the signal is not an object which can change value.

Note

Prefer await signal.value_change to await ValueChange(signal).

Added in version 2.0.

class cocotb.triggers.Edge(signal)[source]#

Fires on any value change of signal.

Parameters:: signal (_SignalType) – The signal upon which to wait for a value change.
Raises:: TypeError – If the signal is not an object which can change value.

Deprecated since version 2.0: Use signal.value_change instead.

Timing Triggers#

class cocotb.triggers.Timer(time, unit='step', *, round_mode=None)[source]#

Fire after the specified simulation time period has elapsed.

This trigger will always consume some simulation time and will return control to the awaiting task at the beginning of the time step.

Parameters:

time (float | Fraction | Decimal) –
The time value.

Changed in version 1.5: Previously this argument was misleadingly called time_ps.
unit (Literal["step"] | TimeUnitWithoutStep) –
The unit of the time value.

One of 'step', 'fs', 'ps', 'ns', 'us', 'ms', 'sec'. When unit is 'step', the timestep is determined by the simulator (see COCOTB_HDL_TIMEPRECISION).

Changed in version 2.0: Renamed from units.
round_mode (str) – String specifying how to handle time values that sit between time steps (one of 'error', 'round', 'ceil', 'floor').

Raises:

ValueError – If a non-positive value is passed for Timer setup.

Usage:

>>> await Timer(100, unit="ps")

The time can also be a float:

>>> await Timer(100e-9, unit="sec")

which is particularly convenient when working with frequencies:

>>> freq = 10e6  # 10 MHz
>>> await Timer(1 / freq, unit="sec")

Other built-in exact numeric types can be used too:

>>> from fractions import Fraction
>>> await Timer(Fraction(1, 10), unit="ns")

>>> from decimal import Decimal
>>> await Timer(Decimal("100e-9"), unit="sec")

These are most useful when using computed durations while avoiding floating point inaccuracies.

Changed in version 1.5: Raise an exception when Timer uses a negative value as it is undefined behavior. Warn for 0 as this will cause erratic behavior in some simulators as well.

Changed in version 1.5: Support 'step' as the unit argument to mean “simulator time step”.

Changed in version 1.6: Support rounding modes.

Removed in version 2.0: Passing None as the unit argument was removed, use 'step' instead.

Removed in version 2.0: The time_ps parameter was removed, use the time parameter instead.

Changed in version 2.0: Passing 0 as the time argument now raises a ValueError.

round_mode: str = 'error'#: The default rounding mode.

class cocotb.triggers.ReadOnly(**kwargs)[source]#

Fires when the current simulation timestep moves to the read-only phase.

The read-only phase is entered when the current timestep no longer has any further delta steps. This will be a point where all the signal values are stable as there are no more RTL events scheduled for the timestep. The simulator will not allow scheduling of more events in this timestep. Useful for monitors which need to wait for all processes to execute (both RTL and cocotb) to ensure sampled signal values are final.

class cocotb.triggers.ReadWrite(**kwargs)[source]#: Fires when the read-write simulation phase is reached.

class cocotb.triggers.NextTimeStep(**kwargs)[source]#: Fires when the next time step is started.

Concurrency Triggers#

Triggers dealing with Tasks or running multiple Tasks concurrently.

class cocotb.task.Join(task)[source]#

Fires when a Task completes and returns the Task’s result.

Equivalent to calling task.join().

async def coro_inner():
    await Timer(1, unit="ns")
    return "Hello world"


task = cocotb.start_soon(coro_inner())
result = await Join(task)
assert result == "Hello world"

Parameters:: task (Task[ResultType]) – The Task upon which to wait for completion.
Returns:: Object that can be awaited or passed into First or Combine; the result of which will be the result of the Task.
Return type:: Join[ResultType]

Deprecated since version 2.0: Using task directly is preferred to Join(task) in all situations where the latter could be used.

Expand for references to cocotb.task.Join

Library Reference / Triggers / Concurrency Triggers

class cocotb.task.TaskComplete(task)[source]#

Fires when a Task completes.

Unlike Join, this Trigger does not return the result of the Task when awaited. See Task.complete for more information.

Warning

This class cannot be instantiated in the normal way. You must use Task.complete.

Added in version 2.0.

property task: Task[ResultType]#: The Task associated with this completion event.

class cocotb.triggers.NullTrigger(name=None)[source]#

Trigger that fires immediately.

Mostly useful when building or using higher-order functions which need to take or return Triggers.

The scheduling order of the Task awaiting this Trigger with respect to any other Task is not deterministic and should generally not be relied upon. Instead of using this Trigger to push the Task until “after” another Task has run, use other synchronization techniques, such as using an Event.

Do not do this:

transaction_data = None


def monitor():
    while dut.valid.value != 1 and dut.ready.value != 1:
        await RisingEdge(dut.clk)
    transaction_data = dut.data.value


def use_transaction():
    while True:
        await RisingEdge(dut.clk)
        # We need the NullTrigger here because both Tasks react to RisingEdge,
        # but there's no guarantee about which Task is run first,
        # so we need to force this one to run "later" using NullTrigger.
        await NullTrigger()
        if transaction_data is not None:
            process(transaction_data)


use_task = cocotb.start_soon(use_transaction())
monitor_task = cocotb.start_soon(monitor())

Instead use an .Event to explicitly synchronize the two Tasks, like so:

transaction_data = None
transaction_event = Event()


def monitor():
    while dut.valid.value != 1 and dut.ready.value != 1:
        await RisingEdge(dut.clk)
    transaction_data = dut.data.value
    transaction_event.set()


def use_transaction():
    # Now we don't need the NullTrigger.
    # This Task will wake up *strictly* after `monitor_task` sets the transaction.
    await transaction_event.wait()
    process(transaction_data)


use_task = cocotb.start_soon(use_transaction())
monitor_task = cocotb.start_soon(monitor())

Removed in version 2.0: The outcome parameter was removed. There is no alternative.

class cocotb.triggers.Combine(*trigger)[source]#

Trigger that fires when all triggers have fired.

awaiting this returns the Combine object. This is similar to Verilog’s join. See Waiting For Multiple Events for an example.

Parameters:: trigger (Trigger | Waitable[Any] | Task[Any]) – One or more awaitable objects.
Raises:: TypeError – When an unsupported trigger object is passed.

class cocotb.triggers.First(*trigger)[source]#

Fires when the first trigger in triggers fires.

awaiting this object returns the result of the first trigger that fires. This is similar to Verilog’s join_any. See Waiting For One Of Multiple Events for an example.

Parameters:

trigger (Trigger | Waitable[Any] | Task[Any]) – One or more awaitable objects.

Raises:

TypeError – When an unsupported trigger object is passed.
ValueError – When no triggers are passed.

Note

The event loop is single threaded, so while events may be simultaneous in simulation time, they can never be simultaneous in real time. For this reason, the value of t_ret is t1 in the following example is implementation-defined, and will vary by simulator:

t1 = Timer(10, unit="ps")
t2 = Timer(10, unit="ps")
t_ret = await First(t1, t2)

Note

In the old-style generator-based coroutines, t = yield [a, b] was another spelling of t = yield First(a, b). This spelling is no longer available when using await-based coroutines.

Synchronization Triggers#

The following objects are not Triggers themselves, but contain methods that can be used as triggers. They are used to synchronize coroutines with each other.

class cocotb.triggers.Event(name=None)[source]#

A way to signal an event across Tasks.

awaiting the result of wait() will block the awaiting Task until set() is called.

Parameters:: name (str | None) – Name for the Event.

Usage:

e = Event()


async def task1():
    await e.wait()
    print("resuming!")


cocotb.start_soon(task1())
# do stuff
e.set()
await NullTrigger()  # allows task1 to execute
# resuming!

Removed in version 2.0: Removed the undocumented data attribute and argument to set(), and the name attribute and argument to the constructor.

property name: str | None#: Name of the Event.

Deprecated since version 2.0: The name field will be removed in a future release.

property data: Any#: The data associated with the Event.

Deprecated since version 2.0: The data field will be removed in a future release. Use a separate variable to store the data instead.

set(data=None)[source]#

Set the Event and unblock all Tasks blocked on this Event.

wait()[source]#

Block the current Task until the Event is set.

If the event has already been set, the trigger will fire immediately.

To set the Event call set(). To reset the Event (and enable the use of wait() again), call clear().

clear()[source]#

Clear this event that has been set.

Subsequent calls to wait() will block until set() is called again.

is_set()[source]#

Return True if event has been set.

class cocotb.triggers.Lock(name=None)[source]#

A mutual exclusion lock.

Guarantees fair scheduling. Lock acquisition is given in order of attempted lock acquisition.

Usage:

By directly calling acquire() and release().

await lock.acquire()
try:
    # do some stuff
    ...
finally:
    lock.release()

Or…

async with lock:
    # do some stuff
    ...

Changed in version 1.4: The lock can be used as an asynchronous context manager in an async with statement

locked()[source]#

Return True if the lock has been acquired.

Changed in version 2.0: This is now a method to match asyncio.Lock.locked(), rather than an attribute.

acquire()[source]#

Produce a trigger which fires when the lock is acquired.

release()[source]#

Release the lock.

class cocotb.triggers.SimTimeoutError[source]#: Exception thrown when a timeout, in terms of simulation time, occurs.

async cocotb.triggers.with_timeout(trigger: Trigger, timeout_time: float | Decimal, timeout_unit: TimeUnit = 'step', round_mode: str | None = None) → None[source]#

async cocotb.triggers.with_timeout(trigger: Waitable[T], timeout_time: float | Decimal, timeout_unit: TimeUnit = 'step', round_mode: str | None = None) → T

async cocotb.triggers.with_timeout(trigger: Task[T], timeout_time: float | Decimal, timeout_unit: TimeUnit = 'step', round_mode: str | None = None) → T

async cocotb.triggers.with_timeout(trigger: Coroutine[Any, Any, T], timeout_time: float | Decimal, timeout_unit: TimeUnit = 'step', round_mode: str | None = None) → T

Wait on triggers or coroutines, throw an exception if it waits longer than the given time.

When a coroutine is passed, the callee coroutine is started, the caller blocks until the callee completes, and the callee’s result is returned to the caller. If timeout occurs, the callee is killed and SimTimeoutError is raised.

When a task is passed, the caller blocks until the callee completes and the callee’s result is returned to the caller. If timeout occurs, the callee continues to run and SimTimeoutError is raised.

If a Trigger or Waitable is passed, the caller blocks until the trigger fires, and the trigger is returned to the caller. If timeout occurs, the trigger is cancelled and SimTimeoutError is raised.

Usage:

await with_timeout(coro, 100, "ns")
await with_timeout(First(coro, event.wait()), 100, "ns")

Parameters:

trigger – A single object that could be right of an await expression in cocotb.
timeout_time – Simulation time duration before timeout occurs.
timeout_unit – Unit of timeout_time, accepts any unit that Timer does.
round_mode – String specifying how to handle time values that sit between time steps (one of 'error', 'round', 'ceil', 'floor').

Returns:

First trigger that completed if timeout did not occur.

Raises:

SimTimeoutError – If timeout occurs.

Added in version 1.3.

Changed in version 1.7: Support passing coroutines.

Changed in version 2.0: Passing None as the timeout_unit argument was removed, use 'step' instead.

Expand for references to cocotb.triggers.with_timeout

Library Reference / Triggers / Synchronization Triggers

Coroutines and Tasks / Waiting For Multiple Events Simultaneously / Waiting For Multiple Events

Abstract Triggers#

The following are internal classes used within cocotb.

class cocotb.triggers.Trigger[source]#

A future event that a Task can wait upon.

class cocotb.triggers.GPITrigger[source]#

A trigger for a simulation event.

class cocotb.triggers.Waitable[source]#

A Trigger-like object that can be implemented using coroutines.

This converts a _wait abstract method into a suitable __await__.

abstract async _wait()[source]#

The coroutine function which implements the functionality of the Waitable.

Test Utilities#

Clock Driver#

class cocotb.clock.Clock(signal, period, unit='step', impl=None)[source]#

Simple 50:50 duty cycle clock driver.

c = Clock(dut.clk, 10, "ns")
c.start()

Parameters:

signal (LogicObject) – The clock pin/signal to be driven.
period (float | Fraction | Decimal) –
The clock period.

Note

Must convert to an even number of timesteps.
unit (Literal["step"] | TimeUnitWithoutStep) –
One of 'step', 'fs', 'ps', 'ns', 'us', 'ms', 'sec'. When unit is 'step', the timestep is determined by the simulator (see COCOTB_HDL_TIMEPRECISION).

Changed in version 2.0: Renamed from units.
impl (Impl | None) –
One of 'auto', 'gpi', 'py'. Specify whether the clock is implemented with a GpiClock (faster), or with a Python coroutine. When 'auto' is used (default), the fastest implementation that supports your environment and use case is picked.

Added in version 2.0.

When impl is 'auto', if COCOTB_TRUST_INERTIAL_WRITES is defined, the GpiClock implementation will be used. Otherwise, the Python coroutine implementation will be used. See the environment variable’s documentation for more information on the consequences of using the simulator’s inertial write mechanism.

If you need more features like a phase shift and an asymmetric duty cycle, it is simple to create your own clock generator (that you then cocotb.start_soon()):

async def custom_clock():
    # pre-construct triggers for performance
    high_time = Timer(high_delay, unit="ns")
    low_time = Timer(low_delay, unit="ns")
    await Timer(initial_delay, unit="ns")
    while True:
        dut.clk.value = 1
        await high_time
        dut.clk.value = 0
        await low_time

If you also want to change the timing during simulation, use this slightly more inefficient example instead where the Timers inside the while loop are created with current delay values:

async def custom_clock():
    while True:
        dut.clk.value = 1
        await Timer(high_delay, unit="ns")
        dut.clk.value = 0
        await Timer(low_delay, unit="ns")


high_delay = low_delay = 100
cocotb.start_soon(custom_clock())
await Timer(1000, unit="ns")
high_delay = low_delay = 10  # change the clock speed
await Timer(1000, unit="ns")

Changed in version 1.5: Support 'step' as the unit argument to mean “simulator time step”.

Changed in version 2.0: Passing None as the unit argument was removed, use 'step' instead.

Changed in version 2.0: start() now automatically calls cocotb.start_soon() and stores the Task on the Clock object, so that it may later be stop()ped.

Expand for references to cocotb.clock.Clock

Library Reference / Test Utilities / Clock Driver

Library Reference / Test Utilities / Clock Driver

property signal: LogicObject#: The clock signal being driven.

property period: float | Fraction | Decimal#: The clock period (unit-less).

property unit: Literal['step'] | TimeUnitWithoutStep#: The unit of the clock period.

property impl: Impl#

The concrete implementation of the clock used.

"gpi" if the clock is implemented in C in the GPI layer, or "py" if the clock is implemented in Python using cocotb Tasks.

start(start_high=True)[source]#

Start driving the clock signal.

You can later stop the clock by calling stop().

Parameters:

start_high (bool) –

Whether to start the clock with a 1 for the first half of the period. Default is True.

Added in version 1.3.

Raises:

RuntimeError – If attempting to start a clock that has already been started.

Returns:

Object which can be passed to cocotb.start_soon() or ignored.

Return type:

Task[None]

Changed in version 2.0: Removed cycles arguments for toggling for a finite amount of cycles. Use stop() to stop a clock from running.

Changed in version 2.0: Previously, this method returned a coroutine which needed to be passed to cocotb.start_soon(). Now the Clock object keeps track of its own driver Task, so this is no longer necessary. Simply call clock.start() to start running the clock.

Expand for references to cocotb.clock.Clock.start

stop()[source]#

Stop driving the clock signal.

You can later start the clock again by calling start().

Raises:: RuntimeError – If attempting to stop a clock that has never been started.

Added in version 2.0.

Expand for references to cocotb.clock.Clock.stop

async cycles(num_cycles, edge_type=<class 'cocotb._gpi_triggers.RisingEdge'>)[source]#

Wait for a number of clock cycles.

Asynchronous Queues#

Expand for references to cocotb.queue

exception cocotb.queue.QueueFull[source]#: Raised when the Queue.put_nowait() method is called on a full Queue.

exception cocotb.queue.QueueEmpty[source]#: Raised when the Queue.get_nowait() method is called on an empty Queue.

class cocotb.queue.AbstractQueue(maxsize=0)[source]#

A queue, useful for coordinating producer and consumer coroutines.

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().

qsize()[source]#

Number of items in the queue.

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

empty()[source]#

Return True if the queue is empty, False otherwise.

full()[source]#

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.

async put(item)[source]#

Put an item into the queue.

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

Expand for references to cocotb.queue.AbstractQueue.put

put_nowait(item)[source]#

Put an item into the queue without blocking.

If no free slot is immediately available, raise QueueFull.

async get()[source]#

Remove and return an item from the queue.

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

Expand for references to cocotb.queue.AbstractQueue.get

get_nowait()[source]#

Remove and return an item from the queue.

Return an item if one is immediately available, else raise QueueEmpty.

class cocotb.queue.Queue(maxsize=0)[source]#

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

Expand for references to cocotb.queue.Queue

class cocotb.queue.PriorityQueue(maxsize=0)[source]#

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

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

class cocotb.queue.LifoQueue(maxsize=0)[source]#

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

Simulation Time Utilities#

Utility functions for dealing with simulation time.

cocotb.utils.get_sim_time(unit='step')[source]#

Retrieve the simulation time from the simulator.

Parameters:

unit (Literal["step"] | TimeUnitWithoutStep) –

String specifying the unit of the result (one of 'step', 'fs', 'ps', 'ns', 'us', 'ms', 'sec'). 'step' will return the raw simulation time.

Changed in version 2.0: Passing None as the unit argument was removed, use 'step' instead.

Changed in version 2.0: Renamed from units.

Raises:

ValueError – If unit is not a valid unit (see Args section).

Returns:

The simulation time in the specified unit.

Return type:

float

Changed in version 1.6: Support 'step' as the the unit argument to mean “simulator time step”.

Expand for references to cocotb.utils.get_sim_time