Trixy3/trixy_core/network/cmd/wakeword.py

"""
Wakeword Detection Command Classes

This module implements commands for wakeword detection and management in the
Trixy system. These commands handle:

- Wakeword detection events from satellites
- Wakeword configuration and model management  
- Detection sensitivity and threshold settings
- Custom wakeword training and validation

These commands implement the wakeword workflow specified in CLAUDE.md.
"""

import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from .base import TrixyCommand, CommandResponse, ValidationError, CommandStatus


def pprint(message: str) -> None:
    """Wakeword command logging function."""
    print(f"[CMD.WAKEWORD] {message}")


@dataclass
class WakewordDetectedCommand(TrixyCommand):
    """
    Command sent when a satellite detects the wakeword.
    
    This is a critical command that triggers the conversation workflow.
    Contains information about the detection event and speaker.
    """
    
    wakeword_id: str = field(default="")
    speaker_id: str = field(default="") 
    speaker_name: str = field(default="")
    satellite_id: str = field(default="")
    confidence: float = field(default=0.0)
    volume: float = field(default=0.0)
    audio_buffer_size: int = field(default=0)
    
    def validate(self) -> bool:
        """Validate wakeword detection command."""
        if not self.wakeword_id:
            raise ValidationError("WakewordDetectedCommand must have a wakeword_id")
        if not self.satellite_id:
            raise ValidationError("WakewordDetectedCommand must have a satellite_id")
        if not (0.0 <= self.confidence <= 1.0):
            raise ValidationError("Confidence must be between 0.0 and 1.0")
        if not (0.0 <= self.volume <= 1.0):
            raise ValidationError("Volume must be between 0.0 and 1.0")
        return True
    
    def get_command_name(self) -> str:
        """Get command name."""
        return "WakewordDetectedCommand"
    
    def get_response_class(self):
        """Expect a detection response.""" 
        return WakewordDetectedResponse
    
    def get_timeout(self) -> float:
        """Short timeout for wakeword detection."""
        return 5.0
    
    def get_priority(self) -> int:
        """Very high priority for wakeword detection."""
        return 1


@dataclass
class WakewordConfigCommand(TrixyCommand):
    """
    Command to configure wakeword detection settings.
    
    Used to update sensitivity, thresholds, model settings, etc.
    """
    
    satellite_id: str = field(default="")
    wakeword_models: List[str] = field(default_factory=list)
    sensitivity: float = field(default=0.5)
    threshold: float = field(default=0.5)
    buffer_duration: float = field(default=2.0)
    config_changes: Dict[str, Any] = field(default_factory=dict)
    
    def validate(self) -> bool:
        """Validate wakeword config command."""
        if not (0.0 <= self.sensitivity <= 1.0):
            raise ValidationError("Sensitivity must be between 0.0 and 1.0")
        if not (0.0 <= self.threshold <= 1.0):
            raise ValidationError("Threshold must be between 0.0 and 1.0")
        if self.buffer_duration <= 0:
            raise ValidationError("Buffer duration must be positive")
        return True
    
    def get_command_name(self) -> str:
        """Get command name."""
        return "WakewordConfigCommand"
    
    def get_response_class(self):
        """Expect a config response."""
        return WakewordConfigResponse
    
    def get_timeout(self) -> float:
        """Timeout for config update."""
        return 15.0
    
    def get_priority(self) -> int:
        """Normal priority for config."""
        return 40


@dataclass
class WakewordStatusCommand(TrixyCommand):
    """
    Command to request wakeword detection status.
    
    Used for monitoring and debugging wakeword detection.
    """
    
    satellite_id: str = field(default="")
    status_type: str = field(default="general")  # general, models, performance, history
    
    def validate(self) -> bool:
        """Validate wakeword status command."""
        valid_types = ["general", "models", "performance", "history", "config"]
        if self.status_type not in valid_types:
            raise ValidationError(f"status_type must be one of: {valid_types}")
        return True
    
    def get_command_name(self) -> str:
        """Get command name."""
        return "WakewordStatusCommand"
    
    def get_timeout(self) -> float:
        """Timeout for status request."""
        return 10.0
    
    def get_priority(self) -> int:
        """Normal priority for status."""
        return 50


# Response Classes

@dataclass
class WakewordDetectedResponse(CommandResponse):
    """Response to wakeword detection command."""
    
    conversation_id: str = field(default="")
    selected_satellite: bool = field(default=False)
    recording_requested: bool = field(default=False)
    timeout_seconds: float = field(default=10.0)
    
    def validate(self) -> bool:
        """Validate detection response."""
        if self.recording_requested and not self.conversation_id:
            raise ValidationError("Recording request must include conversation_id")
        return True
    
    def get_response_name(self) -> str:
        """Get response name."""
        return "WakewordDetectedResponse"


@dataclass
class WakewordConfigResponse(CommandResponse):
    """Response to wakeword config command."""
    
    config_applied: bool = field(default=False)
    applied_changes: Dict[str, Any] = field(default_factory=dict)
    rejected_changes: Dict[str, str] = field(default_factory=dict)  # change -> reason
    requires_restart: bool = field(default=False)
    
    def validate(self) -> bool:
        """Validate config response."""
        return True
    
    def get_response_name(self) -> str:
        """Get response name."""
        return "WakewordConfigResponse"


# Utility functions

def create_wakeword_detection(
    wakeword_id: str,
    satellite_id: str,
    speaker_id: str = "",
    speaker_name: str = "",
    confidence: float = 0.8,
    volume: float = 0.5,
    audio_buffer_size: int = 0
) -> WakewordDetectedCommand:
    """
    Create a wakeword detection command.
    
    Args:
        wakeword_id: Detected wakeword identifier
        satellite_id: Satellite that detected the wakeword
        speaker_id: Identified speaker ID
        speaker_name: Identified speaker name
        confidence: Detection confidence (0.0-1.0)
        volume: Detection volume (0.0-1.0)
        audio_buffer_size: Size of audio buffer
        
    Returns:
        WakewordDetectedCommand: Detection command
    """
    return WakewordDetectedCommand(
        wakeword_id=wakeword_id,
        satellite_id=satellite_id,
        speaker_id=speaker_id,
        speaker_name=speaker_name,
        confidence=confidence,
        volume=volume,
        audio_buffer_size=audio_buffer_size
    )


def create_wakeword_config(
    satellite_id: str = "",
    wakeword_models: Optional[List[str]] = None,
    sensitivity: float = 0.5,
    threshold: float = 0.5,
    buffer_duration: float = 2.0,
    config_changes: Optional[Dict[str, Any]] = None
) -> WakewordConfigCommand:
    """
    Create a wakeword configuration command.
    
    Args:
        satellite_id: Target satellite ID (empty for broadcast)
        wakeword_models: List of wakeword model files
        sensitivity: Detection sensitivity (0.0-1.0)
        threshold: Detection threshold (0.0-1.0)
        buffer_duration: Audio buffer duration in seconds
        config_changes: Additional configuration changes
        
    Returns:
        WakewordConfigCommand: Configuration command
    """
    return WakewordConfigCommand(
        satellite_id=satellite_id,
        wakeword_models=wakeword_models or [],
        sensitivity=sensitivity,
        threshold=threshold,
        buffer_duration=buffer_duration,
        config_changes=config_changes or {}
    )


def create_detection_response(
    command_id: str,
    conversation_id: str = "",
    selected_satellite: bool = False,
    recording_requested: bool = False,
    timeout_seconds: float = 10.0
) -> WakewordDetectedResponse:
    """
    Create a wakeword detection response.
    
    Args:
        command_id: Original command ID
        conversation_id: Assigned conversation ID
        selected_satellite: Whether this satellite was selected
        recording_requested: Whether recording is requested
        timeout_seconds: Recording timeout
        
    Returns:
        WakewordDetectedResponse: Detection response
    """
    status = CommandStatus.SUCCESS if selected_satellite else CommandStatus.SUCCESS
    
    return WakewordDetectedResponse(
        command_id=command_id,
        status=status,
        conversation_id=conversation_id,
        selected_satellite=selected_satellite,
        recording_requested=recording_requested,
        timeout_seconds=timeout_seconds
    )


def create_config_response(
    command_id: str,
    config_applied: bool,
    applied_changes: Optional[Dict[str, Any]] = None,
    rejected_changes: Optional[Dict[str, str]] = None,
    requires_restart: bool = False
) -> WakewordConfigResponse:
    """
    Create a wakeword configuration response.
    
    Args:
        command_id: Original command ID
        config_applied: Whether config was applied
        applied_changes: Successfully applied changes
        rejected_changes: Rejected changes with reasons
        requires_restart: Whether restart is required
        
    Returns:
        WakewordConfigResponse: Configuration response
    """
    status = CommandStatus.SUCCESS if config_applied else CommandStatus.ERROR
    
    return WakewordConfigResponse(
        command_id=command_id,
        status=status,
        config_applied=config_applied,
        applied_changes=applied_changes or {},
        rejected_changes=rejected_changes or {},
        requires_restart=requires_restart
    )


# Module exports
__all__ = [
    "WakewordDetectedCommand",
    "WakewordConfigCommand", 
    "WakewordStatusCommand",
    "WakewordDetectedResponse",
    "WakewordConfigResponse",
    "create_wakeword_detection",
    "create_wakeword_config",
    "create_detection_response",
    "create_config_response",
]