分贝检测程序，超过75dB马上警报

好恨男的打游戏，像头大叫驴

缘起：室友在夜间同兄弟game，战况激烈时嗷嗷怪叫，我胆子很小，每次惊弓之鸟。随后同舍友打商量，叫声高于80罚款，大家各退一步，你好我也好，舍友勉强答应。然后我开始搜索噪声监测APP，搜索结果不太满意，第一个问题是APP仅能记录一个峰值，如果舍友怪叫两声也只能得到一个最大分贝；第二个是没有录音功能，如果我没紧盯着屏幕，可能就给他漏掉了。

需求：一个具有录音功能的分贝检测程序，每检测到大于80分贝的声音就发出警报并且统计罚款次数；

一、录音功能

搜了一下，比较主流的模块是 pyaudio ，操作简单，官网的引导就可以解决我录音的需求。

比较麻烦的是这个模块不兼容 Python3.7 ，用 pip 下载的话老是报错。说我缺少 Visual Studio 14 C++ Build Tool ，下载安装后还是报这个错。后面去Stack Overflow查了一下解决方案，有两个：

  • a) 去下 https://download.lfd.uci.edu/pythonlibs/t7epjj8p/PyAudio-0.2.11-cp37-cp37m-win_amd64.whl ，然后到下载目录 pip install PyAudio-0.2.11-cp37-cp37m-win_amd64.whl
  • b) pip install pipwin => pipwin install pyaudio
    第一个网址我看的时候404了，我用第二种方式解决的。第二种方式也很悬，我看它是从服务器缓存里下的，要是缓存里没有那就没办法了，如果第一个有资源还是第一个靠谱一些。

接下来使用pyaudio录音:

1
2
import pyaudio
p=pyaudio.PyAudio()
  • 实例化PyAudio类，PyAudio用于连接Python和PortAudio，PortAudio是真正实现音频输入输出的库。
    1
    2
    3
    4
    5
    stream=p.open(format=FORMAT, 
                  channels=CHANNELS, 
                  rate=RATE, 
                  input=True, 
                  frames_per_buffer=CHUNK)
  • 创建输入输出流，format是取样值的量化格式；channels是声道，1为单声道，2为立体声；rate是每秒钟取样频率；input是输入流标志，如果需要输出，将输出流标志output设置为True；frames_per_buffer是缓存块大小
    1
    data=stream.wirte()
  • 记录音频，返回byte类型的记录

二、输出分贝

输出分贝使用pydub库的AudioSegment模块，这个模块专门用于音频处理，可以放大调小、音频切片、连接、新建空音源等等，它本身支持多种格式的音频文件，除了.wav还支持 .mp3 .mp4 .raw .flv等等后缀…

在这个程序中我希望能直接从刚刚 stream.write() 的结果创建一个AudioSegment对象实时输出当前分贝，恰好它支持byte类型的输入

1
2
3
result=AudioSegment(data=data,
                    sample_width=p.get_sample_size(FORMAT), 
                    frame_rate=RATE, channels=CHANNELS)
  • data为刚刚录音的byte类型数据，sample_width、frame_rate、channels是必要的参数，他们需要和data的参数一致。
1
db=(20 * log10(result.rms))
  • 这是一个求分贝值的公式，rms是当前音频的响度，值得一提的是有一个名叫dBFs的属性，看上去它更像是当前音频的分贝，但事实上它是“相对于最大可能响度”的值，基本上都是负数，非常残忍，我也不知道它具体是用来计算什么的，但是它一度让我觉得我全写错了，很挫败。

三、具体代码

整个代码是这样的: 主线程一直监测声音，每0.2毫秒得出一次当前的分贝。如果当前分贝大小大于75就使用子线程调用beep函数发出警报（且罚款)。

但是怪叫一声的时间一定大于0.2毫秒，如果不做限制的话叫一声起码要罚8000次款，为了避免这种冤假错案，通过锁和time.sleep为beep函数设置一个冷却时间。全局设置一个锁，如果当前分贝大于75且锁是空闲的，调用beep函数，罚款一次；如果持续发出一个75分贝以上的叫声，后面的叫声会被忽略，不会触发beep函数，直到冷却时间过去，锁重新空闲，随后的叫声算新的一次叫声。

下面是我的实现代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
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
59
60
61
62
63
64
65
66
67
68
69
70
import pyaudio
import wave
from pydub import AudioSegment
from math import log10
import threading
from time import sleep
#设置音频参数
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 8000
RECORD_SECONDS = 30

lock=threading.Lock()#这个是beep函数的锁
def beep():
    '''
    这个函数播放我事先录好的警报声音。假如当前分贝超过75，一个新的线程将被调用执行该函数
    该函数的冷却时间为 1s ，分贝每 1/8000s 输出一次，但一秒内发出的声音只触发一次本函数
    '''
    wf = wave.open('caution.wav', 'rb')#打开我的录音
    p = pyaudio.PyAudio()
    stream = p.open( format=p.get_format_from_width(wf.getsampwidth()),
                     channels=wf.getnchannels(), 
                     rate=wf.getframerate(), 
                     output=True)
    data = wf.readframes(CHUNK)
    while data:
        stream.write(data)#播放我的录音
        data = wf.readframes(CHUNK)
    stream.stop_stream()
    stream.close()
    p.terminate()
    sleep(1)#函数冷却1秒
    lock.release()#释放锁。

def main():
    count=1
    p = pyaudio.PyAudio()
    stream = p.open(format=FORMAT, 
                    channels=CHANNELS, rate=RATE, input=True,
                    frames_per_buffer=CHUNK)#创建写入流

    print("* recording")

    while True:
        data = stream.read(CHUNK)
        #通过AudioSegment得到响度
        res = AudioSegment(data=data, 
                           sample_width=p.get_sample_size(FORMAT),
                           frame_rate=RATE, channels=CHANNELS).rms
        if res>0:
            #计算dB的公式
            out = 20 * log10(res)
            
            #如果当前音量大于75dB而且函数不在冷却时间，警报并罚款
            if out >= 75:
                if not lock.locked():
                    print(count)#输出当前罚款次数
                    lock.acquire()
                    threading.Thread(target=beep).start()
                    count+=1
                print("%.2f" % out)#输出当前分贝大小
    print(' * done record')

    stream.stop_stream()
    stream.close()
    p.terminate()

if __name__ == "__main__":
    main()