universalrobots/single_ursim_control
3
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Added plugin conflict check

Browse Source
This commit is contained in:
Pünkösd Marcell 2021-04-09 00:15:32 +02:00
parent 49a5c07f98
commit fb2c243255
1 changed files with 28 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
single_ursim_control/plugin_repository.py
View File

@ -3,11 +3,15 @@ from plugins import AbstractPlugin, AbstractCommandCompiler
import logging import logging
class ConflictingPlugins(BaseException):
pass
class PluginRepository: class PluginRepository:
def __init__(self): def __init__(self):
self._registered_plugins: Dict[str, type(AbstractPlugin)] = {} self._registered_plugins: Dict[str, type(AbstractPlugin)] = {}
self._loaded_plugins: List[AbstractPlugin] = [] self._loaded_plugins: Dict[str, AbstractPlugin] = {}
self._command_compilers: Dict[str, AbstractCommandCompiler] = {} self._command_compilers: Dict[str, AbstractCommandCompiler] = {}
self._logger = logging.getLogger("plugin_repository") self._logger = logging.getLogger("plugin_repository")
@ -16,20 +20,38 @@ class PluginRepository:
self._logger.debug(f"Registered plugin: {plugin_class.plugin_name}") self._logger.debug(f"Registered plugin: {plugin_class.plugin_name}")
def load_plugin(self, plugin: str): def load_plugin(self, plugin: str):
if plugin in self._loaded_plugins.keys():
self._logger.warning(f"Plugin {plugin} already loaded!")
return
# create instance
plugin_instance = self._registered_plugins[plugin]() # config is statically loaded plugin_instance = self._registered_plugins[plugin]() # config is statically loaded
self._loaded_plugins.append(plugin_instance) # store instance
self._loaded_plugins[plugin] = plugin_instance
# load compilers
compilers = plugin_instance.load_compilers() compilers = plugin_instance.load_compilers()
if set(compilers.keys()).intersection(self._command_compilers.keys()):
raise ConflictingPlugins(f"Conflicting plugin is tried to be loaded: {plugin}")
# Store commands
self._command_compilers.update(compilers) self._command_compilers.update(compilers)
# log stuff
self._logger.info(f"Loaded plugin: {plugin}") self._logger.info(f"Loaded plugin: {plugin}")
self._logger.debug(f"Plugin {plugin} loaded the following commands: {', '.join(compilers.keys())}") self._logger.debug(f"Plugin {plugin} loaded the following commands: {', '.join(compilers.keys())}")
def get_compiler(self, command: str) -> AbstractCommandCompiler: def get_compiler(self, command: str) -> AbstractCommandCompiler:
return self._command_compilers[command] return self._command_compilers[command]
def get_plugin_instance(self, plugin: str) -> AbstractPlugin:
return self._loaded_plugins[plugin]
def close(self): def close(self):
self._command_compilers = [] self._command_compilers = {}
for plugin in self._loaded_plugins: for plugin_name, plugin_instance in self._loaded_plugins:
plugin.close() plugin_instance.close()
self._logger.info(f"Unloaded plugin: {plugin.plugin_name}") self._logger.info(f"Unloaded plugin: {plugin_name}")