Sliding tensorflow classifier over large input

Viewed 189

I have a classifier (implemented in tensorflow 2.2) with a small input of float vectors shaped (None,500,1) and one hot output of (None,100). I want to train the same architecture when "Sliding" it across a larger vector: the input should be (None,5000) and output will be the outcome of its results over the large vector: (None,10,100) that will be aggregated with some function like average to get the one hot result (None,100). I wish to implement this inside the model without splitting the data vectors to several pieces.

#A small input simplified working model:

from tensorflow.keras.layers import Input,Conv1D,LSTM,Dense,Flatten
from tensorflow.keras import Model
import tensorflow as tf
import numpy as np

input_vector_length=500
output_size=100
filters, units,kernel_size=4,4,4

small_inp = Input((input_vector_length, 1))
x = Conv1D(filters, kernel_size=kernel_size, activation='relu')(small_inp)
x = LSTM(units, return_sequences=True)(x)
x = Flatten()(x)
preds= Dense(output_size, activation='softmax')(x)
model = Model(inputs=small_inp, outputs=preds)
model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy']) 

#Generating data and training:

X=np.random.normal(size=(200,500,1))
labels=np.random.randint(0,100,(200,))
y=np.eye(100)[labels]
model.fit(X,y,epochs=5)
2 Answers

You can use Lambda combined with Concatenate, the idea is to divide the input tensor with Lambda and combine the outputs with Concatenate at the end. Here is a working example

from tensorflow.keras.layers import Input,Conv1D,LSTM,Dense,Flatten
from tensorflow.keras import Model
import tensorflow as tf
import numpy as np

# the additional modules you need to import
from tensorflow.keras.layers import Lambda,Concatenate
from tensorflow.keras import Sequential


filters,units,kernel_size = 4,4,4

# create a smaller example dataset
input_vector_length = 10
output_size = 4
num_samples = 5

X = np.ones((1,input_vector_length*num_samples,1))
labels = np.random.randint(0,output_size,(num_samples,))
y = np.eye(output_size)[labels]
y = np.concatenate(y,axis=0)[None,:]

# get the slices of the input layer
start_inds = range(input_vector_length*num_samples+input_vector_length)[::input_vector_length]
start_pairs = [[start_inds[ind],start_inds[ind+1]] for ind in range(len(start_inds)-1)]

conv_layer = Conv1D(filters,kernel_size=kernel_size,activation='relu')
lstm_layer = LSTM(units,return_sequences=True)
flatten_layer = Flatten()
dense_layer = Dense(output_size,activation='softmax')
block_op = Sequential([conv_layer,lstm_layer,flatten_layer,dense_layer])
# create slices of the input layer, return a list of input Tensors
slice_input = Lambda(lambda x: [x[:,inds[0]:inds[1]] for inds in start_pairs])
concat_layer = Concatenate(axis=1)

small_inp = Input(shape=(input_vector_length*num_samples,1),batch_size=1)
preds = concat_layer([block_op(x_slice) for x_slice in slice_input(small_inp)])
model = Model(inputs=small_inp,outputs=preds)
model.compile(optimizer='adam',loss='categorical_crossentropy',metrics=['accuracy'])
model.fit(X,y,epochs=5)

pred = model.predict(X)

You can use tf.data.Dataset.window for this.

import tensorflow as tf

x = tf.reshape(tf.range(1, 1000 + 1), (-1, 1))
y = tf.random.uniform((1000, 1), minval=0, maxval=2, dtype=tf.int32)

ds = tf.data.Dataset.from_tensor_slices((x, y))
ds = ds.window(size=10, shift=1, stride=1)
ds = ds.flat_map(lambda x, y: tf.data.Dataset.zip((x, y)).take(4).batch(4))

for batch in ds.take(10):
    print(batch[0])
tf.Tensor(
[[1], [2], [3], [4]], shape=(4, 1), dtype=int32)
tf.Tensor(
[[2], [3], [4], [5]], shape=(4, 1), dtype=int32)
tf.Tensor(
[[3], [4], [5], [6]], shape=(4, 1), dtype=int32)
tf.Tensor(
[[4], [5], [6], [7]], shape=(4, 1), dtype=int32)
tf.Tensor(
[[5], [6], [7], [8]], shape=(4, 1), dtype=int32)
tf.Tensor(
[[6], [7], [8], [9]], shape=(4, 1), dtype=int32)
tf.Tensor(
[[7], [8], [9], [10]], shape=(4, 1), dtype=int32)

As you can see, the batches are "sliding", in the sense that they are overlapping.

Related