""" Example usage of GoogleSpeechProvider. This script demonstrates how to use the Google Speech-to-Text provider with proper configuration and error handling. """ import asyncio import logging from pathlib import Path from voice_control.providers.google_speech_provider import create_google_speech_provider from voice_control.models import ProviderConfig, ProviderType from voice_control.exceptions import ( ProviderError, QuotaExceededError, UnsupportedFormatError ) # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) async def main(): """Demonstrate GoogleSpeechProvider usage.""" logger.info("=== Google Speech-to-Text Provider Example ===") # Create provider with service account credentials # Note: In production, use proper service account JSON file provider = create_google_speech_provider( name="example_google_speech", priority=2, api_credentials={ # Option 1: Service account JSON string "service_account_key": '{"type": "service_account", "project_id": "your-project"}' # Option 2: Path to service account file # "service_account_key": "/path/to/service-account.json" # Option 3: API key (less secure, not recommended for production) # "api_key": "your-api-key" } ) logger.info(f"Created provider: {provider.name}") logger.info(f"Supported formats: {provider.config.supported_formats}") logger.info(f"Supported languages: {len(provider.config.supported_languages)} languages") # Check provider health logger.info("\n=== Health Check ===") try: health = await provider.check_health() logger.info(f"Provider health: {'Healthy' if health else 'Unhealthy'}") except Exception as e: logger.error(f"Health check failed: {e}") # Check quota status logger.info("\n=== Quota Status ===") try: quota_status = await provider.get_quota_status() for quota_type, status in quota_status.items(): logger.info(f"{quota_type}:") logger.info(f" Current usage: {status.current_usage}") logger.info(f" Limit: {status.limit}") logger.info(f" Remaining: {status.remaining}") logger.info(f" Percentage used: {status.percentage_used:.1%}") except Exception as e: logger.error(f"Failed to get quota status: {e}") # Test format and language support logger.info("\n=== Format and Language Support ===") test_formats = ["wav", "mp3", "flac", "aac", "webm"] for format in test_formats: supported = provider.supports_format(format) logger.info(f"Format {format}: {'Supported' if supported else 'Not supported'}") test_languages = ["en-US", "es-ES", "fr-FR", "de-DE", "ja-JP", "xx-XX"] for language in test_languages: supported = provider.supports_language(language) logger.info(f"Language {language}: {'Supported' if supported else 'Not supported'}") # Cost estimation logger.info("\n=== Cost Estimation ===") durations = [30, 300, 1800, 3600] # 30s, 5min, 30min, 1hour for duration in durations: try: cost = await provider.estimate_cost(duration) minutes = duration / 60 logger.info(f"{minutes:.1f} minutes: ${cost:.4f}") except Exception as e: logger.error(f"Cost estimation failed for {duration}s: {e}") # Simulate transcription (with mock audio data) logger.info("\n=== Transcription Simulation ===") # Create sample audio data (in production, this would be real audio) sample_audio = b'\x00\x01' * 8000 # 16KB of sample data try: logger.info("Starting transcription...") result = await provider.transcribe_audio( audio_data=sample_audio, format="wav", language="en-US", enable_word_time_offsets=True, enable_automatic_punctuation=True ) logger.info(f"Transcription result:") logger.info(f" Text: '{result.text}'") logger.info(f" Confidence: {result.confidence:.2f}") logger.info(f" Processing time: {result.processing_time:.2f}s") logger.info(f" Audio duration: {result.audio_duration:.2f}s") logger.info(f" Language: {result.language}") logger.info(f" Is final: {result.is_final}") if result.word_timestamps: logger.info(f" Word timestamps: {len(result.word_timestamps)} words") for word in result.word_timestamps[:3]: # Show first 3 words logger.info(f" '{word.word}': {word.start_time:.2f}s - {word.end_time:.2f}s (conf: {word.confidence:.2f})") except QuotaExceededError as e: logger.error(f"Quota exceeded: {e}") except UnsupportedFormatError as e: logger.error(f"Unsupported format: {e}") except ProviderError as e: logger.error(f"Provider error: {e}") except Exception as e: logger.error(f"Transcription failed: {e}") # Test streaming transcription logger.info("\n=== Streaming Transcription Simulation ===") async def mock_audio_stream(): """Generate mock audio chunks.""" for i in range(3): yield b'\x00\x01' * 2000 # 4KB chunks await asyncio.sleep(0.1) # Simulate real-time streaming try: logger.info("Starting streaming transcription...") chunk_count = 0 async for result in provider.transcribe_streaming( audio_stream=mock_audio_stream(), format="wav", language="en-US", interim_results=True ): chunk_count += 1 logger.info(f"Chunk {chunk_count}: '{result.text}' (final: {result.is_final}, conf: {result.confidence:.2f})") logger.info(f"Streaming completed: {chunk_count} chunks processed") except Exception as e: logger.error(f"Streaming transcription failed: {e}") # Final quota check logger.info("\n=== Final Quota Status ===") try: quota_status = await provider.get_quota_status() for quota_type, status in quota_status.items(): logger.info(f"{quota_type}: {status.current_usage}/{status.limit} ({status.percentage_used:.1%})") except Exception as e: logger.error(f"Failed to get final quota status: {e}") logger.info("\n=== Example Complete ===") if __name__ == "__main__": # Note: This example uses mock data and won't make real API calls # To use with real Google Speech API: # 1. Set up Google Cloud project # 2. Enable Speech-to-Text API # 3. Create service account and download JSON key # 4. Set GOOGLE_APPLICATION_CREDENTIALS environment variable # 5. Install google-cloud-speech: pip install google-cloud-speech asyncio.run(main())