Trixy3/main.py

#!/usr/bin/env python3
"""
Trixy Voice Assistant - Main Entry Point

This module serves as the entry point for the Trixy voice assistant application,
supporting three deployment modes: server, client/satellite, and standalone.

Usage:
    python3 main.py [server|client|standalone] [--debug] [--config custom_config.json]

Modes:
    server     - Central hub with event handler, plugins, satellite management
    client     - Satellite mode with wakeword detection and audio streaming
    standalone - Full system that can optionally connect to server for sync

Arguments:
    --debug    - Enable development mode (no TUI, print statements enabled)
    --config   - Specify custom configuration file path

Author: Generated with Claude Code
"""

import sys
import os
import argparse
import signal
import logging
import traceback
from pathlib import Path
from typing import Optional, Dict, Any

# Import trainer CLI functionality for training mode
try:
    from trainer.cli import (
        add_wakeword_arguments, add_voice_recognition_arguments,
        add_validation_arguments, add_list_arguments, add_convert_arguments,
        train_wakeword, train_voice_recognition, validate_model,
        list_models, convert_model, setup_cli_logging
    )
    TRAINER_AVAILABLE = True
except ImportError:
    TRAINER_AVAILABLE = False
    # Note: Trainer modules not available - training mode disabled

# Global application container reference for cleanup
_application_container = None
_debug_mode = False


def pprint(message: str) -> None:
    """
    Adaptive print function that behaves differently in debug vs production mode.
    
    In debug mode: Prints directly to stdout
    In production mode: Uses logging system with Textual TUI
    
    Args:
        message: The message to print/log
    """
    if _debug_mode:
        print(f"[TRIXY] {message}")
    else:
        # In production mode, use logging system
        # The TUI will capture and display log messages appropriately
        logging.info(message)


def create_directory_structure() -> None:
    """
    Create the required directory structure for Trixy if it doesn't exist.
    
    Creates all necessary directories as specified in CLAUDE.md:
    - trixy_core/ with all component subdirectories
    - plugins/
    - models/ with subdirectories
    - config/
    - assets/ with default profile
    - trainer/ with data subdirectories
    """
    directories = [
        # Core system modules
        "trixy_core",
        "trixy_core/events",
        "trixy_core/network",
        "trixy_core/network/cmd",
        "trixy_core/satellites", 
        "trixy_core/plugins",
        "trixy_core/config",
        "trixy_core/assets",
        "trixy_core/scheduler",
        "trixy_core/conversation",
        "trixy_core/arbitration",
        
        # Plugin directory
        "plugins",
        
        # ML model storage
        "models",
        "models/wakeword",
        "models/voice_recognition",
        
        # Configuration files
        "config",
        
        # Profile-based assets
        "assets",
        "assets/default",
        
        # ML training system
        "trainer",
        "trainer/data",
        "trainer/data/wakeword",
        "trainer/data/wakeword/raw",
        "trainer/data/voice_recognition",
        "trainer/data/voice_recognition/raw",
        "trainer/wakeword",
        "trainer/voice_recognition",
    ]
    
    base_path = Path.cwd()
    created_dirs = []
    
    for directory in directories:
        dir_path = base_path / directory
        if not dir_path.exists():
            try:
                dir_path.mkdir(parents=True, exist_ok=True)
                created_dirs.append(str(dir_path))
                pprint(f"Created directory: {dir_path}")
            except OSError as e:
                pprint(f"Failed to create directory {dir_path}: {e}")
                sys.exit(1)
    
    if created_dirs:
        pprint(f"Successfully created {len(created_dirs)} directories")
    else:
        pprint("All required directories already exist")


def setup_logging(debug: bool = False) -> None:
    """
    Initialize the logging system based on the mode.
    
    In debug mode: Basic console logging
    In production mode: Structured logging for TUI integration
    
    Args:
        debug: Whether debug mode is enabled
    """
    global _debug_mode
    _debug_mode = debug
    
    if debug:
        # Debug mode: Simple console logging
        logging.basicConfig(
            level=logging.DEBUG,
            format='[%(asctime)s] [%(levelname)s] %(name)s: %(message)s',
            datefmt='%H:%M:%S'
        )
        pprint("Debug mode enabled - console logging active")
    else:
        # Production mode: Structured logging for TUI
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.StreamHandler(sys.stdout)
            ]
        )
        pprint("Production mode enabled - TUI logging active")


def add_training_arguments(parser: argparse.ArgumentParser) -> None:
    """
    Add training-specific arguments to the parser.
    
    Args:
        parser: The argument parser to add training arguments to
    """
    if not TRAINER_AVAILABLE:
        return
        
    # Add global training arguments
    parser.add_argument(
        '--log-level', choices=["DEBUG", "INFO", "WARNING", "ERROR"],
        default="INFO", help="Logging level for training"
    )
    parser.add_argument(
        '--log-file', type=str,
        help="Path to log file for training (default: console only)"
    )
    parser.add_argument(
        '--device', choices=["auto", "cpu", "cuda"],
        default="auto", help="Device to use for training/inference"
    )
    
    # Create subparsers for training types
    train_subparsers = parser.add_subparsers(dest="train_type", help="Type of model to train")
    
    # Wakeword training
    wakeword_parser = train_subparsers.add_parser("wakeword", help="Train wakeword detection model")
    add_wakeword_arguments(wakeword_parser)
    
    # Voice recognition training  
    voice_parser = train_subparsers.add_parser("voice-recognition", help="Train voice recognition model")
    add_voice_recognition_arguments(voice_parser)
    
    # Validation command
    validate_parser = train_subparsers.add_parser("validate", help="Validate a trained model")
    add_validation_arguments(validate_parser)
    
    # List models command
    list_parser = train_subparsers.add_parser("list-models", help="List available trained models")
    add_list_arguments(list_parser)
    
    # Convert model command
    convert_parser = train_subparsers.add_parser("convert", help="Convert model between formats")
    add_convert_arguments(convert_parser)


def parse_arguments() -> argparse.Namespace:
    """
    Parse command line arguments.
    
    Returns:
        Parsed arguments namespace
    """
    parser = argparse.ArgumentParser(
        description="Trixy Voice Assistant - Professional voice assistant system",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python3 main.py server --debug
  python3 main.py client --config /path/to/client_config.json
  python3 main.py standalone
  python3 main.py train wakeword --data-dir ./data --num-epochs 100
  python3 main.py train voice-recognition --data-dir ./data --model-type ecapa_tdnn
  python3 main.py train validate ./models/my_model.pth --test-data ./data/test
  python3 main.py train list-models --models-dir ./models
        """
    )
    
    # Create subparsers for different modes
    subparsers = parser.add_subparsers(dest="mode", help="Deployment mode")
    
    # Server mode
    server_parser = subparsers.add_parser("server", help="Central hub with event handler, plugins, satellite management")
    server_parser.add_argument('--debug', action='store_true', help='Enable debug mode (no TUI, print statements enabled)')
    server_parser.add_argument('--config', type=str, metavar='CONFIG_FILE', help='Path to custom configuration file (overrides default config)')
    
    # Client mode
    client_parser = subparsers.add_parser("client", help="Satellite mode with wakeword detection and audio streaming")
    client_parser.add_argument('--debug', action='store_true', help='Enable debug mode (no TUI, print statements enabled)')
    client_parser.add_argument('--config', type=str, metavar='CONFIG_FILE', help='Path to custom configuration file (overrides default config)')
    
    # Standalone mode
    standalone_parser = subparsers.add_parser("standalone", help="Full system that can optionally connect to server for sync")
    standalone_parser.add_argument('--debug', action='store_true', help='Enable debug mode (no TUI, print statements enabled)')
    standalone_parser.add_argument('--config', type=str, metavar='CONFIG_FILE', help='Path to custom configuration file (overrides default config)')
    
    # Training mode
    train_parser = subparsers.add_parser("train", help="ML model training")
    train_parser.add_argument('--debug', action='store_true', help='Enable debug mode (no TUI, print statements enabled)')
    train_parser.add_argument('--config', type=str, metavar='CONFIG_FILE', help='Path to custom configuration file (overrides default config)')
    add_training_arguments(train_parser)
    
    return parser.parse_args()


def validate_environment() -> None:
    """
    Validate the runtime environment and dependencies.
    
    Raises:
        SystemExit: If critical dependencies are missing
    """
    python_version = sys.version_info
    if python_version < (3, 8):
        pprint(f"Error: Python 3.8+ required, found {python_version.major}.{python_version.minor}")
        sys.exit(1)
    
    pprint(f"Python {python_version.major}.{python_version.minor}.{python_version.micro} detected")
    
    # Check for write permissions in current directory
    try:
        test_file = Path.cwd() / ".trixy_write_test"
        test_file.touch()
        test_file.unlink()
        pprint("Write permissions verified")
    except (OSError, PermissionError) as e:
        pprint(f"Error: No write permissions in current directory: {e}")
        sys.exit(1)


def initialize_application_container(mode: str, config_path: Optional[str] = None):
    """
    Initialize and return the application container based on the deployment mode.
    
    This is a placeholder for the actual application container implementation.
    The container will be implemented in trixy_core/ modules.
    
    Args:
        mode: Deployment mode (server, client, standalone)
        config_path: Optional path to custom configuration file
        
    Returns:
        Application container instance
        
    Raises:
        ImportError: If core modules are not yet implemented
    """
    try:
        # This import will be available once the core modules are implemented
        from trixy_core.application import ApplicationContainer
        
        pprint(f"Initializing {mode} mode application container")
        
        container = ApplicationContainer(mode=mode, config_path=config_path)
        pprint("Application container initialized successfully")
        
        return container
        
    except ImportError as e:
        pprint(f"Core modules not yet implemented: {e}")
        pprint("This is expected during initial setup phase")
        pprint("The application container will be available once core modules are created")
        return None


def setup_signal_handlers() -> None:
    """
    Setup signal handlers for graceful shutdown.
    """
    def signal_handler(signum, frame):
        pprint(f"Received signal {signum}, initiating graceful shutdown...")
        shutdown_application()
        sys.exit(0)
    
    # Handle common termination signals
    signal.signal(signal.SIGINT, signal_handler)   # Ctrl+C
    signal.signal(signal.SIGTERM, signal_handler)  # Termination signal
    
    # Handle SIGQUIT on Unix systems
    if hasattr(signal, 'SIGQUIT'):
        signal.signal(signal.SIGQUIT, signal_handler)
    
    pprint("Signal handlers configured for graceful shutdown")


def execute_training_mode(args: argparse.Namespace) -> bool:
    """
    Execute training mode operations.
    
    Args:
        args: Parsed command line arguments
        
    Returns:
        bool: True if training completed successfully, False otherwise
    """
    if not TRAINER_AVAILABLE:
        pprint("Error: Training modules not available")
        pprint("Please ensure the trainer module is properly installed")
        return False
    
    # Clear root logger handlers to prevent duplicate logging
    # The root logger was configured earlier via logging.basicConfig()
    # but we want clean training-specific logging only
    root_logger = logging.getLogger()
    root_logger.handlers.clear()

    # Setup training-specific logging
    logger = setup_cli_logging(
        level=getattr(args, 'log_level', 'INFO'),
        log_file=getattr(args, 'log_file', None)
    )
    
    # Handle device selection for training
    if hasattr(args, 'device') and args.device == "auto":
        try:
            import torch
            device = "cuda" if torch.cuda.is_available() else "cpu"
            args.device = device
            logger.info(f"Auto-selected device: {device}")
        except ImportError:
            args.device = "cpu"
            logger.warning("PyTorch not available, defaulting to CPU")
    
    success = False
    
    try:
        if not hasattr(args, 'train_type') or args.train_type is None:
            pprint("Error: Please specify training type (wakeword, voice-recognition, validate, list-models, convert)")
            pprint("Usage: python3 main.py train <type> [arguments...]")
            return False
            
        if args.train_type == "wakeword":
            success = train_wakeword(args, logger)
        elif args.train_type == "voice-recognition":
            success = train_voice_recognition(args, logger)
        elif args.train_type == "validate":
            success = validate_model(args, logger)
        elif args.train_type == "list-models":
            success = list_models(args, logger)
        elif args.train_type == "convert":
            success = convert_model(args, logger)
        else:
            pprint(f"Error: Unknown training type '{args.train_type}'")
            pprint("Available types: wakeword, voice-recognition, validate, list-models, convert")
            return False
            
    except KeyboardInterrupt:
        logger.info("Training cancelled by user")
        return True
    except Exception as e:
        logger.error(f"Training failed with error: {e}")
        if _debug_mode:
            traceback.print_exc()
        return False
    
    return success


def shutdown_application() -> None:
    """
    Perform graceful shutdown of the application.
    
    This function will:
    1. Stop all running services
    2. Close network connections
    3. Save state and configuration
    4. Clean up resources
    """
    global _application_container
    
    pprint("Starting graceful shutdown sequence...")
    
    if _application_container:
        try:
            # Call the application container's shutdown method
            # This will be implemented when the container is created
            if hasattr(_application_container, 'shutdown'):
                _application_container.shutdown()
                pprint("Application container shutdown completed")
            else:
                pprint("Application container shutdown method not available")
        except Exception as e:
            pprint(f"Error during application shutdown: {e}")
            if _debug_mode:
                traceback.print_exc()
    
    pprint("Graceful shutdown completed")


def main() -> None:
    """
    Main entry point for the Trixy voice assistant application.
    
    This function:
    1. Parses command line arguments
    2. Validates the environment
    3. Creates directory structure
    4. Initializes logging
    5. Sets up signal handlers
    6. Initializes the application container
    7. Starts the application
    8. Handles graceful shutdown
    """
    global _application_container
    
    try:
        # Parse command line arguments
        args = parse_arguments()
        
        # Setup logging early
        debug_mode = getattr(args, 'debug', False)
        setup_logging(debug_mode)
        
        # Print startup banner
        pprint("="*50)
        pprint("Trixy Voice Assistant")
        pprint("Professional Voice Assistant System")
        pprint("="*50)
        pprint(f"Mode: {args.mode.upper()}")
        pprint(f"Debug: {'ENABLED' if debug_mode else 'DISABLED'}")
        config_path = getattr(args, 'config', None)
        if config_path:
            pprint(f"Config: {config_path}")
        pprint("="*50)
        
        # Validate environment
        pprint("Validating runtime environment...")
        validate_environment()
        
        # Create directory structure
        pprint("Creating directory structure...")
        create_directory_structure()
        
        # Setup signal handlers for graceful shutdown
        pprint("Setting up signal handlers...")
        setup_signal_handlers()
        
        # Handle different modes
        if args.mode == 'train':
            # Training mode - bypass application container
            pprint("Starting Trixy training mode...")
            success = execute_training_mode(args)
            if success:
                pprint("Training completed successfully!")
                sys.exit(0)
            else:
                pprint("Training failed!")
                sys.exit(1)
        else:
            # Standard modes (server, client, standalone)
            # Initialize application container
            pprint("Initializing application container...")
            _application_container = initialize_application_container(
                mode=args.mode,
                config_path=config_path
            )
            
            if _application_container:
                # Start the application
                pprint(f"Starting Trixy {args.mode} mode...")
                
                # The actual application start will be implemented
                # when the application container is created
                if hasattr(_application_container, 'start'):
                    _application_container.start()
                else:
                    pprint("Application container start method not available")
                    pprint("Core system implementation is required")
                    pprint("This main.py serves as the foundation for future development")
                    
            else:
                pprint("Application container not available")
                pprint("Core modules need to be implemented")
                pprint("This main.py is ready for integration with core system")
            
    except KeyboardInterrupt:
        pprint("Received keyboard interrupt")
        shutdown_application()
        sys.exit(0)
        
    except Exception as e:
        pprint(f"Fatal error during startup: {e}")
        if _debug_mode:
            traceback.print_exc()
        shutdown_application()
        sys.exit(1)


if __name__ == "__main__":
    main()