blob: cd9a281964a7a7e7f2b38141b404de1c03e28319 [file] [log] [blame]
Ole Troana7564e82018-06-12 21:06:44 +02001#
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
16import struct
17import collections
18from enum import IntEnum
19import 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#
26logger = logging.getLogger(__name__)
27
28
29class 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):
Ole Troana7564e82018-06-12 21:06:44 +020048 return self.packer.pack(data)
49
50 def unpack(self, data, offset, result=None):
Ole Troana7564e82018-06-12 21:06:44 +020051 return self.packer.unpack_from(data, offset)[0]
52
53
54types = {}
55types['u8'] = BaseTypes('u8')
56types['u16'] = BaseTypes('u16')
57types['u32'] = BaseTypes('u32')
58types['i32'] = BaseTypes('i32')
59types['u64'] = BaseTypes('u64')
60types['f64'] = BaseTypes('f64')
61
62
63class 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."""
Ole Troana7564e82018-06-12 21:06:44 +020073 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
87class 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):
Ole Troana7564e82018-06-12 21:06:44 +020094 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
112class 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):
Ole Troana7564e82018-06-12 21:06:44 +0200120 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):
Ole Troana7564e82018-06-12 21:06:44 +0200134 # 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
152class 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):
Ole Troana7564e82018-06-12 21:06:44 +0200158 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 = []
Ole Troana7564e82018-06-12 21:06:44 +0200172 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
179class 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):
Ole Troana7564e82018-06-12 21:06:44 +0200198 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
205class 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):
Ole Troana7564e82018-06-12 21:06:44 +0200232 for k, v in data.items():
233 logger.debug("Key: {} Value: {}".format(k, v))
234 b = self.packers[k].pack(v, kwargs)
Ole Troana7564e82018-06-12 21:06:44 +0200235 break
236 r = bytearray(self.size)
Ole Troanb199e982018-08-02 19:19:21 +0200237 r[:len(b)] = b
Ole Troana7564e82018-06-12 21:06:44 +0200238 return r
239
240 def unpack(self, data, offset=0, result=None):
241 r = []
242 for k, p in self.packers.items():
Ole Troanb199e982018-08-02 19:19:21 +0200243 r.append(p.unpack(data, offset))
Ole Troana7564e82018-06-12 21:06:44 +0200244 return self.tuple._make(r)
245
246
247class VPPType():
248 # Set everything up to be able to pack / unpack
249 def __init__(self, name, msgdef):
250 self.name = name
251 self.msgdef = msgdef
252 self.packers = []
253 self.fields = []
254 self.fieldtypes = []
255 self.field_by_name = {}
256 size = 0
257 for i, f in enumerate(msgdef):
258 if type(f) is dict and 'crc' in f:
259 self.crc = f['crc']
260 continue
261 f_type, f_name = f[:2]
262 self.fields.append(f_name)
263 self.field_by_name[f_name] = None
264 self.fieldtypes.append(f_type)
265 if f_type not in types:
266 logger.debug('Unknown type {}'.format(f_type))
267 raise ValueError('Unknown message type {}'.format(f_type))
268 if len(f) == 3: # list
269 list_elements = f[2]
270 if list_elements == 0:
271 p = VLAList_legacy(f_name, f_type)
272 self.packers.append(p)
273 elif f_type == 'u8':
274 p = FixedList_u8(f_name, f_type, list_elements)
275 self.packers.append(p)
276 size += p.size
277 else:
278 p = FixedList(f_name, f_type, list_elements)
279 self.packers.append(p)
280 size += p.size
281 elif len(f) == 4: # Variable length list
282 # Find index of length field
283 length_index = self.fields.index(f[3])
284 p = VLAList(f_name, f_type, f[3], length_index)
285 self.packers.append(p)
286 else:
287 self.packers.append(types[f_type])
288 size += types[f_type].size
289
290 self.size = size
291 self.tuple = collections.namedtuple(name, self.fields, rename=True)
292 types[name] = self
293 logger.debug('Adding type {}'.format(name))
294
295 def pack(self, data, kwargs=None):
296 if not kwargs:
297 kwargs = data
Ole Troana7564e82018-06-12 21:06:44 +0200298 b = bytes()
299 for i, a in enumerate(self.fields):
300 if a not in data:
Ole Troana7564e82018-06-12 21:06:44 +0200301 b += b'\x00' * self.packers[i].size
302 continue
303
304 if isinstance(self.packers[i], VPPType):
305 b += self.packers[i].pack(data[a], kwargs[a])
306 else:
307 b += self.packers[i].pack(data[a], kwargs)
308 return b
309
310 def unpack(self, data, offset=0, result=None):
311 # Return a list of arguments
312 result = []
313 for p in self.packers:
314 x = p.unpack(data, offset, result)
315 if type(x) is tuple and len(x) == 1:
316 x = x[0]
317 result.append(x)
318 offset += p.size
319 return self.tuple._make(result)
320
321
322class VPPMessage(VPPType):
323 pass