Source code for spotware_connect.protobuf
[docs]class Protobuf(object):
_protos = dict()
_names = dict()
_abbr_names = dict()
[docs] @classmethod
def populate(cls):
import re
from .messages import OpenApiCommonMessages_pb2 as o1
from .messages import OpenApiMessages_pb2 as o2
for name in dir(o1) + dir(o2):
if not name.startswith("Proto"):
continue
m = o1 if hasattr(o1, name) else o2
klass = getattr(m, name)
cls._protos[klass().payloadType] = klass
cls._names[klass.__name__] = klass().payloadType
abbr_name = re.sub(r'^Proto(OA)?(.*)', r'\2', klass.__name__)
cls._names[abbr_name] = klass().payloadType
return cls._protos
[docs] @classmethod
def get(cls, payload, fail=True, **params):
if not cls._protos:
cls.populate()
if payload in cls._protos:
return cls._protos[payload](**params)
for d in [cls._names, cls._abbr_names]:
if payload in d:
payload = d[payload]
return cls._protos[payload](**params)
if fail: # pragma: nocover
raise IndexError("Invalid payload: " + str(payload))
return None # pragma: nocover
[docs] @classmethod
def get_type(cls, payload, **params):
p = cls.get(payload, **params)
return p.payloadType