I2C FIFO Target¶

The PxI2CFifoTarget class emulates an I2C target with host-accessible RX and TX FIFOs. A controller writes data on the bus into the RX FIFO (delivered to Python as events); Python enqueues data into the TX FIFO for the controller to read on the bus.

FIFO depth is fixed at 4 KB and is not configurable from Python. A single enqueue call may transfer up to the available TX FIFO capacity. Each receive event delivers up to 4 KB of data; each sent event reports the byte count drained in a bus read.

Data flow¶

Controller WRITE  →  RX FIFO  →  fifo_target_recv events  →  Python
Python enqueue    →  TX FIFO  →  Controller READ
Controller READ   →  (drains TX FIFO)  →  fifo_target_sent events  →  Python

Quick start¶

Attach a target to an I2C bus and enqueue TX data:

from aqpxlib import AqProtocolExerciser
from aqpxlib.i2c import PxI2CBus, PxI2CFifoTarget

with AqProtocolExerciser.connect(port=60600) as px:
    i2c_bus = PxI2CBus(px, scl=0, sda=1)
    target = PxI2CFifoTarget(px, address=0x50)
    target.attach_to_bus(i2c_bus)

    target.enqueue([0x01, 0x02, 0x03])

    state = target.state
    print(f"TX fill: {state.tx_fill}")
    print(f"Total read: {state.total_read}")
    print(f"Total written: {state.total_written}")

    target.detach_from_bus()

Enqueue (host → bus)¶

Use enqueue to load the TX FIFO. This does not perform an I2C transaction itself. Use a PxI2CController on the same bus to read the enqueued bytes.

target.enqueue(b"\xde\xad\xbe\xef")

Bus receive events¶

When a controller writes to the target, the exerciser emits fifo_target_recv events. Each event carries a data field with the bytes received in that event packet, and a last flag that marks the final event for the current bus write transaction (STOP or repeated-START).

Field	Type	Description
`data`	`bytes`	Bytes received in this event
`last`	`bool`	`True` when this is the last event for the current bus write

from aqpxlib.event import PxFifoTargetRecvEvent

@target.on_event(PxFifoTargetRecvEvent)
def on_recv(recv_event: PxFifoTargetRecvEvent):
    print(f"Recv {recv_event.data.hex()} last={recv_event.last}")

Bus sent events¶

When a controller reads from the target, the exerciser emits fifo_target_sent events. Each event reports how many bytes were drained from the TX FIFO in the completed bus read, with last marking the end of that read transaction.

Field	Type	Description
`count`	`int`	Bytes sent on the bus in this read transaction
`last`	`bool`	`True` when this is the last event for the current bus read

from aqpxlib.event import PxFifoTargetSentEvent

@target.on_event(PxFifoTargetSentEvent)
def on_sent(sent_event: PxFifoTargetSentEvent):
    print(f"Sent {sent_event.count} bytes last={sent_event.last}")

enqueue and state queries do not emit recv or sent events.

State¶

Query the current FIFO state via target.state:

Property	Description
`tx_fill`	Bytes currently queued in the TX FIFO
`total_read`	Cumulative bytes read by the controller since configure
`total_written`	Cumulative bytes written by the controller since configure

state = target.state
assert state.tx_fill >= 0
assert state.total_read >= 0
assert state.total_written >= 0

Controller access¶

Use a PxI2CController to exercise the target over the bus:

from aqpxlib import AqProtocolExerciser
from aqpxlib.event import PxFifoTargetRecvEvent
from aqpxlib.i2c import PxI2CBus, PxI2CController, PxI2CFifoTarget

with AqProtocolExerciser.connect(port=60600) as px:
    i2c_bus = PxI2CBus(px, scl=0, sda=1)
    target = PxI2CFifoTarget(px, address=0x50)
    target.attach_to_bus(i2c_bus)

    controller = PxI2CController(px)
    controller.attach_to_bus(i2c_bus)

    received: list[int] = []

    @target.on_event(PxFifoTargetRecvEvent)
    def on_recv(recv_event: PxFifoTargetRecvEvent) -> None:
        received.extend(recv_event.data)

    target.enqueue([0xAA, 0xBB, 0xCC])
    read_back = controller.i2c_read(target.address, 3)
    assert read_back == [0xAA, 0xBB, 0xCC]

    controller.i2c_write(target.address, [0x11, 0x22])
    assert received == [0x11, 0x22]

    controller.detach_from_bus()
    target.detach_from_bus()

I3C bus support¶

PxI2CFifoTarget may be attached to an I3C bus (valid_buses includes "i3c" and "i2c"). The device behaves as an I2C-style target on the I3C SDA line.

API reference¶

PxI2CFifoTarget ¶

PxI2CFifoTarget(
    exerciser: AqProtocolExerciser,
    address: int = 0,
    config: PxI2CFifoTargetConfig | None = None,
    *args,
    **kwargs
)

Bases: PxAbstractTarget[PxI2CFifoTargetConfig, PxI2CFifoTargetState]

I2C FIFO target.

Emulates an I2C target with host-accessible RX and TX FIFOs. Bytes written by a controller on the bus are delivered to Python as packeted fifo_target_recv events. Bytes enqueued from Python are presented on subsequent bus reads; completed bus reads are reported as fifo_target_sent events with the byte count drained in each transaction.

Use enqueue to load the TX FIFO. Use self.state for the current device state snapshot (tx_fill, total_read, total_written).

METHOD	DESCRIPTION
`add_event_handler`	Add an event handler to the current device instance, by event class or event name.
`remove_event_handler`	Remove an event handler from the current device instance, by event class or event name.
`get_event_handlers`	Retreive handlers registerd on this device.
`on_event`	Add an event handler to the instance by decorator, by event class or event name.
`match`	Return whether a device config describes a device of this class.
`update_additional_data`	Atomically update additional_data with revision-aware retry.
`update_config`	Set the config of the target.
`set_config`	Set the config of the target.
`attach_to_bus`	Attach the target to a bus.
`detach_from_bus`	Detach the device from the bus.
`perform_operation`	Send an operation.
`enqueue`	Enqueue bytes into the host-side TX FIFO.

ATTRIBUTE	DESCRIPTION
`available_events_map`	Map each available event name to its id in event topics. TYPE: `dict[str, str]`
`is_attached`	Check if the device is attached to a bus. TYPE: `bool`
`state`	Get the state of the device. TYPE: `DeviceStateT \| None`
`name`	Get the name of the device. TYPE: `str`
`cts_op_name`	Get the name of CTS operation. Currently not available now. TYPE: `str \| None`
`additional_data`	Get the opaque additional data blob associated with this device. TYPE: `bytes`
`config`	Get the config of the device. TYPE: `DeviceConfigT \| None`
`address`	Static 7-bit I2C address of the target. TYPE: `int`
`tx_fill`	Number of bytes currently in the TX FIFO. TYPE: `int`
`total_read`	Total bytes read by the controller from the target since configuration. TYPE: `int`
`total_written`	Total bytes written by the controller to the target since configuration. TYPE: `int`

available_events_map ¶

available_events_map: dict[str, str]

Map each available event name to its id in event topics.

is_attached ¶

is_attached: bool

Check if the device is attached to a bus.

state ¶

state: DeviceStateT | None

Get the state of the device.

name ¶

name: str

Get the name of the device.

cts_op_name ¶

cts_op_name: str | None

Get the name of CTS operation. Currently not available now.

additional_data ¶

additional_data: bytes

Get the opaque additional data blob associated with this device.

config ¶

config: DeviceConfigT | None

Get the config of the device.

address ¶

address: int

Static 7-bit I2C address of the target.

tx_fill ¶

tx_fill: int

Number of bytes currently in the TX FIFO.

total_read ¶

total_read: int

Total bytes read by the controller from the target since configuration.

total_written ¶

total_written: int

Total bytes written by the controller to the target since configuration.

add_event_handler ¶

add_event_handler(event: str, handler: Callable[[Any], Any]) -> None

add_event_handler(event: type[EventT], handler: Callable[[EventT], Any]) -> None

add_event_handler(
    event: str | type[PxAbstractEvent], handler: Callable[[Any], Any]
) -> None

Add an event handler to the current device instance, by event class or event name.

remove_event_handler ¶

remove_event_handler(
    event: str | type[PxAbstractEvent], handler: Callable[[Any], Any] | None = None
) -> None

Remove an event handler from the current device instance, by event class or event name.

get_event_handlers ¶

get_event_handlers(
    filter_by: Callable[[str], bool] | None = None,
) -> dict[str, list[Callable[[Any], Any]]]

Retreive handlers registerd on this device.

Handlers are always scoped to this device; an optional filter_by is applied on friendly event names.

on_event ¶

on_event(event_type: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]

on_event(
    event_type: type[EventT],
) -> Callable[[Callable[[EventT], Any]], Callable[[EventT], Any]]

on_event(
    event_type: str | type[PxAbstractEvent],
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Add an event handler to the instance by decorator, by event class or event name.

match ¶

match(config: PxAbstractDeviceConfig) -> bool

Return whether a device config describes a device of this class.

update_additional_data ¶

update_additional_data(updater: Callable[[bytes], bytes]) -> None

Atomically update additional_data with revision-aware retry.

update_config ¶

update_config(updater: Callable[[Any], Any]) -> None

Set the config of the target.

set_config ¶

set_config(config: DeviceConfigT) -> None

Set the config of the target.

attach_to_bus ¶

attach_to_bus(bus: PxAbstractBus) -> None

Attach the target to a bus.

This method is used to make the device aware of the bus it is connected to. It is used to set the bus attribute of the device.

Note

This method will not automatically create an actual instance on the Protocol Exerciser until user call the attach_to_bus method.

PARAMETER	DESCRIPTION
`bus` ¶	The bus to attach the device to TYPE: `PxAbstractBus`

RAISES	DESCRIPTION
`ValueError`	If the device is already attached to a bus

detach_from_bus ¶

detach_from_bus() -> None

Detach the device from the bus.

The opposite of the attach_to_bus method.

perform_operation ¶

perform_operation(
    operation: PxDeviceOperation, *, timeout: int | None = None
) -> PxDeviceOperation

Send an operation.

PARAMETER	DESCRIPTION
`operation` ¶	Device operation to perform. TYPE: `PxDeviceOperation`
`timeout` ¶	Optional base STOP-handshake wait in milliseconds. When omitted, the instrument uses subsystem defaults. TYPE: `int \| None` DEFAULT: `None`

enqueue ¶

enqueue(data: list[int] | bytes) -> None

Enqueue bytes into the host-side TX FIFO.

This does not perform an I2C bus transaction. Enqueued bytes are returned to a controller on subsequent bus read transfers.

PARAMETER	DESCRIPTION
`data` ¶	Bytes to enqueue (`0`-`255` per byte). May exceed 4 KB in a single call; size is limited only by available TX FIFO space. TYPE: `list[int] \| bytes`

I2C FIFO Target¶

Data flow¶

Quick start¶

Enqueue (host → bus)¶

Bus receive events¶

Bus sent events¶

State¶

Controller access¶

I3C bus support¶

API reference¶

PxI2CFifoTarget ¶

available_events_map ¶

is_attached ¶

state ¶

name ¶

cts_op_name ¶

additional_data ¶

config ¶

address ¶

tx_fill ¶

total_read ¶

total_written ¶

add_event_handler ¶

remove_event_handler ¶

get_event_handlers ¶

on_event ¶

match ¶

update_additional_data ¶

update_config ¶

set_config ¶

attach_to_bus ¶

`bus` ¶

detach_from_bus ¶

perform_operation ¶

`operation` ¶

`timeout` ¶

enqueue ¶

`data` ¶

I2C FIFO Target¶

Data flow¶

Quick start¶

Enqueue (host → bus)¶

Bus receive events¶

Bus sent events¶

State¶

Controller access¶

I3C bus support¶

API reference¶

PxI2CFifoTarget ¶

available_events_map ¶

is_attached ¶

state ¶

name ¶

cts_op_name ¶

additional_data ¶

config ¶

address ¶

tx_fill ¶

total_read ¶

total_written ¶

add_event_handler ¶

remove_event_handler ¶

get_event_handlers ¶

on_event ¶

match ¶

update_additional_data ¶

update_config ¶

set_config ¶

attach_to_bus ¶

bus ¶

detach_from_bus ¶

perform_operation ¶

operation ¶

timeout ¶

enqueue ¶

data ¶

`bus` ¶

`operation` ¶

`timeout` ¶

`data` ¶