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 读取已入队的字节。
总线接收事件¶
当控制器写入目标时,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. |
is_attached |
Check if the device is attached to a bus.
TYPE:
|
state |
Get the state of the device.
TYPE:
|
name |
Get the name of the device.
TYPE:
|
cts_op_name |
Get the name of CTS operation. Currently not available now.
TYPE:
|
additional_data |
Get the opaque additional data blob associated with this device.
TYPE:
|
config |
Get the config of the device.
TYPE:
|
address |
Static 7-bit I2C address of the target.
TYPE:
|
tx_fill |
Number of bytes currently in the TX FIFO.
TYPE:
|
total_read |
Total bytes read by the controller from the target since configuration.
TYPE:
|
total_written |
Total bytes written by the controller to the target since configuration.
TYPE:
|
available_events_map
¶
Map each available event name to its id in event topics.
additional_data
¶
additional_data: bytes
Get the opaque additional data blob associated with this device.
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 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 | 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
¶
Atomically update additional_data with revision-aware retry.
attach_to_bus
¶
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 |
|---|---|
|
The bus to attach the device to
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If the device is already attached to a bus |
detach_from_bus
¶
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
enqueue
¶
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 |
|---|---|
|
Bytes to enqueue ( |