在数据流中查找一系列模式

Question

（这是用 Python 编写的，代码会很棒，但我主要对算法感兴趣。）

我正在监视一个音频流 (PyAudio) 并寻找一系列 5 个弹出声（参见底部的可视化）。我正在读取（）流并获取我刚刚读取的块的 RMS 值（类似于this question）。我的问题是我不是在寻找单个事件，而是一系列具有某些特征但不像我想要的布尔值的事件（pop）。检测这五个爆裂声的最直接（和高效）的方法是什么？

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) 定义什么是 pop：Block 的 RMS 超过 0.002。至少 2 个区块，但不超过 4 个区块。以沉默开始，以沉默结束。

此外，我很想定义什么是静音（忽略不太响亮但不太静音的块，但我不确定这是否比考虑 'pop' 为布尔值更有意义）。

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

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

然后您可以使用平滑平均值的最大值作为弹出窗口。定义一个可以看到五次爆裂声的最大间隔，这可能是你想要的。

调整 P 以获得最佳拟合。

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

Answer 2

在我看来，我最终得到了一种天真的方法，它有一个持续的循环和一些变量来维护和过渡到新状态。不过，完成后我突然想到，我应该探索一下热门词检测，因为 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, ...））并确保它检测到（或不检测到） ) 适当。