example with valid return

[mmerchant]

Can you please provide another clear example of a streaming transcription that has a return. At present, the return of alt.transcript when reaching it's final state (i.e. not is_partial) results a [None, None]. Example code as modified:

class MyEventHandler(TranscriptResultStreamHandler):
    async def handle_transcript_event(self, transcript_event: TranscriptEvent):
        # This handler can be implemented to handle transcriptions as needed.
        # Here's an example to get started.
        results = transcript_event.transcript.results
        for result in results:
            if not result.is_partial:
                for alt in result.alternatives:
                    print(alt.transcript)
                    return alt.transcript

[Xinghui-Wu]

Hi! I have encountered the same problem and I just found a solution.

First, you can introduce a variable field named asr_result to this inherited class by overriding the init function, and update its value in the loop.

class MyEventHandler(TranscriptResultStreamHandler):
    def __init__(self, transcript_result_stream: TranscriptResultStream):
        super().__init__(transcript_result_stream)
        self.asr_result = ""
    
    async def handle_transcript_event(self, transcript_event: TranscriptEvent):
        # This handler can be implemented to handle transcriptions as needed.
        # Here's an example to get started.
        results = transcript_event.transcript.results

        for result in results:
            for alt in result.alternatives:
                self.asr_result = alt.transcript
                print(alt.transcript)

Next, in the end of the basic_transcribe function, return the asr_result value of the handler object.

# Instantiate our handler and start processing events
handler = MyEventHandler(stream.output_stream)
await asyncio.gather(write_chunks(), handler.handle_events())

return handler.asr_result

Finally, when using asyncio, create a task and return the result of the task.

def amazon(audio_path, language, region):
    loop = asyncio.get_event_loop()
    task = loop.create_task(basic_transcribe(audio_path, language, region))
    # loop.run_until_complete(basic_transcribe(audio_path, language, region))
    loop.run_until_complete(task)
    loop.close()

    return task.result()

[wallaceblaia]

Hello. I'm new to Python programming and I'm having a little trouble making the correct adjustments to these examples. In my case I am using the example to capture audio by microphone and I would like to have the return only of the last line of final transcription and not of all the alternatives. I know that the last answer explained how to do it. But it doesn't seem to work with my example file.

import asyncio

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

class MyEventHandler(TranscriptResultStreamHandler): async def handle_transcript_event(self, transcript_event: TranscriptEvent): # This handler can be implemented to handle transcriptions as needed. # Here's an example to get started. results = transcript_event.transcript.results for result in results: for alt in result.alternatives: print(alt.transcript)

async def mic_stream(): # This function wraps the raw input stream from the microphone forwarding # the blocks to an asyncio.Queue. loop = asyncio.get_event_loop() input_queue = asyncio.Queue()

def callback(indata, frame_count, time_info, status):
    loop.call_soon_threadsafe(input_queue.put_nowait, (bytes(indata), status))

# Be sure to use the correct parameters for the audio stream that matches
# the audio formats described for the source language you'll be using:
# https://docs.aws.amazon.com/transcribe/latest/dg/streaming.html
stream = sounddevice.RawInputStream(
    channels=1,
    samplerate=16000,
    callback=callback,
    blocksize=1024 * 2,
    dtype="int16",
)
# Initiate the audio stream and asynchronously yield the audio chunks
# as they become available.
with stream:
    while True:
        indata, status = await input_queue.get()
        yield indata, status

async def write_chunks(stream): # This connects the raw audio chunks generator coming from the microphone # and passes them along to the transcription stream. async for chunk, status in mic_stream(): await stream.input_stream.send_audio_event(audio_chunk=chunk) await stream.input_stream.end_stream()

async def basic_transcribe(): # Setup up our client with our chosen AWS region client = TranscribeStreamingClient(region="us-west-2")

# Start transcription to generate our async stream
stream = await client.start_stream_transcription(
    language_code="pt-BR",
    media_sample_rate_hz=16000,
    media_encoding="pcm",
)

# Instantiate our handler and start processing events
handler = MyEventHandler(stream.output_stream)
await asyncio.gather(write_chunks(stream), handler.handle_events())

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

[hongdangshao]

Hi! I have encountered the same problem and I just found a solution.

First, you can introduce a variable field named asr_result to this inherited class by overriding the init function, and update its value in the loop.

class MyEventHandler(TranscriptResultStreamHandler):
    def __init__(self, transcript_result_stream: TranscriptResultStream):
        super().__init__(transcript_result_stream)
        self.asr_result = ""
    
    async def handle_transcript_event(self, transcript_event: TranscriptEvent):
        # This handler can be implemented to handle transcriptions as needed.
        # Here's an example to get started.
        results = transcript_event.transcript.results

        for result in results:
            for alt in result.alternatives:
                self.asr_result = alt.transcript
                print(alt.transcript)

Next, in the end of the basic_transcribe function, return the asr_result value of the handler object.

# Instantiate our handler and start processing events
handler = MyEventHandler(stream.output_stream)
await asyncio.gather(write_chunks(), handler.handle_events())

return handler.asr_result

Finally, when using asyncio, create a task and return the result of the task.

def amazon(audio_path, language, region):
    loop = asyncio.get_event_loop()
    task = loop.create_task(basic_transcribe(audio_path, language, region))
    # loop.run_until_complete(basic_transcribe(audio_path, language, region))
    loop.run_until_complete(task)
    loop.close()

    return task.result()

i have used your way but i got only the last sentence result. how can i get the intermediate result and final result?