Support Proto Descriptors

[MSSandroid]

Summary

The Messages do not have Proto Descriptors like the protobuf implementation. These are required e.g., to write protobuf into MCAP file format.

What is the feature request for?

The core library

The Problem

I am looking into creating MCAP files from python code. MCAP is a container format that stores the schema together with the messages easing the reading of data. To write the schema to the MCAP file it utilizes the Protobuf Descriptors, see mcap-protobuf-support and where schema is writen to mcap. The betterproto2 message classes do not provide such descriptors.

Is it in the scope of this project to provide descriptors for the messages? What do you think would be a good way to implement the feature?

The Ideal Solution

The descriptor could be either added during compile and would then be implemented in betterproto2_compiler or be more of a "decompiler" from the python classes directly (decompiler could be an extra package).

The Current Solution

No response

[MSSandroid]

I found the workaround to serialize the FileDescirptors of the official python protobuf code to a json and then create mock FileDescriptors from that json at "runtime" and assign them to the DESCRIPTOR attribute of the respective betterproto2 generated class. Code example is targeted to messages of https://github.com/OpenSimulationInterface/open-simulation-interface

import base64
import json
from dataclasses import dataclass
from typing import Any


class Dependency():
    def __init__(self, name):
        super().__init__()
        self.name = name
        
@dataclass    
class File():
    name: str
    serialized_pb: Any
    dependencies: list["File"]
    
    def CopyToProto(self, proto):
        return proto.ParseFromString(self.serialized_pb)

@dataclass  
class Descriptor():
    full_name: str
    file: File
        
    @classmethod
    def create_descriptors_from_json(cls, filepath: str ='descriptors.json'):
        with open(filepath, 'r') as f:
            osi3_descriptors = json.load(f)
        result = {}
        def parse_file(file_dict):
            return File(
                name = file_dict['name'],
                serialized_pb = base64.b64decode(file_dict['serialized_pb_base64'].encode()),
                dependencies = [parse_file(o) for o in file_dict['dependencies']]
            )
        for k, v in osi3_descriptors.items():
            result[k] = cls(full_name=v['full_name'], 
                                file=parse_file(v))
        return result
    
    @staticmethod
    def create_descriptor_json(filepath: str = 'descriptors.json'):
        import osi3
        def relsolve_dependencies(filedescriptor):
            return dict(
                name = filedescriptor.name,
                serialized_pb_base64 = base64.b64encode(filedescriptor.serialized_pb).decode(),
                dependencies = [relsolve_dependencies(d) for d in filedescriptor.dependencies]
            )
        osi3_descriptors = {o.DESCRIPTOR.full_name: dict(   
            full_name = o.DESCRIPTOR.full_name,
            name = o.DESCRIPTOR.file.name,
            serialized_pb_base64 = base64.b64encode(o.DESCRIPTOR.file.serialized_pb).decode(),
            dependencies = [relsolve_dependencies(d) for d in o.DESCRIPTOR.file.dependencies]
        ) for o in [osi3.SensorView, osi3.GroundTruth]}

        with open(filepath, 'w') as f:
            json.dump(osi3_descriptors, f)

[MichaelSchuldes]

I have found a neater solution. I have made adaptions in python-betterproto2-compiler that add the serialized Descriptor to the python classes. Then with the following code I am able to regenerate the the proto Descriptors:

def get_proto_dependencies(obj):
    if hasattr(obj, '_betterproto'):
        return set([o for os in [[v]+list(get_proto_dependencies(v)) for v in obj._betterproto.cls_by_field.values() if hasattr(v, '_serialized_pb')] for o in os])
    else:
        return []

def insert_into_hierarchy_dict(d, name, cls):
    first = name.split('.')[0]
    if first not in d:
        d[first] = {'cls': None,  'children': {}}
    if len(name.split('.'))>1:
        insert_into_hierarchy_dict(d[first]['children'], '.'.join(name.split('.')[1:]), cls)
    else:
        d[first]['cls'] = cls


def cls_to_proto(message, children):
    if isinstance(message, EnumType):
        proto = EnumDescriptorProto()
        proto.ParseFromString(message._serialized_pb())
        proto.name = proto.name.split('.')[-1]
        return None, proto
    else:
        proto = DescriptorProto()
        proto.ParseFromString(message._serialized_pb())
        proto.name = proto.name.split('.')[-1]
        proto.ClearField("enum_type")  # Remove all enums first
        proto.ClearField("nested_type")  # Remove all enums first
        for k, v in children.items():
            m, e = cls_to_proto(v['cls'], v['children'])
            if m is not None:
                proto.nested_type.add().MergeFrom(m)
            if e is not None:
                proto.enum_type.add().MergeFrom(e)
        return proto, None

def get_descriptor(cls, package='osi3'):
    fd = FileDescriptorProto(name=cls.__name__, package=package)
    
    package_hierarchy = {} 
    for o in list(get_proto_dependencies(cls))+[cls]:
        proto = DescriptorProto()
        proto.ParseFromString(o._serialized_pb())
        insert_into_hierarchy_dict(package_hierarchy, proto.name, o)
        
    messages = []
    enums = []
    for k, v in package_hierarchy.items():
        m,e = cls_to_proto(v['cls'], v['children'])
        if m is not None:
            messages.append(m)
        if e is not None:
            enums.append(e)
    fd = FileDescriptorProto(name=cls.__name__, package=package)
    fd.message_type.extend(messages)
    fd.enum_type.extend(enums)
    pool = DescriptorPool()
    pool.Add(fd)

    return pool.FindMessageTypeByName(f'{package}.{cls.__name__}')

for c_name in MESSAGES_TYPE:
    o = getattr(generated.osi3, c_name)
    setattr(o, 'DESCRIPTOR', get_descriptor(o))

I am unsure how to neatly integrate it into this library.

[edaniels]

@MichaelSchuldes I think I found a good solve for this using your work as inspiration in #112. I tested this with mcap and it works great