testClient.py

#!/usr/bin/env python3
"""
Test client for MPC Speech-to-Text service
"""

import asyncio
import sys
from pathlib import Path
from mcp_speech_client import transcribe_audio_file, check_service_status

async def main():
    """Main test function"""
    print("🧪 Testing MPC Speech-to-Text Client")
    print("=" * 50)
    
    # First, check if the services are running
    print("1️⃣ Checking service status...")
    status = await check_service_status()
    print(f"   Service status: {status.get('status', 'unknown')}")
    
    if status.get("status") == "error":
        print("\n⚠️  Service appears to be down!")
        print("💡 To start the services, run:")
        print("   python startup.py")
        print("\n   Or start them manually:")
        print("   1. Start Whisper HTTP service: python whisper_http_wrapper.py")
        print("   2. Start MPC service: python mpc_speech_service.py")
        return
    
    print("✅ Services are running!")
    print(f"   Active sessions: {status.get('active_sessions', 0)}")
    
    print("\n2️⃣ Testing audio file transcription...")
    
    # Look for available audio files
    test_files = [
        "audio.wav",
        "speech_test.wav", 
        "test_audio.wav",
        "silent_test.wav"
    ]
    
    audio_file = None
    for file in test_files:
        if Path(file).exists():
            audio_file = file
            break
    
    if not audio_file:
        print("❌ No test audio files found!")
        print("💡 Create test audio files by running:")
        print("   python create_test_audio.py")
        print("\n   Or provide your own audio file and update the test_files list.")
        return
    
    print(f"🎵 Using audio file: {audio_file}")
    
    try:
        # Test transcription
        print("   Sending transcription request...")
        result = await transcribe_audio_file(audio_file)
        
        print(f"\n📋 Transcription result:")
        print(f"   Status: {result.get('status')}")
        print(f"   Session ID: {result.get('session_id')}")
        
        if result.get("status") == "success":
            transcription = result.get("transcription", {})
            
            if isinstance(transcription, dict):
                text = transcription.get("result", {}).get("text", "No text found")
                processing_time = transcription.get("result", {}).get("processing_time", 0)
                model_info = transcription.get("result", {}).get("model_info", {})
                
                print(f"✅ Transcription successful!")
                print(f"   Text: '{text}'")
                print(f"   Processing time: {processing_time:.2f}s")
                print(f"   Model: {model_info.get('model', 'unknown')}")
                print(f"   Device: {model_info.get('device', 'unknown')}")
                
                # If it's synthetic audio, explain the result
                if "test" in audio_file.lower() or "speech" in audio_file.lower():
                    print(f"\n💡 Note: This was synthetic test audio, so the transcription")
                    print(f"   result may not be meaningful. Try with real speech audio")
                    print(f"   for better results.")
            else:
                print(f"   Raw transcription: {transcription}")
                
        else:
            error_msg = result.get('error', 'Unknown error')
            print(f"❌ Transcription failed: {error_msg}")
            
            # Provide specific help for common errors
            if "Cannot connect" in error_msg and "8000" in error_msg:
                print("\n💡 The Whisper HTTP service (port 8000) is not running.")
                print("   Start it with: python whisper_http_wrapper.py")
            elif "Authentication" in error_msg:
                print("\n💡 Authentication issue with the MPC service.")
                print("   Check if the MPC service is running properly.")
            
    except FileNotFoundError:
        print(f"❌ Audio file '{audio_file}' not found.")
    except Exception as e:
        print(f"❌ Unexpected error during transcription: {e}")
    
    print(f"\n🏁 Test completed!")

def check_dependencies():
    """Check if required dependencies are available"""
    missing_deps = []
    
    try:
        import aiohttp
    except ImportError:
        missing_deps.append("aiohttp")
    
    try:
        import soundfile
    except ImportError:
        missing_deps.append("soundfile")
    
    if missing_deps:
        print("❌ Missing dependencies:")
        for dep in missing_deps:
            print(f"   • {dep}")
        print(f"\n💡 Install them with: pip install {' '.join(missing_deps)}")
        return False
    
    return True

if __name__ == "__main__":
    if not check_dependencies():
        sys.exit(1)
    
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n👋 Test interrupted by user")
    except Exception as e:
        print(f"\n❌ Test failed with error: {e}")
        sys.exit(1)