Вычисляйте массив numpy только по определенным индексам, заданным маской.

Содержание

Вопрос или проблема
Ответ или решение
Проблема
Решение
Оптимизированный подход
Объяснение решения
Преимущества
Заключение

Вопрос или проблема

В большом проекте мне нужно было обработать двумерный массив numpy определённым образом. Изначально я реализовал это с помощью вложенного вчетверо цикла for, с проверкой if, которая проверяет, выполнено ли условие для заданных индексов. Ничего не должно происходить, если проверка if возвращает False. Вот минимальный воспроизводимый пример:

import numpy as np

L=8; K=20
data = np.random.randint(-10,10,(L,K)) + np.random.randint(-10,10,(L,K))*1j

check_pos = data > 0
check_neg_imag = np.imag(data) < 0

def for_looped(data,L,K,check_pos,check_neg_imag):
    processed = np.zeros((L,K,L,K), dtype=complex)
    for l in range(L):
        for k in range(K):
            for ll in range(L):
                for kk in range(K):
                    if (check_pos[l,k] and check_neg_imag[ll,kk]) or (check_neg_imag[l,k] and check_pos[ll,kk]):
                        processed[l,k,ll,kk] = data[l,k] * data[ll,kk].conj()
    return processed
processed = for_looped(data,L,K,check_pos,check_neg_imag)

Это становится довольно медленно по мере увеличения размеров данных, поэтому я попытался векторизовать циклы for. Однако у меня возникли некоторые проблемы с проверкой if. Я попробовал просто векторизовать без неё:

def vectorized(data):
    return data[:,:,None,None] * data[None,None,:,:].conj()
processed_vec = vectorized(data)

Это работает, но также тратит много вычислительного времени на вычисление значений, которые не используются. Я прочитал в одном месте, что np.where() можно использовать в сочетании с маской, например:

mask = (check_pos[:,:,None,None] & check_neg_imag[None,None,:,:]) | (check_neg_imag[:,:,None,None] & check_pos[None,None,:,:])
def vectorized_where(data,mask):
    return np.where(mask, data[:,:,None,None] * data[None,None,:,:].conj(), 0)
processed_vec_where = vectorized_where(data,mask)

Однако я проверил время выполнения с помощью timeit, и код с np.where() на самом деле медленнее:

%timeit for_looped(data,L,K)
14.8 ms ± 44.2 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)

%timeit vectorized(data)
45.5 µs ± 2.62 µs per loop (mean ± std. dev. of 7 runs, 10,000 loops each)

%timeit vectorized_where(data,mask)
65.9 µs ± 2.6 µs per loop (mean ± std. dev. of 7 runs, 10,000 loops each)

Что я подозреваю, так это то, что np.where() вычисляет data[:,:,None,None] * data[None,None,:,:].conj() для всех индексов, а затем заменяет те, где маска равна True, на 0. Это также тратит время вычисления на то, для чего я его использую.

Есть ли способ вычислить data[:,:,None,None] * data[None,None,:,:].conj() только для индексов, заданных маской?

Ответ или решение

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

Проблема

Вы хотите обработать двумерный массив NumPy, используя условия, которые проверяются через маску. Вы уже предприняли шаги к векторизации вашего кода, однако, как вы заметили, использование np.where() приводит к ненужным вычислениям для тех индексов, где маска не выполняется.

Решение

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

Оптимизированный подход

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

import numpy as np

L = 8
K = 20
data = np.random.randint(-10, 10, (L, K)) + np.random.randint(-10, 10, (L, K)) * 1j

check_pos = data > 0
check_neg_imag = np.imag(data) < 0

def optimized_vectorized(data):
    # Создаём маску условий
    mask = (check_pos[:, :, None, None] & check_neg_imag[None, None, :, :]) | \
           (check_neg_imag[:, :, None, None] & check_pos[None, None, :, :])

    # Следовательно, используем np.nonzero для получения индексов истинных значений маски
    indices = np.nonzero(mask)

    # Вычисляем произведения только для нужных индексов
    processed = np.zeros((L, K, L, K), dtype=complex)
    processed[indices] = data[indices[0], indices[1]] * data[indices[2], indices[3]].conj()

    return processed

processed_optimized = optimized_vectorized(data)

Объяснение решения

Создание маски: Мы создаем маску, как и в вашем примере, но затем извлекаем индексы, где маска истинна, с помощью np.nonzero(). Это позволяет нам воспользоваться только нужными индексами.
Применение индексов: Использовав индексы из маски, мы можем заполнить массив processed только в тех местах, где условия выполняются. Таким образом, вычисление произведения выполняется только для тех элементов, которые соответствуют указанной маске.

Преимущества

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

Заключение

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