I want to design a metric to estimate the extent to which the two distributions generated from a Siamese model overlap during training, but don't find any solution.
I have referenced this discussion to construct the metric using callbacks argument of model.fit:
class Metrics(tf.keras.callbacks.Callback):
def on_train_begin(self, logs={}):
self._data = []
def on_epoch_end(self, batch, logs={}):
x1, x2 = self.validation_data[0], self.validation_data[1]
predictions = np.asarray(siamese.predict([x1, x2]))
data_distribution1 = predictions[np.squeeze(np.where(labels_data[train_index] == 1))]
data_distribution2 = predictions[np.squeeze(np.where(labels_data[train_index] != 1))]
d1=list(data_distribution1)
d2=list(data_distribution2)
data_list = [d1, d2]
robjects.globalenv['data_list'] = robjects.ListVector({f'data_vector{i+1}': robjects.FloatVector(data_vector) for i, data_vector in enumerate(data_list)})
overlap_coef = robjects.r('overlap(data_list)')
#print('overlap:',overlap_coef[0][0])
# y_val = np.argmax(y_val, axis=1)
overlap_indx = np.argmax(overlap_coef[0][0], axis=1)
self._data.append({
'overlap': overlap_indx,
})
return
def get_data(self):
return self._data
def create_model(distance_metric):
input = layers.Input((photo_size, photo_size, channel))
x = layers.Conv2D(32, (3, 3), activity_regularizer=l2(0.001), activation = 'relu')(input)
x = layers.MaxPooling2D(pool_size=(2, 2))(x)
x = layers.Dropout(0.7)(x)
x = layers.Conv2D(64, (3, 3), activation = 'relu',padding='same')(x)
x = layers.MaxPooling2D(pool_size=(2, 2))(x)
x = layers.Dropout(0.7)(x)
x = layers.Conv2D(128, (3, 3), activation = 'relu',padding='same')(x)
x = layers.MaxPooling2D(pool_size=(2, 2))(x)
x = layers.Dropout(0.7)(x)
x = layers.Flatten()(x)
x = layers.Dense(512, activation = 'relu', kernel_initializer='glorot_uniform')(x)
embedding_network = keras.Model(input, x)
input_1 = layers.Input((photo_size, photo_size, channel))
input_2 = layers.Input((photo_size, photo_size, channel))
tower_1 = embedding_network(input_1)
tower_2 = embedding_network(input_2)
if distance_metric == 'cosine':
merge_layer = layers.Lambda(cosine_similarity)([tower_1, tower_2])
else:
merge_layer = layers.Lambda(euclidean_distance)([tower_1, tower_2])
normal_layer = layers.BatchNormalization()(merge_layer)
model = keras.Model(inputs=[input_1, input_2], outputs=normal_layer)
return model
distance = ['euclidean']
siamese = create_model(distance)
siamese.compile(loss=loss, optimizer="SGD", metrics=\["accuracy"\])
metrics = Metrics()
history = siamese.fit(
[x_train_1, x_train_2],
y_train,
batch_size=batch_size,
epochs=epochs,
validation_data=(x_train_1, x_train_2),
callbacks=[metrics]
)
metrics.get_data()
Here are the problems I encountered:
- It does not seem to work, as I got this error message "AssertionError: Could not compute output Tensor("batch_normalization/batchnorm/add_1:0", shape=(None, 1), dtype=float32)"
- Is it possible to directly read the data [x_train_1, x_train_2]] in the Metrci class?
Many thanks!