找到一个数据流

Question

中的一系列模式（这是在Python和代码将是巨大的，但我的算法感兴趣。）

我监控的音频流（PyAudio）并寻找一系列5个流行音乐（查看底部的可视化）。我正在阅读（）流，并获取刚才读取的块的RMS值（类似于this question）。我的问题是，我不是在寻找一个单一的事件，而是一系列具有某些特征但并不像我想要的布尔值的事件（弹出）。检测这五种流行音乐最直接的方法是什么？

的RMS功能给了我这样的流：

0.000580998485254, 0.00045098391298, 0.00751436443973, 0.002733730043, 0.00160775708652, 0.000847808804511 

它看起来有点更加有用的，如果我圆（类似的流）为您提供：

0.001, 0.001, 0.018, 0.007, 0.003, 0.001, 0.001 

你可以看到弹出在项目3中，大概是因为它在项目4中静止下来，并且可能尾部在项目5的一小部分中。

我想检测连续中的5个。

我的天真做法是： a）定义什么是弹出：块的RMS超过.002。至少2块但不超过4块。沉默开始，沉默结束。另外，我很想定义什么是沉默（忽略不是很响，但不是很沉默的块，但我不确定这是否更有意义，然后考虑'流行'是布尔值）。

b）然后有一个状态机跟踪一堆变量，并有一堆if语句。像：

while True: 
    is_pop = isRMSAmplitudeLoudEnoughToBeAPop(stream.read()) 

    if is_pop: 
    if state == 'pop': 
     #continuation of a pop (or maybe this continuation means 
     #that it's too long to be a pop 
     if num_pop_blocks <= MAX_POP_RECORDS: 
     num_pop_blocks += 1 
     else: 
     # too long to be a pop 
     state = 'waiting' 
     num_sequential_pops = 0 
    else if state == 'silence': 
     #possible beginning of a pop 
     state = 'pop' 
     num_pop_blocks += 1 
     num_silence_blocks = 0 
    else: 
    #silence 
    if state = 'pop': 
     #we just transitioned from pop to silence 
     num_sequential_pops += 1 

     if num_sequential_pops == 5: 
     # we did it 
     state = 'waiting' 
     num_sequential_pops = 0 
     num_silence_blocks = 0 

     fivePopsCallback() 
    else if state = 'silence': 
     if num_silence_blocks >= MAX_SILENCE_BLOCKS: 
     #now we're just waiting 
     state = 'waiting' 
     num_silence_blocks = 0 
     num_sequential_pops = 0 

该代码并不完全（可能有一个或两个错误），但说明了我的思路。这当然比我想要的要复杂，这就是为什么我要求提出建议。

Answer 1

对于我来说，结束了对于持续循环和一些变量以保持和转换到新状态的感觉，这是一种天真的方法。但是，在我完成之后，我发现我应该探索热门词汇检测，因为连续5次点击基本上是一个热门词汇。他们有一个我必须寻找的模式。

不管怎么说，这里是我的代码：

POP_MIN_MS = 50 
POP_MAX_MS = 150 

POP_GAP_MIN_MS = 50 
POP_GAP_MAX_MS = 200 

POP_BORDER_MIN_MS = 500 

assert POP_BORDER_MIN_MS > POP_GAP_MAX_MS 

POP_RMS_THRESHOLD_MIN = 100 

FORMAT = pyaudio.paInt16 
CHANNELS = 2 
RATE = 44100 # Sampling Rate -- frames per second 
INPUT_BLOCK_TIME_MS = 50 
INPUT_FRAMES_PER_BLOCK = int(RATE*INPUT_BLOCK_TIME_MS/1000) 

POP_MIN_BLOCKS = POP_MIN_MS/INPUT_BLOCK_TIME_MS 
POP_MAX_BLOCKS = POP_MAX_MS/INPUT_BLOCK_TIME_MS 

POP_GAP_MIN_BLOCKS = POP_GAP_MIN_MS/INPUT_BLOCK_TIME_MS 
POP_GAP_MAX_BLOCKS = POP_GAP_MAX_MS/INPUT_BLOCK_TIME_MS 

POP_BORDER_MIN_BLOCKS = POP_BORDER_MIN_MS/INPUT_BLOCK_TIME_MS 


def listen(self): 
    pops = 0 
    sequential_loud_blocks = 0 
    sequential_notloud_blocks = 0 

    stream = self.pa.open(
     format=FORMAT, 
     channels=CHANNELS, 
     rate=RATE, 
     input=True, 
     frames_per_buffer=INPUT_FRAMES_PER_BLOCK 
    ) 

    states = { 
     'PENDING': 1, 
     'POPPING': 2, 
     'ENDING': 3, 
    } 

    state = states['PENDING'] 

    while True: 
     amp = audioop.rms(stream.read(INPUT_FRAMES_PER_BLOCK), 2) 

     is_loud = (amp >= POP_RMS_THRESHOLD_MIN) 

     if state == states['PENDING']: 
     if is_loud: 
      # Only switch to POPPING if it's been quiet for at least the border 
      # period. Otherwise stay in PENDING. 
      if sequential_notloud_blocks >= POP_BORDER_MIN_BLOCKS: 
      state = states['POPPING'] 
      sequential_loud_blocks = 1 

      # If it's now loud then reset the # of notloud blocks 
      sequential_notloud_blocks = 0 
     else: 
      sequential_notloud_blocks += 1 

     elif state == states['POPPING']: 

     if is_loud: 
      sequential_loud_blocks += 1 
      # TODO: Is this necessary? 
      sequential_notloud_blocks = 0 

      if sequential_loud_blocks > POP_MAX_BLOCKS: 
      # it's been loud for too long; this isn't a pop 
      state = states['PENDING'] 
      pops = 0 
      #print "loud too long" 
      # since it has been loud and remains loud then no reason to reset 
      # the notloud_blocks count 

     else: 
      # not loud 
      if sequential_loud_blocks: 
      # just transitioned from loud. was that a pop? 
      # we know it wasn't too long, or we would have transitioned to 
      # PENDING during the pop 
      if sequential_loud_blocks < POP_MIN_BLOCKS: 
       # wasn't long enough 
       # go to PENDING 
       state = states['PENDING'] 
       pops = 0 
       #print "not loud long enough" 
      else: 
       # just right 
       pops += 1 
       logging.debug("POP #%s", pops) 

      sequential_loud_blocks = 0 
      sequential_notloud_blocks += 1 

      else: 
      # it has been quiet. and it's still quiet 
      sequential_notloud_blocks += 1 

      if sequential_notloud_blocks > POP_GAP_MAX_BLOCKS: 
       # it was quiet for too long 
       # we're no longer popping, but we don't know if this is the 
       # border at the end 
       state = states['ENDING'] 

     elif state == states['ENDING']: 
     if is_loud: 
      # a loud block before the required border gap. reset 
      # since there wasn't a gap, this couldn't be a valid pop anyways 
      # so just go back to PENDING and let it monitor for the border 
      sequential_loud_blocks = 1 
      sequential_notloud_blocks = 0 
      pops = 0 

      state = states['PENDING'] 
     else: 
      sequential_notloud_blocks += 1 

      # Is the border time (500 ms right now) enough of a delay? 
      if sequential_notloud_blocks >= POP_BORDER_MIN_BLOCKS: 
      # that's a bingo! 
      if pops == 5: 

       stream.stop_stream() 

       # assume that starting now the channel is not silent 
       start_time = time.time() 


       print ">>>>> 5 POPS" 

       elapsed = time.time() - start_time 

       #time.time() may return fractions of a second, which is ideal  
       stream.start_stream() 

       # do whateve we need to do 

      state = states['PENDING'] 
      pops = 0 

它需要一些正式的测试。昨晚我发现了一个问题，那就是在流行音乐之后没有重新设置自己，然后安静得太久。我的计划是重构，然后为其提供一串模拟RMS'（例如（0,0,0,500,200,0,200,0，...）），并确保它检测到（或未检测到）适当。

Answer 2

您可能要计算最后的P点的simple moving average其中P〜= 4，并将结果与​​您的原始输入数据一起绘制。

然后，您可以使用平滑平均值的最大值作为弹出窗口。定义一个最大时间间隔，以查看五个弹出窗口，这可能是您之后的内容。

调整P以达到最佳状态。

如果还没有Python模块，我不会感到惊讶，但我没看过。