amazon-transcribe

import asyncio import os import time # NOTE: aiofile is not a direct dependency but is commonly used # for asynchronous file reads in examples. Install with `pip install aiofile`. # For a minimal example, we'll simulate an audio stream. # import aiofile from amazon_transcribe.client import TranscribeStreamingClient from amazon_transcribe.handlers import TranscriptResultStreamHandler from amazon_transcribe.model import TranscriptEvent # Configure AWS credentials from environment variables for quickstart. # In a real application, consider using AWS CLI config or IAM roles. # os.environ['AWS_ACCESS_KEY_ID'] = os.environ.get('AWS_ACCESS_KEY_ID', '') # os.environ['AWS_SECRET_ACCESS_KEY'] = os.environ.get('AWS_SECRET_ACCESS_KEY', '') # os.environ['AWS_SESSION_TOKEN'] = os.environ.get('AWS_SESSION_TOKEN', '') # Optional class MyEventHandler(TranscriptResultStreamHandler): async def handle_transcript_event(self, transcript_event: TranscriptEvent): results = transcript_event.transcript.results for result in results: if not result.is_partial: for alternative in result.alternatives: print(f"[Transcription]: {alternative.transcript}") async def basic_transcribe_stream(client: TranscribeStreamingClient, region: str): # Simulate a stream of audio bytes (replace with actual audio source) async def get_audio_stream(): # For a real application, read from microphone or file (e.g., using aiofile) # For this example, we'll send a few empty bytes to keep the stream open briefly for _ in range(5): yield b'\x00' * 1024 # Simulate 1KB of silence/empty data await asyncio.sleep(0.1) print("\n--- Audio stream ended ---") stream = await client.start_stream_transcription( language_code="en-US", media_sample_rate_hz=16000, media_encoding="pcm", ) # Instantiate our handler and start processing events handler = MyEventHandler(stream.output_stream) await asyncio.gather(stream.input_stream.send_from_iterable(get_audio_stream()), handler.handle_events()) async def main(): # Ensure AWS_REGION is set, e.g., in your environment variables aws_region = os.environ.get('AWS_REGION', 'us-east-1') print(f"Connecting to Amazon Transcribe in region: {aws_region}") client = TranscribeStreamingClient(region=aws_region) await basic_transcribe_stream(client, aws_region) if __name__ == "__main__": # Set dummy credentials if not set for local testing, replace with actual for production if not os.environ.get('AWS_ACCESS_KEY_ID'): os.environ['AWS_ACCESS_KEY_ID'] = 'AKIAIOSFODNN7EXAMPLE' os.environ['AWS_SECRET_ACCESS_KEY'] = 'wJalrXUtnFEMI/K7MDENG/bPxRfiorexamplekey' os.environ['AWS_REGION'] = 'us-east-1' print("Warning: Using dummy AWS credentials and region. Set environment variables AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and AWS_REGION for actual use.") asyncio.run(main())