transactron.lib package

Submodules

transactron.lib.adapters module

class transactron.lib.adapters.Adapter

Bases: AdapterBase

Adapter method.

Creates a method controlled by plain Amaranth signals. One of the possible uses is to mock a method in a testbench.

Attributes

en: Signal, in: Activates the method (sets the ready signal).
done: Signal, out: Signals that the method is called (returns the run signal).
data_in: View, in: Data returned from the defined method.
data_out: View, out: Data passed as argument to the defined method.
validators: list of tuples of View, out and Signal, in: Hooks for validate_arguments.

__init__(**kwargs)

Parameters

**kwargs: Keyword arguments for Method that will be created. See transactron.core.Method.__init__ for parameters description.

set(with_validate_arguments: Optional[bool])

class transactron.lib.adapters.AdapterBase

Bases: Component

__init__(iface: Method, layout_in: StructLayout, layout_out: StructLayout)

data_in: View[StructLayout]

data_out: View[StructLayout]

debug_signals() → amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]] | collections.abc.Mapping[str, amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]]

done: Signal

en: Signal

class transactron.lib.adapters.AdapterTrans

Bases: AdapterBase

Adapter transaction.

Creates a transaction controlled by plain Amaranth signals. Allows to expose a method to plain Amaranth code, including testbenches.

Attributes

en: Signal, in: Activates the transaction (sets the request signal).
done: Signal, out: Signals that the transaction is performed (returns the grant signal).
data_in: View, in: Data passed to the iface method.
data_out: View, out: Data returned from the iface method.

__init__(iface: Method, *, src_loc: int | tuple[str, int] = 0)

Parameters

iface: Method: The method to be called by the transaction.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

transactron.lib.buttons module

class transactron.lib.buttons.ClickIn

Bases: Elaboratable

Clicked input.

Useful for interactive simulations or FPGA button/switch interfaces. On a rising edge (tested synchronously) of btn, the get method is enabled, which returns the data present on dat at the time. Inputs are synchronized.

Attributes

get: Method: The method for retrieving data from the input. Accepts an empty argument, returns a structure.
btn: Signal, in: The button input.
dat: MethodStruct, in: The data input.

__init__(layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], src_loc: int | tuple[str, int] = 0)

Parameters

layout: method layout: The data format for the input.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.buttons.ClickOut

Bases: Elaboratable

Clicked output.

Useful for interactive simulations or FPGA button/LED interfaces. On a rising edge (tested synchronously) of btn, the put method is enabled, which, when called, changes the value of the dat signal.

Attributes

put: Method: The method for retrieving data from the input. Accepts a structure, returns empty result.
btn: Signal, in: The button input.
dat: MethodStruct, out: The data output.

__init__(layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], *, src_loc: int | tuple[str, int] = 0)

Parameters

layout: method layout: The data format for the output.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

transactron.lib.connectors module

class transactron.lib.connectors.Connect

Bases: Elaboratable

Forwarding by transaction simultaneity

Provides a means to connect two transactions with forwarding by means of the transaction simultaneity mechanism. It provides two methods: read, and write, which always execute simultaneously. Typical use case is for moving data from write to read, but data flow in the reverse direction is also possible.

Attributes

read: Method: The read method. Accepts a (possibly empty) structure, returns a structure.
write: Method: The write method. Accepts a structure, returns a (possibly empty) structure.

__init__(layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']] = (), rev_layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']] = (), *, src_loc: int | tuple[str, int] = 0)

Parameters

layout: method layout: The format of structures forwarded.
rev_layout: method layout: The format of structures forwarded in the reverse direction.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.connectors.ConnectTrans

Bases: Elaboratable

Simple connecting transaction.

Takes two methods and creates a transaction which calls both of them. Result of the first method is connected to the argument of the second, and vice versa. Allows easily connecting methods with compatible layouts.

__init__(method1: Method, method2: Method, *, src_loc: int | tuple[str, int] = 0)

Parameters

method1: Method: First method.
method2: Method: Second method.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.connectors.FIFO

Bases: Elaboratable

FIFO module.

Provides a transactional interface to Amaranth FIFOs. Exposes two methods: read, and write. Both methods are ready only when they can be executed – i.e. the queue is respectively not empty / not full. It is possible to simultaneously read and write in a single clock cycle, but only if both readiness conditions are fulfilled.

Attributes

read: Method: The read method. Accepts an empty argument, returns a structure.
write: Method: The write method. Accepts a structure, returns empty result.

__init__(layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], depth: int, fifo_type=<class 'amaranth.lib.fifo.SyncFIFO'>, *, src_loc: int | tuple[str, int] = 0)

Parameters

layout: method layout: The format of structures stored in the FIFO.
depth: int: Size of the FIFO.
fifoType: Elaboratable: FIFO module conforming to Amaranth library FIFO interface. Defaults to SyncFIFO.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.connectors.Forwarder

Bases: Elaboratable

Forwarding with overflow buffering

Provides a means to connect two transactions with forwarding. Exposes two methods: read, and write. When both of these methods are executed simultaneously, data is forwarded between them. If write is executed, but read is not, the value cannot be forwarded, but is stored into an overflow buffer. No further writes are possible until the overflow buffer is cleared by read.

The write method is scheduled before read.

Attributes

read: Method: The read method. Accepts an empty argument, returns a structure.
write: Method: The write method. Accepts a structure, returns empty result.

__init__(layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], *, src_loc: int | tuple[str, int] = 0)

Parameters

layout: method layout: The format of structures forwarded.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.connectors.ManyToOneConnectTrans

Bases: Elaboratable

Many-to-one method connection.

Connects each of a set of methods to another method using separate transactions. Equivalent to a set of ConnectTrans.

__init__(*, get_results: list[transactron.core.method.Method], put_result: Method, src_loc: int | tuple[str, int] = 0)

Parameters

get_results: list[Method]: Methods to be connected to the put_result method.
put_result: Method: Common method for each of the connections created.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.connectors.Pipe

Bases: Elaboratable

This module implements a Pipe. It is a halfway between Forwarder and 2-FIFO. In the Pipe data is always stored localy, so the critical path of the data is cut, but there is a combinational path between the control signals of the read and the write methods. For comparison: - in Forwarder there is both a data and a control combinational path - in 2-FIFO there are no combinational paths

The read method is scheduled before the write.

Attributes

read: Method: Reads from the pipe. Accepts an empty argument, returns a structure. Ready only if the pipe is not empty.
write: Method: Writes to the pipe. Accepts a structure, returns empty result. Ready only if the pipe is not full.
clean: Method: Cleans the pipe. Has priority over read and write methods.

__init__(layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']])

Parameters

layout: record layout: The format of records forwarded.

class transactron.lib.connectors.StableSelectingNetwork

Bases: Elaboratable

A network that groups inputs with a valid bit set.

The circuit takes n inputs with a valid signal each and on the output returns a grouped and consecutive sequence of the provided input signals. The order of valid inputs is preserved.

For example for input (0 is an invalid input): 0, a, 0, d, 0, 0, e

The circuit will return: a, d, e, 0, 0, 0, 0

The circuit uses a divide and conquer algorithm. The recursive call takes two bit vectors and each of them is already properly sorted, for example: v1 = [a, b, 0, 0]; v2 = [c, d, e, 0]

Now by shifting left v2 and merging it with v1, we get the result: v = [a, b, c, d, e, 0, 0, 0]

Thus, the network has depth log_2(n).

__init__(n: int, layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']])

transactron.lib.dependencies module

class transactron.lib.dependencies.DependencyKey

Bases: Generic[T, U], ABC

Base class for dependency keys.

Dependency keys are used to access dependencies in the DependencyManager. Concrete instances of dependency keys should be frozen data classes.

Parameters

lock_on_get: bool, default: True: Specifies if no new dependencies should be added to key if it was already read by get_dependency.
cache: bool, default: True: If true, result of the combine method is cached and subsequent calls to get_dependency will return the value in the cache. Adding a new dependency clears the cache.
empty_valid: bool, defaultFalse: Specifies if getting key dependency without any added dependencies is valid. If set to False, that action would cause raising KeyError.

cache: bool = True

abstract combine(data: list[T]) → U

Combine multiple dependencies with the same key.

This method is used to generate the value returned from get_dependency in the DependencyManager. It takes dependencies added to the key using add_dependency and combines them to a single result.

Different implementations of combine give different combining behavior for different kinds of keys.

empty_valid: bool = False

lock_on_get: bool = True

class transactron.lib.dependencies.DependencyManager

Bases: object

Dependency manager.

Tracks dependencies across the core.

__init__()

add_dependency(key: DependencyKey[T, Any], dependency: T) → None

Adds a new dependency to a key.

Depending on the key type, a key can have a single dependency or multple dependencies added to it.

dependency_provided(key: DependencyKey) → bool: Checks if any dependency for a key is provided (ignores empty_valid parameter)

get_dependency(key: DependencyKey[Any, U]) → U

Gets the dependency for a key.

The way dependencies are interpreted is dependent on the key type.

class transactron.lib.dependencies.ListKey

Bases: Generic[T], DependencyKey[T, list[T]]

Base class for list key.

List keys are used when there is an one-to-many relation between keys and dependecies. Provides list of dependencies.

combine(data: list[T]) → list[T]

Combine multiple dependencies with the same key.

This method is used to generate the value returned from get_dependency in the DependencyManager. It takes dependencies added to the key using add_dependency and combines them to a single result.

Different implementations of combine give different combining behavior for different kinds of keys.

empty_valid: bool = True

class transactron.lib.dependencies.SimpleKey

Bases: Generic[T], DependencyKey[T, T]

Base class for simple dependency keys.

Simple dependency keys are used when there is an one-to-one relation between keys and dependencies. If more than one dependency is added to a simple key, an error is raised.

Parameters

default_value: T: Specifies the default value returned when no dependencies are added. To enable it empty_valid must be True.

combine(data: list[T]) → T

Combine multiple dependencies with the same key.

This method is used to generate the value returned from get_dependency in the DependencyManager. It takes dependencies added to the key using add_dependency and combines them to a single result.

Different implementations of combine give different combining behavior for different kinds of keys.

default_value: T

class transactron.lib.dependencies.UnifierKey

Bases: DependencyKey[Method, tuple[Method, dict[str, Unifier]]]

Base class for method unifier dependency keys.

Method unifier dependency keys are used to collect methods to be called by some part of the core. As multiple modules may wish to be called, a method unifier is used to present a single method interface to the caller, which allows to customize the calling behavior.

combine(data: list[transactron.core.method.Method]) → tuple[transactron.core.method.Method, dict[str, transactron.lib.transformers.Unifier]]

Combine multiple dependencies with the same key.

This method is used to generate the value returned from get_dependency in the DependencyManager. It takes dependencies added to the key using add_dependency and combines them to a single result.

Different implementations of combine give different combining behavior for different kinds of keys.

unifier: Callable[[list[transactron.core.method.Method]], Unifier]

transactron.lib.fifo module

class transactron.lib.fifo.BasicFifo

Bases: Elaboratable

Transactional FIFO queue

Attributes

read: Method: Reads from the FIFO. Accepts an empty argument, returns a structure. Ready only if the FIFO is not empty.
peek: Method: Returns the element at the front (but not delete). Ready only if the FIFO is not empty. The method is nonexclusive.
write: Method: Writes to the FIFO. Accepts a structure, returns empty result. Ready only if the FIFO is not full.
clear: Method: Clears the FIFO entries. Has priority over read and write methods. Note that, clearing the FIFO doesn’t reinitialize it to values passed in init parameter.

__init__(layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], depth: int, *, src_loc: int | tuple[str, int] = 0) → None

Parameters

layout: method layout: Layout of data stored in the FIFO.
depth: int: Size of the FIFO.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.fifo.Semaphore

Bases: Elaboratable

Semaphore

__init__(max_count: int) → None

Parameters

size: int: Size of the semaphore.

transactron.lib.logging module

class transactron.lib.logging.HardwareLogger

Bases: object

A class for creating log messages in the hardware.

Intuitively, the hardware logger works similarly to a normal software logger. You can log a message anywhere in the circuit, but due to the parallel nature of the hardware you must specify a special trigger signal which will indicate if a message shall be reported in that cycle.

Hardware logs are evaluated and printed during simulation, so both the trigger and the format fields are Amaranth values, i.e. signals or arbitrary Amaranth expressions.

Instances of the HardwareLogger class represent a logger for a single submodule of the circuit. Exactly how a “submodule” is defined is up to the developer. Submodule are identified by a unique string and the names can be nested. Names are organized into a namespace hierarchy where levels are separated by periods, much like the Python package namespace. So in the instance, submodules names might be “frontend” for the upper level, and “frontend.icache” and “frontend.bpu” for the sub-levels. There is no arbitrary limit to the depth of nesting.

Attributes

name: str: Name of this logger.

__init__(name: str)

Parameters

name: str: Name of this logger. Hierarchy levels are separated by periods, e.g. “backend.fu.jumpbranch”.

assertion(m: ModuleLike, value: Value, format: str = '', *args, src_loc_at: int = 0, **kwargs)

Define an assertion.

This function might help find some hardware bugs which might otherwise be hard to detect. If value is false, it will terminate the simulation or it can also be used to turn on a warning LED on a board.

Internally, this is a convenience wrapper over log.error.

See HardwareLogger.log function for more details.

debug(m: ModuleLike, trigger: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable, format: str, *args, **kwargs)

Log a message with severity ‘DEBUG’.

See HardwareLogger.log function for more details.

error(m: ModuleLike, trigger: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable, format: str, *args, **kwargs)

Log a message with severity ‘ERROR’.

This severity level has special semantics. If a log with this serverity level is triggered, the simulation will be terminated.

See HardwareLogger.log function for more details.

info(m: ModuleLike, trigger: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable, format: str, *args, **kwargs)

Log a message with severity ‘INFO’.

See HardwareLogger.log function for more details.

log(m: ModuleLike, level: int, trigger: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable, format: str, *args, src_loc_at: int = 0)

Registers a hardware log record with the given severity.

Parameters

m: ModuleLike: The module for which the log record is added.
trigger: ValueLike: If the value of this Amaranth expression is true, the log will reported.
format: str: The format of the message as defined in PEP 3101.
*args: Amaranth values that will be read during simulation and used to format the message.
src_loc_at: int, optional: How many stack frames below to look for the source location, used to identify the failing assertion.

warning(m: ModuleLike, trigger: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable, format: str, *args, **kwargs)

Log a message with severity ‘WARNING’.

See HardwareLogger.log function for more details.

class transactron.lib.logging.LogKey

Bases: ListKey[LogRecord]

__init__() → None

class transactron.lib.logging.LogRecord

Bases: LogRecordInfo

A LogRecord instance represents an event being logged.

Attributes

trigger: Signal: Amaranth signal triggering the log.
fields: Signal: Amaranth signals that will be used to format the message.

__init__(logger_name: str, level: int, format_str: str, location: tuple[str, int], trigger: ~amaranth.hdl._ast.Signal, fields: list[amaranth.hdl._ast.Signal] = <factory>) → None

fields: list[amaranth.hdl._ast.Signal]

trigger: Signal

class transactron.lib.logging.LogRecordInfo

Bases: object

Simulator-backend-agnostic information about a log record that can be serialized and used outside the Amaranth context.

Attributes

logger_name: str
level: LogLevel: The severity level of the log.
format_str: str: The template of the message. Should follow PEP 3101 standard.
location: SrcLoc: Source location of the log.

__init__(logger_name: str, level: int, format_str: str, location: tuple[str, int]) → None

format(*args) → str: Format the log message with a set of concrete arguments.

format_str: str

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

level: int

location: tuple[str, int]

logger_name: str

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → SchemaF[A]

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Optional[Tuple[str, str]] = None, default: Optional[Callable] = None, sort_keys: bool = False, **kw) → str

transactron.lib.logging.get_log_records(level: int, namespace_regexp: str = '.*') → list[transactron.lib.logging.LogRecord]

Get log records in for the given severity level and in the specified namespace.

This function returns all log records with the severity bigger or equal to the specified level and belonging to the specified namespace.

Parameters

level: LogLevel: The minimum severity level.
namespace: str, optional: The regexp of the namespace. If not specified, logs from all namespaces will be processed.

transactron.lib.logging.get_trigger_bit(level: int, namespace_regexp: str = '.*') → Value

Get a trigger bit for logs of the given severity level and in the specified namespace.

The signal returned by this function is high whenever the trigger signal of any of the records with the severity bigger or equal to the specified level is high.

Parameters

level: LogLevel: The minimum severity level.
namespace: str, optional: The regexp of the namespace. If not specified, logs from all namespaces will be processed.

transactron.lib.metrics module

class transactron.lib.metrics.FIFOLatencyMeasurer

Bases: Elaboratable

Measures duration between two events, e.g. request processing latency. It can track multiple events at the same time, i.e. the second event can be registered as started, before the first finishes. However, they must be processed in the FIFO order.

The module exposes an exponential histogram of the measured latencies.

__init__(fully_qualified_name: str, description: str = '', *, slots_number: int, max_latency: int)

Parameters

fully_qualified_name: str: The fully qualified name of the metric.
description: str: A human-readable description of the metric’s functionality.
slots_number: int: A number of events that the module can track simultaneously.
max_latency: int: The maximum latency of an event. Used to set signal widths and number of buckets in the histogram. If a latency turns to be bigger than the maximum, it will overflow and result in a false measurement.

metrics_enabled() → bool

start(m: TModule)

Registers the start of an event. Can be called before the previous events finish. If there are no slots available, the method will be blocked.

Should be called in the body of either a transaction or a method.

Parameters

m: TModule: Transactron module

stop(m: TModule)

Registers the end of the oldest event (the FIFO order). If there are no started events in the queue, the method will block.

Should be called in the body of either a transaction or a method.

Parameters

m: TModule: Transactron module

class transactron.lib.metrics.HardwareMetricsManager

Bases: object

Collects all metrics registered in the circuit and provides an easy access to them.

__init__()

debug_signals() → amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]] | collections.abc.Mapping[str, amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]]: Returns tree-like SignalBundle composed of all metric registers.

get_metrics() → dict[str, transactron.lib.metrics.HwMetric]: Returns all metrics registered in the circuit.

get_register_value(metric_name: str, reg_name: str) → Signal

Returns the signal holding the register value of the given metric.

Parameters

metric_name: str: The fully qualified name of the metric, for example ‘frontend.icache.loads’.
reg_name: str: The name of the register from that metric, for example if the metric is a histogram, the ‘reg_name’ could be ‘min’ or ‘bucket-32’.

class transactron.lib.metrics.HwCounter

Bases: Elaboratable, HwMetric

Hardware Counter

The most basic hardware metric that can just increase its value.

__init__(fully_qualified_name: str, description: str = '', *, width_bits: int = 32)

Parameters

fully_qualified_name: str: The fully qualified name of the metric.
description: str: A human-readable description of the metric’s functionality.
width_bits: int: The bit-width of the register. Defaults to 32 bits.

incr(m: ~transactron.core.tmodule.TModule, *, cond: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable = (const 1'd1))

Increases the value of the counter by 1.

Should be called in the body of either a transaction or a method.

Parameters

m: TModule: Transactron module

class transactron.lib.metrics.HwExpHistogram

Bases: Elaboratable, HwMetric

Hardware Exponential Histogram

Represents the distribution of sampled data through a histogram. A histogram samples observations (usually things like request durations or queue sizes) and counts them in a configurable number of buckets. The buckets are of exponential size. For example, a histogram with 5 buckets would have the following value ranges: [0, 1); [1, 2); [2, 4); [4, 8); [8, +inf).

Additionally, the histogram tracks the number of observations, the sum of observed values, and the minimum and maximum values.

__init__(fully_qualified_name: str, description: str = '', *, bucket_count: int, sample_width: int = 32, registers_width: int = 32)

Parameters

fully_qualified_name: str: The fully qualified name of the metric.
description: str: A human-readable description of the metric’s functionality.
max_value: int: The maximum value that the histogram would be able to count. This value is used to calculate the number of buckets.

add(m: TModule, sample: Value)

Adds a new sample to the histogram.

Should be called in the body of either a transaction or a method.

Parameters

m: TModule: Transactron module
sample: ValueLike: The value that will be added to the histogram

class transactron.lib.metrics.HwMetric

Bases: ABC, MetricModel

A base for all metric implementations. It should be only used for declaring new types of metrics.

It takes care of registering the metric in the dependency manager.

Attributes

signals: dict[str, Signal]: A mapping from a register name to a Signal containing the value of that register.

__init__(fully_qualified_name: str, description: str)

Parameters

fully_qualified_name: str: The fully qualified name of the metric.
description: str: A human-readable description of the metric’s functionality.

add_registers(regs: list[transactron.lib.metrics.HwMetricRegister])

Adds registers to a metric. Should be only called by inheriting classes during initialization.

Parameters

regs: list[HwMetricRegister]: A list of registers to be registered.

metrics_enabled() → bool

class transactron.lib.metrics.HwMetricsEnabledKey

Bases: SimpleKey[bool]

DependencyManager key for enabling hardware metrics. If metrics are disabled, none of theirs signals will be synthesized.

__init__() → None

default_value: T = False

empty_valid: bool = True

lock_on_get: bool = False

class transactron.lib.metrics.MetricModel

Bases: object

Provides information about a metric exposed by the circuit. Each metric comprises multiple registers, each dedicated to storing specific values.

The configuration of registers is internally determined by a specific metric type and is not user-configurable.

Attributes

fully_qualified_name: str: The fully qualified name of the metric, with name components joined by dots (‘.’), e.g., ‘foo.bar.requests’.
description: str: A human-readable description of the metric’s functionality.
regs: list[MetricRegisterModel]: A list of registers associated with the metric.

__init__(fully_qualified_name: str, description: str, regs: dict[str, transactron.lib.metrics.MetricRegisterModel] = <factory>) → None

description: str

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

fully_qualified_name: str

regs: dict[str, transactron.lib.metrics.MetricRegisterModel]

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → SchemaF[A]

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Optional[Tuple[str, str]] = None, default: Optional[Callable] = None, sort_keys: bool = False, **kw) → str

class transactron.lib.metrics.MetricRegisterModel

Bases: object

Represents a single register of a metric, serving as a fundamental building block that holds a singular value.

Attributes

name: str: The unique identifier for the register (among remaning registers of a specific metric).
description: str: A brief description of the metric’s purpose.
width: int: The bit-width of the register.

__init__(name: str, description: str, width: int) → None

description: str

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

name: str

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → SchemaF[A]

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Optional[Tuple[str, str]] = None, default: Optional[Callable] = None, sort_keys: bool = False, **kw) → str

width: int

class transactron.lib.metrics.TaggedCounter

Bases: Elaboratable, HwMetric

Hardware Tagged Counter

Like HwCounter, but contains multiple counters, each with its own tag. At a time a single counter can be increased and the value of the tag can be provided dynamically. The type of the tag can be either an int enum, a range or a list of integers (negative numbers are ok).

Internally, it detects if tag values can be one-hot encoded and if so, it generates more optimized circuit.

Attributes

tag_width: int: The length of the signal holding a tag value.
one_hot: bool: Whether tag values can be one-hot encoded.
counters: dict[int, HwMetricRegisters]: Mapping from a tag value to a register holding a counter for that tag.

__init__(fully_qualified_name: str, description: str = '', *, tags: Union[range, Type[Enum], list[int]], registers_width: int = 32)

Parameters

fully_qualified_name: str: The fully qualified name of the metric.
description: str: A human-readable description of the metric’s functionality.
tags: range | Type[Enum] | list[int]: Tag values.
registers_width: int: Width of the underlying registers. Defaults to 32 bits.

Increases the counter of a given tag by 1.

Should be called in the body of either a transaction or a method.

Parameters

m: TModule: Transactron module
tag: ValueLike: The tag of the counter.
cond: ValueLike: When set to high, the counter will be increased. By default set to high.

class transactron.lib.metrics.TaggedLatencyMeasurer

Bases: Elaboratable

Measures duration between two events, e.g. request processing latency. It can track multiple events at the same time, i.e. the second event can be registered as started, before the first finishes. However, each event needs to have an unique slot tag.

The module exposes an exponential histogram of the measured latencies.

__init__(fully_qualified_name: str, description: str = '', *, slots_number: int, max_latency: int)

Parameters

fully_qualified_name: str: The fully qualified name of the metric.
description: str: A human-readable description of the metric’s functionality.
slots_number: int: A number of events that the module can track simultaneously.
max_latency: int: The maximum latency of an event. Used to set signal widths and number of buckets in the histogram. If a latency turns to be bigger than the maximum, it will overflow and result in a false measurement.

metrics_enabled() → bool

start(m: TModule, *, slot: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable)

Registers the start of an event for a given slot tag.

Should be called in the body of either a transaction or a method.

Parameters

m: TModule: Transactron module
slot: ValueLike: The slot tag of the event.

stop(m: TModule, *, slot: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable)

Registers the end of the event for a given slot tag.

Should be called in the body of either a transaction or a method.

Parameters

m: TModule: Transactron module
slot: ValueLike: The slot tag of the event.

transactron.lib.reqres module

class transactron.lib.reqres.ArgumentsToResultsZipper

Bases: Elaboratable

Zips arguments used to call method with results, cutting critical path.

This module provides possibility to pass arguments from caller and connect it with results from callee. Arguments are stored in 2-FIFO and results in Forwarder. Because of this asymmetry, the callee should provide results as long as they aren’t correctly received.

FIFO is used as rate-limiter, so when FIFO reaches full capacity there should be no new requests issued.

Example topology:

graph LR Caller -- write_arguments --> 2-FIFO; Caller -- invoke --> Callee["Callee \n (1+ cycle delay)"]; Callee -- write_results --> Forwarder; Forwarder -- read --> Zip; 2-FIFO -- read --> Zip; Zip -- read --> User; subgraph ArgumentsToResultsZipper Forwarder; 2-FIFO; Zip; end

Attributes

peek_arg: Method: A nonexclusive method to read (but not delete) the head of the arg queue.
write_args: Method: Method to write arguments with args_layout format to 2-FIFO.
write_results: Method: Method to save results with results_layout in the Forwarder.
read: Method: Reads latest entries from the fifo and the forwarder and return them as a structure with two fields: ‘args’ and ‘results’.

__init__(args_layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], results_layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], src_loc: int | tuple[str, int] = 0)

Parameters

args_layout: method layout: The format of arguments.
results_layout: method layout: The format of results.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.reqres.Serializer

Bases: Elaboratable

Module to serialize request-response methods.

Provides a transactional interface to connect many client Modules (which request somethig using method call) with a server Module which provides method to request operation and method to get response.

Requests are being serialized from many clients and forwarded to a server which can process only one request at the time. Responses from server are deserialized and passed to proper client. Serializer assumes, that responses from the server are in-order, so the order of responses is the same as order of requests.

Attributes

serialize_in: list[Method]: List of request methods. Data layouts are the same as for serialized_req_method.
serialize_out: list[Method]: List of response methods. Data layouts are the same as for serialized_resp_method. i-th response method provides responses for requests from i-th serialize_in method.

__init__(*, port_count: int, serialized_req_method: Method, serialized_resp_method: Method, depth: int = 4, src_loc: int | tuple[str, int] = 0)

Parameters

port_count: int: Number of ports, which should be generated. len(serialize_in)=len(serialize_out)=port_count
serialized_req_method: Method: Request method provided by server’s Module.
serialized_resp_method: Method: Response method provided by server’s Module.
depth: int: Number of requests which can be forwarded to server, before server provides first response. Describe the resistance of Serializer to latency of server in case when server is fully pipelined.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

transactron.lib.simultaneous module

transactron.lib.simultaneous.condition(m: TModule, *, nonblocking: bool = False, priority: bool = False)

Conditions using simultaneous transactions.

This context manager allows to easily define conditions utilizing nested transactions and the simultaneous transactions mechanism. It is similar to Amaranth’s If, but allows to call different and possibly overlapping method sets in each branch. Each of the branches is defined using a separate nested transaction.

Inside the condition body, branches can be added, which are guarded by Boolean conditions. A branch is considered for execution if its condition is true and the called methods can be run. A catch-all, default branch can be added, which can be executed only if none of the other branches execute. The condition of the default branch is the negated alternative of all the other conditions.

Parameters

mTModule: A module where the condition is defined.
nonblockingbool: States that the condition should not block the containing method or transaction from running, even when none of the branch conditions is true. In case of a blocking method call, the containing method or transaction is still blocked.
prioritybool: States that when conditions are not mutually exclusive and multiple branches could be executed, the first one will be selected. This influences the scheduling order of generated transactions.

Examples

with condition(m) as branch:
    with branch(cond1):
        ...
    with branch(cond2):
        ...
    with branch():  # default, optional
        ...

transactron.lib.storage module

class transactron.lib.storage.AsyncMemoryBank

Bases: Elaboratable

AsyncMemoryBank module.

Provides a transactional interface to asynchronous Amaranth Memory with arbitrary number of read and write ports. It supports optionally writing with given granularity.

Attributes

reads: list[Method]: The read methods, one for each read port. Accepts an addr from which data should be read. The read response method. Return data_layout View which was saved on addr given by last write method call.
writes: list[Method]: The write methods, one for each write port. Accepts write address addr, data in form of data_layout and optionally mask if granularity is not None. 1 in mask means that appropriate part should be written.
read: Method: The only method from reads, if the memory has a single read port.
write: Method: The only method from writes, if the memory has a single write port.

__init__(*, data_layout: list[tuple[str, 'ShapeLike | LayoutList']], elem_count: int, granularity: ~typing.Optional[int] = None, read_ports: int = 1, write_ports: int = 1, memory_type: ~amaranth_types.memory.AbstractMemoryConstructor[int, ~amaranth.hdl._ast.Value] = <class 'amaranth.lib.memory.Memory'>, src_loc: int | tuple[str, int] = 0)

Parameters

data_layout: method layout: The format of structures stored in the Memory.
elem_count: int: Number of elements stored in Memory.
granularity: Optional[int]: Granularity of write, forwarded to Amaranth. If None the whole structure is always saved at once. If not, the width of data_layout is split into granularity parts, which can be saved independently.
read_ports: int: Number of read ports.
write_ports: int: Number of write ports.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.storage.ContentAddressableMemory

Bases: Elaboratable

Content addresable memory

This module implements a content-addressable memory (in short CAM) with Transactron interface. CAM is a type of memory where instead of predefined indexes there are used values fed in runtime as keys (similar as in python dictionary). To insert new entry a pair (key, value) has to be provided. Such pair takes an free slot which depends on internal implementation. To read value a key has to be provided. It is compared with every valid key stored in CAM. If there is a hit, a value is read. There can be many instances of the same key in CAM. In such case it is undefined which value will be read.

Warning

Pushing the value with index already present in CAM is an undefined behaviour.

Attributes

readMethod: Nondestructive read
writeMethod: If index present - do update
removeMethod: Remove
pushMethod: Inserts new data.

__init__(address_layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], data_layout: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], entries_number: int)

Parameters

address_layoutLayoutLike: The layout of the address records.
data_layoutLayoutLike: The layout of the data.
entries_numberint: The number of slots to create in memory.

class transactron.lib.storage.MemoryBank

Bases: Elaboratable

MemoryBank module.

Provides a transactional interface to synchronous Amaranth Memory with arbitrary number of read and write ports. It supports optionally writing with given granularity.

Attributes

read_reqs: list[Method]: The read request methods, one for each read port. Accepts an addr from which data should be read. Only ready if there is there is a place to buffer response. After calling read_reqs[i], the result will be available via the method read_resps[i].
read_resps: list[Method]: The read response methods, one for each read port. Return data_layout View which was saved on addr given by last corresponding read_reqs method call. Only ready after corresponding read_reqs call.
writes: list[Method]: The write methods, one for each write port. Accepts write address addr, data in form of data_layout and optionally mask if granularity is not None. 1 in mask means that appropriate part should be written.
read_req: Method: The only method from read_reqs, if the memory has a single read port. If it has more ports, this method is unavailable and read_reqs should be used instead.
read_resp: Method: The only method from read_resps, if the memory has a single read port. If it has more ports, this method is unavailable and read_resps should be used instead.
write: Method: The only method from writes, if the memory has a single write port. If it has more ports, this method is unavailable and writes should be used instead.

__init__(*, data_layout: list[tuple[str, 'ShapeLike | LayoutList']], elem_count: int, granularity: ~typing.Optional[int] = None, transparent: bool = False, read_ports: int = 1, write_ports: int = 1, memory_type: ~amaranth_types.memory.AbstractMemoryConstructor[int, ~amaranth.hdl._ast.Value] = <class 'amaranth.lib.memory.Memory'>, src_loc: int | tuple[str, int] = 0)

Parameters

data_layout: method layout: The format of structures stored in the Memory.
elem_count: int: Number of elements stored in Memory.
granularity: Optional[int]: Granularity of write, forwarded to Amaranth. If None the whole structure is always saved at once. If not, the width of data_layout is split into granularity parts, which can be saved independently.
transparent: bool: Read port transparency, false by default. When a read port is transparent, if a given memory address is read and written in the same clock cycle, the read returns the written value instead of the value which was in the memory in that cycle.
read_ports: int: Number of read ports.
write_ports: int: Number of write ports.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

transactron.lib.transformers module

class transactron.lib.transformers.CatTrans

Bases: Elaboratable

Concatenating transaction.

Concatenates the results of two methods and passes the result to the third method.

__init__(src1: Method, src2: Method, dst: Method)

Parameters

src1: Method: First input method.
src2: Method: Second input method.
dst: Method: The method which receives the concatenation of the results of input methods.

class transactron.lib.transformers.Collector

Bases: Elaboratable, Unifier

Single result collector.

Creates method that collects results of many methods with identical layouts. Each call of this method will return a single result of one of the provided methods.

Attributes

method: Method: Method which returns single result of provided methods.

__init__(targets: list[transactron.core.method.Method], *, src_loc: int | tuple[str, int] = 0)

Parameters

method_list: list[Method]: List of methods from which results will be collected.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.transformers.ConnectAndMapTrans

Bases: Elaboratable

Connecting transaction with mapping functions.

Behaves like ConnectTrans, but modifies the transferred data using functions or Method`s. Equivalent to a combination of `ConnectTrans and MethodMap. The mapping functions take two parameters, a Module and the structure being transformed.

__init__(method1: Method, method2: Method, *, i_fun: Optional[Callable[[TModule, View[StructLayout]], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable | collections.abc.Mapping[str, RecordDict]]] = None, o_fun: Optional[Callable[[TModule, View[StructLayout]], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable | collections.abc.Mapping[str, RecordDict]]] = None, src_loc: int | tuple[str, int] = 0)

Parameters

method1: Method: First method.
method2: Method: Second method, and the method being transformed.
i_fun: function or Method, optional: Input transformation (method1 to method2).
o_fun: function or Method, optional: Output transformation (method2 to method1).
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.transformers.MethodFilter

Bases: Elaboratable, Transformer

Method filter.

Takes a target method and creates a method which calls the target method only when some condition is true. The condition function takes two parameters, a module and the input structure of the method. Non-zero return value is interpreted as true. Alternatively to using a function, a Method can be passed as a condition. By default, the target method is locked for use even if it is not called. If this is not the desired effect, set use_condition to True, but this will cause that the provided method will be single_caller and all other condition drawbacks will be in place (e.g. risk of exponential complexity).

Attributes

method: Method: The transformed method.

__init__(target: Method, condition: Callable[[TModule, View[StructLayout]], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable], default: Optional[Union[Value, int, Enum, ValueCastable, Mapping[str, RecordDict]]] = None, *, use_condition: bool = False, src_loc: int | tuple[str, int] = 0)

Parameters

target: Method: The target method.
condition: function or Method: The condition which, when true, allows the call to target. When false, default is returned.
default: Value or dict, optional: The default value returned from the filtered method when the condition is false. If omitted, zero is returned.
use_conditionbool: Instead of m.If use simultaneus condition which allow to execute this filter if the condition is False and target is not ready. When use_condition is true, condition must not be a Method.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.transformers.MethodMap

Bases: Elaboratable, Transformer

Bidirectional map for methods.

Takes a target method and creates a transformed method which calls the original target method, mapping the input and output values with functions. The mapping functions take two parameters, a Module and the structure being transformed. Alternatively, a Method can be passed.

Attributes

method: Method: The transformed method.

__init__(target: Method, *, i_transform: Optional[tuple[amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], collections.abc.Callable[[transactron.core.tmodule.TModule, 'View[StructLayout]'], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable | collections.abc.Mapping[str, 'RecordDict']]]] = None, o_transform: Optional[tuple[amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], collections.abc.Callable[[transactron.core.tmodule.TModule, 'View[StructLayout]'], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable | collections.abc.Mapping[str, 'RecordDict']]]] = None, src_loc: int | tuple[str, int] = 0)

Parameters

target: Method: The target method.
i_transform: (method layout, function or Method), optional: Input mapping function. If specified, it should be a pair of a function and a input layout for the transformed method. If not present, input is passed unmodified.
o_transform: (method layout, function or Method), optional: Output mapping function. If specified, it should be a pair of a function and a output layout for the transformed method. If not present, output is passed unmodified.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

class transactron.lib.transformers.MethodProduct

Bases: Elaboratable, Unifier

__init__(targets: list[transactron.core.method.Method], combiner: Optional[tuple[amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], collections.abc.Callable[[transactron.core.tmodule.TModule, list['View[StructLayout]']], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable | collections.abc.Mapping[str, 'RecordDict']]]] = None, *, src_loc: int | tuple[str, int] = 0)

Method product.

Takes arbitrary, non-zero number of target methods, and constructs a method which calls all of the target methods using the same argument. The return value of the resulting method is, by default, the return value of the first of the target methods. A combiner function can be passed, which can compute the return value from the results of every target method.

Parameters

targets: list[Method]: A list of methods to be called.
combiner: (int or method layout, function), optional: A pair of the output layout and the combiner function. The combiner function takes two parameters: a Module and a list of outputs of the target methods.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

Attributes

method: Method: The product method.

class transactron.lib.transformers.MethodTryProduct

Bases: Elaboratable, Unifier

__init__(targets: list[transactron.core.method.Method], combiner: Optional[tuple[amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']], collections.abc.Callable[[transactron.core.tmodule.TModule, list[tuple[amaranth.hdl._ast.Value, 'View[StructLayout]']]], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable | collections.abc.Mapping[str, 'RecordDict']]]] = None, *, src_loc: int | tuple[str, int] = 0)

Method product with optional calling.

Takes arbitrary, non-zero number of target methods, and constructs a method which tries to call all of the target methods using the same argument. The methods which are not ready are not called. The return value of the resulting method is, by default, empty. A combiner function can be passed, which can compute the return value from the results of every target method.

Parameters

targets: list[Method]: A list of methods to be called.
combiner: (int or method layout, function), optional: A pair of the output layout and the combiner function. The combiner function takes two parameters: a Module and a list of pairs. Each pair contains a bit which signals that a given call succeeded, and the result of the call.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

Attributes

method: Method: The product method.

class transactron.lib.transformers.Transformer

Bases: HasElaborate, Protocol

Method transformer abstract class.

Method transformers construct a new method which utilizes other methods.

Attributes

method: Method: The method.

method: Method

use(m: ModuleLike)

Returns the method and adds the transformer to a module.

Parameters

m: Module or TModule: The module to which this transformer is added as a submodule.

class transactron.lib.transformers.Unifier

Bases: Transformer, Protocol

__init__(targets: list[transactron.core.method.Method])

method: Method

transactron.lib package

Submodules

transactron.lib.adapters module

transactron.lib.buttons module

transactron.lib.connectors module

transactron.lib.dependencies module

transactron.lib.fifo module

transactron.lib.logging module

transactron.lib.metrics module

transactron.lib.reqres module

transactron.lib.simultaneous module

transactron.lib.storage module

transactron.lib.transformers module

Module contents