Apollo 笔记（03）— Cyber RT Python 接口（channel 读和写、server/client 通信、record 文件读写信息查询、timer 时间定时器操作）

仲孙向明

2023-12-01

https://cyber-rt.readthedocs.io/en/latest/CyberRT_Python_API.html
https://cyber-rt.readthedocs.io/en/latest/api/pythonapi.html

1. 背景

Cyber RT 核心代码是由 C++ 开发，同时为了方便开发者，提供了 Python 接口。

2. Cyber RT Python 接口实现思路

Cyber RT Python 接口的实现思路是在 Cyber RT C++ 实现的基础上，做了一层 Python 的封装，由 Python 来调用 C++ 的实现函数。Cyber RT 中 Python Wrapper 的实现没有使用 swig 等第三方工具，完全自主实现，以此保证代码的高可维护性和可读性。

3. 主要接口

目前提供的主要接口包括：

channel 读、写
server/client 通信
record 信息查询
record 文件读、写
Time/Duration/Rate 时间操作
Timer 定时器

3.1 Channel 读写接口

使用步骤是：

首先创建 Node;
创建对应的 reader 或 writer；
如果是向 channel 写数据，调用 writer 的 write 接口；
如果是从 channel 读数据，调用 node 的 spin，并在回调函数 callback 中对消息进行处理；

接口定义如下：

class Node:
    """
    Class for Cyber RT Node wrapper.
    """

    def create_writer(self, name, data_type, qos_depth=1):
        """
        create a topic writer for send message to topic.
        @param self
        @param name str: topic name
        @param data_type proto: message class for serialization
        """

   def create_reader(self, name, data_type, callback, args=None):
        """
        create a topic reader for receive message from topic.
        @param self
        @param name str: topic name
        @param data_type proto: message class for serialization
        @callback fn: function to call (fn(data)) when data is
                   received. If args is set, the function must
                   accept the args as a second argument,
                   i.e. fn(data, args)
        @args any: additional arguments to pass to the callback
        """
   def create_client(self, name, request_data_type, response_data_type):
   """
   """
   def create_service(self, name, req_data_type, res_data_type, callback, args=None):

    def spin(self):
        """
        spin in wait and process message.
        @param self
        """

class Writer(object):
    """
    Class for Cyber RT writer wrapper.
    """
    def write(self, data):
        """
        writer msg string
        """

3.2 Record 接口

Record 读的操作是：

创建一个 RecordReader;
对 Record 进行迭代读；

Record 写的操作是：

创建一个 RecordReader;
对 Record 进行写消息；

接口定义如下：

class RecordReader(object):
    """
    Class for Cyber RT RecordReader wrapper.
    """
    def read_messages(self, start_time=0, end_time=18446744073709551615):
        """
        read message from bag file.
        @param self
        @param start_time:
        @param end_time:
        @return: generator of (message, data_type, timestamp)
        """
	def get_messagenumber(self, channel_name):
        """
        return message count.
        """
	def get_messagetype(self, channel_name):
        """
        return message type.
        """
	def get_protodesc(self, channel_name):
        """
        return message protodesc.
        """
	def get_headerstring(self):
        """
        return message header string.
        """
	def reset(self):
        """
        return reset.
        """
        return _Cyber_RECORD.PyRecordReader_Reset(self.record_reader)

     def get_channellist(self):
        """
        return channel list.
        """
        return _Cyber_RECORD.PyRecordReader_GetChannelList(self.record_reader)


class RecordWriter(object):
    """
    Class for Cyber RT RecordWriter wrapper.
    """
	def open(self, path):
        """
        open record file for write.
        """
	def write_channel(self, channel_name, type_name, proto_desc):
        """
        writer channel by channelname,typename,protodesc
        """
	def write_message(self, channel_name, data, time, raw = True):
        """
        writer msg:channelname,data,time,is data raw
        """

	def set_size_fileseg(self, size_kilobytes):
        """
        return filesegment size.
        """

    def set_intervaltime_fileseg(self, time_sec):
        """
        return file interval time.
        """

    def get_messagenumber(self, channel_name):
        """
        return message count.
        """

    def get_messagetype(self, channel_name):
        """
        return message type.
        """

    def get_protodesc(self, channel_name):
        """
        return message protodesc.
        """

3.3 Time 接口

class Time(object):
	@staticmethod
    def now():
        time_now = Time(_CYBER_TIME.PyTime_now())
        return time_now

    @staticmethod
    def mono_time():
        mono_time = Time(_CYBER_TIME.PyTime_mono_time())
        return mono_time

    def to_sec(self):
        return _CYBER_TIME.PyTime_to_sec(self.time)

    def to_nsec(self):
        return _CYBER_TIME.PyTime_to_nsec(self.time)

    def sleep_until(self, nanoseconds):
        return _CYBER_TIME.PyTime_sleep_until(self.time, nanoseconds)

3.4 Timer 接口


class Timer(object):
    
    def set_option(self, period, callback, oneshot=0):
        '''
        period The period of the timer, unit is ms
        callback The tasks that the timer needs to perform
        oneshot 1: perform the callback only after the first timing cycle
        0:perform the callback every timed period
        '''
        
        
    def start(self):
            
            
    def stop(self):

4. 例子

4.1 读 channel （参见 cyber/python/cyber_py3/examples/listener.py)

"""Module for example of listener."""
from cyber.python.cyber_py3 import cyber
from cyber.proto.unit_test_pb2 import ChatterBenchmark


def callback(data):
    """
    Reader message callback.
    """
    print("=" * 80)
    print("py:reader callback msg->:")
    print(data)
    print("=" * 80)


def test_listener_class():
    """
    Reader message.
    """
    print("=" * 120)
    test_node = cyber.Node("listener")
    test_node.create_reader("channel/chatter", ChatterBenchmark, callback)
    test_node.spin()


if __name__ == '__main__':
    cyber.init()
    test_listener_class()
    cyber.shutdown()

在 apollo 镜像中执行命令

export PYTHONPATH=/apollo/bazel-bin/cyber/python/internal:/apollo:/apollo/cyber/proto

在 /apollo/cyber 目录下执行

python -m grpc_tools.protoc -I=./proto --python_out=./proto --grpc_python_out=./proto  ./proto/unit_test.proto

用于生成以下两个文件 unit_test_pb2.py 和 unit_test_pb2_grpc.py。

4.2 写 channel（参见 cyber/python/cyber_py3/examples/talker.py)

"""Module for example of talker."""
import time

from cyber.python.cyber_py3 import cyber
from cyber.proto.unit_test_pb2 import ChatterBenchmark


def test_talker_class():
    """
    Test talker.
    """
    msg = ChatterBenchmark()
    msg.content = "py:talker:send Alex!"
    msg.stamp = 9999
    msg.seq = 0
    print(msg)
    test_node = cyber.Node("node_name1")
    g_count = 1

    writer = test_node.create_writer("channel/chatter", ChatterBenchmark, 6)
    while not cyber.is_shutdown():
        time.sleep(1)
        g_count = g_count + 1
        msg.seq = g_count
        msg.content = "I am python talker."
        print("=" * 80)
        print("write msg -> %s" % msg)
        writer.write(msg)


if __name__ == '__main__':
    cyber.init("talker_sample")
    test_talker_class()
    cyber.shutdown()

分别执行

python cyber/python/cyber_py3/examples/listener.py

和

python cyber/python/cyber_py3/examples/talker.py

listener.py 打印如下：

========================================================================================================================
================================================================================
py:reader callback msg->:
stamp: 9999
seq: 2
content: "I am python talker."

================================================================================
================================================================================
py:reader callback msg->:
stamp: 9999
seq: 3
content: "I am python talker."

================================================================================

talker.py 打印如下：

stamp: 9999
seq: 0
content: "py:talker:send Alex!"

================================================================================
write msg -> stamp: 9999
seq: 2
content: "I am python talker."

================================================================================
write msg -> stamp: 9999
seq: 3
content: "I am python talker."

================================================================================

4.3 读写消息到 Record 文件（参见 cyber/python/cyber_py3/examples/record.py)

"""
Module for example of record.
Run with:
    bazel run //cyber/python/cyber_py3/examples:record
"""

import time

from google.protobuf.descriptor_pb2 import FileDescriptorProto

from cyber.proto.unit_test_pb2 import Chatter
from cyber.python.cyber_py3 import record
from modules.common.util.testdata.simple_pb2 import SimpleMessage


MSG_TYPE = "apollo.common.util.test.SimpleMessage"
MSG_TYPE_CHATTER = "apollo.cyber.proto.Chatter"


def test_record_writer(writer_path):
    """
    Record writer.
    """
    fwriter = record.RecordWriter()
    fwriter.set_size_fileseg(0)
    fwriter.set_intervaltime_fileseg(0)

    if not fwriter.open(writer_path):
        print('Failed to open record writer!')
        return
    print('+++ Begin to writer +++')

    # Writer 2 SimpleMessage
    msg = SimpleMessage()
    msg.text = "AAAAAA"

    file_desc = msg.DESCRIPTOR.file
    proto = FileDescriptorProto()
    file_desc.CopyToProto(proto)
    proto.name = file_desc.name
    desc_str = proto.SerializeToString()
    print(msg.DESCRIPTOR.full_name)
    fwriter.write_channel(
        'simplemsg_channel', msg.DESCRIPTOR.full_name, desc_str)
    fwriter.write_message('simplemsg_channel', msg, 990, False)
    fwriter.write_message('simplemsg_channel', msg.SerializeToString(), 991)

    # Writer 2 Chatter
    msg = Chatter()
    msg.timestamp = 99999
    msg.lidar_timestamp = 100
    msg.seq = 1

    file_desc = msg.DESCRIPTOR.file
    proto = FileDescriptorProto()
    file_desc.CopyToProto(proto)
    proto.name = file_desc.name
    desc_str = proto.SerializeToString()
    print(msg.DESCRIPTOR.full_name)
    fwriter.write_channel('chatter_a', msg.DESCRIPTOR.full_name, desc_str)
    fwriter.write_message('chatter_a', msg, 992, False)
    msg.seq = 2
    fwriter.write_message("chatter_a", msg.SerializeToString(), 993)

    fwriter.close()


def test_record_reader(reader_path):
    """
    Record reader.
    """
    freader = record.RecordReader(reader_path)
    time.sleep(1)
    print('+' * 80)
    print('+++ Begin to read +++')
    count = 0
    for channel_name, msg, datatype, timestamp in freader.read_messages():
        count += 1
        print('=' * 80)
        print('read [%d] messages' % count)
        print('channel_name -> %s' % channel_name)
        print('msgtime -> %d' % timestamp)
        print('msgnum -> %d' % freader.get_messagenumber(channel_name))
        print('msgtype -> %s' % datatype)
        print('message is -> %s' % msg)
        print('***After parse(if needed),the message is ->')
        if datatype == MSG_TYPE:
            msg_new = SimpleMessage()
            msg_new.ParseFromString(msg)
            print(msg_new)
        elif datatype == MSG_TYPE_CHATTER:
            msg_new = Chatter()
            msg_new.ParseFromString(msg)
            print(msg_new)


if __name__ == '__main__':
    test_record_file = "/tmp/test_writer.record"

    print('Begin to write record file: {}'.format(test_record_file))
    test_record_writer(test_record_file)

    print('Begin to read record file: {}'.format(test_record_file))
    test_record_reader(test_record_file)