Source code for openalea.core.plugin.manager

"""
This plugin manager is inspired by nose PluginManager(s) released under LGPL license.
You can get a full copy of this license on nosetest repository:`lgpl.txt <https://github.com/nose-devs/nose/blob/master/lgpl.txt>`_
You can get original nose code on `github <https://github.com/nose-devs/nose/blob/master/nose/plugins/manager.py>`_


Plugin Manager
--------------

A plugin manager class is used to load plugins, search among it and manage the list of loaded plugins.
Plugins are loaded from entry points or can be added dynamically to manager.
  - To *list* plugins, see :meth:`PluginManager.plugin` and :meth:`PluginManager.plugins`.
  - To *add* plugins dynamically, see :meth:`PluginManager.add_plugin` and :meth:`PluginManager.add_plugins`.

All plugin are sorted in categories, each group defining a contract.
This contract is generally described in an interface class or documentation.

If you want to use third party plugins that doesn't fit perfectly to your contract,
you can embed its in plugin proxies.
To do that, you can specify a proxy class for an entire group or for one plugin.
See :meth:`PluginManager.set_proxy` and "plugin_proxy" parameter in :meth:`PluginManager.add_plugin`.

"""

import inspect
from warnings import warn

from openalea.core import logger
from openalea.core.manager import GenericManager
from openalea.core.plugin.plugin import PluginDef
from openalea.core.service.introspection import name
from openalea.core.util import camel_case_to_lower

from pkg_resources import iter_entry_points

__all__ = ['PluginManager']


class UnknownItemError(Exception):
    pass


def get_criteria(plugin):
    criteria = {}
    for criterion in dir(plugin):
        if criterion.startswith('_'):
            continue
        elif criterion in ('implementation', 'name_conversion', 'identifier', 'tags', 'criteria'):
            continue
        criteria[criterion] = getattr(plugin, criterion)
    return criteria


def get_implementation(plugin):
    if hasattr(plugin, 'modulename') and hasattr(plugin, 'objectname'):
        modulename = plugin.modulename
        objectname = plugin.objectname
        module = __import__(modulename, fromlist=[objectname])
        return getattr(module, objectname)
    else:
        return plugin()


def plugin_name(plugin):
    return plugin.name if hasattr(plugin, 'name') else plugin.__class__.__name__


def drop_plugin(name):
    try:
        idx = name.lower().index('plugin')
    except ValueError:
        pass
    else:
        name = name[0:idx] + name[idx + 6:]
    return name


[docs]class PluginManager(GenericManager): @classmethod
[docs] def generate_item_name(cls, item): try: name = item.name except AttributeError: try: name = item.__class__.__name__ except AttributeError: name = str(item.__class__) if hasattr(item, 'name_conversion'): if item.name_conversion == PluginDef.UNCHANGED: return name elif item.name_conversion == PluginDef.DROP_PLUGIN: return drop_plugin(name) else: return camel_case_to_lower(drop_plugin(name)) else: return name
[docs] def generate_item_id(self, plugin): return ':'.join([plugin.__class__.__module__, plugin.__class__.__name__])
[docs] def discover(self, group=None, item_proxy=None): if "entry_points" in self._autoload: for ep in iter_entry_points(group): self._load_entry_point_plugin(group, ep, item_proxy=item_proxy)
[docs] def instantiate(self, item): if inspect.isclass(item): return item() else: raise NotImplementedError
[docs] def patch_item(self, item): if not hasattr(item, "name_conversion"): item.name_conversion = 2 if not hasattr(item, "criteria"): item.__class__.criteria = property(fget=get_criteria) GenericManager.patch_item(self, item) # Look in class dict instead of hasattr(item, 'implementation') to avoid loading implementation if not hasattr(item.__class__, 'implementation'): item.__class__.implementation = property(fget=get_implementation)
[docs] def patch_ep_plugin(self, plugin, ep): plugin.plugin_ep = ep.name plugin.plugin_modulename = ep.module_name plugin.plugin_dist = ep.dist
def _is_plugin_class(self, obj): if hasattr(obj, '__plugin__'): return True elif hasattr(obj, 'modulename') and hasattr(obj, 'objectname'): return True else: return False def _add_plugin_from_ep(self, group, ep, plugin_class, plugin_proxy=None): logger.debug('%s load plugin %s' % (self.__class__.__name__, ep)) from time import time if inspect.ismodule(plugin_class): plugin_classes = [] for pl_name in dir(plugin_class): pl = getattr(plugin_class, pl_name) if self._is_plugin_class(pl): plugin_classes.append(pl) elif isinstance(plugin_class, list): plugin_classes = plugin_class else: plugin_classes = [plugin_class] for plugin_class in plugin_classes: name = plugin_class.name if hasattr(plugin_class, 'name') else plugin_class.__name__ parts = [str(s) for s in (ep.dist.egg_name(), group, ep.module_name, ep.name, name)] identifier = ':'.join(parts) item = self.add(plugin_class, group, item_proxy=plugin_proxy, identifier=identifier) self.patch_ep_plugin(item, ep) def _load_entry_point_plugin(self, group, entry_point, item_proxy=None): ep = entry_point plugin_class = None if self.debug: plugin_class = ep.load() logger.debug('%s load plugin %s' % (self.__class__.__name__, ep)) self._add_plugin_from_ep(group, ep, plugin_class, item_proxy) else: try: plugin_class = ep.load() except Exception: logger.error('%s: error loading %s ' % (group, ep)) else: self._add_plugin_from_ep(group, ep, plugin_class, item_proxy)
class SimpleClassPluginProxy(object): """ Plugin approach used in OpenAlea is : entry_point --(load)--> plugin_class --(instantiate)--> plugin --call--> RealClass If you want to use third party plugins that follow approach: entry_point --(load)--> RealClass You can use this class as a proxy. RealClass is now embeded in a SimpleClassPluginProxy and can be reached with "klass" attribute. Plugin is now compatible with pluginmanager. Then, you can define meta-information, generally generated from RealClass attributes. By default, plugin name is "RealClass" name class ThirdPartyProxy(SimpleClassPluginProxy): label = property(fget=lambda self: self.klass.title) .. warning:: You should not use this proxy because the plugin may slow down the entire application: all code and imports defined in module containing "RealClass" are loaded at first query or listing instead of only when used """ __plugin__ = True def __init__(self, klass): self.klass = klass self.name = klass.name if hasattr(klass, "name") else klass.__class__.__name__ def __call__(self): return self def implementation(self): return self.klass