Function Subscribe

src/runtime/python_runtime/aimrt_py_chn.py:149–209

Subscribe a message from a channel. Args: subscriber (aimrt_python_runtime.SubscriberRef): channel subscriber msg_type: protobuf message type or ROS2 message type callback (Callable): callback function Raises: ValueError: if the callback is invalid T

(subscriber: aimrt_python_runtime.SubscriberRef,
              msg_type: google._upb._message.MessageMeta | Ros2MsgType,
              callback: Callable)

def Subscribe(subscriber: aimrt_python_runtime.SubscriberRef,
              msg_type: google._upb._message.MessageMeta | Ros2MsgType,
              callback: Callable):
    """Subscribe a message from a channel.

    Args:
        subscriber (aimrt_python_runtime.SubscriberRef): channel subscriber
        msg_type: protobuf message type or ROS2 message type
        callback (Callable): callback function

    Raises:
        ValueError: if the callback is invalid
        TypeError: if the message type is invalid

    Callback function signature (both for Protobuf and ROS2 messages):
    - callback(msg)
    - callback(ctx, msg)
    """
    # Check callback signature
    sig = inspect.signature(callback)
    required_param_count = sum(1 for param in sig.parameters.values() if param.default == param.empty)

    if not (1 <= required_param_count <= 2):
        raise ValueError("Invalid callback: expected 1 or 2 parameters, with at most one optional parameter")

    if isinstance(msg_type, google._upb._message.MessageMeta):
        py_pb_ts = aimrt_python_runtime.PyPbTypeSupport()
        py_pb_ts.SetTypeName(GetPbMessageTypeName(msg_type))
        py_pb_ts.SetSerializationTypesSupportedList(["pb", "json"])

        def handle_callback(ctx_ref: aimrt_python_runtime.ContextRef, msg_buf: bytes):
            try:
                msg = _DeserializeProtobufMessage(msg_buf, ctx_ref.GetSerializationType(), msg_type)
                if required_param_count == 1:
                    callback(msg)
                else:
                    callback(ctx_ref, msg)
            except Exception as e:
                print(f"AimRT channel handle get exception, {e}", file=sys.stderr)

        subscriber.PbSubscribeWithCtx(py_pb_ts, handle_callback)

    elif check_is_valid_ros2_msg_type(msg_type):
        from . import aimrt_python_runtime_ros2 as aimrt_py_ros2
        py_ros2_ts = aimrt_py_ros2.PyRos2TypeSupport(msg_type)
        py_ros2_ts.SetTypeName(GetRos2MessageTypeName(msg_type))
        py_ros2_ts.SetSerializationTypesSupportedList(["ros2"])

        def ros2_callback_wrapper(ctx_ref: aimrt_python_runtime.ContextRef, msg):
            try:
                if required_param_count == 1:
                    callback(msg)
                else:
                    callback(ctx_ref, msg)
            except Exception as e:
                print(f"AimRT channel handle get exception, {e}", file=sys.stderr)

        aimrt_py_ros2.Ros2SubscribeWithCtx(subscriber, py_ros2_ts, msg_type, ros2_callback_wrapper)