"""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.plugins.manager import PluginManagerCompanion
mgr = PluginManagerCompanion()
mgr.proxy.attach(viewer)
mgr.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')
# Inventory for docs / notebooks / diagnostics:
rows = mgr.plugin_manager.list_available(refresh=True)
# each row has: name, display_name, loaded, enabled, active, activable, reason
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 .abc import AbstractUICompanion, MenuItem, SEPARATOR
from .loader import PLUGINS_DIR, PluginInfo, discover_plugins, discover_all_plugins
from ..PyTranslate import _
if TYPE_CHECKING:
from ..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, AbstractUICompanion]
Mapping of plugin name → the currently active companion instance.
"""
def __init__(self, viewer: 'WolfMapViewer') -> None:
[docs]
self.plugins: list[PluginInfo] = []
[docs]
self.active: dict[str, 'AbstractUICompanion'] = {}
# -- 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_all_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
[docs]
def list_available(
self,
*,
refresh: bool = True,
directory: Path | str | None = None,
) -> list[dict[str, object]]:
"""Return a normalized inventory of discovered plugins.
This is the preferred routine to know which plugin names are currently
activable from the registry.
:param refresh: When ``True`` (default), perform discovery first.
:param directory: Optional discovery override, forwarded to :meth:`discover`.
:return: List of rows with these keys:
``name``, ``display_name``, ``loaded``, ``enabled``, ``active``,
``activable``, ``reason``, ``load_error``.
"""
if refresh:
self.discover(directory)
rows: list[dict[str, object]] = []
for info in self.plugins:
is_active = info.name in self.active
if not info.loaded:
reason = 'load_error'
activable = False
elif is_active:
reason = 'already_active'
activable = False
else:
reason = 'ok'
activable = True
rows.append({
'name': info.name,
'display_name': info.display_name,
'loaded': info.loaded,
'enabled': info.enabled,
'active': is_active,
'activable': activable,
'reason': reason,
'load_error': info.load_error,
})
return rows
# -- 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
[docs]
def sync_enabled_state(self) -> None:
"""Read the persisted disabled list and apply changes to live plugins.
Called after the global options dialog has been closed so that any
enable/disable changes made there take effect immediately in the
running viewer, without requiring a restart.
"""
disabled = self._disabled_names()
for info in self.plugins:
should_be_enabled = info.name not in disabled
if should_be_enabled and not info.enabled:
self.enable(info.name)
elif not should_be_enabled and info.enabled:
self.disable(info.name)
# -- internal ------------------------------------------------------------
[docs]
def _activate(self, info: PluginInfo) -> None:
"""Instantiate, wire up menu, and store the companion."""
try:
companion = info.companion_class() # type: ignore[misc]
self._viewer.attach_companion(companion)
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 _disabled_names(self) -> list[str]:
cfg = self._get_config()
if cfg is None:
return []
try:
from ..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 ..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 AbstractUICompanion
# ---------------------------------------------------------------------------
[docs]
class PluginManagerCompanion(AbstractUICompanion):
"""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.plugins.manager import PluginManagerCompanion
mgr = PluginManagerCompanion()
mgr.proxy.attach(viewer)
mgr.build() # installs the "Plugins" menu
mgr.discover() # scans directory and activates enabled plugins
"""
def __init__(self) -> None:
[docs]
self._plugin_manager: PluginManager | None = None
@property
[docs]
def plugin_manager(self) -> PluginManager:
"""Underlying plugin lifecycle manager (created lazily after attach)."""
if self._plugin_manager is None:
viewer = self.proxy._viewer
if viewer is None:
raise RuntimeError(
'PluginManagerCompanion must be attached to a viewer before use.'
)
self._plugin_manager = PluginManager(viewer)
return self._plugin_manager
# -- AbstractUICompanion interface ---------------------------------------
[docs]
def start(self) -> None:
"""No-op — the manager operates through the menu, not interactively."""
# -- 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
[docs]
def list_available_plugins(
self,
*,
refresh: bool = True,
directory: Path | str | None = None,
) -> list[dict[str, object]]:
"""Return plugin inventory rows from the underlying registry.
Convenience wrapper used by notebooks/scripts that access the viewer's
already-instantiated plugin registry.
"""
return self.plugin_manager.list_available(refresh=refresh, directory=directory)
# -- private -------------------------------------------------------------
[docs]
def _make_toggle(self, name: str):
"""Return a menu-event handler that toggles plugin *name*."""
def _toggle(_ctx) -> 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, _ctx) -> None:
n_before = len(self.plugin_manager.plugins)
self.discover()
n_after = len(self.plugin_manager.plugins)
self.proxy.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, _ctx) -> None:
try:
chosen_str = self.proxy.ask_directory(_('Choose plugins directory'))
if chosen_str is not None:
chosen = Path(chosen_str)
self._persist_directory(chosen)
self.discover(chosen)
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 ..PyConfig import ConfigurationKeys
cfg[ConfigurationKeys.PLUGINS_DIRECTORY] = str(path)
cfg.save()
except Exception as exc:
_logger.warning("Could not persist plugins directory: %s", exc)