transactron.evlog package

Submodules

transactron.evlog.consumer module

class transactron.evlog.consumer.EventConsumer

Bases: object

Base class for event log consumers.

Subclasses implement one handles-decorated method per consumed event type, e.g. a converter to a trace format:

class KonataConverter(EventConsumer):
    @handles(InsnStageEnter)
    def on_stage_enter(self, rec: DecodedEvent): ...

Handlers are dispatched by the registered event name, so the consumer works with logs decoded from any capture backend. Events without a handler are passed to on_unhandled, which does nothing by default.

dispatch(rec: DecodedEvent): Calls the handler registered for the record’s event type.

on_unhandled(rec: DecodedEvent): Called for events without a registered handler.

run(records: Iterable[DecodedEvent])

Dispatches all records, sorted by cycle.

Sorting makes the consumer independent of the capture order, which matters for consumers of formats requiring monotonic timestamps.

transactron.evlog.consumer.handles(event_type: type[Event])

Marks an EventConsumer method as the handler for an event type.

The decorated method is called with a single DecodedEvent argument for every log record of the given event type.

transactron.evlog.emit module

class transactron.evlog.emit.EmittedEvent

Bases: object

A single event emission site registered during elaboration.

Attributes:

source_name: str: Name of the EventSource which registered this site.
event_type: type[Event]: The event type emitted at this site.
location: SrcLoc: Source location of the emission site.
trigger: Value: Single-bit Amaranth value. The event is recorded in every cycle in which the trigger is high.
fields: dict[str, Value]: Amaranth values backing the dynamic fields.
statics: dict[str, Any]: JSON-serializable values of the static fields.

__init__(source_name: str, event_type: type[Event], location: tuple[str, int], trigger: Value, fields: dict[str, Value], statics: dict[str, Any]) → None

event_type: type[Event]

fields: dict[str, Value]

location: tuple[str, int]

source_name: str

statics: dict[str, Any]

trigger: Value

class transactron.evlog.emit.EvLogEnabledKey

Bases: SimpleKey[bool]

DependencyManager key for enabling the event log. If the event log is disabled (the default), EventSource.emit is a no-op and no signals are synthesized.

__init__() → None

default_value: T = False

empty_valid: bool = True

class transactron.evlog.emit.EvLogKey

Bases: ListKey[EmittedEvent]

DependencyManager key collecting all event emission sites.

__init__() → None

class transactron.evlog.emit.EventSource

Bases: object

Emission point for structured hardware events.

An EventSource emits typed events (see Event) from the hardware description. Events are captured during simulation into an event log, which can be consumed by downstream tools (trace converters, statistics scripts) without touching the hardware description again.

Like HardwareLogger, event sources are identified by a hierarchical dotted name, e.g. “frontend.ftq”. Emitting the same event type multiple times (e.g. once per superscalar lane) is allowed; each emit call creates a separate emission site, typically distinguished by a static field. Usage:

self.evlog = EventSource("exec.alu")
...
self.evlog.emit(m, ExecUnitStart.hw(insn_tag=tag, lane=0), when=start)

Attributes:

name: str: Name of this event source.

__init__(name: str)

Parameters:

name: str: Name of this event source. Hierarchy levels are separated by periods, e.g. “frontend.ftq”.

emit(m: ModuleLike, ev: Event, *, when: Value | int | Enum | ValueCastable = 1, src_loc: int | tuple[str, int] = 0)

Registers an event emission site.

The event is recorded in every cycle in which when is true and the surrounding module context (m.If etc., transaction or method body) is active.

Parameters:

m: ModuleLike: The module in which the event is emitted.
ev: Event: The emitted event. Dynamic fields must be Amaranth values; static fields must be plain Python values.
when: ValueLike: The event is recorded in cycles in which this Amaranth expression is true. Defaults to always (gated only by the module context).
src_loc: int | SrcLoc, optional: How many stack frames below to look for the source location.

top_emit(ev: Event, *, when: Value | int | Enum | ValueCastable = 1, src_loc: int | tuple[str, int] = 0)

Registers an event emission site.

Unlike EventSource.emit, this function ignores m.If etc. and can be used in contexts where a module is not available.

See EventSource.emit for details.

transactron.evlog.emit.evlog_enabled() → bool: Checks if event logging is enabled in the current dependency context.

transactron.evlog.emit.get_emitted_events() → list[EmittedEvent]: Returns all event emission sites registered during elaboration, in registration order.

transactron.evlog.event module

class transactron.evlog.event.Event

Bases: object

Base class for event types.

Concrete event types subclass Event and are declared using the event decorator, which turns them into dataclasses. Fields annotated with Static[…] are static; all other fields are dynamic and are backed by hardware signals sampled during simulation.

The same class is used in two contexts:

during elaboration, instances are constructed with Amaranth values for the dynamic fields and passed to EventSource.emit;
when reading a captured event log, decoded instances carry concrete Python values.

Dynamic fields should be annotated with int, bool or an enum.Enum subclass; raw integers are converted according to the annotation when an event log is decoded.

event_name: ClassVar[str]: Unique dotted name of the event type, e.g. “fetch.block_started”.

classmethod from_raw(dynamics: Mapping[str, int], statics: Mapping[str, Any]) → Self

Reconstructs a typed event instance from raw field values.

Parameters:

dynamics: Mapping[str, int]: Raw values of the dynamic fields, as sampled from hardware signals.
statics: Mapping[str, Any]: Raw values of the static fields, as stored in the event schema.

classmethod hw(**fields: Any) → Self

Constructs an event instance for emission from hardware.

Behaves exactly like the regular constructor, but accepts arbitrary values, as during elaboration the dynamic fields hold Amaranth values instead of the declared (decoded) field types. Field values are validated by EventSource.emit.

transactron.evlog.event.event(name: str) → Callable[[_T], _T]

Class decorator declaring a new event type.

The decorated class must subclass Event. It is converted to a dataclass and registered globally under the given name, which must be unique.

Parameters:

name: str: Unique dotted name of the event type, e.g. “fetch.block_started”. Recorded in event log schemas and used to look up the event type when a log is decoded.

transactron.evlog.event.get_event_class(name: str) → type[Event]

Looks up an event type by its registered name.

The module defining the event type must have been imported beforehand, as event types register themselves at class definition time.

transactron.evlog.log module

class transactron.evlog.log.DecodedEvent

Bases: object

A single decoded event from an event log.

Attributes:

cycle: int: The clock cycle in which the event was recorded.
site: EventSiteSchema: Schema of the emission site which produced the event.
event: Event: The typed event instance, with dynamic and static fields filled in.

__init__(cycle: int, site: EventSiteSchema, event: Event) → None

cycle: int

event: Event

site: EventSiteSchema

property source_name: str

class transactron.evlog.log.EventDecoder

Bases: object

Decodes raw event records into typed Event instances using a schema.

The modules defining the used event types must be imported beforehand, as event types are looked up in the global registry by name.

__init__(schema: EvLogSchema)

decode(cycle: int, site_idx: int, values: Sequence[int]) → DecodedEvent

class transactron.evlog.log.EventLog

Bases: object

An in-memory event log: raw event records paired with their schema.

Events are stored raw (as integers) and decoded on demand. The log is serialized in the JSON-lines format: a header line containing the schema, followed by one line per event.

Attributes:

schema: EvLogSchema: Schema used to decode the raw records.
raw: list[RawEvent]: The raw event records, in capture order.

__init__(schema: EvLogSchema)

decoded() → list[DecodedEvent]: Decodes all events, in capture order.

emit_raw(cycle: int, site: int, values: Sequence[int]) → None

static load(filename: str) → EventLog: Loads an event log from a JSON-lines file.

save(filename: str): Saves the event log to a JSON-lines file.

class transactron.evlog.log.EventLogReader

Bases: object

Streaming reader of JSON-lines event log files.

Decodes events lazily while iterating, so arbitrarily long event logs can be processed with constant memory usage.

Attributes:

schema: EvLogSchema: The schema read from the event log header.

__init__(filename: str)

class transactron.evlog.log.EventLogWriter

Bases: object

Streams raw event records directly to a JSON-lines file.

Unlike EventLog.save, records are not kept in memory, so arbitrarily long simulations can be captured. Can be used as a context manager.

__init__(filename: str, schema: EvLogSchema)

close()

emit_raw(cycle: int, site: int, values: Sequence[int]) → None

transactron.evlog.log.RawEvent(iterable=(), /)

A raw event record: (cycle, site index, dynamic field values in schema order).

alias of tuple[int, int, list[int]]

class transactron.evlog.log.RawEventSink

Bases: Protocol

Protocol for consumers of raw event records produced by samplers.

__init__(*args, **kwargs)

emit_raw(cycle: int, site: int, values: Sequence[int]) → None

transactron.evlog.sampler module

class transactron.evlog.sampler.GeneratedEvLogSampler

Bases: object

Simulator-agnostic event sampler for generated (Verilog) designs.

The sampler reads event signals through readers obtained from a HandleResolver, so the same code works with any backend able to read signals by their hierarchical location.

sample should be called once per clock cycle, at a moment when the sampled signals are stable (e.g. on the falling clock edge). If the generated design provides a packed trigger vector, only a single signal read per cycle is needed to check all emission sites.

__init__(generated: GeneratedEvLog, resolve: Callable[[list[str]], Callable[[], int]])

Parameters:

generated: GeneratedEvLog: Event log information from GenerationInfo.
resolve: HandleResolver: Backend-specific signal location resolver.

sample(cycle: int, sink: RawEventSink) → None: Samples all emission sites and reports fired events to the sink.

transactron.evlog.sampler.HandleResolver

Resolves the location of a signal in generated Verilog code to a reader of its value. Resolution happens once per signal, so the resolver can cache simulator handles.

alias of Callable[[list[str]], Callable[[], int]]

transactron.evlog.sampler.SignalReader

Reads the current value of a single signal in a running (or replayed) simulation.

alias of Callable[[], int]

transactron.evlog.schema module

class transactron.evlog.schema.EvLogSchema

Bases: object

The decoding contract between an elaborated design and event log consumers.

A schema lists all event emission sites of a single elaboration. Raw event records reference sites by their index in sites. The schema is stored together with captured event logs, so that they can be decoded offline.

Attributes:

sites: list[EventSiteSchema]: All event emission sites, in registration order.
metadata: dict[str, Any]: Free-form, JSON-serializable metadata describing the elaborated design (e.g. the core configuration), for use by downstream consumers.

__init__(sites: list[~transactron.evlog.schema.EventSiteSchema], metadata: dict[str, ~typing.Any] = <factory>) → None

classmethod from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) → A

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

metadata: dict[str, Any]

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]

sites: list[EventSiteSchema]

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

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

class transactron.evlog.schema.EventFieldSchema

Bases: object

Schema of a single dynamic event field.

Attributes:

name: str: Name of the field.
width: int: Bit width of the backing hardware signal.
signed: bool: Whether the backing hardware signal is signed.

__init__(name: str, width: int, signed: bool = False) → None

classmethod from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) → A

classmethod from_json(s: 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]

signed: bool = False

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

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

width: int

class transactron.evlog.schema.EventSiteLocation

Bases: object

Locations of one emission site’s signals in generated Verilog code.

Attributes:

trigger: SignalHandle: Location of the trigger signal.
fields: list[SignalHandle]: Locations of the dynamic field signals, in schema field order.

__init__(trigger: list[str], fields: list[list[str]]) → None

fields: list[list[str]]

classmethod from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) → A

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

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, dict | list | str | int | float | bool | None]

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

trigger: list[str]

class transactron.evlog.schema.EventSiteSchema

Bases: object

Schema of a single event emission site.

Attributes:

source_name: str: Name of the EventSource which registered this site.
event_name: str: Registered name of the emitted event type.
location: SrcLoc: Source location of the emission site.
fields: list[EventFieldSchema]: Dynamic fields, in declaration order. Raw event records store field values in this order.
statics: dict[str, Any]: Values of the static fields.

__init__(source_name: str, event_name: str, location: tuple[str, int], fields: list[EventFieldSchema], statics: dict[str, Any]) → None

event_name: str

fields: list[EventFieldSchema]

classmethod from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) → A

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

location: tuple[str, int]

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]

source_name: str

statics: dict[str, Any]

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

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

class transactron.evlog.schema.GeneratedEvLog

Bases: object

Event log information for a generated (Verilog) design.

Attributes:

schema: EvLogSchema: The event log schema.
site_locations: list[EventSiteLocation]: Signal locations for each site, indexed like schema.sites.
triggers_location: Optional[SignalHandle]: Location of a packed vector of all site triggers (bit i is the trigger of site i). Allows a sampler to check all sites with a single signal read per cycle. None if there are no sites.

__init__(schema: EvLogSchema, site_locations: list[EventSiteLocation], triggers_location: list[str] | None = None) → None

classmethod from_dict(kvs: dict | list | str | int | float | bool | None, *, infer_missing=False) → A

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

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]

site_locations: list[EventSiteLocation]

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

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

triggers_location: list[str] | None = None

transactron.evlog.schema.SignalHandle(iterable=(), /)

The location of a signal in generated Verilog code: a list of Verilog identifiers denoting a path of module names with the signal name at the end.

alias of list[str]

transactron.evlog.schema.schema_from_records(records: Iterable[EmittedEvent], metadata: dict[str, Any] | None = None) → EvLogSchema: Builds an event log schema from the emission sites registered during elaboration (see get_emitted_events).

transactron.evlog package

Submodules

transactron.evlog.consumer module

transactron.evlog.emit module

transactron.evlog.event module

transactron.evlog.log module

transactron.evlog.sampler module

transactron.evlog.schema module

Module contents