Trouver une série de motifs dans un flux de données

Question

(Ceci est en Python, et le code serait génial, mais je suis surtout intéressé par l'algorithme.)

Je suis la surveillance d'un flux audio (PyAudio) et à la recherche d'une série de 5 pops (voir en bas pour une visualisation). Je suis en train de lire() le flux et d'obtenir la valeur RMS pour le bloc que je viens de lire (similaire à this question). Mon problème est que je ne cherche pas un seul événement, mais plutôt une série d'événements (pops) qui ont quelques caractéristiques mais qui ne sont pas aussi booléens que je le voudrais. Quel est le moyen le plus simple (et performant) pour détecter ces cinq pops?

La fonction RMS me donne un cours d'eau comme ceci:

0.000580998485254, 0.00045098391298, 0.00751436443973, 0.002733730043, 0.00160775708652, 0.000847808804511 

Il semble un peu plus utile si je compléterai (un flux similaire) pour vous:

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

Vous pouvez voir la pop à l'article 3, et sans doute comme il calme dans l'article 4, et peut-être la queue était pendant une fraction de l'article 5.

Je veux détecter 5 de ceux d'affilée.

Mon approche naïve est de: a) définir ce qu'est un pop: Le RMS du bloc est supérieur à .002. Pour au moins 2 blocs mais pas plus de 4 blocs. Commencé par le silence et se termine par le silence. En outre, je suis tenté de définir ce qu'est le silence (pour ignorer les blocs pas très forts mais pas tout à fait silencieux, mais je ne suis pas sûr que cela ait plus de sens que de considérer que pop est booléen). B) Ensuite, ayez une machine à états qui garde la trace d'un tas de variables et qui a un tas d'instructions if. Comme:

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 

Ce code n'est pas du tout complète (et peut-être un bug ou deux), mais illustre ma ligne de pensée. C'est certainement plus complexe que je ne le voudrais, et c'est pourquoi je demande des suggestions.

Answer 1

Je me suis retrouvé avec ce qui, pour moi, se sent comme une approche naïve avec une boucle en cours et quelques variables pour maintenir et transition vers de nouveaux états. Il m'est apparu après avoir terminé, cependant, que j'aurais dû explorer la détection de mots-clés parce que 5 clics consécutifs sont fondamentalement un mot-clé. Et ils ont un modèle que je dois rechercher.

Quoi qu'il en soit, voici mon code:

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 

Il a besoin d'un test formel.J'ai trouvé un problème la nuit dernière dans lequel il ne se réinitialisait pas après un bruit puis un silence trop long. Mon plan consiste à refactoriser puis à alimenter un flux de RMS simulé (par exemple, (0, 0, 0, 500, 200, 0, 200, 0, ...)) et à s'assurer qu'il détecte (ou ne détecte pas) de manière appropriée.

Answer 2

Vous pouvez calculer le simple moving average des derniers points P, où P ~ = 4 et tracer le résultat en même temps que vos données d'entrée brutes.

Vous pouvez ensuite utiliser les maxima de la moyenne lissée en tant que pop. Définir un intervalle maximum dans lequel voir cinq pops et qui pourrait être ce que votre après.

Ajuster P pour un meilleur ajustement.

Je ne serais pas surpris s'il n'y avait pas déjà un module Python pour cela, mais je n'ai pas regardé.