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:
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
18 from enum import IntEnum
22 # Set log-level in application by doing e.g.:
23 # logger = logging.getLogger('vpp_serializer')
24 # logger.setLevel(logging.DEBUG)
26 logger = logging.getLogger(__name__)
30 def __init__(self, type, elements=0):
31 base_types = {'u8': '>B',
40 if elements > 0 and type == 'u8':
41 self.packer = struct.Struct('>%ss' % elements)
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]))
48 def pack(self, data, kwargs=None):
49 return self.packer.pack(data)
51 def unpack(self, data, offset, result=None):
52 return self.packer.unpack_from(data, offset)[0], self.packer.size
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')
65 def vpp_get_type(name):
73 def __init__(self, name, field_type, num):
76 self.packer = BaseTypes(field_type, num)
77 self.size = self.packer.size
79 def pack(self, list, kwargs):
80 """Packs a fixed length bytestring. Left-pads with zeros
81 if input data is too short."""
82 if len(list) > self.num:
83 raise ValueError('Fixed list length error for "{}", got: {}'
85 .format(self.name, len(list), self.num))
86 return self.packer.pack(list)
88 def unpack(self, data, offset=0, result=None):
89 if len(data[offset:]) < self.num:
90 raise ValueError('Invalid array length for "{}" got {}'
92 .format(self.name, len(data[offset:]), self.num))
93 return self.packer.unpack(data, offset)
97 def __init__(self, name, field_type, num):
99 self.packer = types[field_type]
100 self.size = self.packer.size * num
102 def pack(self, list, kwargs):
103 if len(list) != self.num:
104 raise ValueError('Fixed list length error, got: {} expected: {}'
105 .format(len(list), self.num))
108 b += self.packer.pack(e)
111 def unpack(self, data, offset=0, result=None):
112 # Return a list of arguments
115 for e in range(self.num):
116 x, size = self.packer.unpack(data, offset)
124 def __init__(self, name, field_type, len_field_name, index):
127 self.packer = types[field_type]
128 self.size = self.packer.size
129 self.length_field = len_field_name
131 def pack(self, list, kwargs=None):
132 if len(list) != kwargs[self.length_field]:
133 raise ValueError('Variable length error, got: {} expected: {}'
134 .format(len(list), kwargs[self.length_field]))
138 if self.packer.size == 1:
139 return bytearray(list)
142 b += self.packer.pack(e)
145 def unpack(self, data, offset=0, result=None):
146 # Return a list of arguments
150 if self.packer.size == 1:
151 if result[self.index] == 0:
153 p = BaseTypes('u8', result[self.index])
154 return p.unpack(data, offset)
157 for e in range(result[self.index]):
158 x, size = self.packer.unpack(data, offset)
165 class VLAList_legacy():
166 def __init__(self, name, field_type):
167 self.packer = types[field_type]
168 self.size = self.packer.size
170 def pack(self, list, kwargs=None):
171 if self.packer.size == 1:
176 b += self.packer.pack(e)
179 def unpack(self, data, offset=0, result=None):
181 # Return a list of arguments
182 if (len(data) - offset) % self.packer.size:
183 raise ValueError('Legacy Variable Length Array length mismatch.')
184 elements = int((len(data) - offset) / self.packer.size)
186 for e in range(elements):
187 x, size = self.packer.unpack(data, offset)
189 offset += self.packer.size
195 def __init__(self, name, msgdef):
196 self.size = types['u32'].size
199 if type(f) is dict and 'enumtype' in f:
200 if f['enumtype'] != 'u32':
201 raise NotImplementedError
204 e_hash[ename] = evalue
205 self.enum = IntEnum(name, e_hash)
207 logger.debug('Adding enum {}'.format(name))
209 def __getattr__(self, name):
210 return self.enum[name]
212 def __nonzero__(self):
215 def pack(self, data, kwargs=None):
216 return types['u32'].pack(data, kwargs)
218 def unpack(self, data, offset=0, result=None):
219 x, size = types['u32'].unpack(data, offset)
220 return self.enum(x), size
223 class VPPUnionType():
224 def __init__(self, name, msgdef):
229 self.packers = collections.OrderedDict()
230 for i, f in enumerate(msgdef):
231 if type(f) is dict and 'crc' in f:
235 if f_type not in types:
236 logger.debug('Unknown union type {}'.format(f_type))
237 raise ValueError('Unknown message type {}'.format(f_type))
238 fields.append(f_name)
239 size = types[f_type].size
240 self.packers[f_name] = types[f_type]
246 self.tuple = collections.namedtuple(name, fields, rename=True)
247 logger.debug('Adding union {}'.format(name))
249 def pack(self, data, kwargs=None):
250 for k, v in data.items():
251 logger.debug("Key: {} Value: {}".format(k, v))
252 b = self.packers[k].pack(v, kwargs)
254 r = bytearray(self.size)
258 def unpack(self, data, offset=0, result=None):
261 for k, p in self.packers.items():
262 x, size = p.unpack(data, offset)
266 return self.tuple._make(r), maxsize
270 # Set everything up to be able to pack / unpack
271 def __init__(self, name, msgdef):
277 self.field_by_name = {}
279 for i, f in enumerate(msgdef):
280 if type(f) is dict and 'crc' in f:
283 f_type, f_name = f[:2]
284 self.fields.append(f_name)
285 self.field_by_name[f_name] = None
286 self.fieldtypes.append(f_type)
287 if f_type not in types:
288 logger.debug('Unknown type {}'.format(f_type))
289 raise ValueError('Unknown message type {}'.format(f_type))
290 if len(f) == 3: # list
292 if list_elements == 0:
293 p = VLAList_legacy(f_name, f_type)
294 self.packers.append(p)
296 p = FixedList_u8(f_name, f_type, list_elements)
297 self.packers.append(p)
300 p = FixedList(f_name, f_type, list_elements)
301 self.packers.append(p)
303 elif len(f) == 4: # Variable length list
304 # Find index of length field
305 length_index = self.fields.index(f[3])
306 p = VLAList(f_name, f_type, f[3], length_index)
307 self.packers.append(p)
309 self.packers.append(types[f_type])
310 size += types[f_type].size
313 self.tuple = collections.namedtuple(name, self.fields, rename=True)
315 logger.debug('Adding type {}'.format(name))
317 def pack(self, data, kwargs=None):
321 for i, a in enumerate(self.fields):
323 b += b'\x00' * self.packers[i].size
326 if isinstance(self.packers[i], VPPType):
327 b += self.packers[i].pack(data[a], kwargs[a])
329 b += self.packers[i].pack(data[a], kwargs)
332 def unpack(self, data, offset=0, result=None):
333 # Return a list of arguments
336 for p in self.packers:
337 x, size = p.unpack(data, offset, result)
338 if type(x) is tuple and len(x) == 1:
343 t = self.tuple._make(result)
347 class VPPMessage(VPPType):