app/browse/pyproto.py

   1 # original (C) 2023 Tekky, MIT licensed. https://pypi.org/project/python-protobuf
   2 from enum import IntEnum, unique
   3
   4
   5 @unique
   6 class ProtoFieldType(IntEnum):
   7     VARINT = 0
   8     INT64 = 1
   9     STRING = 2
  10     GROUPSTART = 3
  11     GROUPEND = 4
  12     INT32 = 5
  13     ERROR1 = 6
  14     ERROR2 = 7
  15
  16
  17 class ProtoField:
  18     def __init__(self, idx, type, val):
  19         self.idx = idx
  20         self.type = type
  21         self.val = val
  22
  23
  24 class ProtoWriter:
  25     def __init__(self):
  26         self.data = bytearray()
  27
  28     def write0(self, byte):
  29         self.data.append(byte & 0xFF)
  30
  31     def write(self, bytes):
  32         self.data.extend(bytes)
  33
  34     def writeInt32(self, int32):
  35         bs = int32.to_bytes(4, byteorder="little", signed=False)
  36         self.write(bs)
  37
  38     def writeInt64(self, int64):
  39         bs = int64.to_bytes(8, byteorder="little", signed=False)
  40         self.write(bs)
  41
  42     def writeVarint(self, vint):
  43         vint = vint & 0xFFFFFFFF
  44         while vint > 0x80:
  45             self.write0((vint & 0x7F) | 0x80)
  46             vint >>= 7
  47         self.write0(vint & 0x7F)
  48
  49     def writeString(self, bytes):
  50         self.writeVarint(len(bytes))
  51         self.write(bytes)
  52
  53     def toBytes(self):
  54         return bytes(self.data)
  55
  56
  57 class ProtoBuf:
  58     def __init__(self, data=None):
  59         self.fields = []
  60         if data != None:
  61             if type(data) != dict:
  62                 raise ValueError("unsupport type(%s) to protobuf" % (type(data)))
  63
  64             self.__parseDict(data)
  65
  66     def toBuf(self):
  67         writer = ProtoWriter()
  68         for field in self.fields:
  69             key = (field.idx << 3) | (field.type & 7)
  70             writer.writeVarint(key)
  71             if field.type == ProtoFieldType.INT32:
  72                 writer.writeInt32(field.val)
  73             elif field.type == ProtoFieldType.INT64:
  74                 writer.writeInt64(field.val)
  75             elif field.type == ProtoFieldType.VARINT:
  76                 writer.writeVarint(field.val)
  77             elif field.type == ProtoFieldType.STRING:
  78                 writer.writeString(field.val)
  79             else:
  80                 raise ValueError(
  81                     "encode to protobuf error, unexpected field type: %s"
  82                     % (field.type.name)
  83                 )
  84         return writer.toBytes()
  85
  86     def put(self, field: ProtoField):
  87         self.fields.append(field)
  88
  89     def putInt32(self, idx, int32):
  90         self.put(ProtoField(idx, ProtoFieldType.INT32, int32))
  91
  92     def putInt64(self, idx, int64):
  93         self.put(ProtoField(idx, ProtoFieldType.INT64, int64))
  94
  95     def putVarint(self, idx, vint):
  96         self.put(ProtoField(idx, ProtoFieldType.VARINT, vint))
  97
  98     def putBytes(self, idx, data):
  99         self.put(ProtoField(idx, ProtoFieldType.STRING, data))
 100
 101     def putUtf8(self, idx, data):
 102         self.put(ProtoField(idx, ProtoFieldType.STRING, data.encode("utf-8")))
 103
 104     def putProtoBuf(self, idx, data):
 105         self.put(ProtoField(idx, ProtoFieldType.STRING, data.toBuf()))
 106
 107     def __parseDict(self, data):
 108         """
 109         Convert dict object to ProtoBuf object
 110         """
 111         for k, v in data.items():
 112             if isinstance(v, int):
 113                 self.putVarint(k, v)
 114             elif isinstance(v, str):
 115                 self.putUtf8(k, v)
 116             elif isinstance(v, bytes):
 117                 self.putBytes(k, v)
 118             elif isinstance(v, dict):
 119                 self.putProtoBuf(k, ProtoBuf(v))
 120             elif v is None:
 121                 continue
 122             else:
 123                 raise ValueError("unsupport type(%s) to protobuf" % (type(v)))