Trixy7/docs/api/plugins.md

trixy_core.plugins

Plugin-System mit Versionierung, Abhängigkeiten, Registry, Hooks, Hot-Reload und Testing.

Übersicht

from trixy_core.plugins import (
    # Core
    TrixyPlugin, PluginManager,
    # Versioning
    SemanticVersion, VersionRange, VersionChecker,
    CompatibilityLevel, CompatibilityResult, parse_version, parse_range,
    # Dependencies
    Dependency, DependencyNode, DependencyGraph,
    DependencyResolver, ResolutionResult, ResolutionError, CycleError,
    # Registry
    PluginManifest, PluginInfo, PluginCapability,
    PluginDiscovery, PluginRegistry, DiscoveryResult,
    load_manifest, save_manifest,
    # Hooks
    PluginHook, HookRegistry, HookHandler,
    HookPriority, HookResult, HookCallback,
    # Hot-Reload
    PluginHotReloader, FileWatcher, ReloadConfig, ReloadResult,
    WatchEvent, WatchEventType,
    # Testing
    PluginTestHarness, TestContext,
    MockApplication, MockEventManager, MockSatelliteManager, MockPlugin,
    PluginTestFixture, create_test_plugin_dir, create_test_manifest,
)

---

TrixyPlugin

Basisklasse für alle Plugins.

Definition

from trixy_core.plugins import TrixyPlugin

class MyPlugin(TrixyPlugin):
    NAME = "my_plugin"
    VERSION = "1.0.0"
    DESCRIPTION = "Mein erstes Plugin"
    AUTHOR = "Max Mustermann"

    async def on_load(self) -> None:
        """Wird beim Laden aufgerufen."""
        self.logger.info("Plugin geladen")

    async def on_unload(self) -> None:
        """Wird beim Entladen aufgerufen."""
        self.logger.info("Plugin entladen")

    async def on_enable(self) -> None:
        """Wird beim Aktivieren aufgerufen."""
        pass

    async def on_disable(self) -> None:
        """Wird beim Deaktivieren aufgerufen."""
        pass

    async def on_config_change(self, old_config: dict) -> None:
        """Wird bei Konfigurationsänderung aufgerufen."""
        pass

Properties

Property	Typ	Beschreibung
application	IApplication	Anwendungsreferenz
plugin_path	Path	Plugin-Verzeichnis
config	dict	Plugin-Konfiguration
enabled	bool	Plugin aktiviert?
name	str	Plugin-Name

Konfiguration

class MyPlugin(TrixyPlugin):
    async def on_load(self) -> None:
        # Wert abrufen (mit Dot-Notation)
        timeout = self.get_config_value("api.timeout", default=30)
        debug = self.get_config_value("debug", default=False)

        # Wert setzen
        self.set_config_value("api.last_call", datetime.now().isoformat())

        # Konfiguration speichern
        self.save_config()

    async def on_config_change(self, old_config: dict) -> None:
        # Konfiguration wurde extern geändert
        new_timeout = self.get_config_value("api.timeout")
        self.logger.info(f"Timeout geändert: {new_timeout}")

Events

from trixy_core.events import TrixyEvent, EventPriority

class MyPlugin(TrixyPlugin):
    @TrixyEvent(["wakeword_received"])
    def on_wakeword(self, event_name: str, data: dict) -> None:
        satellite_id = data.get("satellite_id")
        self.logger.info(f"Wakeword von {satellite_id}")

    @TrixyEvent(["speech_recognized"], priority=EventPriority.HIGH)
    async def on_speech(self, event_name: str, data: dict) -> None:
        text = data.get("text")
        await self.process_command(text)

---

PluginManager

Verwaltet Plugin-Lifecycle.

from trixy_core.plugins import PluginManager

manager = PluginManager(application)

# Statistiken
print(manager.count)          # Geladene Plugins
print(manager.enabled_count)  # Aktivierte Plugins

Laden und Entladen

# Alle Plugins laden
count = await manager.load_all()

# Einzelnes Plugin laden
await manager.load("my_plugin")

# Plugin entladen
await manager.unload("my_plugin")

# Plugin neu laden
await manager.reload("my_plugin")

# Alle entladen
count = await manager.unload_all()

Abrufen

# Nach Name
plugin = manager.get("my_plugin")

# Alle Plugins
all_plugins = manager.get_all()

# Nur aktivierte
enabled = manager.get_enabled()

# Fehlgeschlagene mit Fehler
failed = manager.get_failed()
# {"broken_plugin": "ImportError: ..."}

# Index-Zugriff
plugin = manager["my_plugin"]
exists = "my_plugin" in manager

# Iteration
for plugin in manager:
    print(plugin.name)

Aktivieren/Deaktivieren

# Aktivieren
await manager.enable("my_plugin")

# Deaktivieren
await manager.disable("my_plugin")

---

Versionierung

SemanticVersion

from trixy_core.plugins import SemanticVersion, parse_version

# Parsen
v1 = parse_version("1.2.3")
v2 = parse_version("2.0.0-beta.1+build.123")

print(v1.major)       # 1
print(v1.minor)       # 2
print(v1.patch)       # 3
print(v2.prerelease)  # "beta.1"
print(v2.metadata)    # "build.123"

# Vergleich
v1 < v2   # True
v1 == v2  # False
v1 >= v2  # False

# Kompatibilität
v1.is_compatible(v2, level="major")  # False (Major unterschiedlich)
v1.is_compatible(parse_version("1.3.0"), level="minor")  # True

VersionRange

from trixy_core.plugins import VersionRange, parse_range

# Range erstellen
range1 = parse_range(">=1.0.0")
range2 = parse_range("^1.2.0")    # >=1.2.0 <2.0.0
range3 = parse_range("~1.2.0")    # >=1.2.0 <1.3.0
range4 = parse_range("1.0.0 - 2.0.0")

# Prüfen
v = parse_version("1.5.0")

range1.contains(v)       # True
range2.contains(v)       # True
range2.is_compatible(v)  # True

VersionChecker

from trixy_core.plugins import VersionChecker, CompatibilityLevel

checker = VersionChecker()

# Einzelne Prüfung
result = checker.check(
    plugin_version="1.2.0",
    required_range=">=1.0.0 <2.0.0",
)

print(result.level)     # CompatibilityLevel.COMPATIBLE
print(result.message)   # Beschreibung
print(result.severity)  # Schweregrad

# Alle Plugins prüfen
results = checker.check_all(plugins)

CompatibilityLevel

from trixy_core.plugins import CompatibilityLevel

class CompatibilityLevel(Enum):
    COMPATIBLE = "compatible"       # Vollständig kompatibel
    DEPRECATED = "deprecated"       # Veraltet, aber funktional
    INCOMPATIBLE = "incompatible"   # Nicht kompatibel
    UNKNOWN = "unknown"             # Unbekannt

---

Abhängigkeiten

Dependency

from trixy_core.plugins import Dependency

dep = Dependency(
    name="other_plugin",
    version_requirement=">=1.0.0",
    optional=False,
    type="plugin",  # oder "service", "library"
)

DependencyGraph

from trixy_core.plugins import DependencyGraph, DependencyNode, Dependency

graph = DependencyGraph()

# Knoten hinzufügen
graph.add_node(DependencyNode(name="plugin_a", version="1.0.0"))
graph.add_node(DependencyNode(name="plugin_b", version="2.0.0"))
graph.add_node(DependencyNode(name="plugin_c", version="1.5.0"))

# Abhängigkeiten definieren
dep = Dependency(name="plugin_a", version_requirement=">=1.0.0")
graph.add_edge("plugin_b", "plugin_a", dep)  # B hängt von A ab

dep2 = Dependency(name="plugin_b", version_requirement=">=2.0.0")
graph.add_edge("plugin_c", "plugin_b", dep2)  # C hängt von B ab

# Abfragen
node = graph.get_node("plugin_a")
all_deps = graph.get_all_dependencies("plugin_c")  # ["plugin_b", "plugin_a"]
dependents = graph.get_dependents("plugin_a")      # ["plugin_b"]

# Zyklen finden
cycles = graph.find_cycles()  # Liste von Zyklen

# Topologische Sortierung (Lade-Reihenfolge)
order = graph.topological_sort()  # ["plugin_a", "plugin_b", "plugin_c"]

DependencyResolver

from trixy_core.plugins import DependencyResolver, ResolutionError, CycleError

resolver = DependencyResolver()

# Abhängigkeiten auflösen
try:
    result = resolver.resolve(plugins)

    print(result.load_order)   # Lade-Reihenfolge
    print(result.unresolved)   # Unaufgelöste Abhängigkeiten
    print(result.errors)       # Fehler

except CycleError as e:
    print(f"Zyklus erkannt: {e.cycle}")

except ResolutionError as e:
    print(f"Auflösung fehlgeschlagen: {e}")

# Graph validieren
is_valid = resolver.validate(graph)

---

Registry

PluginManifest

from trixy_core.plugins import PluginManifest, load_manifest, save_manifest

# Laden
manifest = load_manifest(Path("plugins/my_plugin"))

print(manifest.name)          # "my_plugin"
print(manifest.version)       # "1.0.0"
print(manifest.description)   # Beschreibung
print(manifest.author)        # Autor
print(manifest.dependencies)  # Liste von Dependency
print(manifest.capabilities)  # Liste von PluginCapability
print(manifest.hooks)         # Deklarierte Hooks

# Erstellen
manifest = PluginManifest(
    name="my_plugin",
    version="1.0.0",
    description="Mein Plugin",
    author="Max Mustermann",
    dependencies=[
        Dependency(name="core_plugin", version_requirement=">=1.0.0")
    ],
    capabilities=[
        PluginCapability(name="tts", version="1.0", description="TTS Support")
    ],
)

# Speichern
save_manifest(manifest, Path("plugins/my_plugin"))

PluginDiscovery

from trixy_core.plugins import PluginDiscovery

discovery = PluginDiscovery()

# Plugins im Verzeichnis finden
result = discovery.discover(Path("plugins/"))

print(result.found_count)  # Anzahl gefundener Plugins
print(result.plugins)      # Liste von Plugin-Infos
print(result.errors)       # Fehler beim Scannen

# Nur Verzeichnisse scannen
paths = discovery.scan_directory(Path("plugins/"))

PluginRegistry

from trixy_core.plugins import PluginRegistry, PluginInfo

registry = PluginRegistry()

# Plugin registrieren
info = PluginInfo(
    name="my_plugin",
    version="1.0.0",
    author="Max Mustermann",
    capabilities=["tts", "stt"],
    status="active",
)

registry.register(info)

# Abrufen
plugin_info = registry.get("my_plugin")
all_plugins = registry.get_all()

# Suchen
results = registry.search({"author": "Max"})

# Nach Fähigkeit
tts_plugins = registry.get_by_capability("tts")

# Deregistrieren
registry.unregister("my_plugin")

PluginCapability

from trixy_core.plugins import PluginCapability

capability = PluginCapability(
    name="tts",
    version="1.0",
    description="Text-to-Speech Unterstützung",
)

---

Hooks

Inter-Plugin-Kommunikation.

PluginHook

from trixy_core.plugins import PluginHook, HookPriority, HookResult

# Hook erstellen
hook = PluginHook(name="before_response")

# Callback registrieren
def modify_response(data: dict, context: dict = None) -> dict:
    data["modified"] = True
    return data

hook.register_callback(modify_response, priority=HookPriority.NORMAL)

# Hook ausführen
results = hook.execute(
    data={"text": "Hallo"},
    context={"satellite_id": "living_room"},
)

for result in results:
    print(result.hook_name)       # "before_response"
    print(result.callback_name)   # Funktionsname
    print(result.success)         # True/False
    print(result.return_value)    # Rückgabewert
    print(result.error)           # Fehler (falls vorhanden)

# Callback entfernen
hook.unregister_callback(modify_response)

# Alle Callbacks abrufen
callbacks = hook.get_callbacks()

HookRegistry

from trixy_core.plugins import HookRegistry, PluginHook

registry = HookRegistry()

# Hook registrieren
hook = PluginHook(name="before_response")
registry.register_hook("before_response", hook)

# Hook abrufen
hook = registry.get_hook("before_response")

# Alle Hooks
all_hooks = registry.get_all()

# Hook ausführen
results = registry.execute_hook(
    name="before_response",
    data={"text": "Hallo"},
    context={},
)

HookPriority

from trixy_core.plugins import HookPriority

class HookPriority(Enum):
    HIGHEST = 0    # Zuerst ausgeführt
    HIGH = 25
    NORMAL = 50    # Standard
    LOW = 75
    LOWEST = 100   # Zuletzt ausgeführt

In Plugins verwenden

class MyPlugin(TrixyPlugin):
    async def on_load(self) -> None:
        # Hook registrieren
        registry = self.application.get_service("hook_registry")
        hook = registry.get_hook("before_response")

        if hook:
            hook.register_callback(self.modify_response)

    def modify_response(self, data: dict, context: dict = None) -> dict:
        # Response modifizieren
        return data

---

Hot-Reload

FileWatcher

from trixy_core.plugins import FileWatcher, WatchEvent, WatchEventType

watcher = FileWatcher()

# Watch-Callback
def on_change(event: WatchEvent) -> None:
    print(event.event_type)  # WatchEventType.MODIFIED
    print(event.path)        # Path zum geänderten File
    print(event.timestamp)   # Zeitstempel

# Überwachung hinzufügen
watcher.add_watch(Path("plugins/my_plugin"), on_change)

# Starten
watcher.start()

# Stoppen
watcher.stop()

# Überwachung entfernen
watcher.remove_watch(Path("plugins/my_plugin"))

WatchEventType

from trixy_core.plugins import WatchEventType

class WatchEventType(Enum):
    CREATED = "created"    # Datei erstellt
    MODIFIED = "modified"  # Datei geändert
    DELETED = "deleted"    # Datei gelöscht
    MOVED = "moved"        # Datei verschoben

PluginHotReloader

from trixy_core.plugins import PluginHotReloader, ReloadConfig, ReloadResult

config = ReloadConfig(
    enabled=True,
    watch_paths=["*.py", "config.json"],
    debounce_ms=500,    # Entprellen
    auto_reload=True,   # Automatisch neu laden
)

reloader = PluginHotReloader(plugin_manager, config)

# Überwachung für Plugin aktivieren
reloader.enable_watch("my_plugin")

# Überwachung deaktivieren
reloader.disable_watch("my_plugin")

# Manuelles Reload
result = reloader.reload_plugin("my_plugin")

print(result.plugin_name)  # "my_plugin"
print(result.success)      # True/False
print(result.errors)       # Fehler-Liste
print(result.duration_ms)  # Dauer in Millisekunden

# Alle Plugins neu laden
results = reloader.reload_all()

---

Testing

PluginTestHarness

from trixy_core.plugins import PluginTestHarness, TestContext

harness = PluginTestHarness()

# Plugin laden
plugin = harness.load_plugin(Path("plugins/my_plugin"))

# Test-Kontext
ctx = TestContext(
    plugin=plugin,
    application=harness.get_mock_application(),
    event_manager=harness.application.event_manager,
    satellite_manager=harness.application.satellite_manager,
)

# Harness starten
await harness.start()

# Event triggern
harness.trigger_event("wakeword_received", {
    "satellite_id": "test_satellite",
    "confidence": 0.95,
})

# Harness stoppen
await harness.stop()

Mock-Objekte

from trixy_core.plugins import (
    MockApplication,
    MockEventManager,
    MockSatelliteManager,
    MockPlugin,
    MockConfigManager,
    MockScheduler,
)

# MockApplication
app = MockApplication()
print(app.config)
print(app.event_manager)
print(app.services)

# MockEventManager
events = MockEventManager()
events.trigger("test_event", {"data": "value"})
events.subscribe("test_event", handler)
calls = events.verify_calls("test_event")

# MockSatelliteManager
satellites = MockSatelliteManager()
satellites.add_satellite("test_sat", room="test_room")
sat = satellites.get("test_sat")
all_sats = satellites.get_all()
await satellites.disconnect("test_sat")

# MockPlugin
mock_plugin = MockPlugin(name="test_plugin")

Test-Fixtures

from trixy_core.plugins import (
    create_test_plugin_dir,
    create_test_manifest,
    cleanup_test_dir,
    PluginTestFixture,
)

# Temporäres Plugin-Verzeichnis erstellen
plugin_dir = create_test_plugin_dir("test_plugin", version="1.0.0")

# Manifest erstellen
create_test_manifest(plugin_dir, {
    "name": "test_plugin",
    "version": "1.0.0",
    "author": "Test",
})

# Nach Tests aufräumen
cleanup_test_dir(plugin_dir)

# Oder mit Fixture
fixture = PluginTestFixture()
await fixture.setup()

# Tests ausführen...

await fixture.teardown()

Beispiel-Test

import pytest
from trixy_core.plugins import (
    PluginTestHarness,
    create_test_plugin_dir,
    cleanup_test_dir,
)

@pytest.fixture
def harness():
    h = PluginTestHarness()
    yield h

@pytest.fixture
def test_plugin_dir():
    dir = create_test_plugin_dir("test_plugin")
    yield dir
    cleanup_test_dir(dir)

@pytest.mark.asyncio
async def test_plugin_loads(harness, test_plugin_dir):
    """Test ob Plugin lädt."""
    plugin = harness.load_plugin(test_plugin_dir)
    assert plugin is not None
    assert plugin.name == "test_plugin"

@pytest.mark.asyncio
async def test_plugin_handles_event(harness, test_plugin_dir):
    """Test Event-Handling."""
    plugin = harness.load_plugin(test_plugin_dir)
    await harness.start()

    harness.trigger_event("test_event", {"value": 42})

    # Assertions...

    await harness.stop()

---

Plugin-Struktur

plugins/
└── my_plugin/
    ├── main.py           # Plugin-Klasse (erforderlich)
    ├── config.json       # Konfiguration (erforderlich)
    ├── manifest.json     # Metadaten (optional)
    └── ...               # Weitere Dateien

main.py

from trixy_core.plugins import TrixyPlugin

class MyPlugin(TrixyPlugin):
    NAME = "my_plugin"
    VERSION = "1.0.0"

    async def on_load(self) -> None:
        pass

    async def on_unload(self) -> None:
        pass

config.json

{
    "enabled": true,
    "settings": {
        "api_key": "",
        "timeout": 30
    }
}

manifest.json

{
    "name": "my_plugin",
    "version": "1.0.0",
    "description": "Mein Plugin",
    "author": "Max Mustermann",
    "dependencies": [
        {
            "name": "core_plugin",
            "version_requirement": ">=1.0.0"
        }
    ],
    "capabilities": [
        {
            "name": "tts",
            "version": "1.0"
        }
    ]
}