89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
from typing import Dict, List
|
|
from plugins import AbstractPlugin, AbstractCommandCompiler
|
|
import logging
|
|
|
|
|
|
class ConflictingPlugins(Exception):
|
|
pass
|
|
|
|
|
|
class UnknownPlugin(Exception):
|
|
pass
|
|
|
|
|
|
class UnknownCommand(Exception):
|
|
pass
|
|
|
|
|
|
class PluginRepository:
|
|
|
|
def __init__(self):
|
|
self._registered_plugins: Dict[str, type(AbstractPlugin)] = {}
|
|
self._loaded_plugins: Dict[str, AbstractPlugin] = {}
|
|
self._command_compilers: Dict[str, AbstractCommandCompiler] = {}
|
|
self._logger = logging.getLogger("plugin_repository")
|
|
|
|
def register_plugin(self, plugin_class: type(AbstractPlugin)):
|
|
self._registered_plugins[plugin_class.plugin_name] = plugin_class
|
|
self._logger.debug(f"Registered plugin: {plugin_class.plugin_name}")
|
|
|
|
def load_plugins(self, plugins: List[str]):
|
|
for plugin in plugins:
|
|
self.load_plugin(plugin)
|
|
|
|
def load_plugin(self, plugin: str):
|
|
self._logger.debug(f"Loading plugin: {plugin}")
|
|
|
|
if plugin in self._loaded_plugins.keys():
|
|
self._logger.warning(f"Plugin {plugin} already loaded!")
|
|
return
|
|
|
|
# lookup plugin class
|
|
try:
|
|
plugin_cls = self._registered_plugins[plugin]
|
|
except KeyError:
|
|
raise UnknownPlugin(f"Tried to load unknown plugin: {plugin}")
|
|
|
|
# Create instance
|
|
plugin_instance = plugin_cls() # config is statically loaded
|
|
|
|
# load compilers
|
|
compilers = plugin_instance.load_compilers()
|
|
|
|
if set(compilers.keys()).intersection(self._command_compilers.keys()):
|
|
plugin_instance.close() # The program should stop at this point anyways
|
|
raise ConflictingPlugins(f"Conflicting plugin is tried to be loaded: {plugin}")
|
|
|
|
# Store commands
|
|
self._command_compilers.update(compilers)
|
|
|
|
# store instance
|
|
self._loaded_plugins[plugin] = plugin_instance
|
|
|
|
# log stuff
|
|
self._logger.info(f"Loaded plugin: {plugin}")
|
|
self._logger.debug(f"Plugin {plugin} loaded the following commands: {', '.join(compilers.keys())}")
|
|
|
|
def get_loaded_plugins(self) -> List[str]:
|
|
return list(self._loaded_plugins.keys())
|
|
|
|
def get_loaded_compilers(self) -> List[str]:
|
|
return list(self._command_compilers.keys())
|
|
|
|
def get_compiler(self, command: str) -> AbstractCommandCompiler:
|
|
try:
|
|
return self._command_compilers[command]
|
|
except KeyError:
|
|
raise UnknownCommand(f"Could not find compiler for command: {command}! A plugin might be not loaded.")
|
|
|
|
def get_plugin_instance(self, plugin: str) -> AbstractPlugin:
|
|
return self._loaded_plugins[plugin]
|
|
|
|
def close(self):
|
|
self._command_compilers = {}
|
|
for plugin_name, plugin_instance in self._loaded_plugins.items():
|
|
plugin_instance.close()
|
|
self._logger.info(f"Unloaded plugin: {plugin_name}")
|
|
|
|
self._loaded_plugins = {}
|