AWSのAmazon Transcribeを動かしてみる

#AWS
!pip install amazon-transcribe
!python -m pip install amazon-transcribe
!git clone https://github.com/awslabs/amazon-transcribe-streaming-sdk.git
%cd "amazon-transcribe-streaming-sdk"
!python -m pip install .
!pip install aiofile

import os
import asyncio
import aiofile
import nest_asyncio

from amazon_transcribe.client import TranscribeStreamingClient
from amazon_transcribe.handlers import TranscriptResultStreamHandler
from amazon_transcribe.model import TranscriptEvent
from amazon_transcribe.utils import apply_realtime_delay

os.environ['AWS_ACCESS_KEY_ID'] = ''
os.environ['AWS_SECRET_ACCESS_KEY'] = ''

SAMPLE_RATE = 16000
BYTES_PER_SAMPLE = 2
CHANNEL_NUMS = 1

AUDIO_PATH = "アップした音声ファイルのパス"
CHUNK_SIZE = 1024 * 8
REGION = "ap-northeast-1" #音声認識するリージョンの指定(今回はアジアパシフィック(東京))

class MyEventHandler(TranscriptResultStreamHandler):
    async def handle_transcript_event(self, transcript_event: TranscriptEvent):
        results = transcript_event.transcript.results
        for result in results:
          if result.is_partial is False:
            for alt in result.alternatives:
                print(alt.transcript)

async def basic_transcribe():
    # Setup up our client with our chosen AWS region
    client = TranscribeStreamingClient(region=REGION)

    # Start transcription to generate our async stream
    stream = await client.start_stream_transcription(
        language_code="ja-JP", #言語設定
        media_sample_rate_hz=SAMPLE_RATE,
        media_encoding="pcm",
        # vocabulary_name = "my-first-vocabulary"
    )

    async def write_chunks():
        async with aiofile.AIOFile(AUDIO_PATH, "rb") as afp:
            reader = aiofile.Reader(afp, chunk_size=CHUNK_SIZE)
            await apply_realtime_delay(
                stream, reader, BYTES_PER_SAMPLE, SAMPLE_RATE, CHANNEL_NUMS
            )
        await stream.input_stream.end_stream()

    handler = MyEventHandler(stream.output_stream)
    await asyncio.gather(write_chunks(), handler.handle_events())

loop = asyncio.get_event_loop()
loop.run_until_complete(basic_transcribe())

async def basic_transcribe():
    # Setup up our client with our chosen AWS region
    client = TranscribeStreamingClient(region=REGION)

    # Start transcription to generate our async stream
    stream = await client.start_stream_transcription(
        language_code="ja-JP", #言語設定
        media_sample_rate_hz=SAMPLE_RATE,
        media_encoding="pcm",
        vocabulary_name = "作成したカスタムボキャブラリー名"
    )