multi input Subclassing model has some error with inputs

Viewed 30

I have a lot of problem with subclassing models. I want to change this code from functional to subclass model But I received some errors.

class CTCLayer(layers.Layer):
    def __init__(self, name=None):
        super().__init__(name=name)
        self.loss_fn = keras.backend.ctc_batch_cost

    def call(self, y_true, y_pred):
        # Compute the training-time loss value and add it
        # to the layer using `self.add_loss()`.
        batch_len = tf.cast(tf.shape(y_true)[0], dtype="int64")
        input_length = tf.cast(tf.shape(y_pred)[1], dtype="int64")
        label_length = tf.cast(tf.shape(y_true)[1], dtype="int64")

        input_length = input_length * tf.ones(shape=(batch_len, 1), dtype="int64")
        label_length = label_length * tf.ones(shape=(batch_len, 1), dtype="int64")

        loss = self.loss_fn(y_true, y_pred, input_length, label_length)
        self.add_loss(loss)

        # At test time, just return the computed predictions
        return y_pred


def build_model():
    # Inputs to the model
    input_img = layers.Input(
        shape=(img_width, img_height, 1), name="image", dtype="float32"
    )
    labels = layers.Input(name="label", shape=(None,), dtype="float32")

    # First conv block
    x = layers.Conv2D(
        32,
        (3, 3),
        activation="relu",
        kernel_initializer="he_normal",
        padding="same",
        name="Conv1",
    )(input_img)
    x = layers.MaxPooling2D((2, 2), name="pool1")(x)

    # Second conv block
    x = layers.Conv2D(
        64,
        (3, 3),
        activation="relu",
        kernel_initializer="he_normal",
        padding="same",
        name="Conv2",
    )(x)
    x = layers.MaxPooling2D((2, 2), name="pool2")(x)

    # We have used two max pool with pool size and strides 2.
    # Hence, downsampled feature maps are 4x smaller. The number of
    # filters in the last layer is 64. Reshape accordingly before
    # passing the output to the RNN part of the model
    new_shape = ((img_width // 4), (img_height // 4) * 64)
    x = layers.Reshape(target_shape=new_shape, name="reshape")(x)
    x = layers.Dense(64, activation="relu", name="dense1")(x)
    x = layers.Dropout(0.2)(x)

    # RNNs
    x = layers.Bidirectional(layers.LSTM(128, return_sequences=True, dropout=0.25))(x)
    x = layers.Bidirectional(layers.LSTM(64, return_sequences=True, dropout=0.25))(x)

    # Output layer
    x = layers.Dense(
        len(char_to_num.get_vocabulary()) + 1, activation="softmax", name="dense2"
    )(x)

    # Add CTC layer for calculating CTC loss at each step
    output = CTCLayer(name="ctc_loss")(labels, x)

    # Define the model
    model = keras.models.Model(
        inputs=[input_img, labels], outputs=output, name="ocr_model_v1"
    )
    # Optimizer
    opt = keras.optimizers.Adam()
    # Compile the model and return
    model.compile(optimizer=opt)
    return model

I want to convert above code to following code:

class CTCLayer(layers.Layer):
    def __init__(self, name=None):
        super().__init__(name=name)
        self.loss_fn = keras.backend.ctc_batch_cost

    def call(self, y_true, y_pred):
        # Compute the training-time loss value and add it
        # to the layer using `self.add_loss()`.
        batch_len = tf.cast(tf.shape(y_true)[0], dtype="int64")
        input_length = tf.cast(tf.shape(y_pred)[1], dtype="int64")
        label_length = tf.cast(tf.shape(y_true)[1], dtype="int64")

        input_length = input_length * tf.ones(shape=(batch_len, 1), dtype="int64")
        label_length = label_length * tf.ones(shape=(batch_len, 1), dtype="int64")

        loss = self.loss_fn(y_true, y_pred, input_length, label_length)
        self.add_loss(loss)

        # At test time, just return the computed predictions
        return y_pred


class ConvBlock(layers.Layer):
    def __init__(self, n_filter:int, activation:str, kernel_initializer:str):
        super(ConvBlock, self).__init__()
        self.conv2d = Conv2D(n_filter, (3, 3), activation=activation,
                             kernel_initializer=kernel_initializer,
                             padding="same")
        
        self.maxpool = MaxPooling2D()
    
    def call(self, inputs):
        x = self.conv2d(inputs)
        x = self.maxpool(x)
        return x


class RNNBlock(layers.Layer):
    def __init__(self, unit_rnn_1:int, drop_rnn_1:float, unit_rnn_2:int, drop_rnn_2:float):
        super(RNNBlock, self).__init__()
        self.rnn_1 = Bidirectional(LSTM(unit_rnn_1, return_sequences=True, dropout=drop_rnn_1))
        self.rnn_2 = Bidirectional(LSTM(unit_rnn_2, return_sequences=True, dropout=drop_rnn_2))
    
    def call(self, inputs):
        x = self.rnn_1(inputs)
        x = self.rnn_2(x)
        return x


class OCRModel(Model):
    def __init__(self, block_1:int=32, act:str='elu', init_kernel:str='he_normal',
                 block_2:int=64, new_shape:tuple=(50, 768), unit_1:int=64, 
                 drop:float=0.2, unit_rnn_1:int=128, drop_rnn_1:float=0.25, 
                 unit_rnn_2:int=64, drop_rnn_2:float=0.25):
        super(OCRModel, self).__init__()

        # CNNs
        self.b1 = ConvBlock(block_1, act, init_kernel)
        self.b2 = ConvBlock(block_2, act, init_kernel)
        self.reshape = Reshape(target_shape=new_shape)
        self.dense1 = Dense(unit_1, act)
        self.drop = Dropout(drop)
        #RNNs
        self.rnn = RNNBlock(unit_rnn_1, drop_rnn_1, unit_rnn_2, drop_rnn_2)
        # Output layer
        self.dense2 = Dense(len(char_to_num.get_vocabulary()) + 1, activation="softmax")
        # CTC Loss
        self.ctc = CTCLayer()

    def call(self, inputs):
        x = self.b1(inputs[0])
        x = self.b2(x)
        x = self.reshape(x)
        x = self.dense1(x)
        x = self.drop(x)
        x = self.rnn(x)
        x = self.dense2(x)
        output = self.ctc(inputs[1], x)
        return output


model = OCRModel()
img_input = Input(shape=(img_width, img_height, 1))
label_input = Input(shape=(None, ))
output = model([img_input, label_input])

model = Model(inputs=[img_input, label_input], outputs = output)
model.compile(optimizer='adam')
model.summary()

When I want to start training with following code:

epochs = 100
early_stopping_patience = 10
# Add early stopping
early_stopping = keras.callbacks.EarlyStopping(
    monitor="val_loss", patience=early_stopping_patience, restore_best_weights=True
)

# Train the model
history = model.fit(
    train_dataset,
    validation_data=validation_dataset,
    epochs=epochs,
    callbacks=[early_stopping],
)

I received this error:

 ValueError: Missing data for input "input_7". You passed a data dictionary with keys ['image', 'label']. Expected the following keys: ['input_7', 'input_8']

I don't have any ideas to fix my problem for first what is my problem and at second how can I handle it before it I trained functional model and It worked but when I want to train with my subclass model I received error

0 Answers
Related