How to include Pause and Resume Feature with Azure conversation transcriber?

Here you can control the flow of audio data by pausing the input stream (i.e., stop feeding audio to the push stream). This simulates a pause in transcription without completely stopping the transcriber session.

• When we resume, the transcriber can pick up from where it left off, though it’s important to check that the SDK can handle this correctly. You may need to investigate how much state is maintained in the transcription session.

App.py:

import logging
import azure.cognitiveservices.speech as speechsdk
import asyncio
from queue import Queue
import websockets

# Assume CONNECTIONS is a global dict to manage websocket connections.
CONNECTIONS = {}

async def receive_audio(uuid, path):
    audio_queue = Queue(maxsize=0)
    transcriber_state = False  # False means transcriber is paused
    conversation_transcriber = None
    push_stream = None

    try:
        # Get the WebSocket connection and initialize the transcriber
        websocket = CONNECTIONS[uuid]["websocket"]
        connection_details = CONNECTIONS[uuid]
        
        conversation_transcriber, push_stream = create_conversation_transcriber(connection_details)

        # Start continuous recognition
        conversation_transcriber.start_transcribing_async().get()
        transcriber_state = True
        logging.info("Started transcribing...")

        while True:
            # Receive control messages or audio data
            data = await websocket.recv()

            if isinstance(data, str):
                # Handle 'inactive' and 'active' state changes (pause/resume)
                logging.info(f"Received state: {data}")
                if data == "inactive" and transcriber_state:
                    # Pausing: keep the transcriber alive, but stop sending audio
                    logging.info("Pausing the transcriber... (not stopping)")
                    transcriber_state = False

                elif data == "active" and not transcriber_state:
                    # Resuming: continue sending audio to the transcriber
                    logging.info("Resuming the transcriber...")
                    transcriber_state = True
                
                CONNECTIONS[uuid]["state"] = data
            
            # If transcriber is active, continue pushing audio data
            if CONNECTIONS[uuid]["state"] == "active":
                audio_queue.put_nowait(data)
                while not audio_queue.empty():
                    chunk = get_chunk_from_queue(q=audio_queue, chunk_size=4096)
                    CONNECTIONS[uuid]["audio_buffer"] += chunk
                    push_stream.write(chunk)  # Keep writing to the open stream

    except websockets.exceptions.ConnectionClosed as e:
        logging.info("WebSocket connection closed.")
        if conversation_transcriber:
            conversation_transcriber.stop_transcribing_async().get()
        if push_stream:
            push_stream.close()
    except Exception as e:
        logging.error(f"Error in receive_audio: {e}")
    finally:
        await websocket.close(code=1000)
        logging.info("WebSocket closed.")

def create_conversation_transcriber(connection_details):
    """Create a conversation transcriber with Azure speech configuration."""
    speech_config = speechsdk.SpeechConfig(
        subscription=connection_details['subscription_key'],
        region=connection_details['region']
    )
    audio_format = speechsdk.audio.AudioStreamFormat(samples_per_second=16000, bits_per_sample=16, channels=1)
    push_stream = speechsdk.audio.PushAudioInputStream(audio_format)
    audio_config = speechsdk.audio.AudioConfig(stream=push_stream)
    
    transcriber = speechsdk.transcription.ConversationTranscriber(speech_config, audio_config)
    
    return transcriber, push_stream

# Helper function to get chunk from queue
def get_chunk_from_queue(q, chunk_size):
    return q.get_nowait()

async def main():
    # Initialize CONNECTIONS dictionary with a dummy WebSocket and credentials
    CONNECTIONS['dummy_uuid'] = {
        'websocket': await websockets.connect('ws://localhost:8000'),  # Example WebSocket endpoint
        'subscription_key': 'your_azure_subscription_key',
        'region': 'your_azure_region',
        'audio_buffer': bytearray(),
        'state': 'active'  # Initial state
    }

    # Start receiving audio for this connection
    await receive_audio('dummy_uuid', 'path/to/audio')

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    logging.info("Starting the application...")
    
    # Run the asyncio event loop to execute the main function
    asyncio.run(main())

• when you receive a "pause" command, you can buffer the incoming audio data and delay pushing it to the transcriber until a "resume" command is received.

• CONNECTIONS[uuid]["state"] controls the flow of audio to the transcriber. When the state is "inactive," the audio stream is not fed to the transcriber.

Console Log:

. This simulates a pause in transcription without completely stopping the transcriber session.

• When we resume, the transcriber can pick up from where it left off, though it’s important to check that the SDK can handle this correctly. You may need to investigate how much state is maintained in the transcription session.

App.py:

import logging
import azure.cognitiveservices.speech as speechsdk
import asyncio
from queue import Queue
import websockets

# Assume CONNECTIONS is a global dict to manage websocket connections.
CONNECTIONS = {}

async def receive_audio(uuid, path):
    audio_queue = Queue(maxsize=0)
    transcriber_state = False  # False means transcriber is paused
    conversation_transcriber = None
    push_stream = None

    try:
        # Get the WebSocket connection and initialize the transcriber
        websocket = CONNECTIONS[uuid]["websocket"]
        connection_details = CONNECTIONS[uuid]
        
        conversation_transcriber, push_stream = create_conversation_transcriber(connection_details)

        # Start continuous recognition
        conversation_transcriber.start_transcribing_async().get()
        transcriber_state = True
        logging.info("Started transcribing...")

        while True:
            # Receive control messages or audio data
            data = await websocket.recv()

            if isinstance(data, str):
                # Handle 'inactive' and 'active' state changes (pause/resume)
                logging.info(f"Received state: {data}")
                if data == "inactive" and transcriber_state:
                    # Pausing: keep the transcriber alive, but stop sending audio
                    logging.info("Pausing the transcriber... (not stopping)")
                    transcriber_state = False

                elif data == "active" and not transcriber_state:
                    # Resuming: continue sending audio to the transcriber
                    logging.info("Resuming the transcriber...")
                    transcriber_state = True
                
                CONNECTIONS[uuid]["state"] = data
            
            # If transcriber is active, continue pushing audio data
            if CONNECTIONS[uuid]["state"] == "active":
                audio_queue.put_nowait(data)
                while not audio_queue.empty():
                    chunk = get_chunk_from_queue(q=audio_queue, chunk_size=4096)
                    CONNECTIONS[uuid]["audio_buffer"] += chunk
                    push_stream.write(chunk)  # Keep writing to the open stream

    except websockets.exceptions.ConnectionClosed as e:
        logging.info("WebSocket connection closed.")
        if conversation_transcriber:
            conversation_transcriber.stop_transcribing_async().get()
        if push_stream:
            push_stream.close()
    except Exception as e:
        logging.error(f"Error in receive_audio: {e}")
    finally:
        await websocket.close(code=1000)
        logging.info("WebSocket closed.")

def create_conversation_transcriber(connection_details):
    """Create a conversation transcriber with Azure speech configuration."""
    speech_config = speechsdk.SpeechConfig(
        subscription=connection_details['subscription_key'],
        region=connection_details['region']
    )
    audio_format = speechsdk.audio.AudioStreamFormat(samples_per_second=16000, bits_per_sample=16, channels=1)
    push_stream = speechsdk.audio.PushAudioInputStream(audio_format)
    audio_config = speechsdk.audio.AudioConfig(stream=push_stream)
    
    transcriber = speechsdk.transcription.ConversationTranscriber(speech_config, audio_config)
    
    return transcriber, push_stream

# Helper function to get chunk from queue
def get_chunk_from_queue(q, chunk_size):
    return q.get_nowait()

async def main():
    # Initialize CONNECTIONS dictionary with a dummy WebSocket and credentials
    CONNECTIONS['dummy_uuid'] = {
        'websocket': await websockets.connect('ws://localhost:8000'),  # Example WebSocket endpoint
        'subscription_key': 'your_azure_subscription_key',
        'region': 'your_azure_region',
        'audio_buffer': bytearray(),
        'state': 'active'  # Initial state
    }

    # Start receiving audio for this connection
    await receive_audio('dummy_uuid', 'path/to/audio')

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    logging.info("Starting the application...")
    
    # Run the asyncio event loop to execute the main function
    asyncio.run(main())

• when you receive a "pause" command, you can buffer the incoming audio data and delay pushing it to the transcriber until a "resume" command is received.

• CONNECTIONS[uuid]["state"] controls the flow of audio to the transcriber. When the state is "inactive," the audio stream is not fed to the transcriber.

Console Log:

)