Найти большое количество последовательных значений, удовлетворяющих условию в массиве numpy

Question

У меня есть некоторые аудиоданные, загруженные в пустой массив, и я хочу сегментировать данные, находя тихие части, то есть части, где амплитуда звука ниже определенного порога в течение определенного периода времени.

Очень простой способ сделать это примерно так:

values = ''.join(("1" if (abs(x) < SILENCE_THRESHOLD) else "0" for x in samples))
pattern = re.compile('1{%d,}'%int(MIN_SILENCE))                                                                           
for match in pattern.finditer(values):
   # code goes here

Приведенный выше код находит части, в которых по крайней мере MIN_SILENCE последовательных элементов меньше, чем SILENCE_THRESHOLD.

Теперь, очевидно, приведенный выше код ужасно неэффективен и ужасно злоупотребляет регулярными выражениями. Есть ли другой метод, который более эффективен, но все же приводит к одинаково простому и короткому коду?

Answer 1

Вот решение на основе numpy.

Я думаю (?) Это должно быть быстрее, чем другие варианты. Надеюсь, это довольно ясно.

Однако для этого требуется вдвое больше памяти, чем для различных генераторных решений. Пока вы можете хранить одну временную копию ваших данных в памяти (для diff) и логический массив той же длины, что и ваши данные (1 бит на элемент), это должно быть довольно эффективно

import numpy as np

def main():
    # Generate some random data
    x = np.cumsum(np.random.random(1000) - 0.5)
    condition = np.abs(x) < 1

    # Print the start and stop indicies of each region where the absolute 
    # values of x are below 1, and the min and max of each of these regions
    for start, stop in contiguous_regions(condition):
        segment = x[start:stop]
        print start, stop
        print segment.min(), segment.max()

def contiguous_regions(condition):
    """Finds contiguous True regions of the boolean array "condition". Returns
    a 2D array where the first column is the start index of the region and the
    second column is the end index."""

    # Find the indicies of changes in "condition"
    d = np.diff(condition)
    idx, = d.nonzero() 

    # We need to start things after the change in "condition". Therefore, 
    # we'll shift the index by 1 to the right.
    idx += 1

    if condition[0]:
        # If the start of condition is True prepend a 0
        idx = np.r_[0, idx]

    if condition[-1]:
        # If the end of condition is True, append the length of the array
        idx = np.r_[idx, condition.size] # Edit

    # Reshape the result into two columns
    idx.shape = (-1,2)
    return idx

main()

Answer 2

Немного небрежно, но просто и быстро, если вы не возражаете против использования scipy:

from scipy.ndimage import gaussian_filter
sigma = 3
threshold = 1
above_threshold = gaussian_filter(data, sigma=sigma) > threshold

Идея состоит в том, что тихие части данных будут сглаживаться до низкой амплитуды, а громкие областине будет.Настройте «сигму», чтобы определить, насколько длинным должен быть «тихий» регион;Настройте «Порог», чтобы повлиять на то, насколько тихо должно бытьЭто замедляет работу больших сигм, и в этот момент сглаживание на основе БПФ может быть быстрее.

Это дает дополнительное преимущество, заключающееся в том, что одиночные «горячие пиксели» не нарушают процесс поиска тишины, поэтому вынемного менее чувствителен к определенным типам шума.

Answer 3

Для этого есть очень удобное решение с использованием scipy.ndimage.Для массива:

a = array([1, 1, 1, 1, 0, 0, 0, 1, 1, 1, 0])

, который может быть результатом условия, примененного к другому массиву, найти смежные области так же просто, как:

regions = scipy.ndimage.find_objects(scipy.ndimage.label(a)[0])

Затем применить любую функцию кэти регионы могут быть сделаны, например:

[np.sum(a[r]) for r in regions]

Answer 4

Я не проверял это, но вы должны быть близки к тому, что вы ищете.Чуть больше строк кода, но он должен быть более эффективным, читабельным и не использовать регулярные выражения: -)

def find_silent(samples):
    num_silent = 0
    start = 0
    for index in range(0, len(samples)):
        if abs(samples[index]) < SILENCE_THRESHOLD:
            if num_silent == 0:
                start = index
            num_silent += 1
        else:
            if num_silent > MIN_SILENCE:
                yield samples[start:index]
            num_silent = 0
    if num_silent > MIN_SILENCE:
        yield samples[start:]

for match in find_silent(samples):
    # code goes here

Answer 5

другой способ сделать это быстро и лаконично:

import pylab as pl

v=[0,0,1,1,0,0,1,1,1,1,1,0,1,0,1,1,0,0,0,0,0,1,0,0]
vd = pl.diff(v)
#vd[i]==1 for 0->1 crossing; vd[i]==-1 for 1->0 crossing
#need to add +1 to indexes as pl.diff shifts to left by 1

i1=pl.array([i for i in xrange(len(vd)) if vd[i]==1])+1
i2=pl.array([i for i in xrange(len(vd)) if vd[i]==-1])+1

#corner cases for the first and the last element
if v[0]==1:
  i1=pl.hstack((0,i1))
if v[-1]==1:
  i2=pl.hstack((i2,len(v)))

теперь i1 содержит начальный индекс и i2 конечный индекс 1, ..., 1 областей

Answer 6

Это должно вернуть список (start,length) пар:

def silent_segs(samples,threshold,min_dur):
  start = -1
  silent_segments = []
  for idx,x in enumerate(samples):
    if start < 0 and abs(x) < threshold:
      start = idx
    elif start >= 0 and abs(x) >= threshold:
      dur = idx-start
      if dur >= min_dur:
        silent_segments.append((start,dur))
      start = -1
  return silent_segments

и простой тест:

>>> s = [-1,0,0,0,-1,10,-10,1,2,1,0,0,0,-1,-10]
>>> silent_segs(s,2,2)
[(0, 5), (9, 5)]

Answer 7

@ joe-kington У меня повышение скорости примерно на 20% -25% по сравнению с решением np.diff / np.nonzero благодаря использованию argmax (см. Код ниже, condition - логическое значение)

def contiguous_regions(condition):
    idx = []
    i = 0
    while i < len(condition):
        x1 = i + condition[i:].argmax()
        try:
            x2 = x1 + condition[x1:].argmin()
        except:
            x2 = x1 + 1
        if x1 == x2:
            if condition[x1] == True:
                x2 = len(condition)
            else:
                break
        idx.append( [x1,x2] )
        i = x2
    return idx

Конечно, ваш пробег может варьироваться в зависимости от ваших данных.

Кроме того, я не совсем уверен, но я предполагаю, что numpy может оптимизировать argmin/argmax по логическим массивам, чтобы остановить поиск при первом появлении True/False. Это может объяснить это.