<jupyter_start><jupyter_text>Loading the model.<jupyter_code>model_arch_name = "vgg16_pretrained_imagenet" # "vgg16_batchnorm"
trained_model_weights = "../trained_models/baseline_4/best_model_dir-acc.ckpt/variables/variables" # "../trained_models/baseline_3/best_model_dir-acc.ckpt/variables/variables"
!ls ../trained_models/baseline_3/best_model_dir-acc.ckpt/variables/
model_factory = ModelFactory()
model = model_factory.get_model(model_arch_name,
                                (224, 224, 3),
                                is_training=False,
                                num_classes=1,
                                learning_rate=0.001)
loss, acc = model.evaluate(test_images,  test_labels, verbose=2)
print("Untrained model, accuracy: {:5.2f}%".format(100*acc))
model.load_weights(trained_model_weights)
loss, acc = model.evaluate(test_images,  test_labels, verbose=2)
print("Trained model, accuracy: {:5.2f}%".format(100*acc))
preds = model.predict(test_images)
preds_1 = model.predict(test_images)
preds_2 = model.predict(test_images)
preds_3 = model.predict(test_images)
np.allclose(preds_1, preds_2), np.allclose(preds_2, preds_3), np.allclose(preds_1, preds_3)
model_pass_preds = model(test_images)
model_pass_preds_1 = model(test_images, training=False)
model_pass_preds_2 = model(test_images, training=True)
np.allclose(preds, model_pass_preds), np.allclose(preds, model_pass_preds_1), np.allclose(preds, model_pass_preds_2)
preds_labels = list(zip(preds[:, 0], test_labels.numpy()))
preds_labels<jupyter_output><empty_output><jupyter_text>Do not load the model directly as the batchnorm layers will also be active.<jupyter_code>direct_model = tf.keras.models.load_model("../trained_models/baseline_3/best_model_dir-acc.ckpt")
direct_preds = direct_model.predict(test_images)
direct_preds_1 = direct_model.predict(test_images)
direct_preds_2 = direct_model.predict(test_images)

np.allclose(direct_preds, direct_preds_1), np.allclose(direct_preds, direct_preds_2)
direct_model = tf.keras.models.load_model("../trained_models/baseline_3/best_model_dir-acc.ckpt")
direct_preds_model_pass_preds = direct_model(test_images)
direct_preds_model_pass_preds_1 = direct_model(test_images, training=False)
direct_preds_model_pass_preds_2 = direct_model(test_images, training=True)

np.allclose(direct_preds_model_pass_preds, direct_preds_model_pass_preds_1), np.allclose(direct_preds_model_pass_preds, direct_preds_model_pass_preds_2)
# Trying by setting the learning phase.
tf.keras.backend.set_learning_phase(0)
direct_model = tf.keras.models.load_model("../trained_models/baseline_3/best_model_dir-acc.ckpt")
direct_preds = direct_model.predict(test_images)
direct_preds_1 = direct_model.predict(test_images)
direct_preds_2 = direct_model.predict(test_images)

np.allclose(direct_preds, direct_preds_1), np.allclose(direct_preds, direct_preds_2)<jupyter_output><empty_output><jupyter_text>We are able to get consistent prediction results!<jupyter_code># Taken from n2cholas, from link: https://github.com/tensorflow/tensorflow/issues/33478.
class LayerWithHooks(tf.keras.layers.Layer):
    def __init__(self, layer: tf.keras.layers.Layer, hooks: List[Callable[[tf.Tensor, tf.Tensor], Optional[tf.Tensor]]] = None):
        super().__init__()
        self._layer = layer
        self._hooks = hooks or []
  
    def call(self, input: tf.Tensor) -> tf.Tensor:
        output = self._layer(input)
        for hook in self._hooks:
            hook_result = hook(input, output)
            if hook_result is not None:
                output = hook_result
        return output
  
    def register_hook(self, hook: Callable[[tf.Tensor, tf.Tensor], Optional[tf.Tensor]]) -> None:
        self._hooks.append(hook)


def get_call_fn(layer: tf.keras.layers.Layer) -> Callable[[tf.Tensor], tf.Tensor]:
    old_call_fn = layer.call
    def call(input: tf.Tensor) -> tf.Tensor:
        output = old_call_fn(input)
        for hook in layer._hooks:
            hook_result = hook(input, output)
            if hook_result is not None:
                output = hook_result
        return output
    return call


class InputOutputSaver:
    def __call__(self, input: tf.Tensor, output: tf.Tensor) -> None:
        self.input = input
        self.output = output


def get_layer_outputs(model, required_layers, input_batch):
    """
    Extract the outputs of intermediate layers provided in required_layers from the model,
    after running the input_batch through the model.
    Returns a dictionary of 'savers'. Access a particular layer output as followed:
        savers[layer_name].output

    """
    # Register hooks and savers.
    savers = {}
    for layer_name in required_layers:
        layer = model.get_layer(layer_name)
        layer._hooks = []
        layer.call = get_call_fn(layer)
        layer.register_hook = lambda hook: layer._hooks.append(hook)
        saver = InputOutputSaver()
        layer.register_hook(saver)
        savers[layer_name] = saver

    model_outputs = model(input_batch)
    return savers

model.summary()
required_layers = ["block1_conv1-Relu", "block3_conv3-Relu", "block5_conv3-Relu"]
savers = get_layer_outputs(model, required_layers, test_images)
savers["block5_conv3-Relu"].output.shape
print(model(test_images))
print(savers[layer_name].output)
layer_out = savers[layer_name].output
layer_out.shape
type(layer_out)<jupyter_output><empty_output><jupyter_text>Testing with sequence mode of data pipeline.<jupyter_code>pipeline_gen = PipelineGenerator(metadata_file,
                                 wellington_images_dir,
                                 is_training=False,
                                 sequence_image_count=3,
                                 label_name="has_animal",
                                 mode="mode_sequence")

test_dataset_raw = pipeline_gen.get_pipeline()
test_dataset_batches = test_dataset_raw.batch(32).prefetch(tf.data.experimental.AUTOTUNE)
num_test_sequences = pipeline_gen.get_size()
print("There are %s test sequences." % num_test_sequences)
for batch in test_dataset_batches:
    print("Loaded first batch to 'batch'.")
    break
test_sequences, labels = batch
test_sequences.shape
a = test_sequences[:, 0, :, :, :]
a.shape
a = [1,2,3]
a.extend([4,5,6])
a
[[]] * 3
import pickle
pickle.dump?
all_pred_labels = [preds_labels, preds_labels, preds_labels]
individual_preds_0 = [x[0] for x in preds_labels]
individual_labels_0 = [x[1] for x in preds_labels]
individual_preds_0
a = np.vstack((individual_preds_0, individual_preds_0, individual_preds_0))
a.shape
b = np.max(a, axis=0)
b
np.allclose?
np.allclose(individual_preds_0, individual_preds_0)
np.allclose(individual_preds_0, individual_preds_0)
b.shape
individual_labels_0
list(zip(b, individual_labels_0))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10499926264562749, 'fraction_numerical': 0.014894558324730866, 'mean_word_length': 4.245166279969064, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 22, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '600530', 'license': 'MIT', 'n_tokens_mistral': 2327, 'n_tokens_neox': 2266, 'n_words': 398, 'path': 'code/inference_pipeline_test.ipynb', 'repo_name': 'bhuvi3/data590_capstone_project', 'size': '6781'}