Audio Stream

The Audio Stream pipeline is a threaded pipeline that plays audio segments. This pipeline is designed to run on local machines given that it requires access to write to an output device.

Example

The following shows a simple example using this pipeline.

from txtai.pipeline import AudioStream

# Create and run pipeline
audio = AudioStream()
audio(data)

This pipeline may require additional system dependencies. See this section for more.

See the link below for a more detailed example.

Notebook	Description
Speech to Speech RAG ▶️	Full cycle speech to speech workflow with RAG

Configuration-driven example

Pipelines are run with Python or configuration. Pipelines can be instantiated in configuration using the lower case name of the pipeline. Configuration-driven pipelines are run with workflows or the API.

config.yml

# Create pipeline using lower case class name
audiostream:

# Run pipeline with workflow
workflow:
  audiostream:
    tasks:
      - action: audiostream

Run with Workflows

from txtai import Application

# Create and run pipeline with workflow
app = Application("config.yml")
list(app.workflow("audiostream", [["numpy data", "sample rate"]]))

Run with API

CONFIG=config.yml uvicorn "txtai.api:app" &

curl \
  -X POST "http://localhost:8000/workflow" \
  -H "Content-Type: application/json" \
  -d '{"name":"audiostream", "elements":[["numpy data", "sample rate"]]}'

Methods

Python documentation for the pipeline.

__init__(rate=None)

Creates an AudioStream pipeline.

Parameters:

Name	Type	Description	Default
rate		optional target sample rate, otherwise uses input target rate with each audio segment	None

Source code in txtai/pipeline/audio/audiostream.py

def __init__(self, rate=None):
    """
    Creates an AudioStream pipeline.

    Args:
        rate: optional target sample rate, otherwise uses input target rate with each audio segment
    """

    if not AUDIOSTREAM:
        raise ImportError(
            (
                'AudioStream pipeline is not available - install "pipeline" extra to enable. '
                "Also check that the portaudio system library is available."
            )
        )

    # Target sample rate
    self.rate = rate

    self.queue = Queue()
    self.thread = Thread(target=self.play)
    self.thread.start()

__call__(segment)

Queues audio segments for the audio player.

Parameters:

Name	Type	Description	Default
segment		(audio, sample rate)|list	required

Returns:

Type	Description
	segment

Source code in txtai/pipeline/audio/audiostream.py

def __call__(self, segment):
    """
    Queues audio segments for the audio player.

    Args:
        segment: (audio, sample rate)|list

    Returns:
        segment
    """

    # Convert single element to list
    segments = [segment] if isinstance(segment, tuple) else segment

    for x in segments:
        self.queue.put(x)

    # Return single element if single element passed in
    return segments[0] if isinstance(segment, tuple) else segments