path: root/plugins/plugin_manager.py

"""
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.metadata
import importlib.util
import json
import os
import re
import sys
import traceback
from dataclasses import dataclass, field
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
    requires:    List[str] = field(default_factory=list)  # pip-style reqs, e.g. "opencv-python>=4.8.0"
    plugin_dir:  str = ""


def _dist_name(requirement: str) -> str:
    """Extract the distribution name from a requirement string, e.g.
    "opencv-python>=4.8.0" -> "opencv-python"."""
    return re.split(r"[<>=!~\[; ]", requirement.strip(), maxsplit=1)[0]


def missing_requirements(requires: List[str]) -> List[str]:
    """Return the subset of `requires` whose distribution isn't installed.

    Checked by distribution name via importlib.metadata (matches what pip
    installed it as), not by import name — those differ for packages like
    opencv-python (imports as cv2) or pyserial (imports as serial).
    """
    missing = []
    for req in requires:
        name = _dist_name(req)
        if not name:
            continue
        try:
            importlib.metadata.version(name)
        except importlib.metadata.PackageNotFoundError:
            missing.append(req)
    return missing


# ── 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"),
                    requires    = data.get("requires", []),
                    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

        missing = missing_requirements(manifest.requires)
        if missing:
            print(f"[Plugin] '{plugin_id}' missing dependencies: {', '.join(missing)}")
            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_missing_dependencies(self, plugin_id: str) -> List[str]:
        manifest = self._manifests.get(plugin_id)
        return missing_requirements(manifest.requires) if manifest else []

    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)