robotframework-onap/ONAPLibrary/VESProtobuf.py

   1 # Copyright 2019 AT&T Intellectual Property. All rights reserved.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 # noinspection PyPackageRequirements
  16 from google.protobuf import descriptor
  17 # noinspection PyPackageRequirements
  18 from google.protobuf import descriptor_pb2
  19 # noinspection PyPackageRequirements
  20 from google.protobuf import message_factory
  21 # noinspection PyPackageRequirements
  22 from google.protobuf.json_format import MessageToJson
  23
  24
  25 class VESProtobuf(object):
  26     """ non keywords methods related to VES """
  27
  28     def __init__(self):
  29         super(VESProtobuf, self).__init__()
  30         self.message_descriptors = VESProtobuf.get_message_definitions()
  31
  32     @staticmethod
  33     def create_ves_event():
  34         file_descriptor_proto = descriptor_pb2.FileDescriptorProto()
  35         file_descriptor_proto.name = 'VesEvent'
  36         VESProtobuf.create_commoneventheader(file_descriptor_proto)
  37         VESProtobuf.create_vesevent(file_descriptor_proto)
  38         return file_descriptor_proto
  39
  40     @staticmethod
  41     def create_vesevent(file_descriptor_proto):
  42         message_type = file_descriptor_proto.message_type.add()
  43         message_type.name = "VesEvent"
  44         VESProtobuf.create_message_field(message_type, 1, "commonEventHeader", "CommonEventHeader")
  45         VESProtobuf.create_field(message_type, 2, "eventFields", descriptor.FieldDescriptor.TYPE_BYTES)
  46         return message_type
  47
  48     @staticmethod
  49     def create_commoneventheader(file_descriptor_proto):
  50         message_type = file_descriptor_proto.message_type.add()
  51         message_type.name = "CommonEventHeader"
  52         enum_type = VESProtobuf.create_enum_type(message_type, 'Priority')
  53         VESProtobuf.create_enum_type_value(enum_type, ["PRIORITY_NOT_PROVIDED", "HIGH", "MEDIUM", "NORMAL", "LOW"])
  54         VESProtobuf.create_field(message_type, 1, "version", descriptor.FieldDescriptor.TYPE_STRING)
  55         VESProtobuf.create_field(message_type, 2, "domain", descriptor.FieldDescriptor.TYPE_STRING)
  56         VESProtobuf.create_field(message_type, 3, "sequence", descriptor.FieldDescriptor.TYPE_UINT32)
  57         VESProtobuf.create_enum_field(message_type, 4, "priority", "Priority")
  58         VESProtobuf.create_field(message_type, 5, "eventId", descriptor.FieldDescriptor.TYPE_STRING)
  59         VESProtobuf.create_field(message_type, 6, "eventName", descriptor.FieldDescriptor.TYPE_STRING)
  60         VESProtobuf.create_field(message_type, 7, "eventType", descriptor.FieldDescriptor.TYPE_STRING)
  61         VESProtobuf.create_field(message_type, 8, "lastEpochMicrosec", descriptor.FieldDescriptor.TYPE_UINT64)
  62         VESProtobuf.create_field(message_type, 9, "startEpochMicrosec", descriptor.FieldDescriptor.TYPE_UINT64)
  63         VESProtobuf.create_field(message_type, 10, "nfNamingCode", descriptor.FieldDescriptor.TYPE_STRING)
  64         VESProtobuf.create_field(message_type, 11, "nfcNamingCode", descriptor.FieldDescriptor.TYPE_STRING)
  65         VESProtobuf.create_field(message_type, 12, "nfVendorName", descriptor.FieldDescriptor.TYPE_STRING)
  66         VESProtobuf.create_field(message_type, 13, "reportingEntityId", descriptor.FieldDescriptor.TYPE_BYTES)
  67         VESProtobuf.create_field(message_type, 14, "reportingEntityName", descriptor.FieldDescriptor.TYPE_STRING)
  68         VESProtobuf.create_field(message_type, 15, "sourceId", descriptor.FieldDescriptor.TYPE_BYTES)
  69         VESProtobuf.create_field(message_type, 16, "sourceName", descriptor.FieldDescriptor.TYPE_STRING)
  70         VESProtobuf.create_field(message_type, 17, "timeZoneOffset", descriptor.FieldDescriptor.TYPE_STRING)
  71         VESProtobuf.create_field(message_type, 18, "vesEventListenerVersion",
  72                                  descriptor.FieldDescriptor.TYPE_STRING)
  73         return message_type
  74
  75     @staticmethod
  76     def create_enum_type(desc_proto, name):
  77         enum_type = desc_proto.enum_type.add()
  78         enum_type.name = name
  79         return enum_type
  80
  81     @staticmethod
  82     def create_enum_type_value(enum_type, value_list):
  83         for i in range(len(value_list)):
  84             enum_type_val = enum_type.value.add()
  85             enum_type_val.name = value_list[i]
  86             enum_type_val.number = i
  87
  88     @staticmethod
  89     def create_field(desc_proto, number, name, field_type):
  90         field = desc_proto.field.add()
  91         field.number = number
  92         field.name = name
  93         field.type = field_type
  94
  95     @staticmethod
  96     def create_enum_field(desc_proto, number, name, type_name):
  97         field = desc_proto.field.add()
  98         field.number = number
  99         field.name = name
 100         field.type = descriptor.FieldDescriptor.TYPE_ENUM
 101         field.type_name = type_name
 102
 103     @staticmethod
 104     def create_message_field(desc_proto, number, name, type_name):
 105         field = desc_proto.field.add()
 106         field.number = number
 107         field.name = name
 108         field.type = descriptor.FieldDescriptor.TYPE_MESSAGE
 109         field.type_name = type_name
 110
 111     @staticmethod
 112     def get_message_definitions():
 113         messages = message_factory.GetMessages((VESProtobuf.create_ves_event(),))
 114         message_factory._FACTORY = message_factory.MessageFactory()
 115         return messages
 116
 117     def binary_to_json(self, binary_message):
 118         ves = self.message_descriptors['VesEvent']()
 119         ves.MergeFromString(binary_message)
 120         json = MessageToJson(ves)
 121         return json