nifty-coder/stemsplit-backend

1"""2Example usage of WebSpeechProvider.3 4This example demonstrates how to use the WebSpeechProvider for both5single transcription and streaming transcription scenarios.6"""7 8import asyncio9import logging10import sys11import os12 13# Add the parent directory to the path so we can import voice_control14sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))15 16from voice_control.providers.web_speech_provider import create_web_speech_provider17 18# Configure logging19logging.basicConfig(level=logging.INFO)20logger = logging.getLogger(__name__)21 22 23async def example_single_transcription():24    """Example of single audio transcription."""25    print("\n=== Single Transcription Example ===")26    27    # Create provider with default configuration28    provider = create_web_speech_provider(29        name="example_web_speech",30        priority=131    )32    33    # Simulate audio data (1 second of 16kHz, 16-bit audio)34    sample_audio = b'\x00' * (16000 * 2)35    36    try:37        # Check provider health38        is_healthy = await provider.check_health()39        print(f"Provider health: {'OK' if is_healthy else 'FAILED'}")40        41        if not is_healthy:42            print("Provider is not healthy, skipping transcription")43            return44        45        # Get quota status46        quota_status = await provider.get_quota_status()47        print(f"Daily requests used: {quota_status['requests_per_day'].current_usage}")48        print(f"Daily audio minutes used: {quota_status['audio_minutes_per_day'].current_usage:.2f}")49        50        # Perform transcription51        print("Performing transcription...")52        result = await provider.transcribe_audio(53            audio_data=sample_audio,54            format="webm",55            language="en-US",56            audio_duration=1.057        )58        59        # Display results60        print(f"Transcribed text: '{result.text}'")61        print(f"Confidence: {result.confidence:.2f}")62        print(f"Processing time: {result.processing_time:.3f}s")63        print(f"Audio duration: {result.audio_duration:.2f}s")64        print(f"Provider: {result.provider}")65        print(f"Language: {result.language}")66        67    except Exception as e:68        print(f"Transcription failed: {e}")69 70 71async def example_streaming_transcription():72    """Example of streaming audio transcription."""73    print("\n=== Streaming Transcription Example ===")74    75    # Create provider76    provider = create_web_speech_provider(77        name="streaming_web_speech",78        priority=179    )80    81    async def simulate_audio_stream():82        """Simulate streaming audio chunks."""83        print("Generating audio stream...")84        for i in range(5):85            # Simulate 0.5 seconds of audio per chunk86            chunk = b'\x00' * (16000 * 1)  # 0.5 seconds at 16kHz, 16-bit87            print(f"  Sending chunk {i+1}/5")88            yield chunk89            await asyncio.sleep(0.1)  # Simulate real-time streaming90    91    try:92        print("Starting streaming transcription...")93        94        result_count = 095        async for result in provider.transcribe_streaming(96            audio_stream=simulate_audio_stream(),97            format="webm",98            language="en-US",99            interim_results=True100        ):101            result_count += 1102            print(f"Result {result_count}:")103            print(f"  Text: '{result.text}'")104            print(f"  Confidence: {result.confidence:.2f}")105            print(f"  Is final: {result.is_final}")106            print(f"  Session ID: {result.session_id}")107            print(f"  Processing time: {result.processing_time:.3f}s")108        109        print(f"Streaming completed. Total results: {result_count}")110        111    except Exception as e:112        print(f"Streaming transcription failed: {e}")113 114 115async def example_provider_capabilities():116    """Example of checking provider capabilities."""117    print("\n=== Provider Capabilities Example ===")118    119    provider = create_web_speech_provider()120    121    # Check supported formats122    print("Supported audio formats:")123    test_formats = ["webm", "wav", "mp3", "ogg", "flac", "aac"]124    for fmt in test_formats:125        supported = provider.supports_format(fmt)126        print(f"  {fmt}: {'✓' if supported else '✗'}")127    128    # Check supported languages129    print("\nSupported languages (sample):")130    test_languages = ["en-US", "es-ES", "fr-FR", "de-DE", "ja-JP", "zh-CN", "ar-SA", "unknown-XX"]131    for lang in test_languages:132        supported = provider.supports_language(lang)133        print(f"  {lang}: {'✓' if supported else '✗'}")134    135    # Check cost estimation136    print("\nCost estimation:")137    durations = [60, 300, 3600]  # 1 minute, 5 minutes, 1 hour138    for duration in durations:139        cost = await provider.estimate_cost(duration)140        print(f"  {duration}s ({duration/60:.1f} min): ${cost:.4f}")141 142 143async def example_error_handling():144    """Example of error handling scenarios."""145    print("\n=== Error Handling Example ===")146    147    provider = create_web_speech_provider()148    sample_audio = b'\x00' * 1000149    150    # Test unsupported format151    try:152        await provider.transcribe_audio(sample_audio, "unsupported_format", "en-US")153    except Exception as e:154        print(f"Unsupported format error: {type(e).__name__}: {e}")155    156    # Test unsupported language157    try:158        await provider.transcribe_audio(sample_audio, "webm", "unsupported-LANG")159    except Exception as e:160        print(f"Unsupported language error: {type(e).__name__}: {e}")161    162    # Test quota limits (simulate exceeded quota)163    provider.config.free_tier_limits["max_daily_requests"] = 0164    try:165        await provider.transcribe_audio(sample_audio, "webm", "en-US", audio_duration=1.0)166    except Exception as e:167        print(f"Quota exceeded error: {type(e).__name__}: {e}")168 169 170async def main():171    """Run all examples."""172    print("WebSpeechProvider Examples")173    print("=" * 50)174    175    try:176        await example_provider_capabilities()177        await example_single_transcription()178        await example_streaming_transcription()179        await example_error_handling()180        181        print("\n" + "=" * 50)182        print("All examples completed successfully!")183        184    except Exception as e:185        print(f"Example execution failed: {e}")186        import traceback187        traceback.print_exc()188 189 190if __name__ == "__main__":191    asyncio.run(main())