src/bindings/python/bt2/bt2/py_plugin.py

   1 # SPDX-License-Identifier: MIT
   2 #
   3 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
   4
   5 from bt2 import utils
   6 from bt2 import component as bt2_component
   7 import sys
   8
   9
  10 # Python plugin path to `_PluginInfo` (cache)
  11 _plugin_infos = {}
  12
  13
  14 def plugin_component_class(component_class):
  15     if not issubclass(component_class, bt2_component._UserComponent):
  16         raise TypeError("component class is not a subclass of a user component class")
  17
  18     component_class._bt_plugin_component_class = None
  19     return component_class
  20
  21
  22 def register_plugin(
  23     module_name, name, description=None, author=None, license=None, version=None
  24 ):
  25     if module_name not in sys.modules:
  26         raise RuntimeError(
  27             "cannot find module '{}' in loaded modules".format(module_name)
  28         )
  29
  30     utils._check_str(name)
  31
  32     if description is not None:
  33         utils._check_str(description)
  34
  35     if author is not None:
  36         utils._check_str(author)
  37
  38     if license is not None:
  39         utils._check_str(license)
  40
  41     if version is not None:
  42         if not _validate_version(version):
  43             raise ValueError(
  44                 "wrong version: expecting a tuple: (major, minor, patch) or (major, minor, patch, extra)"
  45             )
  46
  47     sys.modules[module_name]._bt_plugin_info = _PluginInfo(
  48         name, description, author, license, version
  49     )
  50
  51
  52 def _validate_version(version):
  53     if version is None:
  54         return True
  55
  56     if not isinstance(version, tuple):
  57         return False
  58
  59     if len(version) < 3 or len(version) > 4:
  60         return False
  61
  62     if not isinstance(version[0], int):
  63         return False
  64
  65     if not isinstance(version[1], int):
  66         return False
  67
  68     if not isinstance(version[2], int):
  69         return False
  70
  71     if len(version) == 4:
  72         if not isinstance(version[3], str):
  73             return False
  74
  75     return True
  76
  77
  78 class _PluginInfo:
  79     def __init__(self, name, description, author, license, version):
  80         self.name = name
  81         self.description = description
  82         self.author = author
  83         self.license = license
  84         self.version = version
  85         self.comp_class_addrs = None
  86
  87
  88 # called by the BT plugin system
  89 def _try_load_plugin_module(path):
  90     if path in _plugin_infos:
  91         # do not load module and create plugin info twice for this path
  92         return _plugin_infos[path]
  93
  94     import importlib.machinery
  95     import inspect
  96     import hashlib
  97
  98     if path is None:
  99         raise TypeError("missing path")
 100
 101     # In order to load the module uniquely from its path, even from
 102     # different files which have the same basename, we hash the path
 103     # and prefix with `bt_plugin_`. This is its key in sys.modules.
 104     h = hashlib.sha256()
 105     h.update(path.encode())
 106     module_name = "bt_plugin_{}".format(h.hexdigest())
 107     assert module_name not in sys.modules
 108     # try loading the module: any raised exception is catched by the caller
 109     mod = importlib.machinery.SourceFileLoader(module_name, path).load_module()
 110
 111     # we have the module: look for its plugin info first
 112     if not hasattr(mod, "_bt_plugin_info"):
 113         raise RuntimeError("missing '_bt_plugin_info' module attribute")
 114
 115     plugin_info = mod._bt_plugin_info
 116
 117     # search for user component classes
 118     def is_user_comp_class(obj):
 119         if not inspect.isclass(obj):
 120             return False
 121
 122         if not hasattr(obj, "_bt_plugin_component_class"):
 123             return False
 124
 125         return True
 126
 127     comp_class_entries = inspect.getmembers(mod, is_user_comp_class)
 128     plugin_info.comp_class_addrs = [entry[1].addr for entry in comp_class_entries]
 129     _plugin_infos[path] = plugin_info
 130     return plugin_info