正文  电脑教程 > 服务器教程 >

python的webrtc库如何实现语音端点检测

  语音端点检测最早应用于电话传输和检测系统当中,用于通信信道的时间分配,提高传输线路的利用效率.端点检测属于语音处理系统的前端操作,在语音检测领域意...

  语音端点检测最早应用于电话传输和检测系统当中,用于通信信道的时间分配,提高传输线路的利用效率.端点检测属于语音处理系统的前端操作,在语音检测领域意义重大.

  但是目前的语音端点检测,尤其是检测 人声 开始和结束的端点始终是属于技术难点,各家公司始终处于 能判断,但是不敢保证 判别准确性 的阶段.

python的webrtc库如何实现语音端点检测

  现在基于云端语义库的聊天机器人层出不穷,其中最著名的当属amazon的 Alexa/Echo 智能音箱.

timg.jpg

  国内如雨后春笋般出现了各种搭载语音聊天的智能音箱(如前几天在知乎上广告的若琪机器人)和各类智能机器人产品.国内语音服务提供商主要面对中文语音服务,由于语音不像图像有分辨率等等较为客观的指标,很多时候凭主观判断,所以较难判断各家语音识别和合成技术的好坏.但是我个人认为,国内的中文语音服务和国外的英文语音服务,在某些方面已经有超越的趋势.

timg (1).jpg

  通常搭建机器人聊天系统主要包括以下三个方面:

  语音转文字(ASR/STT)

  语义内容(NLU/NLP)

  文字转语音(TTS)

  语音转文字(ASR/STT)

  在将语音传给云端API之前,是本地前端的语音采集,这部分主要包括如下几个方面:

  麦克风降噪

  声源定位

  回声消除

  唤醒词

  语音端点检测

  音频格式压缩

  python 端点检测

  由于实际应用中,单纯依靠能量检测特征检测等方法很难判断人声说话的起始点,所以市面上大多数的语音产品都是使用唤醒词判断语音起始.另外加上声音回路,还可以做语音打断.这样的交互方式可能有些傻,每次必须喊一下 唤醒词 才能继续聊天.这种方式聊多了,个人感觉会嘴巴疼:-O .现在github上有snowboy唤醒词的开源库,大家可以登录snowboy官网训练自己的唤醒词模型.

  Kitt-AI : Snowboy

  Sensory : Sensory

  考虑到用唤醒词嘴巴会累,所以大致调研了一下,Python拥有丰富的库,直接import就能食用.这种方式容易受强噪声干扰,适合一个人在家玩玩.

  pyaudio: pip install pyaudio 可以从设备节点读取原始音频流数据,音频编码是PCM格式;

  webrtcvad: pip install webrtcvad 检测判断一组语音数据是否为空语音;

  当检测到持续时间长度 T1 vad检测都有语音活动,可以判定为语音起始;

  当检测到持续时间长度 T2 vad检测都没有有语音活动,可以判定为语音结束;

  完整程序代码可以从我的github下载

  程序很简单,相信看一会儿就明白了

  ‘‘‘

  Requirements:

  + pyaudio - `pip install pyaudio`

  + py-webrtcvad - `pip install webrtcvad`

  ‘‘‘

  import webrtcvad

  import collections

  import sys

  import signal

  import pyaudio

  from array import array

  from struct import pack

  import wave

  import time

  FORMAT = pyaudio.paInt16

  CHANNELS = 1

  RATE = 16000

  CHUNK_DURATION_MS = 30 # supports 10, 20 and 30 (ms)

  PADDING_DURATION_MS = 1500 # 1 sec jugement

  CHUNK_SIZE = int(RATE CHUNK_DURATION_MS / 1000) # chunk to read

  CHUNK_BYTES = CHUNK_SIZE 2 # 16bit = 2 bytes, PCM

  NUM_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 ge

  NUM_WINDOW_CHUNKS_END = NUM_WINDOW_CHUNKS 2

  START_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 = False

  leave = False

  def handle_int(sig, chunk):

  global leave, got_a_sentence

  leave = True

  got_a_sentence = True

  def 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 # 16384

  times = float(MAXIMUM) / max(abs(i) for i in snd_data)

  r = array(‘h‘)

  for i in snd_data:

  r.append(int(i times))

  return r

  signal.signal(signal.SIGINT, handle_int)

  while not leave:

  ring_buffer = collections.deque(maxlen=NUM_PADDING_CHUNKS)

  triggered = False

  voiced_frames = []

  ring_buffer_flags = [0] NUM_WINDOW_CHUNKS

  ring_buffer_index = 0

  ring_buffer_flags_end = [0] NUM_WINDOW_CHUNKS_END

  ring_buffer_index_end = 0

  buffer_in = ‘‘

  # WangS

  raw_data = array(‘h‘)

  index = 0

  start_point = 0

  StartTime = time.time()

  print(" recording: ")

  stream.start_stream()

  while not got_a_sentence and not leave:

  chunk = stream.read(CHUNK_SIZE)

  # add WangS

  raw_data.extend(array(‘h‘, chunk))

  index += CHUNK_SIZE

  TimeUse = time.time() - StartTime

  active = vad.is_speech(chunk, RATE)

  sys.stdout.write(‘1‘ if active else ‘_‘)

  ring_buffer_flags[ring_buffer_index] = 1 if active else 0

  ring_buffer_index += 1

  ring_buffer_index %= NUM_WINDOW_CHUNKS

  ring_buffer_flags_end[ring_buffer_index_end] = 1 if active else 0

  ring_buffer_index_end += 1

  ring_buffer_index_end %= NUM_WINDOW_CHUNKS_END

  # start point detection

  if 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 = True

  start_point = index - CHUNK_SIZE 20 # start point

  # voiced_frames.extend(ring_buffer)

  ring_buffer.clear()

  # end point detection

  else:

  # 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 = False

  got_a_sentence = True

  sys.stdout.flush()

  sys.stdout.write(‘ ‘)

  # data = b‘‘.join(voiced_frames)

  stream.stop_stream()

  print(" done recording")

  got_a_sentence = False

  # write to file

  raw_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 = True

  stream.close()

  程序运行方式sudo python vad.py