nifty-coder/stemsplit-backend

1"""2Example usage of GoogleSpeechProvider.3 4This script demonstrates how to use the Google Speech-to-Text provider5with proper configuration and error handling.6"""7 8import asyncio9import logging10from pathlib import Path11 12from voice_control.providers.google_speech_provider import create_google_speech_provider13from voice_control.models import ProviderConfig, ProviderType14from voice_control.exceptions import (15    ProviderError,16    QuotaExceededError,17    UnsupportedFormatError18)19 20# Configure logging21logging.basicConfig(level=logging.INFO)22logger = logging.getLogger(__name__)23 24 25async def main():26    """Demonstrate GoogleSpeechProvider usage."""27    28    logger.info("=== Google Speech-to-Text Provider Example ===")29    30    # Create provider with service account credentials31    # Note: In production, use proper service account JSON file32    provider = create_google_speech_provider(33        name="example_google_speech",34        priority=2,35        api_credentials={36            # Option 1: Service account JSON string37            "service_account_key": '{"type": "service_account", "project_id": "your-project"}'38            39            # Option 2: Path to service account file40            # "service_account_key": "/path/to/service-account.json"41            42            # Option 3: API key (less secure, not recommended for production)43            # "api_key": "your-api-key"44        }45    )46    47    logger.info(f"Created provider: {provider.name}")48    logger.info(f"Supported formats: {provider.config.supported_formats}")49    logger.info(f"Supported languages: {len(provider.config.supported_languages)} languages")50    51    # Check provider health52    logger.info("\n=== Health Check ===")53    try:54        health = await provider.check_health()55        logger.info(f"Provider health: {'Healthy' if health else 'Unhealthy'}")56    except Exception as e:57        logger.error(f"Health check failed: {e}")58    59    # Check quota status60    logger.info("\n=== Quota Status ===")61    try:62        quota_status = await provider.get_quota_status()63        64        for quota_type, status in quota_status.items():65            logger.info(f"{quota_type}:")66            logger.info(f"  Current usage: {status.current_usage}")67            logger.info(f"  Limit: {status.limit}")68            logger.info(f"  Remaining: {status.remaining}")69            logger.info(f"  Percentage used: {status.percentage_used:.1%}")70    except Exception as e:71        logger.error(f"Failed to get quota status: {e}")72    73    # Test format and language support74    logger.info("\n=== Format and Language Support ===")75    76    test_formats = ["wav", "mp3", "flac", "aac", "webm"]77    for format in test_formats:78        supported = provider.supports_format(format)79        logger.info(f"Format {format}: {'Supported' if supported else 'Not supported'}")80    81    test_languages = ["en-US", "es-ES", "fr-FR", "de-DE", "ja-JP", "xx-XX"]82    for language in test_languages:83        supported = provider.supports_language(language)84        logger.info(f"Language {language}: {'Supported' if supported else 'Not supported'}")85    86    # Cost estimation87    logger.info("\n=== Cost Estimation ===")88    89    durations = [30, 300, 1800, 3600]  # 30s, 5min, 30min, 1hour90    for duration in durations:91        try:92            cost = await provider.estimate_cost(duration)93            minutes = duration / 6094            logger.info(f"{minutes:.1f} minutes: ${cost:.4f}")95        except Exception as e:96            logger.error(f"Cost estimation failed for {duration}s: {e}")97    98    # Simulate transcription (with mock audio data)99    logger.info("\n=== Transcription Simulation ===")100    101    # Create sample audio data (in production, this would be real audio)102    sample_audio = b'\x00\x01' * 8000  # 16KB of sample data103    104    try:105        logger.info("Starting transcription...")106        107        result = await provider.transcribe_audio(108            audio_data=sample_audio,109            format="wav",110            language="en-US",111            enable_word_time_offsets=True,112            enable_automatic_punctuation=True113        )114        115        logger.info(f"Transcription result:")116        logger.info(f"  Text: '{result.text}'")117        logger.info(f"  Confidence: {result.confidence:.2f}")118        logger.info(f"  Processing time: {result.processing_time:.2f}s")119        logger.info(f"  Audio duration: {result.audio_duration:.2f}s")120        logger.info(f"  Language: {result.language}")121        logger.info(f"  Is final: {result.is_final}")122        123        if result.word_timestamps:124            logger.info(f"  Word timestamps: {len(result.word_timestamps)} words")125            for word in result.word_timestamps[:3]:  # Show first 3 words126                logger.info(f"    '{word.word}': {word.start_time:.2f}s - {word.end_time:.2f}s (conf: {word.confidence:.2f})")127        128    except QuotaExceededError as e:129        logger.error(f"Quota exceeded: {e}")130    except UnsupportedFormatError as e:131        logger.error(f"Unsupported format: {e}")132    except ProviderError as e:133        logger.error(f"Provider error: {e}")134    except Exception as e:135        logger.error(f"Transcription failed: {e}")136    137    # Test streaming transcription138    logger.info("\n=== Streaming Transcription Simulation ===")139    140    async def mock_audio_stream():141        """Generate mock audio chunks."""142        for i in range(3):143            yield b'\x00\x01' * 2000  # 4KB chunks144            await asyncio.sleep(0.1)  # Simulate real-time streaming145    146    try:147        logger.info("Starting streaming transcription...")148        149        chunk_count = 0150        async for result in provider.transcribe_streaming(151            audio_stream=mock_audio_stream(),152            format="wav",153            language="en-US",154            interim_results=True155        ):156            chunk_count += 1157            logger.info(f"Chunk {chunk_count}: '{result.text}' (final: {result.is_final}, conf: {result.confidence:.2f})")158        159        logger.info(f"Streaming completed: {chunk_count} chunks processed")160        161    except Exception as e:162        logger.error(f"Streaming transcription failed: {e}")163    164    # Final quota check165    logger.info("\n=== Final Quota Status ===")166    try:167        quota_status = await provider.get_quota_status()168        169        for quota_type, status in quota_status.items():170            logger.info(f"{quota_type}: {status.current_usage}/{status.limit} ({status.percentage_used:.1%})")171    except Exception as e:172        logger.error(f"Failed to get final quota status: {e}")173    174    logger.info("\n=== Example Complete ===")175 176 177if __name__ == "__main__":178    # Note: This example uses mock data and won't make real API calls179    # To use with real Google Speech API:180    # 1. Set up Google Cloud project181    # 2. Enable Speech-to-Text API182    # 3. Create service account and download JSON key183    # 4. Set GOOGLE_APPLICATION_CREDENTIALS environment variable184    # 5. Install google-cloud-speech: pip install google-cloud-speech185    186    asyncio.run(main())