Discover and load all installed plugins via entry_points.
(self)
| 98 | return winners |
| 99 | |
| 100 | def discover(self) -> None: |
| 101 | """Discover and load all installed plugins via entry_points.""" |
| 102 | if self._discovered: |
| 103 | return |
| 104 | |
| 105 | try: |
| 106 | eps = importlib.metadata.entry_points() |
| 107 | if hasattr(eps, "select"): |
| 108 | plugin_eps = eps.select(group=ENTRY_POINT_GROUP) |
| 109 | else: |
| 110 | plugin_eps = eps.get(ENTRY_POINT_GROUP, []) |
| 111 | except Exception: |
| 112 | logger.exception("Failed to query entry_points") |
| 113 | return |
| 114 | |
| 115 | candidates: list[tuple[str, MemOSPlugin]] = [] |
| 116 | for ep in plugin_eps: |
| 117 | try: |
| 118 | plugin_cls = ep.load() |
| 119 | plugin = plugin_cls() |
| 120 | if not isinstance(plugin, MemOSPlugin): |
| 121 | logger.warning("Plugin %s does not extend MemOSPlugin, skipped", ep.name) |
| 122 | continue |
| 123 | candidates.append((ep.name, plugin)) |
| 124 | except Exception: |
| 125 | logger.exception("Failed to load plugin: %s", ep.name) |
| 126 | |
| 127 | winners = self._select_plugin_winners(candidates) |
| 128 | for plugin_name, plugin in winners.items(): |
| 129 | if not self._is_plugin_enabled(plugin): |
| 130 | logger.info( |
| 131 | "Plugin discovered but disabled: %s v%s", |
| 132 | plugin.name, |
| 133 | plugin.version, |
| 134 | ) |
| 135 | continue |
| 136 | plugin.on_load() |
| 137 | self._plugins[plugin_name] = plugin |
| 138 | logger.info( |
| 139 | "Plugin discovered: %s v%s (priority=%s)", |
| 140 | plugin.name, |
| 141 | plugin.version, |
| 142 | plugin.priority, |
| 143 | ) |
| 144 | |
| 145 | self._discovered = True |
| 146 | |
| 147 | def init_components(self, context: dict) -> None: |
| 148 | """Initialize runtime components contributed by loaded plugins.""" |