示例:事件注册与处理函数设置

除了上述基本操作，AQPXLIB 也支持订阅 Protocol Exerciser 送出的事件。

示例代码与配置

示例如下。

run_i3c_event_example.py

import logging
import threading

from aqpxlib import AqProtocolExerciser
from aqpxlib.i3c import (
    ACUTE_DEFAULT_PID,
    PxI3CBus,
    PxI3CBaseController,
    PxI3C32BitRegisterTarget,
)
from aqpxlib.event import PxI3CTargetAssignedAddressEvent
from aqpxlib.exceptions import PxAPIError

logger = logging.getLogger(__name__)

def exec_i3c_event_example():
    """I3C 事件注册与处理函数设置的基本示例.

    步骤按下列顺序执行:

    1. 创建并打开与 Acute Protocol Exerciser 的连接.
    2. 配置总线上正确的电压电平与上拉电阻.
    3. 创建 I3C 总线句柄接口.
    4. 将 I3C 控制器与目标器件挂载到总线.
    5. 订阅目标器件的事件.
    6. 执行 Dynamic Address Assignment(此时会收到已分配地址事件)
    """

    with AqProtocolExerciser.connect(port=60600) as px_session:
        logger.info("Created an Exerciser session")

        # 在 SCL 引脚 0、SDA 引脚 1 上配置 I3C 总线
        i3c_bus = PxI3CBus(px_session, scl=0, sda=1)
        logger.info(f"Created {i3c_bus}")

        # 将电压设为 3.3V,上拉电阻设为 1.5kΩ
        i3c_bus.voltage = 3300
        i3c_bus.pullup_resistance = 1500

        # 创建控制器器件
        i3c_controller = PxI3CBaseController(px_session, name="i3c_controller")
        try:
            i3c_controller.attach_to_bus(i3c_bus)
        except PxAPIError:
            logger.warning("Multiple controller is not supported right now, using the existing controller")
            i3c_controller = i3c_bus.get_current_controller()
        logger.info(f"Added a I3C controller with name {i3c_controller.name}")

        # 创建目标器件
        bcr = 0x00  # Bus Characteristic Register
        dcr = 0x00  # Device Characteristic Register
        pid = ACUTE_DEFAULT_PID | (0x01 << 32) | 0 # Provisioned ID

        i3c_target = PxI3C32BitRegisterTarget(
            px_session,
            name="i3c_target",
            static_addr=0x12,
            sub_address_type=PxI3C32BitRegisterTarget.SubAddressType.NONE,
            bcr=bcr,
            dcr=dcr,
            pid=pid,
        )
        i3c_target.attach_to_bus(i3c_bus)
        logger.info(f"Added a I3C 32-bit register target with name {i3c_target.name}")

        # 订阅目标器件的已分配地址事件
        event = threading.Event()
        @i3c_target.on_event(PxI3CTargetAssignedAddressEvent)
        def i3c_target_assigned_address_handler(assigned: PxI3CTargetAssignedAddressEvent):
            """目标器件已分配地址事件的处理函数."""

            assigned_address = assigned.assigned_address
            logger.info(f"I3C target assigned address: 0x{assigned_address:02x}")
            event.set()

        # 这里直接执行 DAA,会为目标器件分配动态地址,
        # 因此已订阅的处理函数应被调用.
        i3c_controller.do_daa()
        logger.info("Controller performed DAA")

        # 确认我们已收到事件
        assert event.wait(timeout=3), "Event has never been called"

        # 将器件从总线卸载
        i3c_target.detach_from_bus()
        i3c_controller.detach_from_bus()
        logger.info(f"Detached the controller and target from the bus")


if __name__ == "__main__":
    logging.basicConfig(format='[%(levelname)s] %(message)s')
    logger.setLevel(logging.INFO)
    exec_i3c_event_example()

在终端执行下列命令:

$ python run_i3c_event_example.py

事件注册装饰器

总线与器件的配置与前一个示例相同，此处不再重复。主要差异是现在订阅目标器件的已分配地址事件。

以下说明如何用装饰器注册事件。

from aqpxlib.event import PxI3CTargetAssignedAddressEvent

# 订阅目标器件的已分配地址事件
event = threading.Event()
@i3c_target.on_event(PxI3CTargetAssignedAddressEvent)
def i3c_target_assigned_address_handler(assigned: PxI3CTargetAssignedAddressEvent):
    """目标器件已分配地址事件的处理函数."""

    assigned_address = assigned.assigned_address
    logger.info(f"I3C target assigned address: 0x{assigned_address:02x}")
    event.set()

# 这里直接执行 DAA,会为目标器件分配动态地址,
# 因此已订阅的处理函数应被调用.
i3c_controller.do_daa()
logger.info("Controller performed DAA")

# 确认我们已收到事件
assert event.wait(timeout=3), "Event has never been called"

执行 DAA 时，目标器件会收到控制器分配的地址。此时 Protocol Exerciser 会立即通知订阅者地址已分配。

上方示例以装饰器风格编写，你也可以用目标类的 add_event_handler 以声明方式注册。

示例最终输出应类似:

[INFO] Created an Exerciser session
[INFO] Created I3C Bus (SCL: 0, SDA: 1)
[INFO] Added a I3C controller with name i3c_controller
[INFO] Added a I3C 32-bit register target with name i3c_target
[INFO] Controller performed DAA
[INFO] I3C target assigned address: 0x08
[INFO] Detached the controller and target from the bus

这里可以看到目标器件已被分配地址 0x08.

事件类型

使用 aqpxlib.event 中的事件类注册处理函数，处理函数会收到该类的对象。设备事件另带有触发事件的设备的 device_id。第二栏的事件名称仍可代替事件类使用。

事件类	事件名称	字段
PxI3CTargetAssignedAddressEvent	i3_c_target_assigned_address	assigned_address
PxI3CTargetUpdatedEventStateEvent	i3_c_target_updated_event_state	event_state
PxI3CTargetAssignedGroupAddressEvent	i3_c_target_assigned_group_address	assigned_group_address
PxI3CControllerRecvIbiDataEvent	i3_c_controller_recv_ibi_data	address, is_ack, data
PxMemoryTargetReadEvent	memory_target_read_event	addr, count
PxMemoryTargetWriteEvent	memory_target_write_event	addr, data
PxFifoTargetRecvEvent	fifo_target_recv	data, last
PxFifoTargetSentEvent	fifo_target_sent	count, last
PxUartRxDataEvent	uart_rx_data	data, has_parity_error
PxGpioTriggerEvent	gpio_trigger	pin_idx, pulse_duration, pulse_polarity
PxUpdatedTopologyEvent	updated_topology	handle_idx, revision, config

PxUpdatedTopologyEvent.config 为该 handle 的总线配置；handle 槽位被清除时为 None。

在 attach 之前注册处理函数

器件事件处理函数可以在 attach_to_bus() 之前注册。pxlib 会将其保存在逻辑注册表中，仅在器件获得总线 device_id 后才向 Protocol Exerciser 订阅。

from aqpxlib.event import PxI3CTargetAssignedAddressEvent

i3c_target = PxI3C32BitRegisterTarget(px_session, name="i3c_target", ...)
i3c_target.add_event_handler(PxI3CTargetAssignedAddressEvent, handler)
i3c_target.attach_to_bus(i3c_bus)  # 在此完成订阅

调用 detach_from_bus() 会注销当前 device_id 下的 exerciser 订阅。再次调用 attach_to_bus() 会在新分配的 ID 下恢复订阅。remove_event_handler() 接受与 add_event_handler() 相同的事件类(或事件名称)。

全局事件(GPIO 触发)

上方器件示例使用器件范围事件(例如 @i3c_target.on_event(...))。有些通知是全局的: 它们不绑定总线器件 ID.把事件类传给 AqProtocolExerciser.add_event_handler 来注册它们。

当你用 set_gpio_trigger 配置的 GPIO 脉冲宽度触发符合条件时，会发出 gpio_trigger 全局事件。请在 AqProtocolExerciser 实例上注册处理函数，而不是在器件上:

from aqpxlib.event import PxGpioTriggerEvent

def on_gpio_trigger(info: PxGpioTriggerEvent) -> None:
    # info.pin_idx, info.pulse_duration, info.pulse_polarity
    ...

px_session.add_event_handler(PxGpioTriggerEvent, on_gpio_trigger)

一般 GPIO 读取、驱动与脉冲 API 请见 GPIO 读取、驱动与脉冲。引脚限制、持续时间范围、触发模式与完整走查(含独立脉冲示例)请见 GPIO 脉冲宽度触发。