src/vpp-api/python/vpp_papi/vpp_serializer.py

   1 #
   2 # Copyright (c) 2018 Cisco and/or its affiliates.
   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
  16 import struct
  17 import collections
  18 from enum import IntEnum
  19 import logging
  20
  21 #
  22 # Set log-level in application by doing e.g.:
  23 # logger = logging.getLogger('vpp_serializer')
  24 # logger.setLevel(logging.DEBUG)
  25 #
  26 logger = logging.getLogger(__name__)
  27
  28
  29 class BaseTypes():
  30     def __init__(self, type, elements=0):
  31         base_types = {'u8': '>B',
  32                       'u16': '>H',
  33                       'u32': '>I',
  34                       'i32': '>i',
  35                       'u64': '>Q',
  36                       'f64': '>d',
  37                       'header': '>HI'}
  38
  39         if elements > 0 and type == 'u8':
  40             self.packer = struct.Struct('>%ss' % elements)
  41         else:
  42             self.packer = struct.Struct(base_types[type])
  43         self.size = self.packer.size
  44         logger.debug('Adding {} with format: {}'
  45                      .format(type, base_types[type]))
  46
  47     def pack(self, data, kwargs=None):
  48         return self.packer.pack(data)
  49
  50     def unpack(self, data, offset, result=None):
  51         return self.packer.unpack_from(data, offset)[0]
  52
  53
  54 types = {}
  55 types['u8'] = BaseTypes('u8')
  56 types['u16'] = BaseTypes('u16')
  57 types['u32'] = BaseTypes('u32')
  58 types['i32'] = BaseTypes('i32')
  59 types['u64'] = BaseTypes('u64')
  60 types['f64'] = BaseTypes('f64')
  61
  62
  63 class FixedList_u8():
  64     def __init__(self, name, field_type, num):
  65         self.name = name
  66         self.num = num
  67         self.packer = BaseTypes(field_type, num)
  68         self.size = self.packer.size
  69
  70     def pack(self, list, kwargs):
  71         """Packs a fixed length bytestring. Left-pads with zeros
  72         if input data is too short."""
  73         if len(list) > self.num:
  74             raise ValueError('Fixed list length error for "{}", got: {}'
  75                              ' expected: {}'
  76                              .format(self.name, len(list), self.num))
  77         return self.packer.pack(list)
  78
  79     def unpack(self, data, offset=0, result=None):
  80         if len(data[offset:]) < self.num:
  81             raise ValueError('Invalid array length for "{}" got {}'
  82                              ' expected {}'
  83                              .format(self.name, len(data), self.num))
  84         return self.packer.unpack(data, offset)
  85
  86
  87 class FixedList():
  88     def __init__(self, name, field_type, num):
  89         self.num = num
  90         self.packer = types[field_type]
  91         self.size = self.packer.size * num
  92
  93     def pack(self, list, kwargs):
  94         if len(list) != self.num:
  95             raise ValueError('Fixed list length error, got: {} expected: {}'
  96                              .format(len(list), self.num))
  97         b = bytes()
  98         for e in list:
  99             b += self.packer.pack(e)
 100         return b
 101
 102     def unpack(self, data, offset=0, result=None):
 103         # Return a list of arguments
 104         result = []
 105         for e in range(self.num):
 106             x = self.packer.unpack(data, offset)
 107             result.append(x)
 108             offset += self.packer.size
 109         return result
 110
 111
 112 class VLAList():
 113     def __init__(self, name, field_type, len_field_name, index):
 114         self.index = index
 115         self.packer = types[field_type]
 116         self.size = self.packer.size
 117         self.length_field = len_field_name
 118
 119     def pack(self, list, kwargs=None):
 120         if len(list) != kwargs[self.length_field]:
 121             raise ValueError('Variable length error, got: {} expected: {}'
 122                              .format(len(list), kwargs[self.length_field]))
 123         b = bytes()
 124
 125         # u8 array
 126         if self.packer.size == 1:
 127             return bytearray(list)
 128
 129         for e in list:
 130             b += self.packer.pack(e)
 131         return b
 132
 133     def unpack(self, data, offset=0, result=None):
 134         # Return a list of arguments
 135
 136         # u8 array
 137         if self.packer.size == 1:
 138             if result[self.index] == 0:
 139                 return b''
 140             p = BaseTypes('u8', result[self.index])
 141             r = p.unpack(data, offset)
 142             return r
 143
 144         r = []
 145         for e in range(result[self.index]):
 146             x = self.packer.unpack(data, offset)
 147             r.append(x)
 148             offset += self.packer.size
 149         return r
 150
 151
 152 class VLAList_legacy():
 153     def __init__(self, name, field_type):
 154         self.packer = types[field_type]
 155         self.size = self.packer.size
 156
 157     def pack(self, list, kwargs=None):
 158         if self.packer.size == 1:
 159             return bytes(list)
 160
 161         b = bytes()
 162         for e in list:
 163             b += self.packer.pack(e)
 164         return b
 165
 166     def unpack(self, data, offset=0, result=None):
 167         # Return a list of arguments
 168         if (len(data) - offset) % self.packer.size:
 169             raise ValueError('Legacy Variable Length Array length mismatch.')
 170         elements = int((len(data) - offset) / self.packer.size)
 171         r = []
 172         for e in range(elements):
 173             x = self.packer.unpack(data, offset)
 174             r.append(x)
 175             offset += self.packer.size
 176         return r
 177
 178
 179 class VPPEnumType():
 180     def __init__(self, name, msgdef):
 181         self.size = types['u32'].size
 182         e_hash = {}
 183         for f in msgdef:
 184             if type(f) is dict and 'enumtype' in f:
 185                 if f['enumtype'] != 'u32':
 186                     raise NotImplementedError
 187                 continue
 188             ename, evalue = f
 189             e_hash[ename] = evalue
 190         self.enum = IntEnum(name, e_hash)
 191         types[name] = self
 192         logger.debug('Adding enum {}'.format(name))
 193
 194     def __getattr__(self, name):
 195         return self.enum[name]
 196
 197     def pack(self, data, kwargs=None):
 198         return types['u32'].pack(data, kwargs)
 199
 200     def unpack(self, data, offset=0, result=None):
 201         x = types['u32'].unpack(data, offset)
 202         return self.enum(x)
 203
 204
 205 class VPPUnionType():
 206     def __init__(self, name, msgdef):
 207         self.name = name
 208         self.size = 0
 209         self.maxindex = 0
 210         fields = []
 211         self.packers = collections.OrderedDict()
 212         for i, f in enumerate(msgdef):
 213             if type(f) is dict and 'crc' in f:
 214                 self.crc = f['crc']
 215                 continue
 216             f_type, f_name = f
 217             if f_type not in types:
 218                 logger.debug('Unknown union type {}'.format(f_type))
 219                 raise ValueError('Unknown message type {}'.format(f_type))
 220             fields.append(f_name)
 221             size = types[f_type].size
 222             self.packers[f_name] = types[f_type]
 223             if size > self.size:
 224                 self.size = size
 225                 self.maxindex = i
 226
 227         types[name] = self
 228         self.tuple = collections.namedtuple(name, fields, rename=True)
 229         logger.debug('Adding union {}'.format(name))
 230
 231     def pack(self, data, kwargs=None):
 232         for k, v in data.items():
 233             logger.debug("Key: {} Value: {}".format(k, v))
 234             b = self.packers[k].pack(v, kwargs)
 235             offset = self.size - self.packers[k].size
 236             break
 237         r = bytearray(self.size)
 238         r[offset:] = b
 239         return r
 240
 241     def unpack(self, data, offset=0, result=None):
 242         r = []
 243         for k, p in self.packers.items():
 244             union_offset = self.size - p.size
 245             r.append(p.unpack(data, offset + union_offset))
 246         return self.tuple._make(r)
 247
 248
 249 class VPPType():
 250     # Set everything up to be able to pack / unpack
 251     def __init__(self, name, msgdef):
 252         self.name = name
 253         self.msgdef = msgdef
 254         self.packers = []
 255         self.fields = []
 256         self.fieldtypes = []
 257         self.field_by_name = {}
 258         size = 0
 259         for i, f in enumerate(msgdef):
 260             if type(f) is dict and 'crc' in f:
 261                 self.crc = f['crc']
 262                 continue
 263             f_type, f_name = f[:2]
 264             self.fields.append(f_name)
 265             self.field_by_name[f_name] = None
 266             self.fieldtypes.append(f_type)
 267             if f_type not in types:
 268                 logger.debug('Unknown type {}'.format(f_type))
 269                 raise ValueError('Unknown message type {}'.format(f_type))
 270             if len(f) == 3:  # list
 271                 list_elements = f[2]
 272                 if list_elements == 0:
 273                     p = VLAList_legacy(f_name, f_type)
 274                     self.packers.append(p)
 275                 elif f_type == 'u8':
 276                     p = FixedList_u8(f_name, f_type, list_elements)
 277                     self.packers.append(p)
 278                     size += p.size
 279                 else:
 280                     p = FixedList(f_name, f_type, list_elements)
 281                     self.packers.append(p)
 282                     size += p.size
 283             elif len(f) == 4:  # Variable length list
 284                     # Find index of length field
 285                     length_index = self.fields.index(f[3])
 286                     p = VLAList(f_name, f_type, f[3], length_index)
 287                     self.packers.append(p)
 288             else:
 289                 self.packers.append(types[f_type])
 290                 size += types[f_type].size
 291
 292         self.size = size
 293         self.tuple = collections.namedtuple(name, self.fields, rename=True)
 294         types[name] = self
 295         logger.debug('Adding type {}'.format(name))
 296
 297     def pack(self, data, kwargs=None):
 298         if not kwargs:
 299             kwargs = data
 300         b = bytes()
 301         for i, a in enumerate(self.fields):
 302             if a not in data:
 303                 b += b'\x00' * self.packers[i].size
 304                 continue
 305
 306             if isinstance(self.packers[i], VPPType):
 307                 b += self.packers[i].pack(data[a], kwargs[a])
 308             else:
 309                 b += self.packers[i].pack(data[a], kwargs)
 310         return b
 311
 312     def unpack(self, data, offset=0, result=None):
 313         # Return a list of arguments
 314         result = []
 315         for p in self.packers:
 316             x = p.unpack(data, offset, result)
 317             if type(x) is tuple and len(x) == 1:
 318                 x = x[0]
 319             result.append(x)
 320             offset += p.size
 321         return self.tuple._make(result)
 322
 323
 324 class VPPMessage(VPPType):
 325     pass