import ipaddress
from . import macaddress
-try:
- text_type = unicode
-except NameError:
- text_type = str
# Copies from vl_api_address_t definition
ADDRESS_IP4 = 0
def format_vl_api_prefix_t(args):
if isinstance(args, (ipaddress.IPv4Network, ipaddress.IPv6Network)):
return {'address': format_vl_api_address_t(
- text_type(args.network_address)),
+ str(args.network_address)),
'len': int(args.prefixlen)}
p, length = args.split('/')
return {'address': format_vl_api_address_t(p),
def format_vl_api_address_with_prefix_t(args):
if isinstance(args, (ipaddress.IPv4Interface, ipaddress.IPv6Interface)):
return {'address': format_vl_api_address_t(
- text_type(args.network_address)),
+ str(args.network_address)),
'len': int(args.prefixlen)}
p, length = args.split('/')
return {'address': format_vl_api_address_t(p),
import sys
import multiprocessing as mp
import os
+import queue
import logging
import functools
import json
logger = logging.getLogger('vpp_papi')
logger.addHandler(logging.NullHandler())
-if sys.version[0] == '2':
- import Queue as queue
-else:
- import queue as queue
-
__all__ = ('FuncWrapper', 'VPP', 'VppApiDynamicMethodHolder',
'VppEnum', 'VppEnumType',
'VPPIOError', 'VPPRuntimeError', 'VPPValueError',
@metaclass(VppEnumType)
-class VppEnum(object):
+class VppEnum:
pass
vpp_instance.disconnect()
-if sys.version[0] == '2':
- def vpp_iterator(d):
- return d.iteritems()
-else:
- def vpp_iterator(d):
- return d.items()
def add_convenience_methods():
ipaddress._IPAddressBase.vapi_af_name = property(_vapi_af_name)
-class VppApiDynamicMethodHolder(object):
+class VppApiDynamicMethodHolder:
pass
-class FuncWrapper(object):
+class FuncWrapper:
def __init__(self, func):
self._func = func
self.__name__ = func.__name__
pass
-class VPPApiJSONFiles(object):
+class VPPApiJSONFiles:
@classmethod
def find_api_dir(cls, dirs):
"""Attempt to find the best directory in which API definition
return messages, services
-class VPPApiClient(object):
+class VPPApiClient:
"""VPP interface.
This class provides the APIs to VPP. The APIs are loaded
def get_function(self, name):
return getattr(self._api, name)
- class ContextId(object):
+ class ContextId:
"""Multiprocessing-safe provider of unique context IDs."""
def __init__(self):
self.context = mp.Value(ctypes.c_uint, 0)
self.id_names = [None] * (self.vpp_dictionary_maxid + 1)
self.id_msgdef = [None] * (self.vpp_dictionary_maxid + 1)
self._api = VppApiDynamicMethodHolder()
- for name, msg in vpp_iterator(self.messages):
+ for name, msg in self.messages.items():
n = name + '_' + msg.crc[2:]
i = self.transport.get_msg_index(n)
if i > 0:
#
logger = logging.getLogger('vpp_papi.serializer')
-if sys.version[0] == '2':
- def check(d): type(d) is dict
-else:
- def check(d): type(d) is dict or type(d) is bytes
+
+def check(d):
+ return type(d) is dict or type(d) is bytes
def conversion_required(data, field_type):
return vpp_format.conversion_unpacker_table[field_type](data)
-# TODO: post 20.01, remove inherit from object.
-class Packer(object):
+class Packer:
options = {}
def pack(self, data, kwargs):
def __bool__(self):
return True
- # TODO: Remove post 20.01.
- if sys.version[0] == '2':
- __nonzero__ = __bool__
def pack(self, data, kwargs=None):
if data is None: # Default to zero if not specified
pass
-class VPPStats(object):
+class VPPStats:
VPPStatsIOError = VPPStatsIOError
default_socketname = '/run/vpp/stats.sock'
super(VppTransportShmemIOError, self).__init__(rv, descr)
-class VppTransport(object):
+class VppTransport:
VppTransportShmemIOError = VppTransportShmemIOError
def __init__(self, parent, read_timeout, server_address):
import threading
import select
import multiprocessing
-try:
- import queue as queue
-except ImportError:
- import Queue as queue
+import queue
import logging
from . import vpp_papi
pass
-class VppTransport(object):
+class VppTransport:
VppTransportSocketIOError = VppTransportSocketIOError
def __init__(self, parent, read_timeout, server_address):