# -*- coding: utf-8 -*- """ Plugin Hot-Reloader. Ermöglicht das automatische Neuladen von Plugins bei Änderungen. """ from __future__ import annotations import asyncio from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Callable, TYPE_CHECKING from trixy_core.plugins.hotreload.watcher import ( FileWatcher, WatchEvent, WatchEventType, ) if TYPE_CHECKING: from trixy_core.plugins.plugin_manager import PluginManager @dataclass class ReloadConfig: """ Konfiguration für Hot-Reload. Attributes: enabled: Hot-Reload aktiviert poll_interval: Polling-Intervall in Sekunden debounce_delay: Verzögerung vor Reload (sammelt mehrere Änderungen) extensions: Zu überwachende Dateierweiterungen auto_reload_on_error: Auch bei Fehlern neu laden versuchen preserve_state: Plugin-Zustand beim Reload erhalten """ enabled: bool = True poll_interval: float = 1.0 debounce_delay: float = 0.5 extensions: list[str] = field(default_factory=lambda: [".py", ".json"]) auto_reload_on_error: bool = False preserve_state: bool = True @dataclass class ReloadResult: """ Ergebnis eines Reload-Vorgangs. Attributes: success: Reload erfolgreich plugin_name: Plugin-Name old_version: Alte Version new_version: Neue Version duration: Reload-Dauer in Sekunden error: Fehler-Nachricht bei Misserfolg state_preserved: Wurde Zustand erhalten """ success: bool plugin_name: str old_version: str = "" new_version: str = "" duration: float = 0.0 error: str = "" state_preserved: bool = False timestamp: datetime = field(default_factory=datetime.now) def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "success": self.success, "plugin_name": self.plugin_name, "old_version": self.old_version, "new_version": self.new_version, "duration": self.duration, "error": self.error, "state_preserved": self.state_preserved, "timestamp": self.timestamp.isoformat(), } # Callback-Typen ReloadCallback = Callable[[ReloadResult], None] AsyncReloadCallback = Callable[[ReloadResult], Any] class PluginHotReloader: """ Hot-Reloader für Plugins. Überwacht Plugin-Verzeichnisse und lädt Plugins automatisch neu, wenn Änderungen erkannt werden. Features: - Debouncing: Sammelt mehrere Änderungen - Zustandserhaltung: Kann Plugin-Zustand beim Reload erhalten - Fehlerbehandlung: Robuste Reload-Logik Example: # Mit PluginManager verwenden reloader = PluginHotReloader(plugin_manager) # Konfiguration anpassen reloader.config.debounce_delay = 1.0 # Callback für Reload-Events @reloader.on_reload def handle_reload(result: ReloadResult): if result.success: print(f"Plugin {result.plugin_name} neu geladen") else: print(f"Reload fehlgeschlagen: {result.error}") # Starten await reloader.start() # Manueller Reload result = await reloader.reload_plugin("my_plugin") # Stoppen await reloader.stop() """ def __init__( self, plugin_manager: "PluginManager", config: ReloadConfig | None = None, ) -> None: """ Initialisiert den Hot-Reloader. Args: plugin_manager: Der PluginManager config: Reload-Konfiguration """ self._manager = plugin_manager self._config = config or ReloadConfig() self._watcher: FileWatcher | None = None self._callbacks: list[ReloadCallback] = [] self._async_callbacks: list[AsyncReloadCallback] = [] self._pending_reloads: dict[str, asyncio.Task] = {} self._reload_history: list[ReloadResult] = [] self._running = False @property def config(self) -> ReloadConfig: """Reload-Konfiguration.""" return self._config @property def is_running(self) -> bool: """Läuft der Reloader?""" return self._running @property def history(self) -> list[ReloadResult]: """Reload-Historie.""" return self._reload_history.copy() def on_reload(self, callback: ReloadCallback) -> ReloadCallback: """Decorator für Reload-Callbacks.""" self._callbacks.append(callback) return callback def on_reload_async( self, callback: AsyncReloadCallback, ) -> AsyncReloadCallback: """Decorator für async Reload-Callbacks.""" self._async_callbacks.append(callback) return callback async def start(self) -> None: """Startet den Hot-Reloader.""" if not self._config.enabled: return if self._running: return self._running = True # FileWatcher erstellen self._watcher = FileWatcher( poll_interval=self._config.poll_interval, recursive=True, extensions=self._config.extensions, ) # Plugin-Verzeichnis überwachen self._watcher.add_directory(self._manager._directory) # Callback für Änderungen self._watcher.add_async_callback(self._handle_change) await self._watcher.start() async def stop(self) -> None: """Stoppt den Hot-Reloader.""" self._running = False # Pending Reloads abbrechen for task in self._pending_reloads.values(): task.cancel() self._pending_reloads.clear() if self._watcher: await self._watcher.stop() self._watcher = None async def _handle_change(self, event: WatchEvent) -> None: """Behandelt Datei-Änderungen.""" if not self._running: return # Plugin-Namen aus Pfad extrahieren plugin_name = self._get_plugin_from_path(event.path) if not plugin_name: return # Nur geladene Plugins berücksichtigen if plugin_name not in self._manager: return # Debouncing: Vorherigen Reload abbrechen if plugin_name in self._pending_reloads: self._pending_reloads[plugin_name].cancel() # Neuen Reload mit Verzögerung planen task = asyncio.create_task( self._debounced_reload(plugin_name) ) self._pending_reloads[plugin_name] = task def _get_plugin_from_path(self, path: Path) -> str | None: """Extrahiert den Plugin-Namen aus einem Pfad.""" try: # Relativen Pfad zum Plugin-Verzeichnis berechnen rel_path = path.relative_to(self._manager._directory) # Erstes Verzeichnis ist der Plugin-Name return rel_path.parts[0] except (ValueError, IndexError): return None async def _debounced_reload(self, plugin_name: str) -> None: """Führt einen verzögerten Reload durch.""" try: await asyncio.sleep(self._config.debounce_delay) await self.reload_plugin(plugin_name) except asyncio.CancelledError: pass finally: self._pending_reloads.pop(plugin_name, None) async def reload_plugin(self, plugin_name: str) -> ReloadResult: """ Lädt ein Plugin neu. Args: plugin_name: Name des Plugins Returns: ReloadResult """ start_time = asyncio.get_event_loop().time() # Aktuelles Plugin holen old_plugin = self._manager.get(plugin_name) old_version = old_plugin.VERSION if old_plugin else "" # Zustand sichern state = None if self._config.preserve_state and old_plugin: state = self._save_plugin_state(old_plugin) # Reload durchführen try: success = await self._manager.reload(plugin_name) if not success: result = ReloadResult( success=False, plugin_name=plugin_name, old_version=old_version, error="Reload fehlgeschlagen", duration=asyncio.get_event_loop().time() - start_time, ) else: new_plugin = self._manager.get(plugin_name) new_version = new_plugin.VERSION if new_plugin else "" # Zustand wiederherstellen state_preserved = False if state and new_plugin: state_preserved = self._restore_plugin_state(new_plugin, state) result = ReloadResult( success=True, plugin_name=plugin_name, old_version=old_version, new_version=new_version, duration=asyncio.get_event_loop().time() - start_time, state_preserved=state_preserved, ) except Exception as e: result = ReloadResult( success=False, plugin_name=plugin_name, old_version=old_version, error=str(e), duration=asyncio.get_event_loop().time() - start_time, ) # Historie aktualisieren self._reload_history.append(result) # Callbacks benachrichtigen await self._notify(result) return result def _save_plugin_state(self, plugin: Any) -> dict[str, Any]: """Sichert den Plugin-Zustand.""" state = {} # Konfiguration sichern if hasattr(plugin, "config"): state["config"] = plugin.config.copy() # Enabled-Status sichern if hasattr(plugin, "enabled"): state["enabled"] = plugin.enabled # Custom-Zustand sichern wenn vorhanden if hasattr(plugin, "get_state") and callable(plugin.get_state): try: state["custom"] = plugin.get_state() except Exception: pass return state def _restore_plugin_state( self, plugin: Any, state: dict[str, Any], ) -> bool: """Stellt den Plugin-Zustand wieder her.""" try: # Konfiguration wiederherstellen if "config" in state and hasattr(plugin, "config"): plugin.config.update(state["config"]) # Enabled-Status wiederherstellen if "enabled" in state and hasattr(plugin, "enabled"): plugin.enabled = state["enabled"] # Custom-Zustand wiederherstellen if "custom" in state: if hasattr(plugin, "set_state") and callable(plugin.set_state): plugin.set_state(state["custom"]) return True except Exception: return False async def _notify(self, result: ReloadResult) -> None: """Benachrichtigt alle Callbacks.""" for callback in self._callbacks: try: callback(result) except Exception: pass for callback in self._async_callbacks: try: awaitable = callback(result) if asyncio.iscoroutine(awaitable): await awaitable except Exception: pass async def reload_all(self) -> list[ReloadResult]: """ Lädt alle Plugins neu. Returns: Liste der Reload-Ergebnisse """ results = [] for plugin in self._manager.get_all(): result = await self.reload_plugin(plugin.name) results.append(result) return results def get_pending_reloads(self) -> list[str]: """Gibt Namen der Plugins mit ausstehenden Reloads zurück.""" return list(self._pending_reloads.keys()) def cancel_pending(self, plugin_name: str | None = None) -> int: """ Bricht ausstehende Reloads ab. Args: plugin_name: Spezifisches Plugin (None = alle) Returns: Anzahl abgebrochener Reloads """ if plugin_name: if plugin_name in self._pending_reloads: self._pending_reloads[plugin_name].cancel() del self._pending_reloads[plugin_name] return 1 return 0 count = len(self._pending_reloads) for task in self._pending_reloads.values(): task.cancel() self._pending_reloads.clear() return count def clear_history(self) -> None: """Löscht die Reload-Historie.""" self._reload_history.clear() def get_statistics(self) -> dict[str, Any]: """Gibt Reload-Statistiken zurück.""" total = len(self._reload_history) successful = sum(1 for r in self._reload_history if r.success) return { "total_reloads": total, "successful": successful, "failed": total - successful, "success_rate": successful / max(1, total), "pending_reloads": len(self._pending_reloads), "is_running": self._running, }