python - numpy 1D 배열:n 번 이상 반복되는 마스크 요소




arrays binning (6)

주어진 정수 배열

[1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5]

N 번 이상 반복되는 요소를 마스크해야합니다. 명확히하기 위해 : 기본 목표는 나중에 비닝 계산에 사용하기 위해 부울 마스크 배열을 검색하는 것입니다.

나는 다소 복잡한 해결책을 생각해 냈습니다.

import numpy as np

bins = np.array([1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5])

N = 3
splits = np.split(bins, np.where(np.diff(bins) != 0)[0]+1)
mask = []
for s in splits:
    if s.shape[0] <= N:
        mask.append(np.ones(s.shape[0]).astype(np.bool_))
    else:
        mask.append(np.append(np.ones(N), np.zeros(s.shape[0]-N)).astype(np.bool_)) 

mask = np.concatenate(mask)

예를 들어주는

bins[mask]
Out[90]: array([1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4, 5, 5, 5])

더 좋은 방법이 있습니까?

편집, # 2

답변 주셔서 감사합니다! 다음은 MSeifert 벤치 마크 플롯의 슬림 버전입니다. simple_benchmark 를 가리켜 주셔서 감사합니다. 가장 빠른 4 가지 옵션 만 표시 :

결론

Paul Panzer가 수정 한 Florian H가 제안한 아이디어는이 문제를 해결하기위한 좋은 방법 인 것 같습니다. 그러나 numba 를 사용하는 것이 numba MSeifert의 솔루션 이 다른 솔루션 보다 뛰어납니다.

더 일반적인 대답이므로 MSeifert의 대답을 솔루션으로 채택하기로 선택했습니다. 연속적인 반복 요소의 (고유하지 않은) 블록으로 임의의 배열을 올바르게 처리합니다. numbanumba 이없는 경우 numba 의 답변 도 볼 가치가 있습니다!


해결책

numpy.unique 사용할 수 있습니다. final_mask 변수를 사용하여 배열 bins 에서 traget 요소를 추출 할 수 있습니다.

import numpy as np

bins = np.array([1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5])
repeat_max = 3

unique, counts = np.unique(bins, return_counts=True)
mod_counts = np.array([x if x<=repeat_max else repeat_max for x in counts])
mask = np.arange(bins.size)
#final_values = np.hstack([bins[bins==value][:count] for value, count in zip(unique, mod_counts)])
final_mask = np.hstack([mask[bins==value][:count] for value, count in zip(unique, mod_counts)])
bins[final_mask]

출력 :

array([1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4, 5, 5, 5])

면책 조항 : 이것은 @FlorianH의 아이디어를 더 잘 구현 한 것입니다.

def f(a,N):
    mask = np.empty(a.size,bool)
    mask[:N] = True
    np.not_equal(a[N:],a[:-N],out=mask[N:])
    return mask

더 큰 배열의 경우 큰 차이가 있습니다.

a = np.arange(1000).repeat(np.random.randint(0,10,1000))
N = 3

print(timeit(lambda:f(a,N),number=1000)*1000,"us")
# 5.443050000394578 us

# compare to
print(timeit(lambda:[True for _ in range(N)] + list(bins[:-N] != bins[N:]),number=1000)*1000,"us")
# 76.18969900067896 us

상당히 이해하기 쉬운 numba 를 사용하여 솔루션을 제시하고 싶습니다. 연속 반복되는 항목을 "마스킹"한다고 가정합니다.

import numpy as np
import numba as nb

@nb.njit
def mask_more_n(arr, n):
    mask = np.ones(arr.shape, np.bool_)

    current = arr[0]
    count = 0
    for idx, item in enumerate(arr):
        if item == current:
            count += 1
        else:
            current = item
            count = 1
        mask[idx] = count <= n
    return mask

예를 들면 다음과 같습니다.

>>> bins = np.array([1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5])
>>> bins[mask_more_n(bins, 3)]
array([1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4, 5, 5, 5])
>>> bins[mask_more_n(bins, 2)]
array([1, 1, 2, 2, 3, 3, 4, 4, 5, 5])

공연:

simple_benchmark 사용-그러나 모든 접근법을 포함하지는 않았습니다. 로그 로그 스케일입니다.

numba 솔루션이 Paul Panzer의 솔루션을 이길 수없는 것처럼 보입니다 .Paul Panzer는 대형 어레이의 경우 조금 더 빠르며 추가 종속성이 필요하지 않습니다.

그러나 둘 다 다른 솔루션보다 성능이 뛰어나지 만 "필터링 된"배열 대신 마스크를 반환합니다.

import numpy as np
import numba as nb
from simple_benchmark import BenchmarkBuilder, MultiArgument

b = BenchmarkBuilder()

bins = np.array([1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5])

@nb.njit
def mask_more_n(arr, n):
    mask = np.ones(arr.shape, np.bool_)

    current = arr[0]
    count = 0
    for idx, item in enumerate(arr):
        if item == current:
            count += 1
        else:
            current = item
            count = 1
        mask[idx] = count <= n
    return mask

@b.add_function(warmups=True)
def MSeifert(arr, n):
    return mask_more_n(arr, n)

from scipy.ndimage.morphology import binary_dilation

@b.add_function()
def Divakar_1(a, N):
    k = np.ones(N,dtype=bool)
    m = np.r_[True,a[:-1]!=a[1:]]
    return a[binary_dilation(m,k,origin=-(N//2))]

@b.add_function()
def Divakar_2(a, N):
    k = np.ones(N,dtype=bool)
    return a[binary_dilation(np.ediff1d(a,to_begin=a[0])!=0,k,origin=-(N//2))]

@b.add_function()
def Divakar_3(a, N):
    m = np.r_[True,a[:-1]!=a[1:],True]
    idx = np.flatnonzero(m)
    c = np.diff(idx)
    return np.repeat(a[idx[:-1]],np.minimum(c,N))

from skimage.util import view_as_windows

@b.add_function()
def Divakar_4(a, N):
    m = np.r_[True,a[:-1]!=a[1:]]
    w = view_as_windows(m,N)
    idx = np.flatnonzero(m)
    v = idx<len(w)
    w[idx[v]] = 1
    if v.all()==0:
        m[idx[v.argmin()]:] = 1
    return a[m]

@b.add_function()
def Divakar_5(a, N):
    m = np.r_[True,a[:-1]!=a[1:]]
    w = view_as_windows(m,N)
    last_idx = len(a)-m[::-1].argmax()-1
    w[m[:-N+1]] = 1
    m[last_idx:last_idx+N] = 1
    return a[m]

@b.add_function()
def PaulPanzer(a,N):
    mask = np.empty(a.size,bool)
    mask[:N] = True
    np.not_equal(a[N:],a[:-N],out=mask[N:])
    return mask

import random

@b.add_arguments('array size')
def argument_provider():
    for exp in range(2, 20):
        size = 2**exp
        yield size, MultiArgument([np.array([random.randint(0, 5) for _ in range(size)]), 3])

r = b.run()
import matplotlib.pyplot as plt

plt.figure(figsize=[10, 8])
r.plot()

인덱싱 으로이 작업을 수행 할 수 있습니다. N에 대한 코드는 다음과 같습니다.

N = 3
bins = np.array([1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5,6])

mask = [True for _ in range(N)] + list(bins[:-N] != bins[N:])
bins[mask]

산출:

array([1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4, 5, 5, 5, 6]

grouby 를 사용하여 N 보다 긴 공통 요소 및 필터 목록을 그룹화 할 수 있습니다.

import numpy as np
from itertools import groupby, chain

def ifElse(condition, exec1, exec2):

    if condition : return exec1 
    else         : return exec2


def solve(bins, N = None):

    xss = groupby(bins)
    xss = map(lambda xs : list(xs[1]), xss)
    xss = map(lambda xs : ifElse(len(xs) > N, xs[:N], xs), xss)
    xs  = chain.from_iterable(xss)
    return list(xs)

bins = np.array([1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5])
solve(bins, N = 3)

접근법 # 1 : 여기에 벡터화 된 방법이 있습니다.

from scipy.ndimage.morphology import binary_dilation

def keep_N_per_group(a, N):
    k = np.ones(N,dtype=bool)
    m = np.r_[True,a[:-1]!=a[1:]]
    return a[binary_dilation(m,k,origin=-(N//2))]

샘플 런-

In [42]: a
Out[42]: array([1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 5, 5, 5, 5, 5, 5, 5])

In [43]: keep_N_per_group(a, N=3)
Out[43]: array([1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4, 5, 5, 5])

접근법 # 2 : 좀 더 컴팩트 한 버전-

def keep_N_per_group_v2(a, N):
    k = np.ones(N,dtype=bool)
    return a[binary_dilation(np.ediff1d(a,to_begin=a[0])!=0,k,origin=-(N//2))]

접근법 # 3 : 그룹화 된 카운트와 np.repeat (마스크를 제공하지는 않음)-

def keep_N_per_group_v3(a, N):
    m = np.r_[True,a[:-1]!=a[1:],True]
    idx = np.flatnonzero(m)
    c = np.diff(idx)
    return np.repeat(a[idx[:-1]],np.minimum(c,N))

접근법 # 4 : view-based 방법으로-

from skimage.util import view_as_windows

def keep_N_per_group_v4(a, N):
    m = np.r_[True,a[:-1]!=a[1:]]
    w = view_as_windows(m,N)
    idx = np.flatnonzero(m)
    v = idx<len(w)
    w[idx[v]] = 1
    if v.all()==0:
        m[idx[v.argmin()]:] = 1
    return a[m]

접근법 # 5 : flatnonzero 인덱스가없는 view-based 방법-

def keep_N_per_group_v5(a, N):
    m = np.r_[True,a[:-1]!=a[1:]]
    w = view_as_windows(m,N)
    last_idx = len(a)-m[::-1].argmax()-1
    w[m[:-N+1]] = 1
    m[last_idx:last_idx+N] = 1
    return a[m]




binning