EDIT
My first solution was functional, but inefficient. It required five iterations of the tensor (zero-trail, reverse, zero-trail, reverse, where). I now have a solution that requires only two iterations, and is also flexible enough to quickly identify local minima as well...
def get_slope(prev, cur):
# A: Ascending
# D: Descending
# P: PEAK (on previous node)
# V: VALLEY (on previous node)
return tf.cond(prev[0] < cur, lambda: (cur, ascending_or_valley(prev, cur)), lambda: (cur, descending_or_peak(prev, cur)))
def ascending_or_valley(prev, cur):
return tf.cond(tf.logical_or(tf.equal(prev[1], 'A'), tf.equal(prev[1], 'V')), lambda: np.array('A'), lambda: np.array('V'))
def descending_or_peak(prev, cur):
return tf.cond(tf.logical_or(tf.equal(prev[1], 'A'), tf.equal(prev[1], 'V')), lambda: np.array('P'), lambda: np.array('D'))
def label_local_extrema(tens):
"""Return a vector of chars indicating ascending, descending, peak, or valley slopes"""
# initializer element values don't matter, just the type.
initializer = (np.array(0, dtype=np.float32), np.array('A'))
# First, get the slope for each element
slope = tf.scan(get_slope, tens, initializer)
# shift by one, since each slope indicator is the slope
# of the previous node (necessary to identify peaks and valleys)
return slope[1][1:]
def find_local_maxima(tens):
"""Return the indices of the local maxima of the first dimension of the tensor"""
return tf.squeeze(tf.where(tf.equal(label_local_extrema(blur_x_tf), 'P')))
End EDIT
Ok, I've managed to find a solution, but it's not pretty. The following function takes a 1D tensor, and reduces all points that are not local maxima to zero. This will work only for positive numbers, and would require modification for datatypes other than float32, but it meets my needs.
There has to be a better way to do this, though.
def zero_descent(prev, cur):
"""reduces all descent steps to zero"""
return tf.cond(prev[0] < cur, lambda: (cur, cur), lambda: (cur, 0.0))
def skeletonize_1d(tens):
"""reduces all point other than local maxima to zero"""
# initializer element values don't matter, just the type.
initializer = (np.array(0, dtype=np.float32), np.array(0, dtype=np.float32))
# First, zero out the trailing side
trail = tf.scan(zero_descent, tens, initializer)
# Next, let's make the leading side the trailing side
trail_rev = tf.reverse(trail[1], [0])
# Now zero out the leading (now trailing) side
lead = tf.scan(zero_descent, trail_rev, initializer)
# Finally, undo the reversal for the result
return tf.reverse(lead[1], [0])
def find_local_maxima(tens):
return tf.where(skeletonize_1d >0)
