Skip to content

音频混音器

pipeline pipeline

音频混音器管道将多个音频流混合成一个单一的流。

示例

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

from txtai.pipeline import AudioMixer

# 创建并运行管道
mixer = AudioMixer()
mixer(((audio1, rate1), (audio2, rate2)))

配置驱动的示例

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

config.yml

# 使用类名的小写形式创建管道
audiomixer:

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

使用工作流运行

from txtai import Application

# 使用工作流创建并运行管道
app = Application("config.yml")
list(app.workflow("audiomixer", [[[audio1, rate1], [audio2, rate2]]]))

使用API运行

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

curl \
  -X POST "http://localhost:8000/workflow" \
  -H "Content-Type: application/json" \
  -d '{"name":"audiomixer", "elements":[[[audio1, rate1], [audio2, rate2]]]}'

方法

管道的Python文档。

__init__(rate=None)

Creates an AudioMixer 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/audiomixer.py
14
15
16
17
18
19
20
21
22
23
24
25
26
def __init__(self, rate=None):
    """
    Creates an AudioMixer pipeline.

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

    if not SCIPY:
        raise ImportError('AudioMixer pipeline is not available - install "pipeline" extra to enable.')

    # Target sample rate
    self.rate = rate

__call__(segment, scale1=1, scale2=1)

Mixes multiple audio streams into a single stream.

Parameters:

Name Type Description Default
segment

((audio1, sample rate), (audio2, sample rate))|list

required
scale1

optional scaling factor for segment1

1
scale2

optional scaling factor for segment2

1

Returns:

Type Description

list of (audio, sample rate)

Source code in txtai/pipeline/audio/audiomixer.py
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __call__(self, segment, scale1=1, scale2=1):
    """
    Mixes multiple audio streams into a single stream.

    Args:
        segment: ((audio1, sample rate), (audio2, sample rate))|list
        scale1: optional scaling factor for segment1
        scale2: optional scaling factor for segment2

    Returns:
        list of (audio, sample rate)
    """

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

    results = []
    for segment1, segment2 in segments:
        audio1, rate1 = segment1
        audio2, rate2 = segment2

        # Resample audio, as necessary
        target = self.rate if self.rate else rate1
        audio1 = Signal.resample(audio1, rate1, target)
        audio2 = Signal.resample(audio2, rate2, target)

        # Mix audio into single segment
        results.append((Signal.mix(audio1, audio2, scale1, scale2), target))

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