Python / NumPy: реализация промежуточной суммы (но не совсем)

Question

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

a = numpy.array([1, 0, 0, 1, 0, 1, 0, 0, 1, 1])
b = numpy.array([0, 0, 0, 0, 0, 0, 0, 0, 0, 0])

Я бы хотел вычислить сумму всех возможных подмножеств трех смежных элементов в a. Если сумма равна 0 или 1, три соответствующих элемента в b остаются без изменений; только если сумма превышает 1, три соответствующих элемента в b устанавливаются в 1, так что после вычисления b становится

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

Простой цикл выполнит это:

for x in range(len(a)-2):
    if a[x:x+3].sum() > 1:
        b[x:x+3] = 1

После этого b имеет желаемую форму.

Я должен сделать это для большого количества данных, поэтому скорость - это проблема. Есть ли в NumPy более быстрый способ выполнить вышеуказанную операцию?

(я понимаю, что это похоже на свертку, но не совсем то же самое).

Answer 1

Вы можете начать со свертки, выбрать значения, которые превышают 1, и, наконец, использовать «расширение»:

b = numpy.convolve(a, [1, 1, 1], mode="same") > 1
b = b | numpy.r_[0, b[:-1]] | numpy.r_[b[1:], 0]

Поскольку это позволяет избежать цикла Python, он должен быть быстрее, чем ваш подход, но я не делал тайминги.

Альтернативой является использование второй свертки для расширения:

kernel = [1, 1, 1]
b = numpy.convolve(a, kernel, mode="same") > 1
b = numpy.convolve(b, kernel, mode="same") > 0

Если у вас есть SciPy, еще один вариант расширения -

b = numpy.convolve(a, [1, 1, 1], mode="same") > 1
b = scipy.ndimage.morphology.binary_dilation(b)

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

b = numpy.convolve(a, kernel) > 1
b[:-1] |= b[1:]  # Shift and "smearing" to the *left* (smearing with b[1:] |= b[:-1] does not work)
b[:-1] |= b[1:]  # … and again!
b = b[:-2]

Для массива из одного миллиона записей это было более чем в 200 раз быстрее, чем ваш первоначальный подход на моей машине. Как отмечает EOL в комментариях, это решение может считаться немного хрупким, поскольку оно зависит от деталей реализации NumPy.

Answer 2

Вы можете эффективно рассчитать суммы "свертки" с помощью:

>>> a0 = a[:-2]
>>> a1 = a[1:-1]
>>> a2 = a[2:]
>>> a_large_sum = a0 + a1 + a2 > 1

Обновление b может быть эффективно выполнено путем написания чего-то, что означает «по крайней мере одно из трех соседних значений a_large_sum - Истина»: сначала вы расширяете массив a_large_sum до того же числа элементов, что и a (вправо, влево и вправо, а затем влево):

>>> a_large_sum_0 = np.hstack([a_large_sum, [False, False]])
>>> a_large_sum_1 = np.hstack([[False], a_large_sum, [False]])
>>> a_large_sum_2 = np.hstack([[False, False], a_large_sum])

Затем вы получаете b эффективным способом:

>>> b = a_large_sum_0 | a_large_sum_1 | a_large_sum_2

Это дает результат, который вы получаете, но очень эффективным способом, за счет использования внутренних быстрых циклов NumPy.

PS : Этот подход, по сути, такой же, как первое решение Свена, но гораздо более пешеходный, чем элегантный код Свена; это так быстро, как бы то ни было. Второе решение Свена (double convolve()) еще более элегантно, и - в два раза быстрее.

Answer 3

Вам также может понравиться NumPy's stride_tricks. Используя настройку синхронизации Свена (см. Ссылку в ответе Свена), я обнаружил, что для (очень) больших массивов это также быстрый способ сделать то, что вы хотите (т. Е. С вашим определением a):

shape = (len(a)-2,3)
strides = a.strides+a.strides
a_strided = numpy.lib.stride_tricks.as_strided(a, shape=shape, strides=strides)
b = np.r_[numpy.sum(a_strided, axis=-1) > 1, False, False]
b[2:] |= b[1:-1] | b[:-2]

После редактирования (см. Комментарии ниже) это больше не самый быстрый способ.

Это создает специально расширенный вид вашего исходного массива. Данные в a не копируются, а просто просматриваются по-новому. Мы хотим создать новый массив, в котором последний индекс содержит вложенные массивы, которые мы хотим суммировать (т. Е. Три элемента, которые вы хотите суммировать). Таким образом, мы можем легко суммировать в конце последнюю команду.

Таким образом, последний элемент этой новой формы должен быть 3, а первым элементом будет длина старого a минус 2 (поскольку мы можем суммировать только до -2 ый элемент).

Список шагов содержит шаги в байтах, которые новый массив a_strided должен сделать, чтобы перейти к следующему элементу в каждом из измерений формы. Если вы установите их равными, это означает, что a_strided[0,1] и a_strided[1,0] будут оба a[1], что именно то, что мы хотим. В обычном массиве это было бы не так (первым шагом будет «размер первого измерения, умноженный на длину массива первого измерения (= shape [0])»), но в этом случае мы можем хорошо использовать его.

Не уверен, что я все это хорошо объяснил, но просто распечатайте a_strided, и вы увидите, каков результат и насколько легко это делает операцию.