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
  31     @staticmethod
  32     def create_ves_event():
  33         file_descriptor_proto = descriptor_pb2.FileDescriptorProto()
  34         file_descriptor_proto.name = 'VesEvent'
  35         VESProtobuf.create_commoneventheader(file_descriptor_proto)
  36         VESProtobuf.create_vesevent(file_descriptor_proto)
  37         return file_descriptor_proto
  38
  39     @staticmethod
  40     def create_vesevent(file_descriptor_proto):
  41         message_type = file_descriptor_proto.message_type.add()
  42         message_type.name = "VesEvent"
  43         VESProtobuf.create_message_field(message_type, 1, "commonEventHeader", "CommonEventHeader")
  44         VESProtobuf.create_field(message_type, 2, "eventFields", descriptor.FieldDescriptor.TYPE_BYTES)
  45         return message_type
  46
  47     @staticmethod
  48     def create_commoneventheader(file_descriptor_proto):
  49         message_type = file_descriptor_proto.message_type.add()
  50         message_type.name = "CommonEventHeader"
  51         enum_type = VESProtobuf.create_enum_type(message_type, 'Priority')
  52         VESProtobuf.create_enum_type_value(enum_type, ["PRIORITY_NOT_PROVIDED", "HIGH", "MEDIUM", "NORMAL", "LOW"])
  53         VESProtobuf.create_field(message_type, 1, "version", descriptor.FieldDescriptor.TYPE_STRING)
  54         VESProtobuf.create_field(message_type, 2, "domain", descriptor.FieldDescriptor.TYPE_STRING)
  55         VESProtobuf.create_field(message_type, 3, "sequence", descriptor.FieldDescriptor.TYPE_UINT32)
  56         VESProtobuf.create_enum_field(message_type, 4, "priority", "Priority")
  57         VESProtobuf.create_field(message_type, 5, "eventId", descriptor.FieldDescriptor.TYPE_STRING)
  58         VESProtobuf.create_field(message_type, 6, "eventName", descriptor.FieldDescriptor.TYPE_STRING)
  59         VESProtobuf.create_field(message_type, 7, "eventType", descriptor.FieldDescriptor.TYPE_STRING)
  60         VESProtobuf.create_field(message_type, 8, "lastEpochMicrosec", descriptor.FieldDescriptor.TYPE_UINT64)
  61         VESProtobuf.create_field(message_type, 9, "startEpochMicrosec", descriptor.FieldDescriptor.TYPE_UINT64)
  62         VESProtobuf.create_field(message_type, 10, "nfNamingCode", descriptor.FieldDescriptor.TYPE_STRING)
  63         VESProtobuf.create_field(message_type, 11, "nfcNamingCode", descriptor.FieldDescriptor.TYPE_STRING)
  64         VESProtobuf.create_field(message_type, 12, "nfVendorName", descriptor.FieldDescriptor.TYPE_STRING)
  65         VESProtobuf.create_field(message_type, 13, "reportingEntityId", descriptor.FieldDescriptor.TYPE_BYTES)
  66         VESProtobuf.create_field(message_type, 14, "reportingEntityName", descriptor.FieldDescriptor.TYPE_STRING)
  67         VESProtobuf.create_field(message_type, 15, "sourceId", descriptor.FieldDescriptor.TYPE_BYTES)
  68         VESProtobuf.create_field(message_type, 16, "sourceName", descriptor.FieldDescriptor.TYPE_STRING)
  69         VESProtobuf.create_field(message_type, 17, "timeZoneOffset", descriptor.FieldDescriptor.TYPE_STRING)
  70         VESProtobuf.create_field(message_type, 18, "vesEventListenerVersion",
  71                                  descriptor.FieldDescriptor.TYPE_STRING)
  72         return message_type
  73
  74     @staticmethod
  75     def create_enum_type(desc_proto, name):
  76         enum_type = desc_proto.enum_type.add()
  77         enum_type.name = name
  78         return enum_type
  79
  80     @staticmethod
  81     def create_enum_type_value(enum_type, value_list):
  82         for i in range(len(value_list)):
  83             enum_type_val = enum_type.value.add()
  84             enum_type_val.name = value_list[i]
  85             enum_type_val.number = i
  86
  87     @staticmethod
  88     def create_field(desc_proto, number, name, field_type):
  89         field = desc_proto.field.add()
  90         field.number = number
  91         field.name = name
  92         field.type = field_type
  93
  94     @staticmethod
  95     def create_enum_field(desc_proto, number, name, type_name):
  96         field = desc_proto.field.add()
  97         field.number = number
  98         field.name = name
  99         field.type = descriptor.FieldDescriptor.TYPE_ENUM
 100         field.type_name = type_name
 101
 102     @staticmethod
 103     def create_message_field(desc_proto, number, name, type_name):
 104         field = desc_proto.field.add()
 105         field.number = number
 106         field.name = name
 107         field.type = descriptor.FieldDescriptor.TYPE_MESSAGE
 108         field.type_name = type_name
 109
 110     @staticmethod
 111     def get_message_definitions():
 112         return message_factory.GetMessages((VESProtobuf.create_ves_event(),))
 113
 114     @staticmethod
 115     def binary_to_json(binary_message):
 116         defs = VESProtobuf.get_message_definitions()
 117         ves = defs['VesEvent']()
 118         ves.MergeFromString(binary_message)
 119         json = MessageToJson(ves)
 120         return json