Spaces:
Running
Running
File size: 6,225 Bytes
638fd44 5d77e0d 638fd44 5d77e0d 638fd44 5d77e0d 638fd44 5d77e0d 638fd44 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | """Plugin System (BioNexus 2.0, Component 20).
Extends the platform at runtime: lightweight Python plugins declare hook
callbacks that are dispatched by the PluginManager around engine validation,
exports and pipeline events. Plugins live in BIONEXUS_PLUGINS_DIR (default
app/plugins/) as plain .py files; each subclass of BioNexusPlugin is loaded,
registered and enabled. All plugin failures are contained (never break the
host operation).
"""
from __future__ import annotations
import importlib.util
import logging
import os
import threading
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
PLUGIN_DIR = os.environ.get("BIONEXUS_PLUGINS_DIR")
if not PLUGIN_DIR:
PLUGIN_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "plugins")
class BioNexusPlugin:
"""Override the hook methods you care about. `name`/`version` are required."""
name: str = "unnamed"
version: str = "0.0.0"
description: str = ""
#: lifecycle + pipeline hooks (all optional, all contained-safe)
def on_register(self) -> None:
pass
def on_disable(self) -> None:
pass
def before_validate(self, engine_name: str, report: dict, result: Any = None) -> list[dict]:
"""Return extra validation checks [{name, passed, detail}] appended to
the engine's native PASS/FAIL report. `result` is the parsed EngineResult."""
return []
def after_export(self, engine_name: str, payload: dict) -> dict | None:
"""Optionally decorate an exported artifact (returned dict is merged)."""
return None
def on_event(self, event: str, payload: dict) -> list[dict]:
"""Receive arbitrary platform events as a trace record."""
return []
class PluginManager:
def __init__(self, plugin_dir: str | None = None) -> None:
self.dir = plugin_dir or PLUGIN_DIR
self.plugins: dict[str, BioNexusPlugin] = {}
self.enabled: set[str] = set()
self.trace: list[dict] = []
self._lock = threading.Lock()
# --- discovery ----------------------------------------------------------
def load_dir(self, directory: str | None = None) -> list[str]:
"""Import every *.py plugin in the (default) plugin directory."""
scan_dir = directory or self.dir
loaded: list[str] = []
if not os.path.isdir(scan_dir):
return loaded
for path in sorted(Path(scan_dir).glob("*.py")):
if path.name.startswith("_"):
continue
before = set(self.plugins)
self._load_file(path)
if set(self.plugins) - before:
loaded.append(path.stem)
return loaded
def _load_file(self, path: Path) -> None:
try:
spec = importlib.util.spec_from_file_location(f"bionexus_plugin_{path.stem}", path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
except Exception as e:
logger.warning("plugin %s import failed: %s", path.name, e)
return
added = False
for obj in vars(mod).values():
if isinstance(obj, type) and issubclass(obj, BioNexusPlugin) and obj is not BioNexusPlugin:
plugin = obj()
try:
plugin.on_register()
self.plugins[plugin.name] = plugin
self.enabled.add(plugin.name)
except Exception as e:
logger.warning("plugin %s register failed: %s", plugin.name, e)
def reload(self) -> list[str]:
with self._lock:
self.plugins = {}
self.enabled = set()
return self.load_dir()
# --- state --------------------------------------------------------------
def list_plugins(self) -> list[dict]:
with self._lock:
return [{"name": p.name, "version": p.version, "description": p.description,
"enabled": p.name in self.enabled}
for p in self.plugins.values()]
def set_enabled(self, name: str, enabled: bool) -> bool:
with self._lock:
if name not in self.plugins:
return False
if enabled:
self.enabled.add(name)
else:
self.enabled.discard(name)
try:
self.plugins[name].on_disable()
except Exception as e:
logger.warning("plugin %s on_disable failed: %s", name, e)
return True
# --- hooks --------------------------------------------------------------
def _active(self) -> list[BioNexusPlugin]:
with self._lock:
return [p for n, p in self.plugins.items() if n in self.enabled]
def before_validate(self, engine_name: str, report: dict, result: Any = None) -> list[dict]:
"""Collect extra checks from every active plugin, containing failures."""
extra: list[dict] = []
for p in self._active():
try:
extra.extend(p.before_validate(engine_name, report, result))
except Exception as e:
extra.append({"name": f"{p.name}:validate_error", "passed": False, "detail": str(e)})
return extra
def after_export(self, engine_name: str, payload: dict) -> dict:
merged = dict(payload)
for p in self._active():
try:
decor = p.after_export(engine_name, payload)
if isinstance(decor, dict):
merged.update(decor)
except Exception as e:
logger.warning("plugin %s.after_export failed: %s", p.name, e)
return merged
def run_event(self, event: str, payload: dict) -> list[dict]:
records: list[dict] = []
for p in self._active():
try:
records.extend(p.on_event(event, dict(payload)))
except Exception as e:
records.append({"plugin": p.name, "event": event, "error": str(e)})
self.trace.append({"event": event, "payload": payload, "records": records})
return records
plugin_manager = PluginManager() |