跳转至

I2C FIFO 目标

PxI2CFifoTarget 模拟具有主机可访问 RX 与 TX FIFO 的 I2C 目标器件。控制器在总线上写入的数据进入 RX FIFO(以事件交给 Python);Python 将数据入队到 TX FIFO,供控制器在总线上读取。

FIFO 深度固定为 4 KB,无法从 Python 配置。单次 enqueue 最多可传送当前可用的 TX FIFO 容量。每个 receive 事件最多交付 4 KB;每个 sent 事件回报一次总线读取抽走的字节数。

数据流

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

快速开始

将目标挂载到 I2C 总线并入队 TX 数据:

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 载入 TX FIFO。这本身不会执行 I2C 事务。请在同一条总线上使用 PxI2CController 读取已入队的字节。

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

总线接收事件

当控制器写入目标时,exerciser 会发出 fifo_target_recv 事件。每个事件带有 data 字段(该事件分组收到的字节),以及标示当前总线写入事务(STOP 或 repeated-START)最后一个事件的 last 标志。

字段 类型 说明
data bytes 此事件收到的字节
last bool 若为当前总线写入的最后一个事件则为 True
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}")

总线送出事件

当控制器从目标读取时,exerciser 会发出 fifo_target_sent 事件。每个事件回报该次完成的总线读取从 TX FIFO 抽走多少字节,last 标示该读取事务结束。

字段 类型 说明
count int 此次读取事务在总线上送出的字节数
last bool 若为当前总线读取的最后一个事件则为 True
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 与状态查询不会发出 recv 或 sent 事件。

另见: 事件注册与处理函数设置。

状态

通过 target.state 查询当前 FIFO 状态: | 属性 | 说明 | |----------|-------------| | tx_fill | TX FIFO 当前排队中的字节数 | | total_read | 自 configure 以来控制器读取的累计字节数 | | total_written | 自 configure 以来控制器写入的累计字节数 |

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

控制器访问

使用 PxI2CController 通过总线操作目标:

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 总线支持

PxI2CFifoTarget 可挂载到 I3C 总线(valid_buses 包含 "i3c" 与 "i2c")。器件在 I3C SDA 线上以 I2C 风格目标运行。

API 参考

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