Source code for autopilot.utils.plugins

Utility functions for handling plugins, eg. importing, downloading, listing, confirming, etc.
import sys
import inspect
import importlib
from typing import Union, Optional
from pathlib import Path
from multiprocessing import Value
from ctypes import c_bool
from collections import OrderedDict as odict

import requests

from autopilot.core.loggers import init_logger
from autopilot import prefs
from autopilot.utils import wiki

_IMPORTED = Value(c_bool, False)

[docs]def import_plugins(plugin_dir:Optional[Path]=None) -> dict: """ Import all plugins in the plugin (or supplied) directory. There is no specific form for a plugin at the moment, so this function will recursively import all modules and packages within the directory. Plugins can then be accessed by the :func:`~.utils.registry.get` registry functions. Args: plugin_dir (None, :class:`pathlib.Path`): Directory to import. if ``None`` (default), use ``prefs.get('PLUGINDIR')``. Returns: dict: of imported objects with form {"class_name": class_object} """ logger = init_logger(module_name='plugins') if plugin_dir is not None: plugin_dir = Path(plugin_dir) else: plugin_dir = Path(prefs.get('PLUGINDIR')) if not plugin_dir.exists(): logger.exception(f"Plugin directory {plugin_dir} does not exist!") return {} plugins = {} # recursively list python files in plugin directory plugin_files = Path(plugin_dir).glob('**/*.py') for pfile in plugin_files: # prepend autopilot.plugins to avoid namespace clashes and make them uniformly moduled module_name = 'autopilot.plugins.' + inspect.getmodulename(str(pfile)) # import module try: if module_name in sys.modules: mod = sys.modules[module_name] else: mod_spec = importlib.util.spec_from_file_location(module_name, pfile) mod = importlib.util.module_from_spec(mod_spec) mod_spec.loader.exec_module(mod) sys.modules[module_name] = mod except Exception as e: logger.exception(f'plugin file {str(pfile)} could not be imported, got exception: {e}') continue # get the imported modules and check if they're one of ours # filter for classes and then for classes that are declared in the plugin # module itself members = inspect.getmembers(mod, inspect.isclass) members = [member for member in members if member[1].__module__ == module_name] # store in the plugin dictionary to be returned, for member in members: # if a plugin object with the same name was already found, warn the user and don't overwrite if member[0] in plugins.keys(): logger.warning(f'Conflicting plugin object names, importing both but only returning first found: {member[0]}\nfound first in: {plugins[member[0]]}\nand now in: {member[1].__module__}') continue plugins[member[0]] = member[1] globals()['_IMPORTED'].value = True return plugins
[docs]def unload_plugins(): """ Un-import imported plugins (mostly for testing purposes) """ mods = [mod for mod in sys.modules if 'autopilot.plugins' in mod] for mod in mods: del sys.modules[mod] _IMPORTED.value = False
[docs]def list_wiki_plugins(): """ List plugins available on the wiki using :func:`` Returns: dict: {'plugin_name': {'plugin_prop':'prop_value',...} """ plugins = wiki.ask(filters="[[Category:Autopilot Plugin]]", properties=[ 'Created By', 'Has Description', 'For Autopilot Version', 'Has Git Repository', 'Has Contributor', 'Is Autopilot Plugin Type', 'Controls Hardware', 'Has DOI' ]) ordered_plugins = [] # reorder fields in plugins with odicts for plugin in plugins: oplugin = odict() for field in ('name','Has Description', 'Created By', 'Has Contributor', 'url','Has Git Repository', 'Has DOI', 'For Autopilot Version', 'Is Autopilot Plugin Type', 'Controls Hardware'): oplugin[field] = plugin[field] ordered_plugins.append(oplugin) return plugins