Module pipettin-piper.piper.plugins.example

def load_plugin(controller: "Controller", **kwargs):
    """
    Plugins are expected to have a function named 'load_plugin' which will instantiate
    the plugin's class and returning it to the main Commander class.
    If they fail to load, they must raise a PluginError exception.
    """
    logging.debug(f"load_plugin: loading {plugin_name} plugin.")
    try:
        class_instance = PluginMonitor(controller=controller, **kwargs)
    except Exception as e:
        msg = f"Failed to load with error: {e}"
        logging.error(msg)
        raise PluginError(msg) from e

    return class_instance

class PluginMonitor(Plugin):
    """Example coroutine plugin, monitoring the status of other plugins."""
    def __init__(self, controller: Controller, **kwargs) -> None:
        self.controller = controller
        self.kwargs: dict = kwargs
        logging.info("Loading plugin with options: " + pformat(kwargs))

        # Example: add a coroutine for the controller to launch at start.
        self.controller.coroutine_methods.extend([
            self.check_plugins()
        ])

        # Register event handler for status updates of the program.
        self.controller.register_event_callback(
            event_name=self.controller.status_event,
            callback_name="Example status callback :)",
            callback_function=self.parse_status
        )

        # Set status.
        self._status = True

    async def parse_status(self, **kwargs):
        """Event system callbacks must be awaitable
        This is an example!
        """
        status = kwargs.get("status")
        if status:
            logging.debug("Got status information from the controller through the local events system :D")

    async def check_plugins(self):
        """Check if any plugin initialization returned an exception.
        And emit alerts to the GUI."""
        logging.info("Coroutine startup.")

        try:
            while self.controller.run:
                for name, plugin in self.controller.plugins.items():
                    if isinstance(plugin, Exception):
                        await self.controller.comms.sio_emit_alert(
                            text=f"Error when loading {name}: {plugin}",
                            alert_type='error')

                await asyncio.sleep(2)

        except asyncio.exceptions.CancelledError:
            logging.warning("Coroutine cancelled.")

        logging.warning("Coroutine ended.")