Source code for wolfhece._companion_plugin_manager

"""Companion plugin manager.

Manages discovery, loading, enabling and disabling of external companion
plugins, and integrates them into the viewer menu bar.

Quick usage
-----------
::

    from wolfhece._companion_plugin_manager import PluginManagerCompanion

    mgr = PluginManagerCompanion(viewer)
    mgr.menu_build()   # adds "Plugins" top-level menu
    mgr.discover()     # scans default directory and activates enabled plugins

    # Programmatic enable/disable:
    mgr.plugin_manager.disable('dot_picker')
    mgr.plugin_manager.enable('dot_picker')

Persistence
-----------
Enabled/disabled state is stored in the viewer's
:class:`~wolfhece.PyConfig.WolfConfiguration` under the
``PLUGINS_DISABLED`` key (list of disabled plugin names).

The plugins directory is stored under ``PLUGINS_DIRECTORY``
(empty string → built-in ``wolfhece/data/plugins``).
"""
from __future__ import annotations

import logging
from pathlib import Path
from typing import TYPE_CHECKING

from wolfhece._menu_companion_abc import AbstractCompanion, ActionItem, SEPARATOR
from wolfhece._plugin_loader import PLUGINS_DIR, PluginInfo, discover_plugins
from wolfhece.PyTranslate import _

if TYPE_CHECKING:
    from wolfhece.PyDraw import WolfMapViewer

[docs] _logger = logging.getLogger(__name__)
# --------------------------------------------------------------------------- # PluginManager — lifecycle and persistence # ---------------------------------------------------------------------------
[docs] class PluginManager: """Manages the runtime lifecycle of companion plugins for one viewer. Attributes ---------- plugins : list[PluginInfo] All discovered plugins (loaded or failed). active : dict[str, AbstractCompanion] Mapping of plugin name → currently active companion instance. """ def __init__(self, viewer: 'WolfMapViewer') -> None:
[docs] self._viewer = viewer
[docs] self.plugins: list[PluginInfo] = []
[docs] self.active: dict[str, AbstractCompanion] = {}
# -- discovery -----------------------------------------------------------
[docs] def discover(self, directory: Path | str | None = None) -> list[PluginInfo]: """Scan *directory*, load plugins, and apply the stored enable/disable state. :param directory: Override the plugins directory. ``None`` reads the value from the viewer configuration (falling back to the built-in ``wolfhece/data/plugins``). :return: The refreshed :attr:`plugins` list. """ if directory is None: directory = self._configured_directory() self.plugins = discover_plugins(directory) # Apply persisted enable/disable state disabled = self._disabled_names() for info in self.plugins: info.enabled = info.name not in disabled # Activate every enabled, loaded plugin that is not yet active for info in self.plugins: if info.enabled and info.loaded and info.name not in self.active: self._activate(info) return self.plugins
# -- enable / disable ----------------------------------------------------
[docs] def enable(self, name: str) -> bool: """Enable the plugin *name* and activate it if not already running. :return: ``True`` on success, ``False`` if the plugin was not found or could not be loaded. """ info = self._find(name) if info is None: _logger.warning("enable: plugin '%s' not found", name) return False if not info.loaded: _logger.warning( "enable: cannot activate '%s' — load error: %s", name, info.load_error, ) return False info.enabled = True self._save_state() if name not in self.active: self._activate(info) return True
[docs] def disable(self, name: str) -> bool: """Disable the plugin *name* and destroy its active companion. :return: ``True`` on success, ``False`` if the plugin was not found. """ info = self._find(name) if info is None: _logger.warning("disable: plugin '%s' not found", name) return False info.enabled = False self._save_state() companion = self.active.pop(name, None) if companion is not None: try: companion.destroy() except Exception as exc: _logger.warning("Error destroying companion '%s': %s", name, exc) return True
[docs] def is_enabled(self, name: str) -> bool: """Return whether the plugin *name* is currently enabled.""" info = self._find(name) return info.enabled if info is not None else False
# -- internal ------------------------------------------------------------
[docs] def _activate(self, info: PluginInfo) -> None: """Instantiate, wire up menu, and store the companion.""" try: companion = info.companion_class(self._viewer) # type: ignore[misc] companion.menu_build() self.active[info.name] = companion _logger.debug("Activated plugin '%s'", info.name) except Exception as exc: _logger.error( "Failed to activate plugin '%s': %s", info.name, exc, exc_info=True )
[docs] def _find(self, name: str) -> PluginInfo | None: for p in self.plugins: if p.name == name: return p return None
[docs] def _get_config(self): """Return the viewer's WolfConfiguration, or ``None`` if unavailable.""" try: return self._viewer.get_configuration() except Exception: return None
[docs] def _configured_directory(self) -> Path | None: """Read the plugins directory from the viewer config.""" cfg = self._get_config() if cfg is None: return None try: from wolfhece.PyConfig import ConfigurationKeys d = cfg[ConfigurationKeys.PLUGINS_DIRECTORY] return Path(d) if d else None except Exception: return None
[docs] def _disabled_names(self) -> list[str]: cfg = self._get_config() if cfg is None: return [] try: from wolfhece.PyConfig import ConfigurationKeys return list(cfg[ConfigurationKeys.PLUGINS_DISABLED]) except Exception: return []
[docs] def _save_state(self) -> None: cfg = self._get_config() if cfg is None: return try: from wolfhece.PyConfig import ConfigurationKeys disabled = [p.name for p in self.plugins if not p.enabled] cfg[ConfigurationKeys.PLUGINS_DISABLED] = disabled cfg.save() except Exception as exc: _logger.warning("Could not persist plugin state: %s", exc)
# --------------------------------------------------------------------------- # PluginManagerCompanion — viewer integration via AbstractCompanion # ---------------------------------------------------------------------------
[docs] class PluginManagerCompanion(AbstractCompanion): """Companion that manages external companion plugins. Adds a *Plugins* top-level menu to the viewer with items to discover, reload, and toggle individual plugins. Typical integration in a viewer subclass or startup script:: from wolfhece._companion_plugin_manager import PluginManagerCompanion mgr = PluginManagerCompanion(viewer) mgr.menu_build() # installs the "Plugins" menu mgr.discover() # scans directory and activates enabled plugins """ def __init__(self, viewer: 'WolfMapViewer') -> None: super().__init__(viewer) #: The underlying plugin lifecycle manager.
[docs] self.plugin_manager = PluginManager(viewer)
# -- AbstractCompanion interface -----------------------------------------
[docs] def start(self) -> None: """No-op — the manager operates through the menu, not interactively."""
[docs] def menu_build(self) -> None: """Build (or rebuild) the top-level *Plugins* menu.""" self._build_menu(_('Plugins'), self._menu_items())
# -- public API ----------------------------------------------------------
[docs] def discover(self, directory: Path | str | None = None) -> list[PluginInfo]: """Scan *directory*, activate plugins, then refresh the menu. :param directory: Override the plugins directory. ``None`` reads from the viewer configuration. :return: The refreshed list of :class:`~wolfhece._plugin_loader.PluginInfo`. """ plugins = self.plugin_manager.discover(directory) self._rebuild_menu() return plugins
# -- private -------------------------------------------------------------
[docs] def _menu_items(self) -> list: items: list = [ ActionItem( _('Discover & reload\u2026'), self._on_discover, _('Scan the plugins directory and reload all plugins'), ), ActionItem( _('Choose plugins directory\u2026'), self._on_choose_dir, _('Pick a custom directory to scan for plugins'), ), SEPARATOR, ] if not self.plugin_manager.plugins: items.append( ActionItem( _('(no plugins found)'), lambda e: None, _('Use "Discover & reload" to scan for plugins'), ) ) else: for info in self.plugin_manager.plugins: label = info.display_name if not info.loaded: label += _(' \u26a0 error') elif info.enabled: label += ' \u2713' items.append( ActionItem( label, self._make_toggle(info.name), info.manifest.description or _('Toggle this plugin on/off'), ) ) return items
[docs] def _make_toggle(self, name: str): """Return a menu-event handler that toggles plugin *name*.""" def _toggle(event) -> None: if self.plugin_manager.is_enabled(name): self.plugin_manager.disable(name) else: self.plugin_manager.enable(name) self._rebuild_menu() return _toggle
[docs] def _on_discover(self, event) -> None: n_before = len(self.plugin_manager.plugins) self.discover() n_after = len(self.plugin_manager.plugins) self._set_status( _('Plugins: {n} plugin(s) found.').format(n=n_after) if n_after else _('Plugins: no plugins found.') ) if n_after != n_before: _logger.info( "Plugin discovery: %d plugin(s) found (was %d)", n_after, n_before )
[docs] def _on_choose_dir(self, event) -> None: try: import wx dlg = wx.DirDialog( self._viewer, _('Choose plugins directory'), style=wx.DD_DEFAULT_STYLE, ) try: if dlg.ShowModal() == wx.ID_OK: chosen = Path(dlg.GetPath()) self._persist_directory(chosen) self.discover(chosen) finally: dlg.Destroy() except Exception as exc: _logger.error("Error choosing plugins directory: %s", exc, exc_info=True)
[docs] def _persist_directory(self, path: Path) -> None: cfg = self.plugin_manager._get_config() if cfg is None: return try: from wolfhece.PyConfig import ConfigurationKeys cfg[ConfigurationKeys.PLUGINS_DIRECTORY] = str(path) cfg.save() except Exception as exc: _logger.warning("Could not persist plugins directory: %s", exc)
[docs] def _rebuild_menu(self) -> None: try: self.menu_build() except Exception as exc: _logger.warning("Could not rebuild plugins menu: %s", exc)