""" 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)