166 lines
4.8 KiB
Python
166 lines
4.8 KiB
Python
import logging
|
|
import collections
|
|
import copy
|
|
|
|
from importlib import util
|
|
import sys
|
|
import os
|
|
import glob
|
|
|
|
from openflexure_microscope.common.labthings_core.utilities import (
|
|
get_docstring,
|
|
camel_to_snake,
|
|
snake_to_spine,
|
|
)
|
|
|
|
|
|
class BaseExtension:
|
|
"""
|
|
Parent class for all extensions.
|
|
|
|
Handles binding route views and forms.
|
|
"""
|
|
|
|
def __init__(self, name: str, description=""):
|
|
self._views = (
|
|
{}
|
|
) # Key: Full, Python-safe ID. Val: Original rule, and view class
|
|
self._rules = {} # Key: Original rule. Val: View class
|
|
self._gui = None
|
|
|
|
self._cls = str(self) # String description of extension instance
|
|
|
|
self.actions = []
|
|
self.properties = []
|
|
|
|
self.name = name
|
|
self.description = get_docstring(self)
|
|
|
|
self.methods = {}
|
|
|
|
@property
|
|
def views(self):
|
|
return self._views
|
|
|
|
def add_view(self, view_class, rule, **kwargs):
|
|
# Remove all leading slashes from view route
|
|
cleaned_rule = rule
|
|
while cleaned_rule[0] == "/":
|
|
cleaned_rule = cleaned_rule[1:]
|
|
|
|
# Expand the rule to include extension name
|
|
full_rule = "/{}/{}".format(self._name_uri_safe, cleaned_rule)
|
|
|
|
view_id = cleaned_rule.replace("/", "_").replace("<", "").replace(">", "")
|
|
|
|
# Create a Python-safe route ID
|
|
logging.debug(view_id)
|
|
|
|
# Store route information in a dictionary
|
|
d = {"rule": full_rule, "view": view_class, "kwargs": kwargs}
|
|
|
|
# Add view to private views dictionary
|
|
self._views[view_id] = d
|
|
# Store the rule expansion information
|
|
self._rules[rule] = self._views[view_id]
|
|
|
|
def register_action(self, view_class):
|
|
self.actions.append(view_class)
|
|
|
|
def register_property(self, view_class):
|
|
self.properties.append(view_class)
|
|
|
|
@property
|
|
def gui(self):
|
|
print(self._gui)
|
|
# Handle missing/no GUI
|
|
if not self._gui:
|
|
return None
|
|
# Handle GUI as a dictionary-returning callable function
|
|
elif isinstance(self._gui, collections.Callable):
|
|
api_gui = self._gui()
|
|
# Handle GUI as a static dictionary
|
|
elif isinstance(self._gui, dict):
|
|
api_gui = copy.deepcopy(self._gui)
|
|
# Handle borked GUI
|
|
else:
|
|
raise ValueError(
|
|
"GUI must be a dictionary, or a dictionary-returning function with no arguments."
|
|
)
|
|
|
|
api_gui["id"] = self._name
|
|
|
|
if "forms" in api_gui and isinstance(api_gui["forms"], list):
|
|
for form in api_gui["forms"]:
|
|
if "route" in form and form["route"] in self._rules.keys():
|
|
form["route"] = self._rules[form["route"]]["rule"]
|
|
else:
|
|
logging.warn(
|
|
"No valid expandable route found for {}".format(form["route"])
|
|
)
|
|
return api_gui
|
|
|
|
def set_gui(self, form_dictionary: dict):
|
|
self._gui = form_dictionary
|
|
|
|
@property
|
|
def _name(self):
|
|
return self.name
|
|
|
|
@property
|
|
def _name_python_safe(self):
|
|
name = camel_to_snake(self._name) # Camel to snake
|
|
name = name.replace(" ", "_") # Spaces to snake
|
|
return name
|
|
|
|
@property
|
|
def _name_uri_safe(self):
|
|
return snake_to_spine(self._name_python_safe)
|
|
|
|
def add_method(self, method, method_name):
|
|
self.methods[method_name] = method
|
|
|
|
if not hasattr(self, method_name):
|
|
setattr(self, method_name, method)
|
|
else:
|
|
logging.warning(
|
|
"Unable to bind method to extension. Method name already exists."
|
|
)
|
|
|
|
|
|
def find_instances_in_module(module, class_to_find):
|
|
objs = []
|
|
for attribute in dir(module):
|
|
if not attribute.startswith("__"):
|
|
if isinstance(getattr(module, attribute), class_to_find):
|
|
objs.append(getattr(module, attribute))
|
|
return objs
|
|
|
|
|
|
def find_extensions_in_file(extension_path: str, module_name="extensions"):
|
|
logging.debug(f"Loading extensions from {extension_path}")
|
|
|
|
spec = util.spec_from_file_location(module_name, extension_path)
|
|
mod = util.module_from_spec(spec)
|
|
sys.modules[spec.name] = mod
|
|
|
|
spec.loader.exec_module(mod)
|
|
|
|
if hasattr(mod, "__extensions__"):
|
|
return [getattr(mod, ext_name) for ext_name in mod.__extensions__]
|
|
else:
|
|
return find_instances_in_module(mod, BaseExtension)
|
|
|
|
|
|
def find_extensions(extension_dir: str, module_name="extensions"):
|
|
logging.debug(f"Loading extensions from {extension_dir}")
|
|
|
|
extensions = []
|
|
extension_paths = glob.glob(os.path.join(extension_dir, "*.py"))
|
|
|
|
for extension_path in extension_paths:
|
|
extensions.extend(
|
|
find_extensions_in_file(extension_path, module_name=module_name)
|
|
)
|
|
|
|
return extensions
|