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
