Coverage for src / tracekit / extensibility / plugins.py: 28%

1"""Plugin architecture for third-party extensions. 

2 

3This module implements entry point discovery for third-party plugins, 

4allowing custom decoders, measurements, and file formats to be loaded 

5dynamically. 

6""" 

7 

8from __future__ import annotations 

9 

10import importlib.metadata 

11import logging 

12from dataclasses import dataclass 

13from typing import Any 

14 

15logger = logging.getLogger(__name__) 

16 

17 

18class PluginError(Exception): 

19 """Exception raised when plugin loading fails. 

20 

21 This exception is used to isolate plugin failures so they don't 

22 crash the main application. 

23 

24 Example: 

25 >>> try: 

26 ... plugin = tk.load_plugin('tracekit.decoders', 'flexray') 

27 ... except tk.PluginError as e: 

28 ... print(f"Plugin failed: {e}") 

29 

30 References: 

31 API-007: Plugin Architecture 

32 """ 

33 

34 pass 

35 

36 

37@dataclass 

38class PluginMetadata: 

39 """Metadata about a loaded plugin. 

40 

41 Attributes: 

42 name: Plugin name. 

43 entry_point: Entry point group. 

44 version: Plugin version (if available). 

45 module: Module name. 

46 callable: The loaded plugin object. 

47 dependencies: Plugin dependencies (if available). 

48 

49 Example: 

50 >>> plugin = load_plugin('tracekit.decoders', 'can') 

51 >>> print(f"Loaded {plugin.name} v{plugin.version}") 

52 

53 References: 

54 API-007: Plugin Architecture 

55 """ 

56 

57 name: str 

58 entry_point: str 

59 version: str | None = None 

60 module: str | None = None 

61 callable: Any | None = None 

62 dependencies: list[str] | None = None 

63 

64 def __repr__(self) -> str: 

65 """String representation of plugin metadata. 

66 

67 Returns: 

68 String representation showing plugin name, version, and module. 

69 """ 

70 parts = [f"name='{self.name}'"] 

71 if self.version: 

72 parts.append(f"version='{self.version}'") 

73 if self.module: 

74 parts.append(f"module='{self.module}'") 

75 return f"PluginMetadata({', '.join(parts)})" 

76 

77 

78class PluginManager: 

79 """Manager for discovering and loading third-party plugins. 

80 

81 Discovers plugins via setuptools entry points and provides lazy loading 

82 with error isolation. Supports multiple entry point groups for different 

83 plugin types. 

84 

85 Entry point groups: 

86 - tracekit.decoders: Protocol decoders 

87 - tracekit.measurements: Custom measurements 

88 - tracekit.loaders: File format loaders 

89 - tracekit.exporters: Export format handlers 

90 

91 Example: 

92 >>> import tracekit as tk 

93 >>> # Plugins auto-discovered from installed packages 

94 >>> # Use plugin decoder 

95 >>> can_frames = tk.decode(trace, protocol='can', baudrate=500000) 

96 >>> # List available plugins 

97 >>> plugins = tk.list_plugins() 

98 >>> print(f"Installed plugins: {plugins}") 

99 

100 Advanced Example: 

101 >>> # Manually load plugin with error handling 

102 >>> try: 

103 ... plugin = tk.load_plugin('tracekit.decoders', 'flexray') 

104 ... print(f"Plugin loaded: {plugin.name} v{plugin.version}") 

105 ... except tk.PluginError as e: 

106 ... print(f"Plugin failed to load: {e}") 

107 

108 Plugin Package Example: 

109 In your package's pyproject.toml: 

110 ```toml 

111 [project.entry-points."tracekit.decoders"] 

112 flexray = "my_package.flexray:FlexRayDecoder" 

113 ``` 

114 

115 References: 

116 API-007: Plugin Architecture 

117 importlib.metadata entry points 

118 https://packaging.python.org/en/latest/guides/creating-and-discovering-plugins/ 

119 """ 

120 

121 # Standard entry point groups 

122 ENTRY_POINT_GROUPS = [ # noqa: RUF012 

123 "tracekit.decoders", 

124 "tracekit.measurements", 

125 "tracekit.loaders", 

126 "tracekit.exporters", 

127 ] 

128 

129 def __init__(self) -> None: 

130 """Initialize plugin manager.""" 

131 self._loaded_plugins: dict[tuple[str, str], PluginMetadata] = {} 

132 self._failed_plugins: dict[tuple[str, str], Exception] = {} 

133 

134 def discover_plugins(self, group: str | None = None) -> dict[str, list[str]]: 

135 """Discover available plugins via entry points. 

136 

137 Args: 

138 group: Specific entry point group to search. If None, searches 

139 all standard groups. 

140 

141 Returns: 

142 Dictionary mapping group names to lists of plugin names. 

143 

144 Example: 

145 >>> manager = PluginManager() 

146 >>> plugins = manager.discover_plugins() 

147 >>> print(plugins) 

148 {'tracekit.decoders': ['uart', 'spi', 'can'], ...} 

149 

150 References: 

151 importlib.metadata.entry_points 

152 """ 

153 discovered: dict[str, list[str]] = {} 

154 groups = [group] if group else self.ENTRY_POINT_GROUPS 

155 

156 for group_name in groups: 

157 try: 

158 # Get entry points for this group 

159 # Python 3.10+ API 

160 eps = importlib.metadata.entry_points(group=group_name) 

161 discovered[group_name] = [ep.name for ep in eps] 

162 except (AttributeError, TypeError): 

163 # Fallback for Python 3.9 and earlier 

164 try: 

165 eps = importlib.metadata.entry_points().get(group_name, []) # type: ignore[attr-defined] 

166 discovered[group_name] = [ep.name for ep in eps] 

167 except Exception as e: 

168 logger.warning(f"Failed to discover plugins for group '{group_name}': {e}") 

169 discovered[group_name] = [] 

170 

171 return discovered 

172 

173 def load_plugin( 

174 self, 

175 group: str, 

176 name: str, 

177 reload: bool = False, 

178 ) -> PluginMetadata: 

179 """Load a plugin by group and name. 

180 

181 Loads the plugin lazily on first use. Subsequent calls return cached 

182 instance unless reload=True. 

183 

184 Args: 

185 group: Entry point group. 

186 name: Plugin name. 

187 reload: Force reload even if already loaded. Default False. 

188 

189 Returns: 

190 PluginMetadata with loaded plugin information. 

191 

192 Raises: 

193 PluginError: If plugin fails to load. 

194 

195 Example: 

196 >>> manager = PluginManager() 

197 >>> plugin = manager.load_plugin('tracekit.decoders', 'can') 

198 >>> decoder = plugin.callable 

199 

200 References: 

201 API-007: Plugin Architecture 

202 """ 

203 plugin_key = (group, name) 

204 

205 # Check if already loaded 

206 if not reload and plugin_key in self._loaded_plugins: 

207 return self._loaded_plugins[plugin_key] 

208 

209 # Check if previously failed 

210 if not reload and plugin_key in self._failed_plugins: 

211 raise PluginError( 

212 f"Plugin '{name}' in group '{group}' previously failed to load: " 

213 f"{self._failed_plugins[plugin_key]}" 

214 ) 

215 

216 try: 

217 # Find entry point 

218 entry_point = self._find_entry_point(group, name) 

219 

220 if entry_point is None: 

221 raise PluginError(f"Plugin '{name}' not found in group '{group}'") 

222 

223 # Load the plugin 

224 logger.info(f"Loading plugin '{name}' from group '{group}'") 

225 plugin_obj = entry_point.load() 

226 

227 # Get version if available 

228 version = None 

229 if hasattr(entry_point, "dist") and entry_point.dist: 

230 version = entry_point.dist.version 

231 

232 # Create metadata 

233 metadata = PluginMetadata( 

234 name=name, 

235 entry_point=group, 

236 version=version, 

237 module=entry_point.value, 

238 callable=plugin_obj, 

239 ) 

240 

241 # Cache loaded plugin 

242 self._loaded_plugins[plugin_key] = metadata 

243 

244 logger.info(f"Successfully loaded plugin '{name}' v{version}") 

245 return metadata 

246 

247 except Exception as e: 

248 # Cache failure 

249 self._failed_plugins[plugin_key] = e 

250 logger.error(f"Failed to load plugin '{name}': {e}") 

251 raise PluginError(f"Failed to load plugin '{name}' from group '{group}': {e}") from e 

252 

253 def get_plugin(self, group: str, name: str) -> Any: 

254 """Get loaded plugin callable. 

255 

256 Convenience method that loads plugin if needed and returns the 

257 callable object. 

258 

259 Args: 

260 group: Entry point group. 

261 name: Plugin name. 

262 

263 Returns: 

264 The loaded plugin object. 

265 

266 Example: 

267 >>> manager = PluginManager() 

268 >>> decoder = manager.get_plugin('tracekit.decoders', 'can') 

269 >>> frames = decoder.decode(trace) 

270 """ 

271 metadata = self.load_plugin(group, name) 

272 return metadata.callable 

273 

274 def is_loaded(self, group: str, name: str) -> bool: 

275 """Check if plugin is already loaded. 

276 

277 Args: 

278 group: Entry point group. 

279 name: Plugin name. 

280 

281 Returns: 

282 True if plugin is loaded. 

283 

284 Example: 

285 >>> if manager.is_loaded('tracekit.decoders', 'can'): 

286 ... print("CAN decoder already loaded") 

287 """ 

288 return (group, name) in self._loaded_plugins 

289 

290 def list_loaded_plugins(self) -> list[PluginMetadata]: 

291 """List all loaded plugins. 

292 

293 Returns: 

294 List of PluginMetadata for loaded plugins. 

295 

296 Example: 

297 >>> loaded = manager.list_loaded_plugins() 

298 >>> for plugin in loaded: 

299 ... print(f"{plugin.name} v{plugin.version}") 

300 """ 

301 return list(self._loaded_plugins.values()) 

302 

303 def unload_plugin(self, group: str, name: str) -> None: 

304 """Unload a plugin from cache. 

305 

306 Args: 

307 group: Entry point group. 

308 name: Plugin name. 

309 

310 Example: 

311 >>> manager.unload_plugin('tracekit.decoders', 'can') 

312 """ 

313 plugin_key = (group, name) 

314 if plugin_key in self._loaded_plugins: 

315 del self._loaded_plugins[plugin_key] 

316 if plugin_key in self._failed_plugins: 

317 del self._failed_plugins[plugin_key] 

318 

319 def _find_entry_point(self, group: str, name: str) -> Any | None: 

320 """Find entry point by group and name. 

321 

322 Args: 

323 group: Entry point group. 

324 name: Entry point name. 

325 

326 Returns: 

327 Entry point object or None if not found. 

328 """ 

329 try: 

330 # Python 3.10+ API 

331 eps = importlib.metadata.entry_points(group=group) 

332 for ep in eps: 

333 if ep.name == name: 

334 return ep 

335 except (AttributeError, TypeError): 

336 # Fallback for Python 3.9 and earlier 

337 try: 

338 eps = importlib.metadata.entry_points().get(group, []) # type: ignore[attr-defined] 

339 for ep in eps: 

340 if ep.name == name: 

341 return ep 

342 except Exception: 

343 pass 

344 

345 return None 

346 

347 

348# Global plugin manager instance 

349_manager = PluginManager() 

350 

351 

352def load_plugin(group: str, name: str) -> PluginMetadata: 

353 """Load a plugin from the global plugin manager. 

354 

355 Convenience function for loading plugins without accessing the manager 

356 directly. 

357 

358 Args: 

359 group: Entry point group. 

360 name: Plugin name. 

361 

362 Returns: 

363 PluginMetadata with loaded plugin. 

364 

365 Example: 

366 >>> import tracekit as tk 

367 >>> plugin = tk.load_plugin('tracekit.decoders', 'flexray') 

368 >>> print(f"Loaded {plugin.name} v{plugin.version}") 

369 

370 References: 

371 API-007: Plugin Architecture 

372 """ 

373 return _manager.load_plugin(group, name) 

374 

375 

376def list_plugins(group: str | None = None) -> dict[str, list[str]]: 

377 """List available plugins. 

378 

379 Args: 

380 group: Specific group to list. If None, lists all groups. 

381 

382 Returns: 

383 Dictionary mapping group names to plugin names. 

384 

385 Example: 

386 >>> import tracekit as tk 

387 >>> plugins = tk.list_plugins() 

388 >>> print(f"Available decoders: {plugins['tracekit.decoders']}") 

389 

390 References: 

391 API-007: Plugin Architecture 

392 """ 

393 return _manager.discover_plugins(group) 

394 

395 

396def get_plugin_manager() -> PluginManager: 

397 """Get the global plugin manager instance. 

398 

399 Returns: 

400 Global PluginManager instance. 

401 

402 Example: 

403 >>> manager = tk.get_plugin_manager() 

404 >>> loaded = manager.list_loaded_plugins() 

405 """ 

406 return _manager 

407 

408 

409__all__ = [ 

410 "PluginError", 

411 "PluginManager", 

412 "PluginMetadata", 

413 "get_plugin_manager", 

414 "list_plugins", 

415 "load_plugin", 

416]