mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
ChuckandClaude Opus 5.5
b518c51679
- load_plugin: an on_enable() that raises unregisters the instance, so the next load retries instead of returning True "already loaded". - get_plugin_info: guard plugin.get_info(); one plugin raising no longer breaks /api/v3/plugins/installed. - plugin_state.json and the operation history are written with atomic_write_text under their lock. - plugin_loader: module-level lock serialises pip installs across the parallel startup loaders. - store_manager._install_via_download: extract dir cleanup moved to finally. - Test doubles: draw_image() warns (DeprecationWarning; the real DisplayManager has none), MockDisplayManager.draw_text accepts the real signature's optional params, VisualTestDisplayManager logs draw errors at WARNING. - Docs/comments: compatibility.py method name, PluginState.LOADED meaning, brittle schema count, why _report_skip_once uses setdefault. - Remove unused PluginOperationQueue.get_active_operations(). Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
829 lines
34 KiB
Python
829 lines
34 KiB
Python
"""
|
|
Plugin Loader
|
|
|
|
Handles plugin module imports, dependency installation, and class instantiation.
|
|
Extracted from PluginManager to improve separation of concerns.
|
|
"""
|
|
|
|
import errno
|
|
import importlib
|
|
import importlib.metadata
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
import subprocess
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Dict, Any, List, Optional, Tuple, Type
|
|
import logging
|
|
|
|
from packaging.requirements import InvalidRequirement, Requirement
|
|
|
|
from src.exceptions import PluginError
|
|
from src.logging_config import get_logger
|
|
from src.plugin_system.plugin_dirs import resolve_plugin_dir
|
|
|
|
#: Serialises pip runs across threads. Startup loads plugins on a small
|
|
#: thread pool, and two concurrent ``pip install`` processes writing the same
|
|
#: site-packages can corrupt it or fail on each other's partial installs.
|
|
_PIP_INSTALL_LOCK = threading.Lock()
|
|
|
|
|
|
def requirements_has_real_deps(requirements_file: str) -> bool:
|
|
"""
|
|
Check whether a requirements.txt actually specifies anything to install.
|
|
|
|
Plugins that ship all their dependencies with LEDMatrix core often keep a
|
|
requirements.txt where every line is commented out, for documentation
|
|
purposes only. Running pip against such a file still pays the full
|
|
subprocess/resolver cost for zero effect, so callers should skip the
|
|
install step entirely when this returns False.
|
|
"""
|
|
try:
|
|
with open(requirements_file, 'r', encoding='utf-8') as fh:
|
|
for line in fh:
|
|
line = line.strip()
|
|
if line and not line.startswith('#'):
|
|
return True
|
|
except OSError:
|
|
# Let the caller's own file handling report the error.
|
|
return True
|
|
return False
|
|
|
|
|
|
def _extra_dependencies(dist_name: str, extras) -> Optional[List[Requirement]]:
|
|
"""Dependencies a distribution declares *only* behind the given extras.
|
|
|
|
Returns None when the installed metadata cannot be read or parsed, so the
|
|
caller can fall back to running pip rather than assuming anything.
|
|
"""
|
|
try:
|
|
meta = importlib.metadata.metadata(dist_name)
|
|
except importlib.metadata.PackageNotFoundError:
|
|
return None
|
|
|
|
gated: List[Requirement] = []
|
|
for raw in meta.get_all('Requires-Dist') or []:
|
|
try:
|
|
dep = Requirement(raw)
|
|
except InvalidRequirement:
|
|
return None
|
|
if dep.marker is None:
|
|
continue
|
|
# Keep only what the distribution gates behind an extra we asked for:
|
|
# satisfied when `extra` is that name, but not when no extra is
|
|
# requested. A marker that holds either way (python_version, sys_platform)
|
|
# belongs to the base install and is already covered by the version check.
|
|
if dep.marker.evaluate({'extra': ''}):
|
|
continue
|
|
if any(dep.marker.evaluate({'extra': extra}) for extra in extras):
|
|
gated.append(dep)
|
|
return gated
|
|
|
|
|
|
def _extras_are_satisfied(req: Requirement, _visited: Optional[set] = None) -> bool:
|
|
"""Check the dependencies pulled in by req's extras are installed.
|
|
|
|
Follows extras through nested extras. A gated dependency can itself request
|
|
one (`requests[socks]`), and checking only that `requests` is installed at
|
|
an acceptable version says nothing about whether the socks extra's own
|
|
dependency is there -- so the caller would skip pip and the plugin would
|
|
fail at import instead. Plain dependencies are still checked one level
|
|
deep, which is all that is needed to tell "the extra was installed" from
|
|
"the extra was never installed".
|
|
|
|
`_visited` carries the (distribution, extras) pairs already seen, so a
|
|
dependency cycle between extras terminates instead of recursing forever.
|
|
Anything unreadable returns False, so the caller still falls through to pip.
|
|
"""
|
|
if _visited is None:
|
|
_visited = set()
|
|
marker = (req.name.lower(), frozenset(e.lower() for e in req.extras))
|
|
if marker in _visited:
|
|
# Already accounted for higher up the chain; treating a cycle as
|
|
# satisfied here is safe because the outer frame still has to pass.
|
|
return True
|
|
_visited.add(marker)
|
|
|
|
gated = _extra_dependencies(req.name, req.extras)
|
|
if gated is None:
|
|
return False
|
|
|
|
for dep in gated:
|
|
try:
|
|
dep_version = importlib.metadata.version(dep.name)
|
|
except importlib.metadata.PackageNotFoundError:
|
|
return False
|
|
if dep.specifier and not dep.specifier.contains(dep_version, prereleases=True):
|
|
return False
|
|
if dep.extras and not _extras_are_satisfied(dep, _visited):
|
|
return False
|
|
return True
|
|
|
|
|
|
def requirements_are_satisfied(requirements_file: str) -> bool:
|
|
"""
|
|
Check whether every real requirement line in requirements.txt is already
|
|
satisfied by packages installed in the current interpreter.
|
|
|
|
This replaces marker-file tracking with a direct fact check, so it's
|
|
immune to stale/missing/corrupted markers: it looks at what's actually
|
|
importable right now rather than trusting a hash comparison from a
|
|
previous run. Anything ambiguous (pip options, unparseable lines,
|
|
extras, unresolvable versions) conservatively returns False so the
|
|
caller falls through to running pip — this check only ever saves work,
|
|
never masks a real install.
|
|
"""
|
|
try:
|
|
with open(requirements_file, 'r', encoding='utf-8') as fh:
|
|
lines = fh.readlines()
|
|
except OSError:
|
|
return False
|
|
|
|
for raw_line in lines:
|
|
line = raw_line.strip()
|
|
if not line or line.startswith('#'):
|
|
continue
|
|
if line.startswith('-'):
|
|
return False # pip option (-r, --index-url, ...), can't verify
|
|
|
|
try:
|
|
req = Requirement(line)
|
|
except InvalidRequirement:
|
|
return False
|
|
|
|
if req.marker is not None and not req.marker.evaluate():
|
|
continue # not applicable on this platform/interpreter
|
|
|
|
try:
|
|
installed_version = importlib.metadata.version(req.name)
|
|
except importlib.metadata.PackageNotFoundError:
|
|
return False
|
|
|
|
if req.specifier and not req.specifier.contains(installed_version, prereleases=True):
|
|
return False
|
|
|
|
if req.extras and not _extras_are_satisfied(req):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def find_trusted_subdir(trusted_dir: str, name: str) -> Optional[str]:
|
|
"""Return `name` if it names an actual subdirectory of trusted_dir, else None.
|
|
|
|
Used as a containment check for a directory name derived from untrusted
|
|
input (a manifest-declared plugin id, an externally-supplied plugin
|
|
path): the returned value always comes from enumerating trusted_dir
|
|
itself via os.scandir(), so a caller that builds a path by joining
|
|
trusted_dir with this return value is joining against a name the
|
|
filesystem produced under a trusted root -- not the caller's original
|
|
string, which could otherwise smuggle a traversal sequence through.
|
|
"""
|
|
try:
|
|
with os.scandir(trusted_dir) as entries:
|
|
for entry in entries:
|
|
if entry.name == name and entry.is_dir():
|
|
return entry.name
|
|
except OSError:
|
|
pass
|
|
return None
|
|
|
|
|
|
def contained_plugin_dir(plugin_dir: Path, plugins_dir: Path) -> Optional[str]:
|
|
"""``plugin_dir`` rebuilt from an entry enumerated under ``plugins_dir``.
|
|
|
|
Returns None when ``plugin_dir`` is not a subdirectory of ``plugins_dir``.
|
|
Callers derive ``plugin_dir`` from a manifest-declared id, so the path is
|
|
rebuilt from :func:`find_trusted_subdir`'s answer rather than trusted: a
|
|
name that came out of ``os.scandir()`` on the trusted root carries no
|
|
taint, which is a real containment guarantee (and one CodeQL's
|
|
path-injection query can follow), not a string sanitiser.
|
|
"""
|
|
plugin_dir_real = os.path.realpath(str(plugin_dir))
|
|
plugins_dir_real = os.path.realpath(str(plugins_dir))
|
|
matched_name = find_trusted_subdir(plugins_dir_real, os.path.basename(plugin_dir_real))
|
|
if matched_name is None:
|
|
return None
|
|
return os.path.join(plugins_dir_real, matched_name)
|
|
|
|
|
|
def requirements_to_install(plugin_dir: str, logger: logging.Logger,
|
|
label: str) -> Optional[str]:
|
|
"""The plugin's requirements.txt if pip has work to do, else None.
|
|
|
|
None when there is no requirements.txt, when it lists nothing (plugins
|
|
whose dependencies ship with core often keep an all-comments file), or
|
|
when every requirement is already installed. Shared by the loader and the
|
|
store so both skip pip for the same reasons; they differ only in how they
|
|
run it.
|
|
"""
|
|
requirements_file = os.path.join(plugin_dir, "requirements.txt")
|
|
if not os.path.isfile(requirements_file):
|
|
return None
|
|
if not requirements_has_real_deps(requirements_file):
|
|
logger.debug("requirements.txt for %s has no real dependencies, skipping pip", label)
|
|
return None
|
|
if requirements_are_satisfied(requirements_file):
|
|
logger.debug("Dependencies for %s already satisfied, skipping pip", label)
|
|
return None
|
|
return requirements_file
|
|
|
|
|
|
class PluginLoader:
|
|
"""Handles plugin module loading and class instantiation."""
|
|
|
|
def __init__(self, logger: Optional[logging.Logger] = None) -> None:
|
|
"""
|
|
Initialize the plugin loader.
|
|
|
|
Args:
|
|
logger: Optional logger instance
|
|
"""
|
|
self.logger = logger or get_logger(__name__)
|
|
self._loaded_modules: Dict[str, Any] = {}
|
|
self._plugin_module_registry: Dict[str, set] = {} # Maps plugin_id to set of module names
|
|
# Lock to serialize module loading when plugins share module names
|
|
# (e.g., scroll_display.py, game_renderer.py across sport plugins).
|
|
# During exec_module, bare-name sub-modules temporarily appear in
|
|
# sys.modules; the lock prevents concurrent plugins from seeing each
|
|
# other's entries. After exec_module, _namespace_plugin_modules
|
|
# moves those bare names to namespaced keys (e.g.
|
|
# _plg_basketball_scoreboard_scroll_display) so they never collide.
|
|
self._module_load_lock = threading.Lock()
|
|
|
|
def find_plugin_directory(
|
|
self,
|
|
plugin_id: str,
|
|
plugins_dir: Path,
|
|
plugin_directories: Optional[Dict[str, Path]] = None
|
|
) -> Optional[Path]:
|
|
"""
|
|
Find the plugin directory for a given plugin ID.
|
|
|
|
1. The discovery mapping, when it has the id and the path exists.
|
|
2. ``plugins_dir`` only, by the shared rules in
|
|
``src/plugin_system/plugin_dirs.py``: a directory whose manifest
|
|
declares the id wins; otherwise ``<id>`` or ``ledmatrix-<id>``,
|
|
matched case-insensitively. Backup and hidden directories are
|
|
never matched.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
plugins_dir: Base plugins directory
|
|
plugin_directories: Optional mapping of plugin_id to directory
|
|
|
|
Returns:
|
|
Path to plugin directory or None if not found. An id that is not
|
|
one plain path segment finds nothing.
|
|
"""
|
|
# Strategy 1: Use mapping from discovery
|
|
if plugin_directories and plugin_id in plugin_directories:
|
|
plugin_dir = plugin_directories[plugin_id]
|
|
if plugin_dir.exists():
|
|
self.logger.debug("Using plugin directory from discovery mapping: %s", plugin_dir)
|
|
return plugin_dir
|
|
|
|
plugin_dir = resolve_plugin_dir(
|
|
plugin_id, [plugins_dir], prefix=True, case_insensitive=True)
|
|
if plugin_dir is not None and plugin_dir.name != plugin_id:
|
|
self.logger.debug("Found plugin %s in directory %s",
|
|
plugin_id, plugin_dir.name)
|
|
return plugin_dir
|
|
|
|
def install_dependencies(
|
|
self,
|
|
plugin_dir: Path,
|
|
plugin_id: str,
|
|
plugins_dir: Path,
|
|
timeout: int = 300
|
|
) -> bool:
|
|
"""
|
|
Install plugin dependencies from requirements.txt.
|
|
|
|
Args:
|
|
plugin_dir: Plugin directory path
|
|
plugin_id: Plugin identifier
|
|
plugins_dir: Trusted base plugins directory for path containment check.
|
|
Required (not optional) so every caller reconstructs the plugin
|
|
path through the sanitiser below rather than trusting plugin_dir
|
|
directly -- CodeQL's path-injection query (and a malicious
|
|
manifest/plugin_id in practice) can't tell a legitimate
|
|
plugin_dir from one crafted to traverse outside plugins_dir.
|
|
timeout: Installation timeout in seconds
|
|
|
|
Returns:
|
|
True if dependencies installed or not needed, False on error
|
|
"""
|
|
plugin_id = os.path.basename(plugin_id or '')
|
|
if not plugin_id:
|
|
return False
|
|
|
|
safe_plugin_dir = contained_plugin_dir(plugin_dir, plugins_dir)
|
|
if safe_plugin_dir is None:
|
|
self.logger.error(
|
|
"Plugin directory for %s not found inside plugins dir", plugin_id
|
|
)
|
|
return False
|
|
|
|
requirements_file = requirements_to_install(safe_plugin_dir, self.logger, plugin_id)
|
|
if requirements_file is None:
|
|
return True
|
|
|
|
with _PIP_INSTALL_LOCK:
|
|
try:
|
|
self.logger.info("Installing dependencies for plugin %s...", plugin_id)
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "pip", "install", "--break-system-packages", "-r", requirements_file],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
check=False
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
self.logger.info("Dependencies installed successfully for %s", plugin_id)
|
|
return True
|
|
else:
|
|
stderr = result.stderr or ""
|
|
# uninstall-no-record-file means a system-managed copy of a package
|
|
# (e.g. apt's python3-requests, which ships no pip RECORD file) is in
|
|
# the way of the version this requirements.txt pins. Retry with
|
|
# --ignore-installed so pip lays the pinned version down alongside
|
|
# the system copy instead of trying to replace it — matching the
|
|
# retry already used by install_dependencies_apt.py / safe_pip_install.sh.
|
|
# Without this retry, the plugin would silently keep running against
|
|
# whatever version the system happened to ship.
|
|
if "uninstall-no-record-file" in stderr:
|
|
self.logger.warning(
|
|
"Dependencies for %s conflict with a system-managed package "
|
|
"(no pip RECORD); retrying with --ignore-installed: %s",
|
|
plugin_id, stderr.strip()
|
|
)
|
|
# Wrapped in its own try/except so a retry timeout is
|
|
# tolerated the same way as a retry failure, instead of
|
|
# propagating to the outer handler and returning False
|
|
# (which would contradict the "assume satisfied" fallback
|
|
# below).
|
|
try:
|
|
# sys.executable is this process's own interpreter (not
|
|
# attacker-influenced), and requirements_file is rebuilt
|
|
# by contained_plugin_dir() from a trusted listing, never raw
|
|
# external input.
|
|
retry_result = subprocess.run( # nosec B603 - no shell invoked (list-form argv) # nosemgrep
|
|
[sys.executable, "-m", "pip", "install", "--break-system-packages",
|
|
"--ignore-installed", "-r", requirements_file],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
check=False
|
|
)
|
|
if retry_result.returncode != 0:
|
|
self.logger.warning(
|
|
"Retry with --ignore-installed also failed for %s; assuming the "
|
|
"system-managed version satisfies the requirement: %s",
|
|
plugin_id, (retry_result.stderr or "").strip()
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
self.logger.warning(
|
|
"Retry with --ignore-installed timed out for %s; assuming the "
|
|
"system-managed version satisfies the requirement",
|
|
plugin_id
|
|
)
|
|
return True
|
|
self.logger.warning(
|
|
"Dependency installation returned non-zero exit code for %s: %s",
|
|
plugin_id,
|
|
stderr
|
|
)
|
|
return False
|
|
except subprocess.TimeoutExpired:
|
|
self.logger.error("Dependency installation timed out for %s", plugin_id)
|
|
return False
|
|
except FileNotFoundError:
|
|
self.logger.warning("pip not found. Skipping dependency installation for %s", plugin_id)
|
|
return True
|
|
except OSError as e:
|
|
# A broken pipe (EPIPE) happens when pip's output pipe closes
|
|
# mid-download, usually a network interruption.
|
|
if e.errno == errno.EPIPE:
|
|
self.logger.error(
|
|
"Broken pipe error during dependency installation for %s. "
|
|
"This usually indicates a network interruption or pip output buffer issue. "
|
|
"Try installing again or check your network connection.", plugin_id
|
|
)
|
|
else:
|
|
self.logger.error("OS error during dependency installation for %s: %s", plugin_id, e)
|
|
return False
|
|
except Exception as e:
|
|
self.logger.error("Unexpected error installing dependencies for %s: %s", plugin_id, e, exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def _iter_plugin_bare_modules(
|
|
plugin_dir: Path, before_keys: set
|
|
) -> list:
|
|
"""Return bare-name modules from plugin_dir added after before_keys.
|
|
|
|
Returns a list of (mod_name, module) tuples for modules that:
|
|
- Were added to sys.modules after before_keys snapshot
|
|
- Have bare names (no dots)
|
|
- Have a ``__file__`` inside plugin_dir
|
|
"""
|
|
resolved_dir = plugin_dir.resolve()
|
|
result = []
|
|
for key in set(sys.modules.keys()) - before_keys:
|
|
if "." in key:
|
|
continue
|
|
mod = sys.modules.get(key)
|
|
if mod is None:
|
|
continue
|
|
mod_file = getattr(mod, "__file__", None)
|
|
if not mod_file:
|
|
continue
|
|
try:
|
|
if Path(mod_file).resolve().is_relative_to(resolved_dir):
|
|
result.append((key, mod))
|
|
except (ValueError, TypeError):
|
|
continue
|
|
return result
|
|
|
|
def _evict_stale_bare_modules(self, plugin_dir: Path) -> dict:
|
|
"""Temporarily remove bare-name sys.modules entries from other plugins.
|
|
|
|
Before exec_module, scan the current plugin directory for .py files.
|
|
For each, if sys.modules has a bare-name entry whose ``__file__`` lives
|
|
in a *different* directory, remove it so Python's import system will
|
|
load the current plugin's version instead of reusing the stale cache.
|
|
|
|
Returns:
|
|
Dict mapping evicted module names to their module objects
|
|
(for restoration on error).
|
|
"""
|
|
resolved_dir = plugin_dir.resolve()
|
|
evicted: dict = {}
|
|
|
|
for py_file in plugin_dir.glob("*.py"):
|
|
mod_name = py_file.stem
|
|
if mod_name.startswith("_"):
|
|
continue
|
|
existing = sys.modules.get(mod_name)
|
|
if existing is None:
|
|
continue
|
|
existing_file = getattr(existing, "__file__", None)
|
|
if not existing_file:
|
|
continue
|
|
try:
|
|
if not Path(existing_file).resolve().is_relative_to(resolved_dir):
|
|
evicted[mod_name] = sys.modules.pop(mod_name)
|
|
self.logger.debug(
|
|
"Evicted stale bare-name module '%s' before loading plugin", mod_name,
|
|
)
|
|
except (ValueError, TypeError):
|
|
continue
|
|
|
|
return evicted
|
|
|
|
def _namespace_plugin_modules(
|
|
self, plugin_id: str, plugin_dir: Path, before_keys: set
|
|
) -> None:
|
|
"""
|
|
Move bare-name plugin modules to namespaced keys in sys.modules.
|
|
|
|
After exec_module loads a plugin's entry point, Python will have added
|
|
the plugin's local modules (scroll_display, game_renderer, …) to
|
|
sys.modules under their bare names. This method renames them to
|
|
``_plg_<plugin_id>_<module>`` so they cannot collide with identically-
|
|
named modules from other plugins.
|
|
|
|
The plugin code keeps working because ``from scroll_display import X``
|
|
binds ``X`` to the class *object*, not to the sys.modules entry.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
plugin_dir: Plugin directory path
|
|
before_keys: Snapshot of sys.modules keys taken *before* exec_module
|
|
"""
|
|
safe_id = plugin_id.replace("-", "_")
|
|
namespaced_names: set = set()
|
|
|
|
for mod_name, mod in self._iter_plugin_bare_modules(plugin_dir, before_keys):
|
|
namespaced = f"_plg_{safe_id}_{mod_name}"
|
|
sys.modules[namespaced] = mod
|
|
# Remove the bare sys.modules entry. The module object stays
|
|
# alive via the namespaced key and all existing Python-level
|
|
# bindings (``from scroll_display import X`` already bound X
|
|
# to the class object). Leaving bare entries would cause the
|
|
# NEXT plugin's exec_module to find the cached entry and reuse
|
|
# it instead of loading its own version.
|
|
sys.modules.pop(mod_name, None)
|
|
namespaced_names.add(namespaced)
|
|
self.logger.debug(
|
|
"Namespace-isolated module '%s' -> '%s' for plugin %s",
|
|
mod_name, namespaced, plugin_id,
|
|
)
|
|
|
|
# Track for cleanup during unload
|
|
self._plugin_module_registry[plugin_id] = namespaced_names
|
|
|
|
if namespaced_names:
|
|
self.logger.info(
|
|
"Namespace-isolated %d module(s) for plugin %s",
|
|
len(namespaced_names), plugin_id,
|
|
)
|
|
|
|
def unregister_plugin_modules(self, plugin_id: str) -> None:
|
|
"""Remove namespaced sub-modules and cached module for a plugin from sys.modules.
|
|
|
|
Called by PluginManager during unload to clean up all module entries
|
|
that were created when the plugin was loaded.
|
|
"""
|
|
for ns_name in self._plugin_module_registry.pop(plugin_id, set()):
|
|
sys.modules.pop(ns_name, None)
|
|
self._loaded_modules.pop(plugin_id, None)
|
|
|
|
def load_module(
|
|
self,
|
|
plugin_id: str,
|
|
plugin_dir: Path,
|
|
entry_point: str
|
|
) -> Any:
|
|
"""
|
|
Load a plugin module from file.
|
|
|
|
Module loading is serialized via _module_load_lock because plugins are
|
|
loaded in parallel (ThreadPoolExecutor) and multiple sport plugins
|
|
share identically-named local modules (scroll_display.py,
|
|
game_renderer.py, sports.py, etc.).
|
|
|
|
After loading, bare-name modules from the plugin directory are moved
|
|
to namespaced keys in sys.modules (e.g. ``_plg_basketball_scoreboard_scroll_display``)
|
|
so they cannot collide with other plugins.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
plugin_dir: Plugin directory path
|
|
entry_point: Entry point filename (e.g., 'manager.py')
|
|
|
|
Returns:
|
|
The loaded module
|
|
|
|
Raises:
|
|
PluginError: If the plugin id, directory or entry point is
|
|
invalid. Whatever the module raises while executing
|
|
propagates unchanged.
|
|
"""
|
|
plugin_id = os.path.basename(plugin_id or '')
|
|
if not plugin_id:
|
|
raise PluginError("Invalid plugin ID")
|
|
try:
|
|
plugin_dir_resolved = plugin_dir.resolve(strict=True)
|
|
except OSError:
|
|
raise PluginError("Plugin directory not found", plugin_id=plugin_id)
|
|
entry_file = (plugin_dir_resolved / entry_point).resolve()
|
|
try:
|
|
entry_file.relative_to(plugin_dir_resolved)
|
|
except ValueError:
|
|
raise PluginError("Invalid entry point path", plugin_id=plugin_id)
|
|
if not entry_file.exists():
|
|
error_msg = f"Entry point file not found for plugin {plugin_id}"
|
|
self.logger.error(error_msg)
|
|
raise PluginError(error_msg, plugin_id=plugin_id, context={'entry_file': str(entry_file)})
|
|
|
|
with self._module_load_lock:
|
|
# Add plugin directory to sys.path if not already there
|
|
plugin_dir_str = str(plugin_dir)
|
|
if plugin_dir_str not in sys.path:
|
|
sys.path.insert(0, plugin_dir_str)
|
|
self.logger.debug("Added plugin %s's directory to sys.path", plugin_id)
|
|
|
|
# Import the plugin module
|
|
module_name = f"plugin_{plugin_id.replace('-', '_')}"
|
|
|
|
# Check if already loaded
|
|
if module_name in sys.modules:
|
|
self.logger.debug("Module %s already loaded, reusing", module_name)
|
|
return sys.modules[module_name]
|
|
|
|
spec = importlib.util.spec_from_file_location(module_name, entry_file)
|
|
if spec is None or spec.loader is None:
|
|
self.logger.error("Could not create module spec for plugin %s", plugin_id)
|
|
error_msg = f"Could not create module spec for {entry_file}"
|
|
raise PluginError(error_msg, plugin_id=plugin_id, context={'entry_file': str(entry_file)})
|
|
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[module_name] = module
|
|
|
|
# Snapshot AFTER inserting the main module so that
|
|
# _namespace_plugin_modules and error cleanup only target
|
|
# sub-modules, not the main module entry itself.
|
|
before_keys = set(sys.modules.keys())
|
|
|
|
# Evict stale bare-name modules from other plugin directories
|
|
# so Python's import system loads fresh copies from this plugin.
|
|
evicted = self._evict_stale_bare_modules(plugin_dir)
|
|
|
|
try:
|
|
spec.loader.exec_module(module)
|
|
|
|
# Move bare-name plugin modules to namespaced keys so they
|
|
# cannot collide with identically-named modules from other plugins
|
|
self._namespace_plugin_modules(plugin_id, plugin_dir, before_keys)
|
|
except Exception:
|
|
# Restore evicted modules so other plugins are unaffected
|
|
for evicted_name, evicted_mod in evicted.items():
|
|
if evicted_name not in sys.modules:
|
|
sys.modules[evicted_name] = evicted_mod
|
|
# Clean up the partially-initialized main module and any
|
|
# bare-name sub-modules that were added during exec_module
|
|
# so they don't leak into subsequent plugin loads.
|
|
sys.modules.pop(module_name, None)
|
|
for key, _ in self._iter_plugin_bare_modules(plugin_dir, before_keys):
|
|
sys.modules.pop(key, None)
|
|
raise
|
|
|
|
self._loaded_modules[plugin_id] = module
|
|
self.logger.debug("Loaded module %s for plugin %s", module_name, plugin_id)
|
|
|
|
return module
|
|
|
|
def get_plugin_class(
|
|
self,
|
|
plugin_id: str,
|
|
module: Any,
|
|
class_name: str
|
|
) -> Type[Any]:
|
|
"""
|
|
Get the plugin class from a loaded module.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
module: Loaded module
|
|
class_name: Name of the plugin class
|
|
|
|
Returns:
|
|
Plugin class
|
|
|
|
Raises:
|
|
PluginError: If class not found
|
|
"""
|
|
if not hasattr(module, class_name):
|
|
error_msg = f"Class {class_name} not found in module for plugin {plugin_id}"
|
|
self.logger.error(error_msg)
|
|
raise PluginError(
|
|
error_msg,
|
|
plugin_id=plugin_id,
|
|
context={'class_name': class_name, 'module': module.__name__}
|
|
)
|
|
|
|
plugin_class = getattr(module, class_name)
|
|
|
|
# Verify it's a class
|
|
if not isinstance(plugin_class, type):
|
|
error_msg = f"{class_name} is not a class in module for plugin {plugin_id}"
|
|
self.logger.error(error_msg)
|
|
raise PluginError(error_msg, plugin_id=plugin_id, context={'class_name': class_name})
|
|
|
|
return plugin_class
|
|
|
|
def instantiate_plugin(
|
|
self,
|
|
plugin_id: str,
|
|
plugin_class: Type[Any],
|
|
config: Dict[str, Any],
|
|
display_manager: Any,
|
|
cache_manager: Any,
|
|
plugin_manager: Any
|
|
) -> Any:
|
|
"""
|
|
Instantiate a plugin class.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
plugin_class: Plugin class to instantiate
|
|
config: Plugin configuration
|
|
display_manager: Display manager instance
|
|
cache_manager: Cache manager instance
|
|
plugin_manager: Plugin manager instance
|
|
|
|
Returns:
|
|
Plugin instance
|
|
|
|
Raises:
|
|
PluginError: If instantiation fails
|
|
"""
|
|
try:
|
|
plugin_instance = plugin_class(
|
|
plugin_id=plugin_id,
|
|
config=config,
|
|
display_manager=display_manager,
|
|
cache_manager=cache_manager,
|
|
plugin_manager=plugin_manager
|
|
)
|
|
self.logger.debug("Instantiated plugin %s", plugin_id)
|
|
return plugin_instance
|
|
except Exception as e:
|
|
error_msg = f"Failed to instantiate plugin {plugin_id}: {e}"
|
|
self.logger.error(error_msg, exc_info=True)
|
|
raise PluginError(error_msg, plugin_id=plugin_id) from e
|
|
|
|
def _warn_if_incompatible(self, plugin_id: str, manifest: Dict[str, Any]) -> None:
|
|
"""Log one warning when a plugin declares a minimum LEDMatrix version
|
|
newer than the running core. Advisory only — never raises — so a
|
|
plugin that guards optional features with try/except keeps working.
|
|
"""
|
|
from src.plugin_system import compatibility
|
|
core_version = compatibility.current_core_version()
|
|
|
|
compatible, _reason = compatibility.check(manifest, core_version)
|
|
if compatible:
|
|
# Distinguish "fine" from "couldn't tell" for anyone reading logs:
|
|
# a core below the trustworthy floor is skipped, not cleared.
|
|
current = compatibility.parse_semver(core_version)
|
|
if current is None or current < compatibility.TRUSTWORTHY_FLOOR:
|
|
self.logger.debug(
|
|
"Skipping version compatibility check for %s: core __version__ "
|
|
"(%s) is below the ecosystem floor", plugin_id, core_version)
|
|
return
|
|
|
|
declared = compatibility.declared_min_version(manifest)
|
|
self.logger.warning(
|
|
"Plugin %s declares min LEDMatrix version %s but this core is %s — "
|
|
"features it relies on may be missing; update the core or expect "
|
|
"degraded fallbacks", plugin_id, declared, core_version)
|
|
|
|
def load_plugin(
|
|
self,
|
|
plugin_id: str,
|
|
manifest: Dict[str, Any],
|
|
plugin_dir: Path,
|
|
config: Dict[str, Any],
|
|
display_manager: Any,
|
|
cache_manager: Any,
|
|
plugin_manager: Any,
|
|
install_deps: bool = True,
|
|
plugins_dir: Optional[Path] = None,
|
|
) -> Tuple[Any, Any]:
|
|
"""
|
|
Complete plugin loading process.
|
|
|
|
Args:
|
|
plugin_id: Plugin identifier
|
|
manifest: Plugin manifest
|
|
plugin_dir: Plugin directory path
|
|
config: Plugin configuration
|
|
display_manager: Display manager instance
|
|
cache_manager: Cache manager instance
|
|
plugin_manager: Plugin manager instance
|
|
install_deps: Whether to install dependencies
|
|
plugins_dir: Trusted base plugins directory forwarded to install_dependencies
|
|
|
|
Returns:
|
|
Tuple of (plugin_instance, module)
|
|
|
|
Raises:
|
|
PluginError: If loading fails
|
|
"""
|
|
self._warn_if_incompatible(plugin_id, manifest)
|
|
|
|
# Install dependencies if needed
|
|
if install_deps:
|
|
if plugins_dir is None:
|
|
raise PluginError(
|
|
f"plugins_dir is required to install dependencies for plugin {plugin_id} "
|
|
"(needed for path containment; pass install_deps=False if the caller "
|
|
"doesn't have a trusted plugins directory to supply)",
|
|
plugin_id=plugin_id,
|
|
context={'plugin_dir': str(plugin_dir)},
|
|
)
|
|
if not self.install_dependencies(plugin_dir, plugin_id, plugins_dir=plugins_dir):
|
|
raise PluginError(
|
|
f"Dependency installation failed for plugin {plugin_id} in {plugin_dir}",
|
|
plugin_id=plugin_id,
|
|
context={'plugin_dir': str(plugin_dir)},
|
|
)
|
|
|
|
# Load module
|
|
entry_point = manifest.get('entry_point', 'manager.py')
|
|
module = self.load_module(plugin_id, plugin_dir, entry_point)
|
|
|
|
# Get plugin class
|
|
class_name = manifest.get('class_name')
|
|
if not class_name:
|
|
raise PluginError(f"No class_name in manifest for plugin {plugin_id}", plugin_id=plugin_id)
|
|
|
|
plugin_class = self.get_plugin_class(plugin_id, module, class_name)
|
|
|
|
# Instantiate plugin
|
|
plugin_instance = self.instantiate_plugin(
|
|
plugin_id,
|
|
plugin_class,
|
|
config,
|
|
display_manager,
|
|
cache_manager,
|
|
plugin_manager
|
|
)
|
|
|
|
return (plugin_instance, module)
|
|
|