对时间数据进行最大池化操作 - 选择具有最高幅度的信号

Question

是否有任何干净的方法来对时间数据执行最大池操作（即输出幅度最高的信号）。

例如，

# sample four sin signals
a = 2*tf.math.sin(tf.linspace(0, 10, 200))
b = 0.1*tf.math.sin(2*tf.linspace(0, 10, 200))
c = 3*tf.math.sin(0.5*tf.linspace(0, 10, 200))
d = 1*tf.math.sin(5*tf.linspace(0, 10, 200))
# stack the signals
data = tf.stack([a, b, c, d], -1)
# reshape to appropriate timeseries of 2D feature-maps
# (batch_size, sequence length, feature_dim1, feature_dim2, channels)
data = tf.reshape(data, [1, 200, 2, 2, 1])

data

看起来像这样：

现在，我想对

MaxPooling2D((2,2))

执行类似于

data

的操作，以仅获得

c

（因为它具有最高的振幅）。显然，我们不能直接使用

MaxPooling3D

和

TimeDistributed

层，因为它们将在每个时间步执行池化。我尝试了使用

tf.math.reduce_max()

和

tf.nn.max_pool_with_argmax

的替代方案，但它们并不简单。

如有任何建议或意见，我们将不胜感激。预先感谢:)

Answer 1

这是我对上述问题的实现，

def temporal_max_pooler(signal_stack):

  #signal_stack (bs, T, f, f, d)
  overall_stack = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
  for ch in range(signal_stack.shape[-1]):
    ch_signals = signal_stack[:,:,:,:,ch:ch+1]
    patches = tf.extract_volume_patches(ch_signals, (1, 1, 2, 2, 1), 
                                        (1, 1, 2, 2, 1), 'VALID')
    patches = tf.transpose(patches, [0, 1, 4, 2, 3])
    (s0, s1, s2, s3, s4) = patches.shape
    patches = tf.reshape(patches, [s0, s1, s2//2, s2//2, s4*s3])
    ch_stack = tf.TensorArray(dtype=tf.float32, size=0, dynamic_size=True)
    for p in range(patches.shape[-1]):
      p_signals = tf.reshape(patches[:,:,:,:,p],
                             (patches.shape[0], patches.shape[1], -1))
      max_amps = tf.math.reduce_max(p_signals, 1)
      where_is_max = tf.math.argmax(max_amps, -1)
      winners = tf.gather(p_signals, where_is_max, axis=-1, batch_dims=1)
      ch_stack.write(ch_stack.size(), winners)
    ch_stack = tf.transpose(ch_stack.stack(), [1, 2, 0])
    n_d = tf.math.sqrt(tf.cast(ch_stack.shape[-1], 'float32'))
    n_d = tf.cast(n_d, 'int32')
    ch_stack = tf.reshape(ch_stack, [ch_stack.shape[0], ch_stack.shape[1],
                                     n_d, n_d])
    overall_stack.write(overall_stack.size(), ch_stack)
  overall_stack = overall_stack.stack()
  return tf.transpose(overall_stack, [1, 2, 3, 4, 0])