Coverage for src / tracekit / plugins / base.py: 83%

1"""Plugin base classes and metadata. 

2 

3This module defines the base class for TraceKit plugins and 

4the metadata structures for plugin registration. 

5 

6 

7Example: 

8 >>> class MyDecoder(PluginBase): 

9 ... name = "my_decoder" 

10 ... version = "1.0.0" 

11 ... api_version = "1.0.0" 

12 ... 

13 ... def on_load(self): 

14 ... self.register_protocol("my_protocol") 

15""" 

16 

17from __future__ import annotations 

18 

19from abc import ABC 

20from dataclasses import dataclass, field 

21from enum import Enum, auto 

22from typing import TYPE_CHECKING, Any 

23 

24if TYPE_CHECKING: 

25 from collections.abc import Callable 

26 from pathlib import Path 

27 

28 

29class PluginCapability(Enum): 

30 """Plugin capability types.""" 

31 

32 PROTOCOL_DECODER = auto() 

33 """Protocol decoder (UART, SPI, etc.)""" 

34 FILE_LOADER = auto() 

35 """File format loader""" 

36 FILE_EXPORTER = auto() 

37 """File format exporter""" 

38 ANALYZER = auto() 

39 """Signal analyzer""" 

40 ALGORITHM = auto() 

41 """Analysis algorithm""" 

42 VISUALIZATION = auto() 

43 """Visualization component""" 

44 WORKFLOW = auto() 

45 """High-level workflow""" 

46 

47 

48@dataclass 

49class PluginMetadata: 

50 """Plugin metadata and configuration. 

51 

52 Attributes: 

53 name: Unique plugin identifier. 

54 version: Plugin version (semver). 

55 api_version: Required TraceKit API version. 

56 author: Plugin author. 

57 description: Human-readable description. 

58 homepage: Plugin homepage URL. 

59 license: License identifier (SPDX). 

60 capabilities: List of plugin capabilities. 

61 dependencies: Required plugins and packages. 

62 provides: What the plugin provides (protocols, algorithms, etc.). 

63 path: Path to plugin (set during discovery). 

64 enabled: Whether plugin is enabled. 

65 """ 

66 

67 name: str 

68 version: str 

69 api_version: str = "1.0.0" 

70 author: str = "" 

71 description: str = "" 

72 homepage: str = "" 

73 license: str = "" 

74 capabilities: list[PluginCapability] = field(default_factory=list) 

75 dependencies: dict[str, str] = field(default_factory=dict) 

76 provides: dict[str, list[str]] = field(default_factory=dict) 

77 path: Path | None = None 

78 enabled: bool = True 

79 

80 def __post_init__(self) -> None: 

81 """Validate metadata after initialization.""" 

82 if not self.name: 

83 raise ValueError("Plugin name cannot be empty") 

84 if not self.version: 

85 raise ValueError("Plugin version cannot be empty") 

86 

87 @property 

88 def qualified_name(self) -> str: 

89 """Get fully qualified plugin name with version. 

90 

91 Returns: 

92 Name in format "name@version". 

93 """ 

94 return f"{self.name}@{self.version}" 

95 

96 def is_compatible_with(self, api_version: str) -> bool: 

97 """Check if plugin is compatible with given API version. 

98 

99 Uses semver compatibility rules (major version must match). 

100 

101 Args: 

102 api_version: TraceKit API version to check against. 

103 

104 Returns: 

105 True if compatible. 

106 """ 

107 try: 

108 plugin_major = int(self.api_version.split(".")[0]) 

109 target_major = int(api_version.split(".")[0]) 

110 return plugin_major == target_major 

111 except (ValueError, IndexError): 

112 return False 

113 

114 

115class PluginBase(ABC): # noqa: B024 

116 """Base class for all TraceKit plugins. 

117 

118 Subclass this to create a plugin. Define class attributes for 

119 metadata and implement lifecycle methods. 

120 

121 Example: 

122 >>> class UartDecoder(PluginBase): 

123 ... name = "uart_decoder" 

124 ... version = "1.0.0" 

125 ... api_version = "1.0.0" 

126 ... author = "TraceKit Contributors" 

127 ... description = "UART protocol decoder" 

128 ... 

129 ... def on_load(self): 

130 ... self.register_protocol("uart") 

131 ... 

132 ... def on_configure(self, config): 

133 ... self.baud_rate = config.get("baud_rate", 115200) 

134 """ 

135 

136 # Required class attributes (override in subclass) 

137 name: str = "" 

138 version: str = "" 

139 api_version: str = "1.0.0" 

140 

141 # Optional class attributes 

142 author: str = "" 

143 description: str = "" 

144 homepage: str = "" 

145 license: str = "" 

146 capabilities: list[PluginCapability] = [] # noqa: RUF012 

147 requires_plugins: list[tuple[str, str]] = [] # (name, version_spec) # noqa: RUF012 

148 

149 def __init__(self) -> None: 

150 """Initialize plugin instance.""" 

151 self._metadata: PluginMetadata | None = None 

152 self._registered_protocols: list[str] = [] 

153 self._registered_algorithms: list[tuple[str, str, Callable]] = [] # type: ignore[type-arg] 

154 self._config: dict[str, Any] = {} 

155 

156 @property 

157 def metadata(self) -> PluginMetadata: 

158 """Get plugin metadata. 

159 

160 Returns: 

161 PluginMetadata instance. 

162 """ 

163 if self._metadata is None: 

164 self._metadata = PluginMetadata( 

165 name=self.name, 

166 version=self.version, 

167 api_version=self.api_version, 

168 author=self.author, 

169 description=self.description, 

170 homepage=self.homepage, 

171 license=self.license, 

172 capabilities=list(self.capabilities), 

173 dependencies=dict(self.requires_plugins), 

174 ) 

175 return self._metadata 

176 

177 def on_load(self) -> None: # noqa: B027 

178 """Called when plugin is loaded. 

179 

180 Override to register capabilities, initialize resources, etc. 

181 

182 References: 

183 PLUG-002: Plugin Registration - lifecycle hooks 

184 """ 

185 pass 

186 

187 def on_configure(self, config: dict[str, Any]) -> None: 

188 """Called when plugin is configured. 

189 

190 Override to handle configuration changes. 

191 

192 Args: 

193 config: Plugin configuration dictionary. 

194 

195 References: 

196 PLUG-002: Plugin Registration - lifecycle hooks 

197 """ 

198 self._config = config 

199 

200 def on_enable(self) -> None: # noqa: B027 

201 """Called when plugin is enabled. 

202 

203 Override to activate plugin functionality, start services, etc. 

204 

205 References: 

206 PLUG-002: Plugin Registration - lifecycle hooks 

207 """ 

208 pass 

209 

210 def on_disable(self) -> None: # noqa: B027 

211 """Called when plugin is disabled. 

212 

213 Override to pause plugin functionality, stop services, etc. 

214 

215 References: 

216 PLUG-002: Plugin Registration - lifecycle hooks 

217 """ 

218 pass 

219 

220 def on_unload(self) -> None: # noqa: B027 

221 """Called when plugin is unloaded. 

222 

223 Override to clean up resources. 

224 

225 References: 

226 PLUG-002: Plugin Registration - lifecycle hooks 

227 """ 

228 pass 

229 

230 def register_protocol(self, protocol_name: str) -> None: 

231 """Register a protocol decoder capability. 

232 

233 Args: 

234 protocol_name: Protocol identifier (e.g., "uart"). 

235 """ 

236 self._registered_protocols.append(protocol_name) 

237 if "protocols" not in self.metadata.provides: 

238 self.metadata.provides["protocols"] = [] 

239 self.metadata.provides["protocols"].append(protocol_name) 

240 

241 def register_algorithm( 

242 self, 

243 category: str, 

244 name: str, 

245 func: Callable, # type: ignore[type-arg] 

246 ) -> None: 

247 """Register an algorithm implementation. 

248 

249 Args: 

250 category: Algorithm category (e.g., "edge_detection"). 

251 name: Algorithm name. 

252 func: Algorithm function. 

253 """ 

254 self._registered_algorithms.append((category, name, func)) 

255 if "algorithms" not in self.metadata.provides: 

256 self.metadata.provides["algorithms"] = [] 

257 self.metadata.provides["algorithms"].append(f"{category}:{name}") 

258 

259 def get_config(self, key: str, default: Any = None) -> Any: 

260 """Get configuration value. 

261 

262 Args: 

263 key: Configuration key. 

264 default: Default value if key not found. 

265 

266 Returns: 

267 Configuration value. 

268 """ 

269 return self._config.get(key, default) 

270 

271 

272__all__ = [ 

273 "PluginBase", 

274 "PluginCapability", 

275 "PluginMetadata", 

276]