Skip to content

音频流

pipeline pipeline

音频流管道是一个线程化的管道，用于播放音频片段。该管道设计为在本地机器上运行，因为它需要访问输出设备进行写操作。

示例

以下展示了一个使用此管道的简单示例。

from txtai.pipeline import AudioStream

# 创建并运行管道
audio = AudioStream()
audio(data)

此管道可能需要额外的系统依赖项。更多信息请参见此部分。

配置驱动的示例

管道可以通过Python或配置来运行。管道可以通过配置中的小写类名实例化。配置驱动的管道可以通过工作流或API运行。

config.yml

# 使用小写类名创建管道
audiostream:

# 使用工作流运行管道
workflow:
  audiostream:
    tasks:
      - action: audiostream

使用工作流运行

from txtai import Application

# 使用工作流创建并运行管道
app = Application("config.yml")
list(app.workflow("audiostream", [["numpy数据", "采样率"]]))

使用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数据", "采样率"]]}'

方法

管道的Python文档。

__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 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
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 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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