Trixy3/plugins/example_plugin/main.py

"""
Example Plugin for Trixy Application

This plugin demonstrates the basic functionality of the Trixy plugin system:
- Basic plugin structure and inheritance from TrixyPlugin
- Event handling with @TrixyEvent decorator
- Configuration management with auto-loading and type-casting
- Plugin lifecycle management (initialize, enable, disable, cleanup)
- Integration with application container and event system
- Basic health monitoring and error handling

This serves as a template and reference for developing new plugins.

Features demonstrated:
- Multiple event handlers
- Configuration with different data types
- Plugin state management
- Error handling and recovery
- Health status reporting
- Custom plugin methods
"""

from typing import Any, Dict, Optional
import time
import random

# Import plugin base class and event decorator
from trixy_core.plugins import TrixyPlugin
from trixy_core.events import TrixyEvent


def pprint(message: str) -> None:
    """Example plugin logging function."""
    print(f"[EXAMPLE_PLUGIN] {message}")


class ExamplePlugin(TrixyPlugin):
    """
    Example plugin demonstrating basic Trixy plugin functionality.
    
    This plugin:
    - Responds to various system events
    - Maintains configuration settings
    - Demonstrates plugin lifecycle management
    - Shows event handling patterns
    - Provides health monitoring example
    """
    
    def initialize(self) -> None:
        """Initialize the example plugin."""
        pprint("Initializing Example Plugin...")
        
        # Plugin-specific initialization
        self._message_count = 0
        self._last_wakeword_time = None
        self._last_text_received = None
        self._startup_time = time.time()
        
        # Configuration validation and setup
        self._setup_configuration()
        
        # Plugin health status
        self._health_status = "initialized"
        
        pprint(f"Example Plugin initialized with config: {self.config}")
    
    def _setup_configuration(self) -> None:
        """Set up plugin configuration with defaults and validation."""
        # Get configuration values with type casting and defaults
        self.greeting_message = self.get_config_value(
            "greeting_message", 
            default="Hello from Example Plugin!", 
            type_cast=str
        )
        
        self.response_enabled = self.get_config_value(
            "response_enabled", 
            default=True, 
            type_cast=bool
        )
        
        self.response_delay = self.get_config_value(
            "response_delay", 
            default=1.0, 
            type_cast=float
        )
        
        self.max_responses_per_session = self.get_config_value(
            "max_responses_per_session", 
            default=5, 
            type_cast=int
        )
        
        self.allowed_speakers = self.get_config_value(
            "allowed_speakers", 
            default=["all"], 
            type_cast=list
        )
        
        # Validate configuration
        if self.response_delay < 0:
            pprint("Warning: response_delay cannot be negative, setting to 0")
            self.set_config_value("response_delay", 0.0)
            self.response_delay = 0.0
        
        if self.max_responses_per_session < 1:
            pprint("Warning: max_responses_per_session must be at least 1")
            self.set_config_value("max_responses_per_session", 1)
            self.max_responses_per_session = 1
    
    # Event Handlers
    
    @TrixyEvent("system_startup")
    def on_system_startup(self, event_name: str, event_data: Any) -> None:
        """Handle system startup event."""
        if not self.is_enabled():
            return
        
        pprint("System startup detected!")
        pprint(f"Startup mode: {event_data.get('mode', 'unknown')}")
        pprint(f"Debug mode: {event_data.get('debug_mode', False)}")
        
        self._health_status = "running"
        
        # Example: Send a greeting message
        if self.response_enabled:
            try:
                # Get event handler to trigger a custom event
                event_handler = self.application.get_event_handler()
                if event_handler:
                    greeting_data = {
                        "message": self.greeting_message,
                        "plugin": "example_plugin",
                        "timestamp": time.time()
                    }
                    event_handler.trigger_event("example_greeting", greeting_data)
            except Exception as e:
                pprint(f"Error sending greeting: {e}")
    
    @TrixyEvent(["wakeword_received", "text_received"])
    def handle_voice_events(self, event_name: str, event_data: Any) -> None:
        """Handle wakeword and text events."""
        if not self.is_enabled():
            return
        
        try:
            if event_name == "wakeword_received":
                self._handle_wakeword(event_data)
            elif event_name == "text_received":
                self._handle_text_received(event_data)
        except Exception as e:
            pprint(f"Error handling voice event {event_name}: {e}")
    
    def _handle_wakeword(self, event_data: Any) -> None:
        """Handle wakeword detection."""
        speaker_name = event_data.get('speaker_name', 'unknown')
        satellite_id = event_data.get('satellite_id', 'unknown')
        volume = event_data.get('volume', 0)
        
        pprint(f"Wakeword detected! Speaker: {speaker_name}, Satellite: {satellite_id}, Volume: {volume}")
        
        # Check if speaker is allowed
        if not self._is_speaker_allowed(speaker_name):
            pprint(f"Speaker {speaker_name} not in allowed list, ignoring")
            return
        
        self._last_wakeword_time = time.time()
        self._message_count += 1
        
        # Example: Log wakeword statistics
        pprint(f"Total wakewords processed: {self._message_count}")
    
    def _handle_text_received(self, event_data: Any) -> None:
        """Handle text from speech-to-text."""
        text = event_data.get('text', '')
        confidence = event_data.get('confidence', 0.0)
        speaker_name = event_data.get('speaker_info', {}).get('name', 'unknown')
        
        pprint(f"Text received: '{text}' (confidence: {confidence:.2f}) from {speaker_name}")
        
        # Check if speaker is allowed
        if not self._is_speaker_allowed(speaker_name):
            pprint(f"Speaker {speaker_name} not in allowed list, ignoring")
            return
        
        self._last_text_received = text
        
        # Example: Simple command recognition
        if text.lower().startswith("example"):
            self._handle_example_command(text, event_data)
    
    def _handle_example_command(self, text: str, event_data: Any) -> None:
        """Handle commands that start with 'example'."""
        if not self.response_enabled:
            pprint("Response disabled, ignoring command")
            return
        
        # Check session limit
        if self._message_count > self.max_responses_per_session:
            pprint("Maximum responses per session reached, ignoring command")
            return
        
        pprint(f"Handling example command: {text}")
        
        # Add response delay if configured
        if self.response_delay > 0:
            time.sleep(self.response_delay)
        
        # Generate a response
        responses = [
            "Example plugin is working!",
            f"Hello! I received: {text}",
            "Example plugin at your service!",
            f"Message count: {self._message_count}",
            "This is an example response!"
        ]
        
        response = random.choice(responses)
        
        try:
            # Trigger a response event
            event_handler = self.application.get_event_handler()
            if event_handler:
                response_data = {
                    "text": response,
                    "plugin": "example_plugin",
                    "original_command": text,
                    "timestamp": time.time()
                }
                event_handler.trigger_event("example_response", response_data)
                pprint(f"Triggered response: {response}")
        except Exception as e:
            pprint(f"Error triggering response: {e}")
    
    @TrixyEvent("system_shutdown")
    def on_system_shutdown(self, event_name: str, event_data: Any) -> None:
        """Handle system shutdown event."""
        pprint("System shutdown detected, saving statistics...")
        
        # Example: Save plugin statistics
        uptime = time.time() - self._startup_time
        stats = {
            "uptime_seconds": uptime,
            "messages_processed": self._message_count,
            "last_activity": self._last_wakeword_time or self._startup_time
        }
        
        # Save statistics to config for persistence
        self.set_config_value("last_session_stats", stats)
        
        pprint(f"Example plugin shutting down after {uptime:.1f} seconds")
    
    # Lifecycle Hooks
    
    def on_enable(self) -> None:
        """Called when the plugin is enabled."""
        pprint("Example plugin enabled")
        self._health_status = "enabled"
        
        # Reset counters on enable
        self._message_count = 0
        self._startup_time = time.time()
    
    def on_disable(self) -> None:
        """Called when the plugin is disabled."""
        pprint("Example plugin disabled")
        self._health_status = "disabled"
    
    def cleanup(self) -> None:
        """Clean up plugin resources."""
        pprint("Example plugin cleanup")
        
        # Save final statistics
        if hasattr(self, '_startup_time'):
            uptime = time.time() - self._startup_time
            final_stats = {
                "total_uptime_seconds": uptime,
                "total_messages_processed": self._message_count,
                "last_cleanup": time.time()
            }
            self.set_config_value("cleanup_stats", final_stats)
        
        self._health_status = "cleanup_complete"
    
    # Plugin-Specific Methods
    
    def _is_speaker_allowed(self, speaker_name: str) -> bool:
        """Check if a speaker is in the allowed list."""
        if "all" in self.allowed_speakers:
            return True
        return speaker_name in self.allowed_speakers
    
    def get_plugin_stats(self) -> Dict[str, Any]:
        """Get plugin statistics."""
        uptime = time.time() - self._startup_time if hasattr(self, '_startup_time') else 0
        
        return {
            "plugin_name": "example_plugin",
            "enabled": self.is_enabled(),
            "uptime_seconds": uptime,
            "messages_processed": self._message_count,
            "last_wakeword_time": self._last_wakeword_time,
            "last_text_received": self._last_text_received,
            "health_status": getattr(self, '_health_status', 'unknown'),
            "configuration": {
                "greeting_message": self.greeting_message,
                "response_enabled": self.response_enabled,
                "response_delay": self.response_delay,
                "max_responses_per_session": self.max_responses_per_session,
                "allowed_speakers": self.allowed_speakers
            }
        }
    
    def reset_counters(self) -> None:
        """Reset plugin counters (utility method)."""
        self._message_count = 0
        self._last_wakeword_time = None
        self._last_text_received = None
        pprint("Plugin counters reset")
    
    def simulate_error(self) -> None:
        """Simulate an error for testing error handling."""
        pprint("Simulating error for testing...")
        raise Exception("This is a simulated error for testing plugin error handling")
    
    # Override health check to provide plugin-specific health info
    
    def check_health(self) -> Dict[str, Any]:
        """Check plugin health with custom status."""
        # Get base health info
        health = super().check_health()
        
        # Add plugin-specific health information
        health.update({
            'custom_health_status': getattr(self, '_health_status', 'unknown'),
            'message_count': self._message_count,
            'last_activity': self._last_wakeword_time,
            'uptime': time.time() - self._startup_time if hasattr(self, '_startup_time') else 0,
            'configuration_valid': self._validate_current_config(),
            'response_capability': self.response_enabled and self.is_enabled()
        })
        
        return health
    
    def _validate_current_config(self) -> bool:
        """Validate current configuration."""
        try:
            # Check all required config values are present and valid
            required_keys = ['greeting_message', 'response_enabled', 'response_delay', 'max_responses_per_session', 'allowed_speakers']
            
            for key in required_keys:
                if not hasattr(self, key):
                    return False
            
            # Check value ranges
            if self.response_delay < 0 or self.max_responses_per_session < 1:
                return False
            
            return True
        except:
            return False