如何将gRPC返回的Protocol Buffer数据转换为Python字典?
将gRPC返回的RepeatedCompositeContainer转换为Python字典的方法
方法一:使用官方Protobuf工具类
Google Protobuf提供了json_format模块,可直接将Protobuf消息转换为字典,处理重复容器时只需遍历每个元素即可:
- 导入依赖模块
from google.protobuf.json_format import MessageToDict
- 转换处理
假设你的RepeatedCompositeContainer对象名为grpc_data,执行以下代码:
# 遍历容器内的每个Protobuf消息,逐个转为字典 result_list = [MessageToDict(item, preserving_proto_field_name=True) for item in grpc_data]
preserving_proto_field_name=True参数会保留Protobuf原有的下划线命名规则(比如manufacturer_device_id不会被转为驼峰式manufacturerDeviceId)。
方法二:自定义递归转换函数
如果需要更灵活的控制(比如自定义字段处理逻辑),可以编写递归函数处理嵌套消息和重复字段:
def proto_msg_to_dict(proto_obj): result = {} # 遍历消息的所有字段 for field, value in proto_obj.ListFields(): field_name = field.name # 处理重复类型字段 if field.label == field.LABEL_REPEATED: if field.type == field.TYPE_MESSAGE: # 嵌套的重复消息,递归转换每个元素 result[field_name] = [proto_msg_to_dict(item) for item in value] else: # 基础类型的重复字段,直接转为列表 result[field_name] = list(value) elif field.type == field.TYPE_MESSAGE: # 单个嵌套消息,递归转换 result[field_name] = proto_msg_to_dict(value) else: # 基础类型字段,直接取值 result[field_name] = value return result # 处理RepeatedCompositeContainer result_list = [proto_msg_to_dict(item) for item in grpc_data]
转换后的示例结果
针对你提供的示例数据,转换后的字典列表如下:
[ { "manufacturer_device_id": "B0000B0204c45bbefa948b", "scale": "Days", "bucket_epoch_seconds": [1698552000, 1698638400], "channel_usages": [ { "channel": 1, "usages": [798.6454022750977, 1353.0604314283914] } ] } ]
内容的提问来源于stack exchange,提问作者Fahim Ashab
相关产品推荐
相关产品推荐

