Fast DDS-Gen 支持为数据类型生成 Python 绑定，使得开发者可以在 Python 环境中使用 IDL 定义的数据结构进行发布与订阅操作。通过生成的 Python 模块，可以实现类型注册、数据发布和订阅的完整流程。 ### 注册类型 在 Python 中，需要使用 `register_type` 方法将生成的数据类型注册到 Fast DDS 系统中。该方法接受两个参数：数据类型的类和类型名称。例如： ```python from fastdds import DomainParticipant, DomainParticipantQos, Topic, Publisher, DataWriter, Subscriber, DataReader from fastdds import TypeSupport from FloatArrayStructPubSubTypes import FloatArrayStructPubSubType participant = DomainParticipant(0) type_support = TypeSupport(FloatArrayStructPubSubType()) participant.register_type(type_support, "FloatArrayStruct") ``` 上述代码创建了一个 `DomainParticipant` 实例，并通过 `TypeSupport` 将 `FloatArrayStruct` 类型注册到系统中。 ### 创建主题和发布者 定义 `Topic` 并创建 `Publisher` 和 `DataWriter` 以发布数据： ```python topic = Topic(participant, "FloatArrayTopic", "FloatArrayStruct") publisher = Publisher(participant) writer = DataWriter(publisher, topic) ``` ### 发布数据 通过构造 `FloatArrayStruct` 实例并填充数组，然后使用 `DataWriter` 发送数据： ```python from FloatArrayStruct import FloatArrayStruct data = FloatArrayStruct() for i in range(10): data.float_array[i] = float(i) writer.write(data) ``` ### 创建订阅者并接收数据 创建 `Subscriber` 和 `DataReader`，并监听数据到达事件： ```python subscriber = Subscriber(participant) reader = DataReader(subscriber, topic) class DataListener: def on_data_available(self, reader): data = FloatArrayStruct() if reader.take_next_sample(data): print("Received data:", data.float_array) listener = DataListener() reader.set_listener(listener) ``` ### 完整示例代码 以下是一个完整的 Python 示例代码，包含类型注册、发布和订阅操作： ```python from fastdds import DomainParticipant, DomainParticipantQos, Topic, Publisher, DataWriter, Subscriber, DataReader from fastdds import TypeSupport from FloatArrayStructPubSubTypes import FloatArrayStructPubSubType from FloatArrayStruct import FloatArrayStruct # 初始化参与者 participant = DomainParticipant(0) # 注册类型 type_support = TypeSupport(FloatArrayStructPubSubType()) participant.register_type(type_support, "FloatArrayStruct") # 创建主题 topic = Topic(participant, "FloatArrayTopic", "FloatArrayStruct") # 创建发布者和写入器 publisher = Publisher(participant) writer = DataWriter(publisher, topic) # 构造数据并发送 data = FloatArrayStruct() for i in range(10): data.float_array[i] = float(i) writer.write(data) # 创建订阅者和读取器 subscriber = Subscriber(participant) reader = DataReader(subscriber, topic) # 定义监听器并接收数据 class DataListener: def on_data_available(self, reader): data = FloatArrayStruct() if reader.take_next_sample(data): print("Received data:", data.float_array) listener = DataListener() reader.set_listener(listener) # 保持程序运行以接收数据 input("Press Enter to exit...\n") ``` 该示例展示了如何在 Python 中使用 Fast DDS-Gen 生成的模块实现数据类型的注册、发布和订阅操作，适用于基于 Fast DDS 的分布式系统开发。 ---