在 Python 中捕获输出音频频谱

Posted 2023-02-25

【中文标题】在 Python 中捕获输出音频频谱

【英文标题】：Capture output audio spectrum in Python

【发布时间】：2015-10-16 14:27:13

【问题描述】：

挑战

捕捉输出音频的频谱。

第一次尝试

import gtk, gst

def playerbin_message(bus, message):
    if message.type == gst.MESSAGE_ELEMENT:
        struct = message.structure
        if struct.get_name() == 'spectrum':
            print struct['magnitude']
pipeline = gst.parse_launch(
    'pulsesrc device="alsa_output.pci-0000_00_1b.0.analog-stereo.monitor" ! spectrum ! fakesink')
bus = pipeline.get_bus()
bus.add_signal_watch()
bus.connect('message', playerbin_message)
pipeline.set_state(gst.STATE_PLAYING)
gtk.main()

哪里失败了？

出于某种原因，脚本仅在某些时候有效。通常它不会打印任何东西。

@otopolsky 在详细模式下记录以下行：

INFO spectrum gstspectrum.c:1051:gst_spectrum_transform_ip:<spectrum0> interval 0:00:00.100000000, fpi 4410, error 0:00:00.000000000

第二次尝试

按照@otopolsky 的建议，我尝试运行：

GST_DEBUG=4,spectrum:7 gst-launch-0.10 pulsesrc device="alsa_output.pci-0000_00_1b.0.analog-stereo.monitor" ! spectrum ! fakesink > out.log

得到this output。

这会卡住：

gst-launch-0.10 -v -m pulsesrc device="alsa_output.pci-0000_00_1b.0.analog-stereo.monitor" ! spectrum ! fakesink

虽然这有效：

gst-launch-1.0 -v -m pulsesrc device="alsa_output.pci-0000_00_1b.0.analog-stereo.monitor" ! spectrum ! fakesink

（仅更改了版本）。

第三次尝试

移至python gi：

import gi
gi.require_version('Gst', '1.0')
from gi.repository import GObject, Gst

GObject.threads_init()
Gst.init(None)

def handler(bus, msg):
    if msg.type == Gst.MessageType.ELEMENT:
        struct = msg.get_structure()
        print struct.get_value('magnitude')

p = Gst.parse_launch(
    'pulsesrc device="alsa_output.pci-0000_00_1b.0.analog-stereo.monitor" ! spectrum ! fakesink')

bus = p.get_bus()
bus.add_signal_watch()
bus.connect('message', handler)
p.set_state(Gst.State.PLAYING)

ctx = GObject.main_context_default()
while ctx:
    ctx.iteration()

哪里失败了？

TypeError: unknown type GstValueList

问题

如何修复脚本以在每次执行时都能正常工作？ 还有其他方法可以使用 Python 捕获输出频谱吗？

【问题讨论】：

在执行python脚本之前添加export GST_DEBUG=3（或4个）怎么样？

成功日志：pastebin.com/5nBypmtv，不成功日志：pastebin.com/pCfzAfCe

好的，试试 gstreamer 1.2 怎么样？场景2 - 它是否有效或一段时间后它一直都在完成？使用alsasrc 怎么样（我不确定在 gst 0.10.x 中是否可以访问）？

@otopolsky 我已经添加了方案 2 的结果。alsasrc 在 gst 0.10 中可以访问，你知道我如何使用它来捕获内部声音而不是麦克风吗？

添加了 alsasrc 示例来回答

【参考方案1】：

您使用的是 0.10.36 .. 太旧了.. 您不能使用较新的 gstreamer，比如 1.2 吗？

您正在运行 Ubuntu，所以这应该不是问题.. 0.10.x 和 1.2 可以轻松共存.. 我不知道您的 python 绑定是否可以..

这可能是有问题的行（来自您的 pastebin，请将该部分添加到您的问题中......）

INFO spectrum gstspectrum.c:1051:gst_spectrum_transform_ip:<spectrum0> interval 0:00:00.100000000, fpi 4410, error 0:00:00.000000000

这是来源code.. 它说明了舍入误差和未初始化的 FFT：

/* 如果我们还没有 FFT 上下文（或者由于 参数 * 更改）获取一个并为所有内容分配内存 */

/* 我们在发布消息之前处理的样本帧数 * 间隔为 ns */

/*以ns为单位的frames_per_interval的舍入误差， * 将其聚合到accumulated_error */

您可以再次运行以获取更多详细信息：

export GST_DEBUG=4,spectrum:7

你能在 gst-launch 中运行管道吗？

GST_DEBUG=4,spectrum:7 gst-launch-0.10 pulsesrc device="alsa_output.pci-0000_00_1b.0.analog-stereo.monitor" ! spectrum ! fakesink

更新：

如果您听到，请尝试使用 alsasrc..（autoaudiosink 或 alsasink）：

gst-launch-1.0 alsasrc device=hw:0,2 ! spectrum ! autoaudiosink

hw:X,Y格式说明：X = card number, Y = device name

从arecord -l 获取麦克风等输入设备。

和aplay -l 用于输出设备..

> arecord -l | grep card
card 0: PCH [HDA Intel PCH], device 0: ALC887-VD Analog [ALC887-VD Analog]
card 0: PCH [HDA Intel PCH], device 2: ALC887-VD Alt Analog [ALC887-VD Alt Analog]

【讨论】：

执行$ arecord -l | grep card 会得到card 0: PCH [HDA Intel PCH], device 0: 92HD87B1/3 Analog [92HD87B1/3 Analog] 并且只有hw:0,0 有效。但它只提供麦克风。

好的@Michael，当您在代码中使用此 alsasrc 管道时，麦克风是否与频谱一起使用？

是的，在 python 中使用 gst-0.10 的 alsasrc 管道确实会产生频谱 gst.parse_launch('alsasrc device=hw:0,0 ! spectrum ! fakesink')

【参考方案2】：

gst-python 似乎不支持GstValueList。

从GstValueList 获取数据的方式并不优雅。我们可以将消息结构体转换成字符串，然后将字符串转换成浮点数组。

import gi
gi.require_version('Gst', '1.0')
from gi.repository import GObject, Gst
import re

GObject.threads_init()
Gst.init(None)


def on_message(bus, msg):
    struct = msg.get_structure()
    if struct.get_name() == 'spectrum':
        struct_str = struct.to_string()
        magnitude_str = re.match(r'.*magnitude=\(float\)(.*).*', struct_str)
        if magnitude_str:
            magnitude = map(float, magnitude_str.group(1).split(','))
            print magnitude

pipeline = Gst.parse_launch(
    'audiotestsrc ! spectrum interval=1000000000 bands=16 post-messages=true message-magnitude=1 ! directsoundsink'
)

bus = pipeline.get_bus()
bus.add_signal_watch()
bus.connect('message::element', on_message)

pipeline.set_state(Gst.State.PLAYING)

loop = GObject.MainLoop()
loop.run()

【参考方案3】：

意识到这个问题已经很老了（但很有帮助！），但是...

我遇到了同样的问题，发现这只是使用get_list() 绑定而不是get_value() 的问题。这为您提供了一个布尔值和一个GObject.ValueArray，这似乎可用：

struct = msg.get_structure()
success, values = struct.get_list("magnitude")
if struct.get_name() == "spectrum":
    mags = [values.get_nth(i) for i in range(values.n_values)]