Amazon Transcribe Streaming SDK for Python

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())