path: root/plugins/plugin_manager.py

"""
plugins/plugin_manager.py

Discovers and manages the lifecycle of LabUI 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 = ""
    source_url:  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 LabUI plugins.

    Typical usage in MainWindow:

        self._plugins = PluginManager(user_dir, extra_scan_dirs=[project_plugins])
        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, user_dir: str, extra_scan_dirs: List[str] = None):
        self._user_dir     = user_dir
        self._scan_dirs    = [user_dir] + list(extra_scan_dirs or [])
        self._enabled_path = os.path.join(user_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 all plugin directories and return all found manifests."""
        self._manifests.clear()
        for scan_dir in self._scan_dirs:
            if not os.path.isdir(scan_dir):
                continue
            for entry in sorted(os.listdir(scan_dir)):
                plugin_dir = os.path.join(scan_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,
                        source_url  = data.get("source_url", ""),
                    )
                    if m.plugin_id not in self._manifests:
                        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._user_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 = self.get_missing_dependencies(plugin_id)
        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"_labui_plugin_{plugin_id}", module_file
            )
            mod = importlib.util.module_from_spec(spec)
            vendor = os.path.join(manifest.plugin_dir, "vendor")
            if os.path.isdir(vendor) and vendor not in sys.path:
                sys.path.insert(0, vendor)
            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_manifest(self, plugin_id: str) -> Optional[PluginManifest]:
        return self._manifests.get(plugin_id)

    def get_missing_dependencies(self, plugin_id: str) -> List[str]:
        manifest = self._manifests.get(plugin_id)
        if not manifest:
            return []
        vendor = os.path.join(manifest.plugin_dir, "vendor")
        if os.path.isdir(vendor):
            added = vendor not in sys.path
            if added:
                sys.path.insert(0, vendor)
            try:
                return missing_requirements(manifest.requires)
            finally:
                if added:
                    sys.path.remove(vendor)
        return missing_requirements(manifest.requires)

    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)

    # ── Install / uninstall ───────────────────────────────────────────────

    def install_from_zip(self, zip_path: str) -> PluginManifest:
        """Extract a plugin zip into user_dir and re-discover."""
        import shutil
        import zipfile

        with zipfile.ZipFile(zip_path) as zf:
            names = zf.namelist()
            top_dirs = {n.split("/")[0] for n in names if "/" in n}
            if not top_dirs:
                raise ValueError("Plugin zip must contain a top-level directory")
            if len(top_dirs) != 1:
                raise ValueError(
                    f"Plugin zip must contain exactly one top-level directory, found: {sorted(top_dirs)}"
                )
            plugin_folder = top_dirs.pop()
            if f"{plugin_folder}/manifest.json" not in names:
                raise ValueError(f"No {plugin_folder}/manifest.json found in zip")
            dest = os.path.join(self._user_dir, plugin_folder)
            if os.path.exists(dest):
                shutil.rmtree(dest)
            os.makedirs(self._user_dir, exist_ok=True)
            zf.extractall(self._user_dir)

        self.discover()
        for m in self._manifests.values():
            if os.path.basename(m.plugin_dir) == plugin_folder:
                return m
        raise ValueError("Plugin installed but not found after discovery")

    def uninstall(self, plugin_id: str):
        """Delete a user-installed plugin from disk. Must be disabled first."""
        import shutil

        manifest = self._manifests.get(plugin_id)
        if manifest is None:
            raise ValueError(f"Plugin '{plugin_id}' not found")
        if not self.is_user_installed(plugin_id):
            raise ValueError(f"Plugin '{plugin_id}' is not user-installed and cannot be removed")
        if self.is_enabled(plugin_id):
            raise RuntimeError(f"Disable '{plugin_id}' before uninstalling")
        shutil.rmtree(manifest.plugin_dir)
        self._manifests.pop(plugin_id, None)

    def is_user_installed(self, plugin_id: str) -> bool:
        """True if the plugin lives inside user_dir (i.e. can be uninstalled)."""
        manifest = self._manifests.get(plugin_id)
        if manifest is None:
            return False
        return os.path.abspath(manifest.plugin_dir).startswith(
            os.path.abspath(self._user_dir)
        )

    def get_user_dir(self) -> str:
        return self._user_dir