+ vpp_instance = vpp_weakref()
+ if vpp_instance and vpp_instance.transport.connected:
+ logger.debug("Cleaning up VPP on exit")
+ vpp_instance.disconnect()
+
+
+def add_convenience_methods():
+ # provide convenience methods to IP[46]Address.vapi_af
+ def _vapi_af(self):
+ if 6 == self._version:
+ return VppEnum.vl_api_address_family_t.ADDRESS_IP6.value
+ if 4 == self._version:
+ return VppEnum.vl_api_address_family_t.ADDRESS_IP4.value
+ raise ValueError("Invalid _version.")
+
+ def _vapi_af_name(self):
+ if 6 == self._version:
+ return "ip6"
+ if 4 == self._version:
+ return "ip4"
+ raise ValueError("Invalid _version.")
+
+ ipaddress._IPAddressBase.vapi_af = property(_vapi_af)
+ ipaddress._IPAddressBase.vapi_af_name = property(_vapi_af_name)
+
+
+class VppApiDynamicMethodHolder:
+ pass
+
+
+class FuncWrapper:
+ def __init__(self, func):
+ self._func = func
+ self.__name__ = func.__name__
+ self.__doc__ = func.__doc__
+
+ def __call__(self, **kwargs):
+ return self._func(**kwargs)
+
+ def __repr__(self):
+ return "<FuncWrapper(func=<%s(%s)>)>" % (self.__name__, self.__doc__)
+
+
+class VPPApiError(Exception):
+ pass
+
+
+class VPPNotImplementedError(NotImplementedError):
+ pass
+
+
+class VPPIOError(IOError):
+ pass
+
+
+class VPPRuntimeError(RuntimeError):
+ pass
+
+
+class VPPValueError(ValueError):
+ pass
+
+
+class VPPApiJSONFiles:
+ @classmethod
+ def find_api_dir(cls, dirs=[]):
+ """Attempt to find the best directory in which API definition
+ files may reside. If the value VPP_API_DIR exists in the environment
+ then it is first on the search list. If we're inside a recognized
+ location in a VPP source tree (src/scripts and src/vpp-api/python)
+ then entries from there to the likely locations in build-root are
+ added. Finally the location used by system packages is added.
+
+ :returns: A single directory name, or None if no such directory
+ could be found.
+ """
+
+ # perhaps we're in the 'src/scripts' or 'src/vpp-api/python' dir;
+ # in which case, plot a course to likely places in the src tree
+ import __main__ as main
+
+ if os.getenv("VPP_API_DIR"):
+ dirs.append(os.getenv("VPP_API_DIR"))
+
+ if hasattr(main, "__file__"):
+ # get the path of the calling script
+ localdir = os.path.dirname(os.path.realpath(main.__file__))
+ else:
+ # use cwd if there is no calling script
+ localdir = os.getcwd()
+ localdir_s = localdir.split(os.path.sep)
+
+ def dmatch(dir):
+ """Match dir against right-hand components of the script dir"""
+ d = dir.split("/") # param 'dir' assumes a / separator
+ length = len(d)
+ return len(localdir_s) > length and localdir_s[-length:] == d
+
+ def sdir(srcdir, variant):
+ """Build a path from srcdir to the staged API files of
+ 'variant' (typically '' or '_debug')"""
+ # Since 'core' and 'plugin' files are staged
+ # in separate directories, we target the parent dir.
+ return os.path.sep.join(
+ (
+ srcdir,
+ "build-root",
+ "install-vpp%s-native" % variant,
+ "vpp",
+ "share",
+ "vpp",
+ "api",
+ )
+ )
+
+ srcdir = None
+ if dmatch("src/scripts"):
+ srcdir = os.path.sep.join(localdir_s[:-2])
+ elif dmatch("src/vpp-api/python"):
+ srcdir = os.path.sep.join(localdir_s[:-3])
+ elif dmatch("test"):
+ # we're apparently running tests
+ srcdir = os.path.sep.join(localdir_s[:-1])
+
+ if srcdir:
+ # we're in the source tree, try both the debug and release
+ # variants.
+ dirs.append(sdir(srcdir, "_debug"))
+ dirs.append(sdir(srcdir, ""))
+
+ # Test for staged copies of the scripts
+ # For these, since we explicitly know if we're running a debug versus
+ # release variant, target only the relevant directory
+ if dmatch("build-root/install-vpp_debug-native/vpp/bin"):
+ srcdir = os.path.sep.join(localdir_s[:-4])
+ dirs.append(sdir(srcdir, "_debug"))
+ if dmatch("build-root/install-vpp-native/vpp/bin"):
+ srcdir = os.path.sep.join(localdir_s[:-4])
+ dirs.append(sdir(srcdir, ""))
+
+ # finally, try the location system packages typically install into
+ dirs.append(os.path.sep.join(("", "usr", "share", "vpp", "api")))
+
+ # check the directories for existence; first one wins
+ for dir in dirs:
+ if os.path.isdir(dir):
+ return dir
+
+ return None
+
+ @classmethod
+ def find_api_files(cls, api_dir=None, patterns="*"): # -> list
+ """Find API definition files from the given directory tree with the
+ given pattern. If no directory is given then find_api_dir() is used
+ to locate one. If no pattern is given then all definition files found
+ in the directory tree are used.
+
+ :param api_dir: A directory tree in which to locate API definition
+ files; subdirectories are descended into.
+ If this is None then find_api_dir() is called to discover it.
+ :param patterns: A list of patterns to use in each visited directory
+ when looking for files.
+ This can be a list/tuple object or a comma-separated string of
+ patterns. Each value in the list will have leading/trialing
+ whitespace stripped.
+ The pattern specifies the first part of the filename, '.api.json'
+ is appended.
+ The results are de-duplicated, thus overlapping patterns are fine.
+ If this is None it defaults to '*' meaning "all API files".
+ :returns: A list of file paths for the API files found.
+ """
+ if api_dir is None:
+ api_dir = cls.find_api_dir([])
+ if api_dir is None:
+ raise VPPApiError("api_dir cannot be located")
+
+ if isinstance(patterns, list) or isinstance(patterns, tuple):
+ patterns = [p.strip() + ".api.json" for p in patterns]
+ else:
+ patterns = [p.strip() + ".api.json" for p in patterns.split(",")]
+
+ api_files = []
+ for root, dirnames, files in os.walk(api_dir):
+ # iterate all given patterns and de-dup the result
+ files = set(sum([fnmatch.filter(files, p) for p in patterns], []))
+ for filename in files:
+ api_files.append(os.path.join(root, filename))
+
+ return api_files
+
+ @classmethod
+ def process_json_file(self, apidef_file):
+ api = json.load(apidef_file)
+ return self._process_json(api)
+
+ @classmethod
+ def process_json_str(self, json_str):
+ api = json.loads(json_str)
+ return self._process_json(api)
+
+ @classmethod
+ def process_json_array_str(self, json_str):
+ services = {}
+ messages = {}
+
+ apis = json.loads(json_str)
+ for a in apis:
+ m, s = self._process_json(a)
+ messages.update(m)
+ services.update(s)
+ return messages, services
+
+ @staticmethod
+ def _process_json(api): # -> Tuple[Dict, Dict]
+ types = {}
+ services = {}
+ messages = {}
+ try:
+ for t in api["enums"]:
+ t[0] = "vl_api_" + t[0] + "_t"
+ types[t[0]] = {"type": "enum", "data": t}
+ except KeyError:
+ pass
+ try:
+ for t in api["enumflags"]:
+ t[0] = "vl_api_" + t[0] + "_t"
+ types[t[0]] = {"type": "enum", "data": t}
+ except KeyError:
+ pass
+ try:
+ for t in api["unions"]:
+ t[0] = "vl_api_" + t[0] + "_t"
+ types[t[0]] = {"type": "union", "data": t}
+ except KeyError:
+ pass
+
+ try:
+ for t in api["types"]:
+ t[0] = "vl_api_" + t[0] + "_t"
+ types[t[0]] = {"type": "type", "data": t}
+ except KeyError:
+ pass
+
+ try:
+ for t, v in api["aliases"].items():
+ types["vl_api_" + t + "_t"] = {"type": "alias", "data": v}
+ except KeyError:
+ pass
+
+ try:
+ services.update(api["services"])
+ except KeyError:
+ pass
+
+ i = 0
+ while True:
+ unresolved = {}
+ for k, v in types.items():
+ t = v["data"]
+ if not vpp_get_type(k):
+ if v["type"] == "enum":
+ try:
+ VPPEnumType(t[0], t[1:])
+ except ValueError:
+ unresolved[k] = v
+ if not vpp_get_type(k):
+ if v["type"] == "enumflag":
+ try:
+ VPPEnumFlagType(t[0], t[1:])
+ except ValueError:
+ unresolved[k] = v
+ elif v["type"] == "union":
+ try:
+ VPPUnionType(t[0], t[1:])
+ except ValueError:
+ unresolved[k] = v
+ elif v["type"] == "type":
+ try:
+ VPPType(t[0], t[1:])
+ except ValueError:
+ unresolved[k] = v
+ elif v["type"] == "alias":
+ try:
+ VPPTypeAlias(k, t)
+ except ValueError:
+ unresolved[k] = v
+ if len(unresolved) == 0:
+ break
+ if i > 3:
+ raise VPPValueError("Unresolved type definitions {}".format(unresolved))
+ types = unresolved
+ i += 1
+ try:
+ for m in api["messages"]:
+ try:
+ messages[m[0]] = VPPMessage(m[0], m[1:])
+ except VPPNotImplementedError:
+ logger.error("Not implemented error for {}".format(m[0]))
+ except KeyError:
+ pass
+ return messages, services
+
+ @staticmethod
+ def load_api(apifiles=None, apidir=None):
+ messages = {}
+ services = {}
+ if not apifiles:
+ # Pick up API definitions from default directory
+ try:
+ if isinstance(apidir, list):
+ apifiles = []
+ for d in apidir:
+ apifiles += VPPApiJSONFiles.find_api_files(d)
+ else:
+ apifiles = VPPApiJSONFiles.find_api_files(apidir)
+ except (RuntimeError, VPPApiError):
+ raise VPPRuntimeError
+
+ for file in apifiles:
+ with open(file) as apidef_file:
+ m, s = VPPApiJSONFiles.process_json_file(apidef_file)
+ messages.update(m)
+ services.update(s)
Code Review / vpp.git / blobdiff