（これはPythonでであり、そしてコードは素晴らしいことだが、私は、アルゴリズムで主に興味があります。）</p> <p>私は、オーディオストリームを監視しています（PyAudio）データストリーム

Question

以内に一連のパターンを見つけます5つのポップを探しています（ビジュアライゼーションの下部を参照）。私はストリームを読み込み（ブロックされたブロックのRMS値を取得しています（this questionに似ています）。私の問題は、私は単一のイベントではなく、いくつかの特徴を持っているが、私が望むようなブール値ではない一連のイベント（ポップ）を探しているということです。これら5つのポップを検出する最も簡単で（そしてパフォーマンスの高い）方法は何ですか？

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の分数の間にあったとします。

私の素朴なアプローチは、次のことです。 a）ポップの定義：ブロックのRMSが.002を超えています。少なくとも2ブロック、4ブロック以下。無音で始まり、無音で終わる。

さらに、私は無声音を定義するように誘惑されています（あまり大きな音ではないがかなり静かなブロックは無視しますが、 'pop'をブーリアンと考えると意味が分かりません）。

b）次に、変数の束を追跡し、if文の束を持つ状態マシンを用意します。 Like：

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 

このコードは完全ではなく（バグまたは2つの可能性もあります）、私の考え方を示しています。それは確かに私がしたいと思っている以上に複雑です、なぜ私は提案を求めている。

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〜= 4最後のP点のsimple moving averageを計算し、あなたの生の入力データと共に結果をプロットすることができます。

スムージング平均の最大値をポップとして使用できます。 5つのポップを表示する最大間隔を定義します。これはあなたの後になる可能性があります。

ベストフィットを調整します。

このためのPythonモジュールがない場合は驚きませんが、私は見ていません。