"""Plug-in functionality."""
import os
import yaml
import deode
from .logs import logger
[docs]
class DeodePluginRegistry:
"""Registry of plugins."""
def __init__(self, config=None):
"""Construct the plugin registry.
Args:
config (dict, optional): Registry config. Defaults to None.
Raises:
RuntimeError: "Can not set plugin registry"
"""
if config is None:
config = {}
self.config_input = config
self.config = self.get_registry_config()
self.plugins = []
self.deode_plugin()
self.load_plugins()
[docs]
def deode_plugin(self):
"""Base DEODE plugin."""
path = os.path.dirname(deode.__path__[0])
plugin = DeodePlugin("deode", path)
self.register_plugin(plugin)
[docs]
def load_plugin(self, plugin):
"""Load plugin.
Args:
plugin (DeodePlugin): Deode plugin
"""
self.plugins += [plugin]
self.config["plugins"].update({plugin.name: plugin.path})
[docs]
def load_plugins(self):
"""Load all registered plugins."""
plugins = self.config["plugins"]
for name, path in plugins.items():
if name != "deode":
plugin = DeodePlugin(name, path)
self.load_plugin(plugin)
[docs]
def plugin_exists(self, plugin):
"""Check if plugin exists.
Args:
plugin (DeodePlugin): Deode plugin
Returns:
bool: True if already exists in registry.
"""
return any(plg.name == plugin.name for plg in self.plugins)
[docs]
def register_plugin(self, plugin):
"""Register plugin.
Args:
plugin (DeodePlugin): Deode plugin
"""
if not self.plugin_exists(plugin):
self.config["plugins"].update({plugin.name: plugin.path})
self.load_plugin(plugin)
else:
logger.warning("Plugin {} does already exists", plugin.name)
[docs]
def get_registry_config(self):
"""Get the registry config."""
config = self.config_input
if "plugins" not in config:
config.update({"plugins": {}})
return config
[docs]
def save_registry(self, config_file):
"""Save registry config.
Args:
config_file (str): Filename
"""
with open(config_file, mode="w", encoding="utf8") as fh:
yaml.safe_dump(self.config, fh)
[docs]
class DeodePluginRegistryFromFile(DeodePluginRegistry):
"""Registry file of plugins."""
def __init__(self, config_file):
"""Construct the plugin registry.
Args:
config_file (str, optional): Registry config. Defaults to None.
Raises:
RuntimeError: "Can not set plugin registry"
"""
with open(config_file, mode="r", encoding="utf8") as fh:
config = yaml.safe_load(fh)
DeodePluginRegistry.__init__(self, config)
[docs]
class DeodePluginRegistryFromConfig(DeodePluginRegistry):
"""Create a registry from a deode config file."""
def __init__(self, config):
"""Construct a registry from a deode config.
Args:
config (ParsedConfig): Deode config.
"""
try:
plugin_registry = config["general.plugin_registry"].dict()
except KeyError:
plugin_registry = None
DeodePluginRegistry.__init__(self, plugin_registry)
[docs]
class DeodePlugin:
"""Deode plugin."""
def __init__(self, name, path):
"""Construct the plugin.
Args:
name (str): Lower case name of plugin
path (str): Root path to plugin.
"""
self.name = name.lower()
self.path = path
self.tasks_path = f"{self.path}/{self.name}/tasks"
self.suites_path = f"{self.path}/{self.name}/suites"
[docs]
class DeodePluginFromConfigFile(DeodePlugin):
"""Deode plugin."""
def __init__(self, config_file):
"""Construct the plugin from a config file.
Args:
config_file (str): Config file.
"""
config = self.get_plugin_config(config_file)
name = config["name"]
path = config["path"]
DeodePlugin.__init__(self, name, path)
[docs]
@staticmethod
def get_plugin_config(config_file):
"""Get the registry config."""
with open(config_file, mode="r", encoding="utf8") as fh:
config = yaml.safe_load(fh)
return config