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

除了上述基本操作，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 PxEventMsg
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("i3_c_target_assigned_address")
        def i3c_target_assigned_address_handler(event_msg: PxEventMsg):
            """目标器件已分配地址事件的处理函数."""

            assigned_address = event_msg.device_event.i3_c_target_assigned_address.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

事件注册装饰器

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

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

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

    assigned_address = event_msg.device_event.i3_c_target_assigned_address.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.

全局事件(GPIO 触发)

上方器件示例使用器件范围事件(例如 @i3c_target.on_event(...))。有些通知是全局的: 它们不绑定总线器件 ID.Exerciser 以单一逻辑名称发布这些事件，你把它传给 AqProtocolExerciser.add_event_handler。

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

from aqpxlib.event import PxEventMsg

def on_gpio_trigger(event_msg: PxEventMsg) -> None:
    info = event_msg.global_event.gpio_trigger
    # info.pin_idx, info.pulse_duration, info.pulse_polarity
    ...

px_session.add_event_handler("gpio_trigger", on_gpio_trigger)

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