#!/usr/bin/env python
#
# Copyright (c) 2016 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from __future__ import print_function
import sys, os, logging, collections, struct, json, threading
logging.basicConfig(level=logging.DEBUG)
import vpp_api
def eprint(*args, **kwargs):
print(*args, file=sys.stderr, **kwargs)
class VPP():
def __init__(self, apifiles):
self.messages = {}
self.id_names = []
self.id_msgdef = []
self.buffersize = 10000
self.connected = False
self.header = struct.Struct('>HI')
self.results = {}
self.timeout = 5
self.apifile = []
for file in apifiles:
self.apifile.append(file)
with open(file) as apidef_file:
api = json.load(apidef_file)
for t in api['types']:
self.add_type(t[0], t[1:])
for m in api['messages']:
self.add_message(m[0], m[1:])
# Basic sanity check
if not 'control_ping' in self.messages:
raise ValueError(1, 'Error in JSON message definitions')
class ContextId(object):
def __init__(self):
self.context = 0
def __call__(self):
self.context += 1
return self.context
get_context = ContextId()
def status(self):
print('Connected') if self.connected else print('Not Connected')
print('Read API definitions from', self.apifile)
def __struct (self, t, n = None, e = 0, vl = None):
base_types = { 'u8' : 'B',
'u16' : 'H',
'u32' : 'I',
'i32' : 'i',
'u64' : 'Q',
'f64' : 'd',
}
pack = None
if t in base_types:
pack = base_types[t]
if not vl:
if e and t == 'u8':
# Fixed byte array
return struct.Struct('>' + str(e) + 's')
if e:
# Fixed array of base type
return [e, struct.Struct('>' + base_types[t])]
else:
# Variable length array
return [vl, struct.Struct('>s')] if t == 'u8' else [vl, struct.Struct('>' + base_types[t])]
return struct.Struct('>' + base_types[t])
if t in self.messages:
### Return a list in case of array ###
if e and not vl:
return [e, lambda self, encode, buf, offset, args: (
self.__struct_type(encode, self.messages[t], buf, offset, args)
)]
if vl:
return [vl, lambda self, encode, buf, offset, args: (
self.__struct_type(encode, self.messages[t], buf, offset, args)
)]
return lambda self, encode, buf, offset, args: (
self.__struct_type(encode, self.messages[t], buf, offset, args)
)
raise ValueError
def __struct_type(self, encode, msgdef, buf, offset, kwargs):
if encode:
return self.__struct_type_encode(msgdef, buf, offset, kwargs)
else:
return self.__struct_type_decode(msgdef, buf, offset)
def __struct_type_encode(self, msgdef, buf, offset, kwargs):
off = offset
size = 0
for k in kwargs:
if k not in msgdef['args']:
raise ValueError(1, 'Invalid field-name in message call ' + k)
for k,v in msgdef['args'].iteritems():
off += size
if k in kwargs:
if type(v) is list:
if callable(v[1]):
if v[0] in kwargs:
e = kwargs[v[0]]
else:
e = v[0]
size = 0
for i in range(e):
size += v[1](self, True, buf, off + size,
kwargs[k][i])
else:
if v[0] in kwargs:
l = kwargs[v[0]]
else:
l = len(kwargs[k])
if v[1].size == 1:
buf[off:off + l] = bytearray(kwargs[k])
size = l
else:
size = 0
for i in kwargs[k]:
v[1].pack_into(buf, off + size, i)
size += v[1].size
else:
if callable(v):
size = v(self, True, buf, off, kwargs[k])
else:
v.pack_into(buf, off, kwargs[k])
size = v.size
else:
size = v.size if not type(v) is list else 0
return off + size - offset
def __getitem__(self, name):
if name in self.messages:
return self.messages[name]
return None
def encode(self, msgdef, kwargs):
# Make suitably large buffer
buf = bytearray(self.buffersize)
offset = 0
size = self.__struct_type(True, msgdef, buf, offset, kwargs)
return buf[:offset + size]
def decode(self, msgdef, buf):
return self.__struct_type(False, msgdef, buf, 0, None)[1]
def __struct_type_decode(self, msgdef, buf, offset):
res = []
off = offset
size = 0
for k,v in msgdef['args'].iteritems():
off += size
if type(v) is list:
lst = []
if callable(v[1]): # compound type
size = 0
if v[0] in msgdef['args']: # vla
e = int(res[-1])
else: # fixed array
e = v[0]
res.append(lst)
for i in range(e):
(s,l) = v[1](self, False, buf, off + size, None)
lst.append(l)
size += s
continue
if v[1].size == 1:
size = int(res[-1])
res.append(buf[off:off + size])
else:
e = int(res[-1])
if e == 0:
raise ValueError
lst = []
res.append(lst)
size = 0
for i in range(e):
lst.append(v[1].unpack_from(buf, off + size)[0])
size += v[1].size
else:
if callable(v):
(s,l) = v(self, False, buf, off, None)
res.append(l)
size += s
else:
res.append(v.unpack_from(buf, off)[0])
size = v.size
return off + size - offset, msgdef['return_tuple']._make(res)
def ret_tup(self, name):
if name in self.messages and 'return_tuple' in self.messages[name]:
return self.messages[name]['return_tuple']
return None
def add_message(self, name, msgdef):
if name in self.messages:
raise ValueError('Duplicate message name: ' + name)
args = collections.OrderedDict()
fields = []
msg = {}
for f in msgdef:
if type(f) is dict and 'crc' in f:
msg['crc'] = f['crc']
continue
field_name = f[1]
args[field_name] = self.__struct(*f)
fields.append(field_name)
msg['return_tuple']