import importlib import copy import os import inspect import logging from openflexure_microscope.api.views import MicroscopeViewPlugin class ConColors: HEADER = "\033[95m" OKBLUE = "\033[94m" OKGREEN = "\033[92m" WARNING = "\033[93m" FAIL = "\033[91m" ENDC = "\033[0m" BOLD = "\033[1m" UNDERLINE = "\033[4m" def module_from_file(plugin_path): # Expand environment variables in path string plugin_path = os.path.expandvars(plugin_path) # Expand user directory in path string plugin_path = os.path.expanduser(plugin_path) # Check if the path is to a file if not os.path.isfile(plugin_path): logging.warning( ConColors.FAIL + "No valid plugin found at {}.".format(plugin_path) + ConColors.ENDC ) return None, None, None else: # Get name of plugin from the file plugin_name = os.path.splitext(os.path.basename(plugin_path))[0] plugin_spec = importlib.util.spec_from_file_location(plugin_name, plugin_path) plugin_module = importlib.util.module_from_spec(plugin_spec) return plugin_spec, plugin_module, plugin_name def check_module(module_path): """ Used to check if a module exists, without ever importing it. This checks each nested level separately to avoid raising exceptions. For example, checking "mymodule.submodule.subsubmodule" will first check if 'mymodule' exists, then if it does, it will check 'mymodule.submodule', and so on. """ module_split = module_path.split(".") spec_path = "" for module_level in module_split: spec_path += ".{}".format(module_level) spec_path = spec_path.strip(".") logging.debug("Checking {}".format(spec_path)) # Try to find a module loader for the plugin path plugin_spec = importlib.util.find_spec(spec_path) # If plugin spec doesn't exist, return False early if plugin_spec is None: return False # If all checks pass, return True return True def name_from_module(plugin_path): path_array = plugin_path.split(".") # Truncate namespace of default plugins if path_array[0:2] == ["openflexure_microscope", "plugins"]: path_array = path_array[2:] return "/".join(path_array) def load_plugin_module(plugin_path): # If the loader was found (i.e. plugin probably exists) if check_module(plugin_path): try: plugin_module = importlib.import_module(plugin_path) plugin_name = name_from_module(plugin_path) except Exception as e: logging.warning("Error loading plugin.") logging.error(e) return None, None # If no loader was found, try finding a file from path else: plugin_spec, plugin_module, plugin_name = module_from_file(plugin_path) # If a valid plugin was found if plugin_spec and plugin_module: # Execute the module, so we have access to it plugin_spec.loader.exec_module(plugin_module) return plugin_module, plugin_name def load_plugin_class(plugin_path, plugin_class_name): plugin_module, plugin_name = load_plugin_module(plugin_path) if plugin_module: # Now try to extract the class try: plugin_class = getattr(plugin_module, plugin_class_name) except AttributeError: logging.warning( ConColors.FAIL + "Class {} does not exist in plugin {}. Skipping.".format( plugin_class_name, plugin_path ) + ConColors.ENDC ) return None, None else: return plugin_class, plugin_name else: return None, None def class_from_map(plugin_map): plugin_arr = plugin_map.split(":") if not len(plugin_arr) == 2: logging.warning( ConColors.WARNING + "Malformed plugin map {}. Skipping.".format(plugin_map) + ConColors.ENDC ) return None, None else: return load_plugin_class(*plugin_arr) def expand_views(plugin_object, plugin_name): routes = [] # If plugin contains valid endpoints if hasattr(plugin_object, "api_views") and isinstance(plugin_object.api_views, dict): # We'll keep a record of how each route was expanded expanded_routes = {} # For each defined endpoint for view_route, view_class in plugin_object.api_views.items(): # Remove all leading slashes from view route cleaned_route = view_route while cleaned_route[0] == "/": cleaned_route = cleaned_route[1:] # Construct a full view route from the plugin name full_view_route = "/{}/{}".format(plugin_name, cleaned_route) logging.debug(full_view_route) # Record how the view_route got expanded expanded_routes[view_route] = full_view_route # Check if endpoint name clashes if issubclass(view_class, MicroscopeViewPlugin): # Create a Python-safe name for the route plugin_route_id = "plugin{}".format(full_view_route).replace( "/", "_" ) # Add route to the plugin representation dictionary route_representation = { "id": plugin_route_id, "route": full_view_route, "links": {}, } routes.append(route_representation) return routes class PluginLoader(object): """ A mount-point for all loaded plugins. Attaches to a Microscope object. Args: parent (:py:class:`openflexure_microscope.microscope.Microscope`): The parent Microscope object to attach to. """ def __init__(self, parent): self.parent = parent self._plugins = [] # List of plugin objects self.forms = [] # List of plugin forms logging.info("Creating plugin mount") @property def state(self): # DEPRECATED logging.warning( "PluginMount.state is deprecated. Use PluginMount.active instead. State will be removed in a future version." ) return [m["python_name"] for m in self._plugins] @property def active(self): return self._plugins def attach(self, plugin_map): """ Attach a MicroscopePlugin instance to the plugin mount. Args: plugin_map (str): A plugin map describing the file or module to load a MicroscopePlugin child from. Maps should be in the format 'module.to.load:ClassName' or '/path/to/file:ClassName'. """ plugin_class, plugin_name = class_from_map(plugin_map) if plugin_class is not None: plugin_name_python_safe = plugin_name.replace("/", "_") if plugin_class and plugin_name: plugin_object = plugin_class() if hasattr( self, plugin_name ): # If a plugin with the same name is already attached. logging.warning( ConColors.WARNING + "A plugin named {} has already been loaded. Skipping {}.".format( plugin_name, plugin_map ) + ConColors.ENDC ) elif isinstance( plugin_object, MicroscopePlugin ): # If plugin_object is an instance of MicroscopePlugin # Attach plugin_object to the plugin mount setattr(self, plugin_name_python_safe, plugin_object) # Store the plugin object, and it's properties self._plugins.append( { "name": plugin_name, "python_name": plugin_name_python_safe, "plugin": plugin_object, "routes": [], "form": None, } ) # Grant plugin access to the hardware plugin_object.microscope = self.parent logging.info( ConColors.OKGREEN + "Plugin {} loaded as {}.".format(plugin_map, plugin_name) + ConColors.ENDC ) else: logging.warning(f"Error loading plugin {plugin_map}. Moving on.") class MicroscopePlugin: """ Parent class for all microscope plugins. Initially only defines an empty object for microscope. All plugins must be an instance of this class to successfully attach to PluginMount. """ api_views = {} # Initially empty dictionary of API views associated with the plugin def __init__(self): self.microscope = ( None ) #: :py:class:`openflexure_microscope.microscope.Microscope`: Microscope object