diff options
Diffstat (limited to 'plugins/plugin_manager.py')
| -rw-r--r-- | plugins/plugin_manager.py | 225 |
1 files changed, 225 insertions, 0 deletions
diff --git a/plugins/plugin_manager.py b/plugins/plugin_manager.py new file mode 100644 index 0000000..36ae71a --- /dev/null +++ b/plugins/plugin_manager.py @@ -0,0 +1,225 @@ +""" +plugins/plugin_manager.py + +Discovers and manages the lifecycle of LabDAQ plugins. + +Enabled state is persisted to plugins/enabled.json (one dict +{plugin_id: bool}). This survives app restarts; profiles store +per-plugin state separately via get_save_state / apply_save_state. +""" + +from __future__ import annotations + +import importlib.util +import json +import os +import sys +import traceback +from dataclasses import dataclass +from typing import Dict, List, Optional + +from plugins.base_plugin import LabPlugin, PluginContext + + +_MANIFEST_FILE = "manifest.json" +_ENABLED_FILE = "enabled.json" + + +# ── Manifest ────────────────────────────────────────────────────────────────── + +@dataclass +class PluginManifest: + plugin_id: str + name: str + version: str = "1.0.0" + description: str = "" + author: str = "" + entry_point: str = "plugin.Plugin" # "module.ClassName" relative to plugin dir + plugin_dir: str = "" + + +# ── Manager ─────────────────────────────────────────────────────────────────── + +class PluginManager: + """ + Discovers, loads, and lifecycle-manages LabDAQ plugins. + + Typical usage in MainWindow: + + self._plugins = PluginManager(plugins_dir) + self._plugins.discover() + for plugin in self._plugins.load_enabled(context): + self._install_plugin(plugin) + + Enable / disable at runtime: + + plugin = self._plugins.enable("my_plugin", context) + if plugin: + self._install_plugin(plugin) + + self._plugins.disable("my_plugin") + self._uninstall_plugin("my_plugin") + """ + + def __init__(self, plugins_dir: str): + self._dir = plugins_dir + self._enabled_path = os.path.join(plugins_dir, _ENABLED_FILE) + self._manifests: Dict[str, PluginManifest] = {} + self._loaded: Dict[str, LabPlugin] = {} + self._enabled: Dict[str, bool] = {} + self._load_enabled_state() + + # ── Discovery ───────────────────────────────────────────────────────── + + def discover(self) -> List[PluginManifest]: + """Scan plugin directory and return all found manifests.""" + self._manifests.clear() + if not os.path.isdir(self._dir): + return [] + + for entry in sorted(os.listdir(self._dir)): + plugin_dir = os.path.join(self._dir, entry) + if not os.path.isdir(plugin_dir): + continue + manifest_path = os.path.join(plugin_dir, _MANIFEST_FILE) + if not os.path.isfile(manifest_path): + continue + try: + with open(manifest_path) as f: + data = json.load(f) + m = PluginManifest( + plugin_id = data["plugin_id"], + name = data.get("name", entry), + version = data.get("version", "1.0.0"), + description = data.get("description", ""), + author = data.get("author", ""), + entry_point = data.get("entry_point", "plugin.Plugin"), + plugin_dir = plugin_dir, + ) + self._manifests[m.plugin_id] = m + except Exception as exc: + print(f"[PluginManager] Bad manifest in '{entry}': {exc}") + + return list(self._manifests.values()) + + # ── Enabled state ───────────────────────────────────────────────────── + + def _load_enabled_state(self): + if os.path.isfile(self._enabled_path): + try: + with open(self._enabled_path) as f: + self._enabled = json.load(f) + except Exception: + self._enabled = {} + + def _save_enabled_state(self): + os.makedirs(self._dir, exist_ok=True) + with open(self._enabled_path, "w") as f: + json.dump(self._enabled, f, indent=2) + + def is_enabled(self, plugin_id: str) -> bool: + return self._enabled.get(plugin_id, False) + + def get_enabled_ids(self) -> List[str]: + return [pid for pid, on in self._enabled.items() if on] + + # ── Load / unload ───────────────────────────────────────────────────── + + def load_enabled(self, context: PluginContext) -> List[LabPlugin]: + """Load all enabled plugins that have a discovered manifest.""" + result = [] + for plugin_id in self.get_enabled_ids(): + if plugin_id not in self._manifests: + continue + p = self._load_plugin(plugin_id, context) + if p: + result.append(p) + return result + + def _load_plugin(self, plugin_id: str, + context: PluginContext) -> Optional[LabPlugin]: + if plugin_id in self._loaded: + return self._loaded[plugin_id] + + manifest = self._manifests.get(plugin_id) + if not manifest: + print(f"[PluginManager] No manifest for '{plugin_id}'") + return None + + module_name, class_name = manifest.entry_point.rsplit(".", 1) + module_file = os.path.join( + manifest.plugin_dir, *module_name.split("/") + ) + ".py" + + # Each stage named so errors are unambiguous + stage = "locating module file" + try: + if not os.path.isfile(module_file): + raise FileNotFoundError(f"'{module_file}' does not exist") + + stage = "importing module" + spec = importlib.util.spec_from_file_location( + f"_labdaq_plugin_{plugin_id}", module_file + ) + mod = importlib.util.module_from_spec(spec) + if manifest.plugin_dir not in sys.path: + sys.path.insert(0, manifest.plugin_dir) + spec.loader.exec_module(mod) + + stage = f"finding class '{class_name}'" + if not hasattr(mod, class_name): + raise AttributeError( + f"'{module_file}' has no class '{class_name}'" + ) + cls = getattr(mod, class_name) + + stage = "instantiating plugin class" + plugin: LabPlugin = cls() + + stage = "calling on_load()" + plugin.on_load(context) + + self._loaded[plugin_id] = plugin + print(f"[Plugin] Loaded '{plugin.name}' v{plugin.version}") + return plugin + + except Exception as exc: + print(f"[Plugin] ERROR — could not load '{plugin_id}' " + f"(failed at {stage}): {exc}") + traceback.print_exc() + return None + + def _unload_plugin(self, plugin_id: str): + plugin = self._loaded.pop(plugin_id, None) + if plugin is None: + return + try: + plugin.on_unload() + except Exception as exc: + print(f"[Plugin] ERROR — '{plugin_id}' on_unload() raised: {exc}") + traceback.print_exc() + print(f"[Plugin] Unloaded '{plugin_id}'") + + # ── Public API ──────────────────────────────────────────────────────── + + def enable(self, plugin_id: str, + context: PluginContext) -> Optional[LabPlugin]: + """Mark enabled, persist, load and return the plugin (or None on error).""" + self._enabled[plugin_id] = True + self._save_enabled_state() + return self._load_plugin(plugin_id, context) + + def disable(self, plugin_id: str): + """Unload the plugin and persist the disabled state.""" + self._enabled[plugin_id] = False + self._save_enabled_state() + self._unload_plugin(plugin_id) + + def get_manifests(self) -> List[PluginManifest]: + return list(self._manifests.values()) + + def get_loaded(self) -> List[LabPlugin]: + return list(self._loaded.values()) + + def get_plugin(self, plugin_id: str) -> Optional[LabPlugin]: + return self._loaded.get(plugin_id) |