详解python的webrtc库实现语⾳端点检测引⾔语⾳端点检测最早应⽤于电话传输和检测系统当中,⽤于通信信道的时间分配,提⾼传输线路的利⽤效率.端点检测属于语⾳处理系统的前端操作,在语⾳检测领域意义重⼤.但是⽬前的语⾳端点检测,尤其是检测⼈声开始和结束的端点始终是属于技术难点,各家公司始终处于能判断,但是不敢保证判别准确性的阶段.现在基于云端语义库的聊天机器⼈层出不穷,其中最著名的当属amazon的 Alexa/Echo 智能⾳箱.国内如⾬后春笋般出现了各种搭载语⾳聊天的智能⾳箱(如前⼏天在知乎上⼴告的若琪机器⼈)和各类智能机器⼈产品.国内语⾳服务提供商主要⾯对中⽂语⾳服务,由于语⾳不像图像有分辨率等等较为客观的指标,很多时候凭主观判断,所以较难判断各家语⾳识别和合成技术的好坏.但是我个⼈认为,国内的中⽂语⾳服务和国外的英⽂语⾳服务,在某些⽅⾯已经有超越的趋势.通常搭建机器⼈聊天系统主要包括以下三个⽅⾯:1. 语⾳转⽂字(ASR/STT)2. 语义内容(NLU/NLP)3. ⽂字转语⾳(TTS)语⾳转⽂字(ASR/STT)在将语⾳传给云端API之前,是本地前端的语⾳采集,这部分主要包括如下⼏个⽅⾯:1. 麦克风降噪2. 声源定位3. 回声消除4. 唤醒词5. 语⾳端点检测6. ⾳频格式压缩python 端点检测由于实际应⽤中,单纯依靠能量检测特征检测等⽅法很难判断⼈声说话的起始点,所以市⾯上⼤多数的语⾳产品都是使⽤唤醒词判断语⾳起始.另外加上声⾳回路,还可以做语⾳打断.这样的交互⽅式可能有些傻,每次必须喊⼀下唤醒词才能继续聊天.这种⽅式聊多了,个⼈感觉会嘴巴疼:-O .现在github上有snowboy唤醒词的开源库,⼤家可以登录snowboy官⽹训练⾃⼰的唤醒词模型.1. Kitt-AI : Snowboy2. Sensory : Sensory考虑到⽤唤醒词嘴巴会累,所以⼤致调研了⼀下,Python拥有丰富的库,直接import就能⾷⽤.这种⽅式容易受强噪声⼲扰,适合⼀个⼈在家玩玩.1. pyaudio: pip install pyaudio 可以从设备节点读取原始⾳频流数据,⾳频编码是PCM格式;2. webrtcvad: pip install webrtcvad 检测判断⼀组语⾳数据是否为空语⾳;当检测到持续时间长度 T1 vad检测都有语⾳活动,可以判定为语⾳起始;当检测到持续时间长度 T2 vad检测都没有有语⾳活动,可以判定为语⾳结束;完整程序代码可以从我的下载程序很简单,相信看⼀会⼉就明⽩了'''Requirements:+ pyaudio - `pip install pyaudio`+ py-webrtcvad - `pip install webrtcvad`'''import webrtcvadimport collectionsimport sysimport signalimport pyaudiofrom array import arrayfrom struct import packimport waveimport timeFORMAT = pyaudio.paInt16CHANNELS = 1RATE = 16000CHUNK_DURATION_MS = 30 # supports 10, 20 and 30 (ms)PADDING_DURATION_MS = 1500 # 1 sec jugementCHUNK_SIZE = int(RATE CHUNK_DURATION_MS / 1000) # chunk to readCHUNK_BYTES = CHUNK_SIZE 2 # 16bit = 2 bytes, PCMNUM_PADDING_CHUNKS = int(PADDING_DURATION_MS / CHUNK_DURATION_MS)# NUM_WINDOW_CHUNKS = int(240 / CHUNK_DURATION_MS)NUM_WINDOW_CHUNKS = int(400 / CHUNK_DURATION_MS) # 400 ms/ 30ms geNUM_WINDOW_CHUNKS_END = NUM_WINDOW_CHUNKS 2START_OFFSET = int(NUM_WINDOW_CHUNKS CHUNK_DURATION_MS 0.5 RATE)vad = webrtcvad.Vad(1)pa = pyaudio.PyAudio()stream = pa.open(format=FORMAT,channels=CHANNELS,rate=RATE,input=True,start=False,# input_device_index=2,frames_per_buffer=CHUNK_SIZE)got_a_sentence = Falseleave = Falsedef handle_int(sig, chunk):global leave, got_a_sentenceleave = Truegot_a_sentence = Truedef record_to_file(path, data, sample_width):"Records from the microphone and outputs the resulting data to 'path'" # sample_width, data = record()data = pack('<' + ('h' len(data)), data)wf = wave.open(path, 'wb')wf.setnchannels(1)wf.setsampwidth(sample_width)wf.setframerate(RATE)wf.writeframes(data)wf.close()def normalize(snd_data):"Average the volume out"MAXIMUM = 32767 # 16384times = float(MAXIMUM) / max(abs(i) for i in snd_data)r = array('h')for i in snd_data:r.append(int(i times))return rsignal.signal(signal.SIGINT, handle_int)while not leave:ring_buffer = collections.deque(maxlen=NUM_PADDING_CHUNKS) triggered = Falsevoiced_frames = []ring_buffer_flags = [0] NUM_WINDOW_CHUNKSring_buffer_index = 0ring_buffer_flags_end = [0] NUM_WINDOW_CHUNKS_ENDring_buffer_index_end = 0buffer_in = ''# WangSraw_data = array('h')index = 0start_point = 0StartTime = time.time()print(" recording: ")stream.start_stream()while not got_a_sentence and not leave:chunk = stream.read(CHUNK_SIZE)# add WangSraw_data.extend(array('h', chunk))index += CHUNK_SIZETimeUse = time.time() - StartTimeactive = vad.is_speech(chunk, RATE)sys.stdout.write('1' if active else '_')ring_buffer_flags[ring_buffer_index] = 1 if active else 0ring_buffer_index += 1ring_buffer_index %= NUM_WINDOW_CHUNKSring_buffer_flags_end[ring_buffer_index_end] = 1 if active else 0ring_buffer_index_end += 1ring_buffer_index_end %= NUM_WINDOW_CHUNKS_END# start point detectionif not triggered:ring_buffer.append(chunk)num_voiced = sum(ring_buffer_flags)if num_voiced > 0.8 NUM_WINDOW_CHUNKS:sys.stdout.write(' Open ')triggered = Truestart_point = index - CHUNK_SIZE 20 # start point# voiced_frames.extend(ring_buffer)ring_buffer.clear()# end point detectionelse:# voiced_frames.append(chunk)ring_buffer.append(chunk)num_unvoiced = NUM_WINDOW_CHUNKS_END - sum(ring_buffer_flags_end)if num_unvoiced > 0.90 NUM_WINDOW_CHUNKS_END or TimeUse > 10:sys.stdout.write(' Close ')triggered = Falsegot_a_sentence = Truesys.stdout.flush()sys.stdout.write('\n')# data = b''.join(voiced_frames)stream.stop_stream()print(" done recording")got_a_sentence = False# write to fileraw_data.reverse()for index in range(start_point):raw_data.pop()raw_data.reverse()raw_data = normalize(raw_data)record_to_file("recording.wav", raw_data, 2)leave = Truestream.close()程序运⾏⽅式sudo python vad.py以上就是本⽂的全部内容，希望对⼤家的学习有所帮助，也希望⼤家多多⽀持。
