Wsrtjtyk

I am trying to test to see if the values in an array are above some value a consecutive number of times.

For example

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

arr2 = np.array([1,2,1,3,4,2,6,7])

Say I want to test to see if an item in the array is >=3 for four consecutive periods. The test would return true for arr1 but false for arr2.

Here's one way with convolution -

def cross_thresh_convolve(arr, thresh, N):

    # Detect if arr crosses thresh for N consecutive times anywhere

    return (np.convolve(arr>=thresh,np.ones(N,dtype=int))==N).any()

Alternatively with binary-dilation -

from scipy.ndimage.morphology import binary_erosion



def cross_thresh_erosion(arr, thresh, N):

    return binary_erosion(arr>=thresh, np.ones(N)).any()

Sample runs -

In [43]: arr1 = np.array([1,2,1,3,4,5,6,7])

    ...: arr2 = np.array([1,2,1,3,4,2,6,7])



In [44]: print cross_thresh_convolve(arr1, thresh=3, N=4)

    ...: print cross_thresh_erosion(arr1, thresh=3, N=4)

    ...: print cross_thresh_convolve(arr2, thresh=3, N=4)

    ...: print cross_thresh_erosion(arr2, thresh=3, N=4)

True

True

False

False

Generic comparisons

To cover generic comparisons, say if we want to look for greater or less-than or even simply compare for equality against a value, we could use NumPy builtin comparison functions to replace the arr>=thresh part from earlier solutions and hence give ourselves generic implementations, like so -

def consecutive_comp_convolve(arr, comp, N, comparison=np.greater_equal):

    return (np.convolve(comparison(arr,comp),np.ones(N,dtype=int))==N).any()



def consecutive_comp_erosion(arr, comp, N, comparison=np.greater_equal):

    return binary_erosion(comparison(arr,comp), np.ones(N)).any()

Hence, our specific case runs would be -

consecutive_comp_convolve(arr1, comp=3, N=4, comparison=np.greater_equal)

consecutive_comp_erosion(arr1, comp=3, N=4, comparison=np.greater_equal)

consecutive_comp_convolve(arr2, comp=3, N=4, comparison=np.greater_equal)

consecutive_comp_erosion(arr2, comp=3, N=4, comparison=np.greater_equal)

Here is a lowtech but fast method. Construct the boolean array, form the cumsum() and compare each element to the one n places away. If the difference is n this must be a streak of Trues.

def check_streak(a, th, n):

    ps = (a>=th).cumsum()

    return (ps[n:]-ps[:ps.size-n] == n).any()

Another solution (but slower than the others)

import numpy as np

from numpy.lib.stride_tricks import as_strided



def f(arr, threshold=3, n=4):

    arr = as_strided(arr, shape=(arr.shape[0]-n+1, n), strides=2*arr.strides)

    return (arr >= threshold).all(axis=1).any()





# How it works:

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

# n = 4

# threshold = 3



# arr = as_strided(arr, shape=(arr.shape[0]-n+1, n), strides=2*arr.strides)

# print(arr)

# [[1 2 3 4]

#  [2 3 4 5]

#  [3 4 5 6]

#  [4 5 6 7]

#  [5 6 7 8]]



# print(arr >= threshold)

# [[False False  True  True]

#  [False  True  True  True]

#  [ True  True  True  True]

#  [ True  True  True  True]

#  [ True  True  True  True]]



# print((arr >= threshold).all(axis=1))

# [False False  True  True  True]



# print((arr >= threshold).all(axis=1).any())

# True