Получение пересекающихся строк из двух двумерных массивов numpy

Вы можете использовать множества в Python:

import numpy as np
A = np.array([[1, 4], [2, 5], [3, 6]])
B = np.array([[1, 4], [3, 6], [7, 8]])
aset = set(tuple(x) for x in A)
bset = set(tuple(x) for x in B)
result = np.array([x for x in aset & bset])

В результате получится:

array([[1, 4],
       [3, 6]])

Как отмечает Роб Коуи, это можно записать более кратко:

np.array([x for x in set(tuple(x) for x in A) & set(tuple(x) for x in B)])

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

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

a = np.random.randint(10, size=(5, 3))
b = np.zeros_like(a)
b[:4, :] = a[np.random.randint(a.shape[0], size=4), :]

Вот результаты выполнения:

a = array([[5, 6, 3],
           [8, 1, 0],
           [2, 1, 4],
           [8, 0, 6],
           [6, 7, 6]])
b = array([[2, 1, 4],
           [2, 1, 4],
           [6, 7, 6],
           [5, 6, 3],
           [0, 0, 0]])

Процесс можно описать так (массивы могут быть поменяны местами):

# a имеет размерность nxm, а b — kxm
c = np.swapaxes(a[:, :, None], 1, 2) == b  # преобразуем a в nx1xm
# c имеет размерность nxkxm из-за сравнения с широковещательной передачей
# каждый nxixj срез содержит матрицу сравнения между a[j,:] и b[i,:]
# Уменьшаем размерность до nxk с помощью произведения:
c = np.prod(c, axis=2)
# Чтобы избежать дубликатов:
# Вычисляем кумулятивную сумму по k-ой размерности
c = c * np.cumsum(c, axis=0)
# Сравниваем с 1, чтобы получить только одно "True" в строке
c = c == 1
#//
# Суммируем по k-ой размерности, чтобы получить nx1 вектор
c = np.sum(c, axis=1).astype(bool)
# Пересечение между a и b — это a[c]
result = a[c]

Функция, занимающая всего 2 строки для уменьшения используемой памяти (исправьте меня, если я ошибаюсь):

def array_row_intersection(a, b):
    tmp = np.prod(np.swapaxes(a[:, :, None], 1, 2) == b, axis=2)
    return a[np.sum(np.cumsum(tmp, axis=0) * tmp == 1, axis=1).astype(bool)]

Для моего примера это дало:

result = array([[5, 6, 3],
                 [2, 1, 4],
                 [6, 7, 6]])

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

Бенчмарк для проголосованного ответа и моего (количество элементов в каждом измерении играет роль в выборе):

def voted_answer(A, B):
    nrows, ncols = A.shape
    dtype = {'names': ['f{}'.format(i) for i in range(ncols)],
             'formats': ncols * [A.dtype]}
    C = np.intersect1d(A.view(dtype), B.view(dtype))
    return C.view(A.dtype).reshape(-1, ncols)

a_small = np.random.randint(10, size=(10, 10))
b_small = np.zeros_like(a_small)
b_small = a_small[np.random.randint(a_small.shape[0], size=[a_small.shape[0]]), :]
a_big_row = np.random.randint(10, size=(10, 1000))
b_big_row = a_big_row[np.random.randint(a_big_row.shape[0], size=[a_big_row.shape[0]]), :]
a_big_col = np.random.randint(10, size=(1000, 10))
b_big_col = a_big_col[np.random.randint(a_big_col.shape[0], size=[a_big_col.shape[0]]), :]
a_big_all = np.random.randint(10, size=(100, 100))
b_big_all = a_big_all[np.random.randint(a_big_all.shape[0], size=[a_big_all.shape[0]]), :]

print('Маленькие массивы:')
print('\t Проголосованный ответ:', timeit.timeit(lambda: voted_answer(a_small, b_small), number=100) / 100)
print('\t Предложенный ответ:', timeit.timeit(lambda: array_row_intersection(a_small, b_small), number=100) / 100)
print('Большие столбцовые массивы:')
print('\t Проголосованный ответ:', timeit.timeit(lambda: voted_answer(a_big_col, b_big_col), number=100) / 100)
print('\t Предложенный ответ:', timeit.timeit(lambda: array_row_intersection(a_big_col, b_big_col), number=100) / 100)
print('Большие строковые массивы:')
print('\t Проголосованный ответ:', timeit.timeit(lambda: voted_answer(a_big_row, b_big_row), number=100) / 100)
print('\t Предложенный ответ:', timeit.timeit(lambda: array_row_intersection(a_big_row, b_big_row), number=100) / 100)
print('Большие массивы:')
print('\t Проголосованный ответ:', timeit.timeit(lambda: voted_answer(a_big_all, b_big_all), number=100) / 100)
print('\t Предложенный ответ:', timeit.timeit(lambda: array_row_intersection(a_big_all, b_big_all), number=100) / 100)

С результатами:

Маленькие массивы:
     Проголосованный ответ: 7.47108459473e-05
     Предложенный ответ: 2.47001647949e-05
Большие столбцовые массивы:
     Проголосованный ответ: 0.00198730945587
     Предложенный ответ: 0.0560171294212
Большие строковые массивы:
     Проголосованный ответ: 0.00500325918198
     Предложенный ответ: 0.000308241844177
Большие массивы:
     Проголосованный ответ: 0.000864889621735
     Предложенный ответ: 0.00257176160812

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

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

>>> a = np.array([[3, 1, 2], [5, 8, 9], [7, 4, 3]])
>>> b = np.array([[2, 3, 0], [3, 1, 2], [7, 4, 3]])
>>> av = a.view([('', a.dtype)] * a.shape[1]).ravel()
>>> bv = b.view([('', b.dtype)] * b.shape[1]).ravel()
>>> np.intersect1d(av, bv).view(a.dtype).reshape(-1, a.shape[1])
array([[3, 1, 2],
       [7, 4, 3]])

Для наглядности, структурированный вид выглядит так:

>>> a.view([('', a.dtype)] * a.shape[1])
array([[(3, 1, 2)],
       [(5, 8, 9)],
       [(7, 4, 3)]],
       dtype=[('f0', '<i8'), ('f1', '<i8'), ('f2', '<i8')])

В этом примере мы сначала создаем два массива a и b, затем преобразовываем их в структурированные массивы, чтобы упростить поиск пересечений. Функция np.intersect1d позволяет найти пересечения, после чего мы формируем результирующий массив с исходным выделением.

Вы можете использовать следующий код для нахождения элементов, которые присутствуют в коллекции b, но отсутствуют в коллекции a. Этот код преобразует списки в кортежи, а затем использует разность множеств для получения уникальных значений:

np.array(set(map(tuple, b)).difference(set(map(tuple, a))))

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

import numpy as np

# Предположим, a и b - это ваши массивы
result = b[~np.isin(b, a)]

Этот метод также позволит вам получить элементы, которые есть в b, но отсутствуют в a, и может оказаться более удобным, если вы работаете с numpy массивами.

Ваш код выглядит правильно, и давайте по шагам разберем, как он работает. Мы имеем две матрицы A и B, и функция matching_rows, которая ищет совпадающие строки в A и B.

Описание функции:

1. Создание списков совпадений: В списке matches мы будем хранить индексы строк из B, которые полностью совпадают с какой-либо строкой из A. Мы используем генератор списка, который проходит по всем строкам B, и для каждой строки проверяем, есть ли совпадение с A с помощью np.all(A == B[i], axis=1). Функция np.any() определяет, есть ли хотя бы одно совпадение.

2. Возврат результата: Если у нас нет совпадений (len(matches) == 0), функция возвращает пустой результат B[matches]. Однако это будет возвращать пустой массив, так как matches пустой. И в случае наличия совпадений, функция возвращает уникальные строки из B[matches].

Указание о допущении:

Следует отметить, что в данном решении предполагается, что все строки имеют одинаковую длину. Это важное замечание, так как функция не будет корректно работать, если размеры строк различаются.

Пример:

Если вызвать вашу функцию следующим образом:

A = np.array([[1,4],[2,5],[3,6]])
B = np.array([[1,4],[3,6],[7,8]])

print(matching_rows(A, B))

Вы получите:

array([[1, 4],
       [3, 6]])

Таким образом, ваше решение верно определяет совпадающие строки в массивах A и B.