Amazon Transcribe Streaming SDK for Python

0.6.4 · deprecated · verified Thu Apr 16

The `amazon-transcribe` library is an asynchronous Python SDK designed for direct integration with the Amazon Transcribe Streaming service, enabling real-time conversion of audio into text. As of version 0.6.4, this SDK is considered experimental, is no longer actively developed, and is not officially supported by AWS for new projects. It receives infrequent updates and has a stated lack of commitment for ongoing support.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to establish an asynchronous connection to Amazon Transcribe Streaming, send a simulated audio stream, and process the real-time transcription results. It defines a custom event handler to print the final transcription segments. Remember to set your AWS credentials and region via environment variables or AWS CLI configuration for actual use.

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

view raw JSON →