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