找到一個數據流

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模塊，我不會感到驚訝，但我沒看過。