将树莓派用作声源

前一阵子跟刘老师聊天发现，不能总是灌10分低端水了，得进军20分中端水……就着树莓派上的麦阵列作为传感器信号源，送到mbp上做分析试试水先。

1 硬件现状

1.1 源端

硬件：树莓派3B+一枚，还是老文章里麦阵列用的那枚；麦阵列还是ReSpeaker-6mic阵列。

系统：用buster-2020-05的版本，装麦阵列驱动有问题，报错：

E: Unable to locate package dkms
E: Package 'libasound2-plugins' has no installation candidate

先换成旧的stretch……

刷了stretch，更新软件包update & upgrade后貌似也会出问题……干脆：

• 直接用旧版image(2019-04-09)不升级；
• 手动安装旧版kernel-headers(2019-04-01)：dpkg -i raspberrypi-kernel-headers_1.20190401-1_armhf.deb
• 将麦阵列驱动git reset --hard回溯到2019年四月份的版本，然后把install.sh里76-77行的升级和内核安装注释掉，再安装就好了。

毕竟我记得去年四月的时候还能用……估计是新系统和驱动不匹配的锅……其中有用的操作：

• 升级单个软件包sudo apt-get --only-upgrade install apt，全升级的话kernel和kernel-headers也会跟着升级；
• 用pyenv安装和管理Python版本：Install Python 3.8.1 on Raspberry Pi (Raspbian)
• 树莓派时间同步：How to sync time with a server on Raspberry Pi?（后来发现是mbp的时间总不同步：sudo sntp -sS time.apple.com）
• 重启shell的命令是exec "$SHELL"，装完zsh等重启shell会方便很多；

1.2 接收端

我那尚能饭否的mbp……

2 使用PulseAudio进行流媒体传输

2.1 TCP（是不对的）

按照ReSpeaker 6-Mic Circular Array kit for Raspberry Pi的介绍，打算安装最常见的PulseAudio Server，据说可以方便的做流媒体数据源。

• 修改PulseAudio远程配置，匿名网络访问：

# /etc/pulse/default.pa
load-module module-native-protocol-tcp auth-ip-acl=127.0.0.1;10.0.6.0/24 auth-anonymous=1

• 开防火墙端口：sudo ufw allow proto tcp to 0.0.0.0/0 port 4713 comment "pulseaudio tcp port"

后来发现tcp访问pulseaudio的目的貌似是控制而不是传输流媒体数据……进一步才发现原来rtp是传输流媒体数据用的……

2.2 RTP（也没搞定）

一开始想得简单，以为直接默认配置就能搞定：

load-module module-null-sink sink_name=rtp
load-module module-rtp-send source=rtp.monitor
set-default-sink rtp

此时并没有指定广播目的IP和端口，pulseaudio就默认用了224.0.0.56和一个随机五位端口号。出现的现象是：

• 在树莓派上可以通过tcpdump -n net 224.0.0.0/8 -c10看到发包，类似：

  IP 192.168.1.101.44556 > 224.0.0.56.44668: UDP, length 1292

• 我的连在同一个局域网中的mbp里tcpdump一直是空的。
• 不论rpi还是mbp的ping到224.0.0.56都是不通的。

看了Multicast UDP not working后使用netstat -gn发现rpi和mbp的组播地址里都没有224.0.0.56，猜测可能是内核把包扔掉了，按照该文章的方法处理：

# macbookpro
netstat -gn
IPv4 Multicast Group Memberships
Group                   Link-layer Address      Netif
224.0.0.1               1:0:5e:0:0:1            en0
224.0.0.251             1:0:5e:0:0:fb           en0

# raspberrypi
 netstat -gn
IPv6/IPv4 Group Memberships
Interface       RefCnt Group
--------------- ------ ---------------------
wlan0           1      224.0.0.251
wlan0           1      224.0.0.1

ping了一下组播组里出现的224.0.0.1和224.0.0.251发现是通的。本来想按照Linux built-in or open source program to join multicast group?写的把224.0.0.56加入两个机器的组播组里，结果rpi试了不行（命令执行成功，但是用netstat或ip maddr show检查都没法写新加的组播地址），而mbp根本没有ip这个命令，我又不太懂iptable这种高端货，天色已晚我也懒得查解决方案了……干脆就着现有组播组里的224.0.0.251用吧：

load-module module-null-sink sink_name=rtp
load-module module-rtp-send source=rtp.monitor destination_ip="224.0.0.251" port=45678
set-default-sink rtp

此时出现新问题，在mbp上tcpdump已经能看到rpi不停的发包，但是用tcpdump -n net 224.0.0.0/8 -c1 -X查看包内容时，发现内容全零，pulseaudio的配置文件怎么改都不行（就rtp相关的几个模块配置参数怎么调都没用），用pulseaudio -v调试模式启动，也看不太懂，遂早早放弃有（后）空（会）在（无）搞（期）……

2.3 小结：战略失败

当需求很简单时，用别人的软件，有学习如何配置的功夫，还不如自己写个小程序实现功能呢……

3 使用python-sounddevice采集/播放音频流

整理一下思路，目的其实很明确，就是将麦阵列采集的数据通过网络传出去。搜了下，用python的sounddevice（以下简称sd）貌似就可以采集/写入音频流，至于如何发送，想了想就用以前用过的pyzmq吧（高速低延时高可拓展）。

sd的流有两种封装，Stream需要numpy，而RawStream更底层，使用的是buffer。虽然看源码Stream版本是在Raw版本之上做的封装，但是并没感觉慢多少，试用zmq发包的时候，rpi的那颗弱鸡CPU一直在7-9%浮动，完全hold住，那就用numpy输出吧，操作起来会方便很多。

（这次树莓派上的python版本控制选用了pyenv，装的python3.8.5）

3.1 采集音频数据

使用sd.query_devices()在rpi上查询音频设备：

  0 bcm2835 ALSA: - (hw:0,0), ALSA (0 in, 2 out)
  1 bcm2835 ALSA: IEC958/HDMI (hw:0,1), ALSA (0 in, 2 out)
  2 seeed-8mic-voicecard: - (hw:1,0), ALSA (8 in, 8 out)
  3 sysdefault, ALSA (0 in, 128 out)
  4 ac108, ALSA (128 in, 128 out)
  5 dmixer, ALSA (0 in, 128 out)
  6 ac101, ALSA (128 in, 128 out)
  7 dmix, ALSA (0 in, 2 out)
* 8 default, ALSA (32 in, 32 out)

0 {'name': 'bcm2835 ALSA: - (hw:0,0)', 'hostapi': 0, 'max_input_channels': 0, 'max_output_channels': 2, 'default_low_input_latency': -1.0, 'default_low_output_latency': 0.005804988662131519, 'default_high_input_latency': -1.0, 'default_high_output_latency': 0.034829931972789115, 'default_samplerate': 44100.0}
1 {'name': 'bcm2835 ALSA: IEC958/HDMI (hw:0,1)', 'hostapi': 0, 'max_input_channels': 0, 'max_output_channels': 2, 'default_low_input_latency': -1.0, 'default_low_output_latency': 0.005804988662131519, 'default_high_input_latency': -1.0, 'default_high_output_latency': 0.034829931972789115, 'default_samplerate': 44100.0}
2 {'name': 'seeed-8mic-voicecard: - (hw:1,0)', 'hostapi': 0, 'max_input_channels': 8, 'max_output_channels': 8, 'default_low_input_latency': 0.005804988662131519, 'default_low_output_latency': 0.008707482993197279, 'default_high_input_latency': 0.034829931972789115, 'default_high_output_latency': 0.034829931972789115, 'default_samplerate': 44100.0}
3 {'name': 'sysdefault', 'hostapi': 0, 'max_input_channels': 0, 'max_output_channels': 128, 'default_low_input_latency': -1.0, 'default_low_output_latency': 0.005804988662131519, 'default_high_input_latency': -1.0, 'default_high_output_latency': 0.034829931972789115, 'default_samplerate': 44100.0}
4 {'name': 'ac108', 'hostapi': 0, 'max_input_channels': 128, 'max_output_channels': 128, 'default_low_input_latency': 0.005333333333333333, 'default_low_output_latency': 0.008, 'default_high_input_latency': 0.032, 'default_high_output_latency': 0.032, 'default_samplerate': 48000.0}
5 {'name': 'dmixer', 'hostapi': 0, 'max_input_channels': 0, 'max_output_channels': 128, 'default_low_input_latency': -1.0, 'default_low_output_latency': 0.125, 'default_high_input_latency': -1.0, 'default_high_output_latency': 0.125, 'default_samplerate': 48000.0}
6 {'name': 'ac101', 'hostapi': 0, 'max_input_channels': 128, 'max_output_channels': 128, 'default_low_input_latency': 0.005333333333333333, 'default_low_output_latency': 0.008, 'default_high_input_latency': 0.032, 'default_high_output_latency': 0.032, 'default_samplerate': 48000.0}
7 {'name': 'dmix', 'hostapi': 0, 'max_input_channels': 0, 'max_output_channels': 2, 'default_low_input_latency': -1.0, 'default_low_output_latency': 0.021333333333333333, 'default_high_input_latency': -1.0, 'default_high_output_latency': 0.021333333333333333, 'default_samplerate': 48000.0}
8 {'name': 'default', 'hostapi': 0, 'max_input_channels': 32, 'max_output_channels': 32, 'default_low_input_latency': 0.008707482993197279, 'default_low_output_latency': 0.008707482993197279, 'default_high_input_latency': 0.034829931972789115, 'default_high_output_latency': 0.034829931972789115, 'default_samplerate': 44100.0}

其中的seeed-8mic-voicecard是我们需要的输入设备，8路输入（2×AC108 ADC，每芯片4路输出，而每芯片都有一路是playback，因此实际是2*3=6mic信号）。为了方便rpi本地测试，AC101（1×DAC，2输入2输出，我觉得就是左右声道？）要用作输出设备。sd的流要用id指定输入输出设备（也就是上面的[2, 6]）：

iodevs = [0, 0]
for idx, d in enumerate(sd.query_devices()):
    if 'seeed-8mic-voicecard' in d['name']:
        iodevs[0] = idx
    elif 'ac101' in d['name']:
        iodevs[1] = idx
rs = sd.Stream(samplerate=48000, device=iodevs, channels=[8, 2], callback=cb, finished_callback=fcb)

流有两种发送方式：阻塞 和 非阻塞回调，为了方便调试，我用了非阻塞回调的方式（即Stream.start()后立即返回不耽误监控或执行其他命令）。

sd.Stream一旦指定了callback参数就会使用非阻塞模式运行，其函数签名：

callback(indata: ndarray, outdata: ndarray, frames: int, time: CData, status: CallbackFlags) -> None

• indata是输入设备传来的数据，我用的48kHz采样率，每次回调传来的都是(512, 8)的numpy数组，基本上10毫秒一组数据；
• outdata是传给输出设备的数据，我的输入输出配置是[8, 2]，因此不能直接把indata复制给输出，图省事我就前后四个数分别求mean，把8个数强行压成两个数；
• frames是帧数，我这里每次都是512；
• time是一个CFFI的C结构体，能用的属性有time.inputBufferAdcTime（输入开始时间）、time.outputBufferDacTime（输出开始的时间）、time.currentTime（本次callback被调用的时间）。
• status没用过不知道，看起来可以用来在回调里发送指令终止回调或终止流。

一旦完成输入数据到输出数据的复制，插在AC101上的音响就有声音了，能感觉到很微小的延迟，完全够用了。

最后就剩在回调里写上pyzmq的发送，如此发送端就完成了。这里有个小问题，就是pyzmq直接发送numpy是不行滴，直接发送的话代码虽然可以运行，但zmq会使用Python的memoryview直接将numpy数组转换为字节发出去。如此一来，在接收端是无法重建数组的，因为丢失了shape和dtype等元数据。按照官方解决方案，使用多段发送标识SNDMORE先发送数组属性，在发送数组内容即可在接收端重建数组了：

def send_array(socket, A, flags=0, copy=True, track=False):
    """send a numpy array with metadata"""
    md = dict(
        dtype = str(A.dtype),
        shape = A.shape,
    )
    socket.send_json(md, flags|zmq.SNDMORE)
    return socket.send(A, flags, copy=copy, track=track)

def recv_array(socket, flags=0, copy=True, track=False):
    """recv a numpy array"""
    md = socket.recv_json(flags=flags)
    msg = socket.recv(flags=flags, copy=copy, track=track)
    buf = memoryview(msg)
    A = numpy.frombuffer(buf, dtype=md['dtype'])
    return A.reshape(md['shape'])

发送端代码如下：

import sounddevice
import numpy as np
import zmq
import sys


class SdSender(object):
    def __init__(self, host='*', port=9000):
        self.__frame_idx = 0
        self.__frame_time = 0
        self.__device = self.__init_device()

        context = zmq.Context()
        self.__zmq_socket = context.socket(zmq.PUB)
        self.__zmq_socket.bind(f"tcp://{host}:{port}")

        self.__verbose = False

    def __init_device(self):
        iodevs = [0, 0]
        for idx, d in enumerate(sounddevice.query_devices()):
            if 'seeed-8mic-voicecard' in d['name']:
                iodevs[0] = idx
            elif 'ac101' in d['name']:
                iodevs[1] = idx
        return iodevs

    def send_array(self, A, ct, flags=0, copy=True, track=False):
        """send a numpy array with metadata"""
        self.__frame_idx += 1
        md = dict(
            frame_idx = self.__frame_idx,
            tf = ct,
            dtype = str(A.dtype),
            shape = A.shape,
        )
        if self.__verbose:
            print(f'\r{md["self.__frame_idx"]}', end="", flush=True)
        self.__zmq_socket.send_json(md, flags|zmq.SNDMORE)
        return self.__zmq_socket.send(A, flags, copy=copy, track=track)

    def cb(self, indata, outdata, frames, time, status):
        # 六声道强行用平均数合成双声道(512, 8) -> (512, 2)，让rpi本机也输出声音方便调试
        outdata[:] = np.hstack( (np.mean(indata[:, 0:4], axis=1, keepdims=True),
                                np.mean(indata[:, 4:8], axis=1, keepdims=True)) )
        # 发送音频原始8路数据
        self.send_array(indata, time.currentTime)

    def fcb(self):
        print('finished!')

    def run(self, verbose=False):
        self.__verbose = verbose
        try:
            with sounddevice.Stream(samplerate=48000, device=self.__device, channels=[8, 2],
                                    callback=self.cb, finished_callback=self.fcb) as rs:
                input()
        except KeyboardInterrupt as ki:
            self.__zmq_socket.close()
            print('exit!')
            sys.exit(0)
        except Exception as e:
            raise Exception

if __name__ == "__main__":
    sd = SdSender()
    sd.run(verbose=True)

3.2 接收音频数据

使用sd.query_devices()在mbp上查询音频设备，可以看到mbp就1-in-1-out，都是双通道，很朴实（我修改了mbp的midi音频设置，将采样率从默认的44100该为48000，输出44100的话，播放48000的数据明显会…慢…）：

> 0 Built-in Microphone, Core Audio (2 in, 0 out)
< 1 Built-in Output, Core Audio (0 in, 2 out)

0 {'name': 'Built-in Microphone', 'hostapi': 0, 'max_input_channels': 2, 'max_output_channels': 0, 'default_low_input_latency': 0.0027708333333333335, 'default_low_output_latency': 0.01, 'default_high_input_latency': 0.012104166666666666, 'default_high_output_latency': 0.1, 'default_samplerate': 48000.0}
1 {'name': 'Built-in Output', 'hostapi': 0, 'max_input_channels': 0, 'max_output_channels': 2, 'default_low_input_latency': 0.01, 'default_low_output_latency': 0.012291666666666666, 'default_high_input_latency': 0.1, 'default_high_output_latency': 0.021625, 'default_samplerate': 48000.0}

接收端代码如下：

import sounddevice
import numpy as np
import zmq
import sys


class SdReceiver(object):
    def __init__(self, host='10.0.6.223', port=9000):
        self.__device = self.__init_device()

        context = zmq.Context()
        self.__zmq_socket = context.socket(zmq.SUB)
        self.__zmq_socket.connect(f"tcp://{host}:{port}")
        self.__zmq_socket.setsockopt_string(zmq.SUBSCRIBE, '')

        self.__verbose = False

    def __init_device(self):
        iodevs = [0, 0]
        for idx, d in enumerate(sounddevice.query_devices()):
            if 'Built-in Microphone' in d['name']:
                iodevs[0] = idx
            elif 'Built-in Output' in d['name']:
                iodevs[1] = idx
        return iodevs

    def recv_array(self, flags=0, copy=True, track=False):
        """recv a numpy array"""
        md = self.__zmq_socket.recv_json(flags=flags)
        msg = self.__zmq_socket.recv(flags=flags, copy=copy, track=track)
        if self.__verbose:
            print(f'\r{md["frame_idx"]}: {md["tf"]}', end="", flush=True)
        buf = memoryview(msg)
        A = np.frombuffer(buf, dtype=md['dtype'])
        return A.reshape(md['shape'])

    def cb(self, indata, outdata, frames, time, status):
        rpi_8mic_data = self.recv_array()
        outdata[:] = np.hstack( (np.mean(rpi_8mic_data[:, 0:4], axis=1, keepdims=True),
                                np.mean(rpi_8mic_data[:, 4:8], axis=1, keepdims=True)) )

    def fcb(self):
        print('finished!')

    def run(self, verbose=False):
        self.__verbose = verbose
        try:
            with sounddevice.Stream(samplerate=48000, device=self.__device, channels=[2, 2],
                                    callback=self.cb, finished_callback=self.fcb) as ss:
                input()
        except KeyboardInterrupt as ki:
            self.__zmq_socket.close()
            print('exit!')
            sys.exit(0)
        except Exception as e:
            raise Exception

if __name__ == "__main__":
    sd = SdReceiver()
    sd.run(verbose=True)

3.3 小结

吐槽：明确目标直接写代码，比无头苍蝇似的瞎配PulseAudio简单太多了。

4 总结

python-sounddevice每次调用callback传出来的是一个512帧的数组，即采样率48000Hz时约0.01067秒的数据，数据类型为float32，每个数字4字节，一个回调输出512*8*4=16384字节，如果按秒算的话就是48000*8*4=1,536,000字节，大概1.465MB/s。以前在单位都是千万十万的交换机，没仔细抠门过带宽问题，在上海基地用的这个民用wifi很明显受不了这种流量……延迟极大……（这时候想想人家mp3，192kHz几分钟的歌才4/5M，真NB）

这种极端环境下，ØMQ这种“假消息队列”的劣势就显现出来了，作为仅实现了“消息”和“队列”功能的超轻量级库，zmq没有持久化，默认缓存就几十兆，我试了下大概能存不到半分钟数据……订阅者消化能力太差就会导致缓存不够旧数据丢失，然而我就是喜欢zmq的这种朴素感🤦‍。反正树莓派上sd卡也不敢做缓存，搞不好写一写就坏了……后面在研究研究zmq的其他连接模式，毕竟宝藏库。

接下来就是利用音频数据进行分析了，明天试试把mfcc调个参魔改一下，看能不能水一篇20分的中端……