# ================================================================================ #
# Authors: Fabio Frazao and Oliver Kirsebom #
# Contact: fsfrazao@dal.ca, oliver.kirsebom@dal.ca #
# Organization: MERIDIAN (https://meridian.cs.dal.ca/) #
# Team: Data Analytics #
# Project: ketos #
# Project goal: The ketos library provides functionalities for handling #
# and processing acoustic data and applying deep neural networks to sound #
# detection and classification tasks. #
# #
# License: GNU GPLv3 #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# ================================================================================ #
import tensorflow as tf
from .losses import FScoreLoss
from ketos.neural_networks.dev_utils.export import get_export_function
from zipfile import ZipFile, Path as ZipPath
from pathlib import Path
from shutil import rmtree
import numpy as np
import pandas as pd
import json
import os
import warnings
[docs]class RecipeCompat():
""" Makes a loss function, metric or optimizer compatible with the Ketos recipe format.
The resulting object can be included in a ketos recipe and read by the NNInterface (or it's subclasses)
Args:
recipe_name: str
The name to be used in the recipe
template: constructor
The loss function, metric or optimizer constructor
kwargs
Any keyword arguments to be passed to the constructor (func)
Returns:
A RecipeCompat object
Examples:
>>> # Example Metric
>>> p = tf.keras.metrics.Precision
>>> dec_p = RecipeCompat("precision", p)
>>> # Example Optimizer
>>> opt = tf.keras.optimizers.Adam
>>> dec_opt = RecipeCompat("adam", opt, learning_rate=0.001)
>>> # Example Loss
>>> loss = tf.keras.losses.BinaryCrossentropy
>>> dec_loss = RecipeCompat('binary_crossentropy', loss, from_logits=True)
"""
def __repr__(self):
return "{0} ketos recipe".format(self.recipe_name)
def __init__(self, recipe_name, template, **kwargs):
self.recipe_name = recipe_name
self.args = kwargs
self.template = template
self.instance = self.instantiate_template(**kwargs)
[docs] def instantiate_template(self, **template_kwargs):
args = self.args.copy()
args.update(template_kwargs)
inst = self.template(**args)
return inst
def __call__(self, *args, **kwargs):
result = self.instance(*args, **kwargs)
return result
# Every specific architecture class imports this base class
class NNArch(tf.keras.Model):
""" General class for neural network architectures in the ketos.neural_networks module
This class holds general methods that are common to neural network models.
When implementing new neural network architectures, this class should be inherited.
Example:
The following example shows how to define a new architecture that inherits NNArch.
The architecture can then be used to build an NN interface.
>>> class MLP(NNArch): # doctest: +SKIP
... def __init__(self, n_neurons, activation):
... super(MLP, self).__init__()
...
... self.dense = tf.keras.layers.Dense(n_neurons, activation=activation)
... self.final_node = tf.keras.layers.Dense(1)
...
... def call(self, inputs):
... output = self.call_frontend(inputs)
... output = self.dense(output)
... output = self.final_node(output)
... return output
With the architecture, the interface to the MLP can be created by subclassing NNInterface:
>>> class MLPInterface(NNInterface): # doctest: +SKIP
... def __init__(self, n_neurons, activation, optimizer, loss_function, metrics):
... super(MLPInterface, self).__init__(optimizer, loss_function, metrics)
... self.n_neurons = n_neurons
... self.activation = activation
... self.model = MLP(n_neurons=n_neurons, activation=activation)
"""
def __init__(self):
super(NNArch, self).__init__()
self.frontend = None # Initialize as None
# Adds a frontend layer
def add_frontend(self, frontend):
""" Adds a frontend block to the NN architecture. This block can be composed
by any number of layers. add_frontend can be called at any point during
model development by accessing the model instance.
Args:
frontend: A tf layer or sequence of layers.
"""
if self.frontend == None:
self.frontend = tf.keras.models.Sequential(name="frontend") #Here we are creating an EMPTY frontend block
self.frontend.add(frontend)
# This method will be called during call. If no frontend layer is defined, it will simply return the input
def call_frontend(self, inputs):
""" This method is supposed to be called on the call method of a tenserflow model
pipeline. See specfic architecture implementation for examples.
Args:
inputs: Tensor or list of tensors
A tensor or list of tensors
Returns:
A tensor or list of tensors.
"""
if self.frontend == None:
return inputs
else:
return self.frontend(inputs)
[docs]class NNInterface():
""" General interface for neural network architectures in the ketos.neural_networks module.
This class implements common methods for neural network models and is supposed to be subclassed.
When implementing new neural network architectures, the interface implemented in this class can be inherited.
Args:
optimizer: RecipeCompat object
An instance of the RecipeCompat class wrapping a tensorflow(-compatible) optimizer (e.g.:from tensorflow.keras.optimizers)
loss_function: RecipeCompat object
An instance of the RecipeCompat class wrappinf a tensorflow(-compatible) loss-function (e.g.:from tensorflow.keras.losses)
metrics: list of RecipeCompat objects
A list of instances of the RecipeCompat class wrapping a tensorflow(-compatible) metric (e.g.:from tensorflow.keras.metrics)
Examples:
The following example shows how a newly defined network architecture could use the interface provided by NNInterface.
First, the new architecture must be defined. Here, a simple multi-layer perceptron is defined in the following class.
>>> class MLP(tf.keras.Model): # doctest: +SKIP
... def __init__(self, n_neurons, activation):
... super(MLP, self).__init__()
...
... self.dense = tf.keras.layers.Dense(n_neurons, activation=activation)
... self.final_node = tf.keras.layers.Dense(1)
...
... def call(self, inputs):
... output = self.dense(inputs)
... output = self.dense(output)
... output = self.final_node(output)
... return output
With the architecture, the interface to the MLP can be created by subclassing NNInterface:
>>> from ketos.neural_networks.dev_utils import RecipeCompat, NNInterface # doctest: +SKIP
>>> class MLPInterface(NNInterface): # doctest: +SKIP
... @classmethod
... def _build_from_recipe(cls, recipe, recipe_compat=True):
... n_neurons = recipe['n_neurons'] # take the n_neurons parameter from the recipe instead of using the default
... activation = recipe['activation'] # take the activation parameter from the recipe instead of using the default
...
... if recipe_compat == True:
... optimizer = recipe['optimizer']
... loss_function = recipe['loss_function']
... metrics = recipe['metrics']
...
... else:
... optimizer = cls._optimizer_from_recipe(recipe['optimizer'])
... loss_function = cls._loss_function_from_recipe(recipe['loss_function'])
... metrics = cls._metrics_from_recipe(recipe['metrics'])
...
... instance = cls(n_neurons=n_neurons, activation=activation, optimizer=optimizer, loss_function=loss_function, metrics=metrics)
... return instance
...
... @classmethod
... def _read_recipe_file(cls, json_file, return_recipe_compat=True):
... with open(json_file, 'r') as json_recipe:
... recipe_dict = json.load(json_recipe)
...
... optimizer = cls.optimizer_from_recipe(recipe_dict['optimizer'])
... loss_function = cls.loss_function_from_recipe(recipe_dict['loss_function'])
... metrics = cls.metrics_from_recipe(recipe_dict['metrics'])
...
... if return_recipe_compat == True:
... recipe_dict['optimizer'] = optimizer
... recipe_dict['loss_function'] = loss_function
... recipe_dict['metrics'] = metrics
... else:
... recipe_dict['optimizer'] = cls._optimizer_to_recipe(optimizer)
... recipe_dict['loss_function'] = cls._loss_function_to_recipe(loss_function)
... recipe_dict['metrics'] = cls._metrics_to_recipe(metrics)
...
... recipe_dict['n_neurons'] = recipe_dict['n_neurons'] # read the n_neurons parameter from the recipe file
... recipe_dict['activation'] = recipe_dict['activation'] # read the activation parameter from the recipe file
...
... return recipe_dict
...
... def __init__(self, n_neurons, activation, optimizer, loss_function, metrics):
... super(MLPInterface, self).__init__(optimizer, loss_function, metrics)
... self.n_neurons = n_neurons
... self.activation = activation
... self.model = MLP(n_neurons=n_neurons, activation=activation)
...
... def _extract_recipe_dict(self):
... recipe = {}
... recipe['optimizer'] = self._optimizer_to_recipe(self.optimizer)
... recipe['loss_function'] = self._loss_function_to_recipe(self.loss_function)
... recipe['metrics'] = self._metrics_to_recipe(self.metrics)
... recipe['n_neurons'] = self.n_neurons
... recipe['activation'] = self.activation
... return recipe
"""
valid_optimizers = {'Adadelta':tf.keras.optimizers.Adadelta,
'Adagrad':tf.keras.optimizers.Adagrad,
'Adam':tf.keras.optimizers.Adam,
'Adamax':tf.keras.optimizers.Adamax,
'Nadam':tf.keras.optimizers.Nadam,
'RMSprop':tf.keras.optimizers.RMSprop,
'SGD':tf.keras.optimizers.SGD,
}
valid_losses = {'FScoreLoss':FScoreLoss,
'BinaryCrossentropy':tf.keras.losses.BinaryCrossentropy,
'CategoricalCrossentropy':tf.keras.losses.CategoricalCrossentropy,
'CategoricalHinge':tf.keras.losses.CategoricalHinge,
'CosineSimilarity':tf.keras.losses.CosineSimilarity,
'Hinge':tf.keras.losses.Hinge,
'Huber':tf.keras.losses.Huber,
'KLD':tf.keras.losses.KLD,
'LogCosh':tf.keras.losses.LogCosh,
'MAE':tf.keras.losses.MAE,
'MAPE':tf.keras.losses.MAPE,
'MeanAbsoluteError':tf.keras.losses.MeanAbsoluteError,
'MeanAbsolutePercentageError':tf.keras.losses.MeanAbsolutePercentageError,
'MeanSquaredError':tf.keras.losses.MeanSquaredError,
'MeanSquaredLogarithmicError':tf.keras.losses.MeanSquaredLogarithmicError,
'MSE':tf.keras.losses.MSE,
'MSLE':tf.keras.losses.MSLE,
'Poisson':tf.keras.losses.Poisson,
'SparseCategoricalCrossentropy':tf.keras.losses.SparseCategoricalCrossentropy,
}
valid_metrics = {'Accuracy':tf.keras.metrics.Accuracy,
'AUC':tf.keras.metrics.AUC,
'BinaryAccuracy':tf.keras.metrics.BinaryAccuracy,
'BinaryCrossentropy':tf.keras.metrics.BinaryCrossentropy,
'CategoricalAccuracy':tf.keras.metrics.CategoricalAccuracy,
'CategoricalCrossentropy':tf.keras.metrics.CategoricalCrossentropy,
'CategoricalHinge':tf.keras.metrics.CategoricalHinge,
'CosineSimilarity':tf.keras.metrics.CosineSimilarity,
'FalseNegatives':tf.keras.metrics.FalseNegatives,
'FalsePositives':tf.keras.metrics.FalsePositives,
'Hinge':tf.keras.metrics.Hinge,
'KLDivergence':tf.keras.metrics.KLDivergence,
'LogCoshError':tf.keras.metrics.LogCoshError,
'Mean':tf.keras.metrics.Mean,
'MeanAbsoluteError':tf.keras.metrics.MeanAbsoluteError,
'MeanAbsolutePercentageError':tf.keras.metrics.MeanAbsolutePercentageError,
'MeanIoU':tf.keras.metrics.MeanIoU,
'MeanRelativeError':tf.keras.metrics.MeanRelativeError,
'MeanSquaredError':tf.keras.metrics.MeanSquaredError,
'MeanSquaredLogarithmicError':tf.keras.metrics.MeanSquaredLogarithmicError,
'Poisson':tf.keras.metrics.Poisson,
'Precision':tf.keras.metrics.Precision,
'Recall':tf.keras.metrics.Recall,
'RootMeanSquaredError':tf.keras.metrics.RootMeanSquaredError,
'SensitivityAtSpecificity':tf.keras.metrics.SensitivityAtSpecificity,
'SparseCategoricalAccuracy':tf.keras.metrics.SparseCategoricalAccuracy,
'SparseCategoricalCrossentropy':tf.keras.metrics.SparseCategoricalCrossentropy,
'SparseTopKCategoricalAccuracy':tf.keras.metrics.SparseTopKCategoricalAccuracy,
'SpecificityAtSensitivity':tf.keras.metrics.SensitivityAtSpecificity,
'SquaredHinge':tf.keras.metrics.SquaredHinge,
'Sum':tf.keras.metrics.Sum,
'TopKCategoricalAccuracy':tf.keras.metrics.TopKCategoricalAccuracy,
'TrueNegatives':tf.keras.metrics.TrueNegatives,
'TruePositives':tf.keras.metrics.TruePositives,
}
@classmethod
def _to1hot(cls, class_label, n_classes=2):
""" Create the one hot representation of class_label
Args:
class_label: int
An integer number representing the class label
n_class: int
The number of classes available
Returns:
one_hot: numpy.array
The one hot representation of the class_label in a 1 x n_classes array.
Examples:
>>> NNInterface._to1hot(class_label=0, n_classes=2)
array([1., 0.])
>>> NNInterface._to1hot(class_label=1, n_classes=2)
array([0., 1.])
>>> NNInterface._to1hot(class_label=1, n_classes=3)
array([0., 1., 0.])
>>> NNInterface._to1hot(class_label=1, n_classes=5)
array([0., 1., 0., 0., 0.])
"""
one_hot = np.zeros(n_classes)
one_hot[class_label]=1.0
return one_hot
@classmethod
def _transform_input(cls,input):
""" Transforms a training input to the format expected by the network.
Similar to :func:`NNInterface.transform_batch`, but only acts on the
inputs (not labels). Mostly used for inference, rather than training.
When this interface is subclassed to make new neural_network classes, this
method can be overwritten to accomodate any transformations required. Common
operations are reshaping of an input.
Args:
input:numpy.array
An input instance. Must have 2 or 3 dimensions.
Raises:
ValueError if input does not have 2 or 3 dimensions.
Returns:
tranformed_input:numpy.array
The transformed batch of inputs
Examples:
>>> import numpy as np
>>> # Create a batch of 10 5x5 arrays
>>> batch_of_inputs = np.random.rand(10,5,5)
>>> selected_input = batch_of_inputs[0]
>>> selected_input.shape
(5, 5)
>>> transformed_input = NNInterface._transform_input(selected_input)
>>> transformed_input.shape
(1, 5, 5, 1)
# The input can also have shape=(1,n,m)
>>> selected_input = batch_of_inputs[0:1]
>>> selected_input.shape
(1, 5, 5)
>>> transformed_input = NNInterface._transform_input(selected_input)
>>> transformed_input.shape
(1, 5, 5, 1)
"""
if input.ndim == 2:
transformed_input = input.reshape(1, input.shape[0], input.shape[1], 1)
elif input.ndim == 3:
transformed_input = input.reshape(input.shape[0], input.shape[1], input.shape[2], 1)
else:
raise ValueError("Expected input to have 2 or 3 dimensions, got {} ({}) instead".format(input.ndim, input.shape))
return transformed_input
@classmethod
def _transform_output(cls,output):
""" Transforms the network output
When this interface is subclassed to make new neural_network classes, this method
can be overwritten to accomodate any transformations required. Common operations are
reshaping of an input and returning the class wih the highest score instead of a softmax vector.
Args:
output:np.array
The output neural network output. An array of one or more vectors of float scores
that each add to 1.0.
Returns:
transformed_output:tuple
The transformed output, where the first value is the integer representing the highest
classs in the rank the second is the respective score.
Example:
>>> import numpy as np
>>> output = np.array([[0.2,0.1,0.7]])
>>> NNInterface._transform_output(output)
(array([2]), array([0.7]))
>>> output = np.array([[0.2,0.1,0.7],[0.05,0.65,0.3]])
>>> NNInterface._transform_output(output)
(array([2, 1]), array([0.7 , 0.65]))
"""
max_class = np.argmax(output, axis=-1)
if output.shape[0] == 1:
max_class_conf = output[0][max_class]
transformed_output = (max_class[0], max_class_conf[0])
elif output.shape[0] > 1:
max_class_conf = np.array([output[i][c] for i, c in enumerate(max_class)])
transformed_output = (max_class, max_class_conf)
return transformed_output
@classmethod
def _optimizer_from_recipe(cls, optimizer):
""" Create a recipe-compatible optimizer object from an optimizer dictionary
Used when building a model from a recipe dictionary.
Args:
optimizer: optimizer dictionay
A dictionary with the following keys: {'name':..., 'parameters':{...}}.
The 'name' value must be a valid name as defined in the `valid_optimizers`
class attribute. The 'parameters' value is a dictionary of keyword arguments
to be used when building the optimizer (e.g.: {'learning_rate':0.001, 'momentum': 0.01})
Returns:
built_optimizer:
A recipe-compatible optimizer object.
Raises:
ValueError if the optimizer name is not included in the valid_optimizers
class attribute.
"""
recipe_name = optimizer['recipe_name']
kwargs = optimizer['parameters']
if recipe_name not in cls.valid_optimizers.keys():
raise ValueError("Invalid optimizer name '{}'".format(recipe_name))
built_optimizer = RecipeCompat(recipe_name,cls.valid_optimizers[recipe_name],**kwargs)
return built_optimizer
@classmethod
def _optimizer_to_recipe(cls, optimizer):
""" Create an optimizer dictionary from a recipe-compatible optimizer object
Used when creating a ketos recipe that can be used to recreate the model.
Args:
optimizer: instance of RecipeCompat
An optimizer wrapped in a RecipeCompat object
Returns:
recipe_optimizer: dict
A dictionary with the 'name' and 'parameters' keys.
Raises:
ValueError if the optimizer name is not included in the valid_optimizers
class attribute.
"""
recipe_name = optimizer.recipe_name
kwargs = optimizer.args
if recipe_name not in cls.valid_optimizers.keys():
raise ValueError("Invalid optimizer name '{}'".format(recipe_name))
recipe_optimizer = {'recipe_name':recipe_name, 'parameters':kwargs}
return recipe_optimizer
@classmethod
def _loss_function_from_recipe(cls, loss_function):
""" Create a recipe-compatible loss object from a loss function dictionary
Used when building a model from a recipe dictionary.
Args:
loss_function: loss function dictionay
A dictionary with the following keys: {'name':..., 'parameters':{...}}.
The 'name' value must be a valid name as defined in the `valid_losses` class attribute.
The 'parameters' value is a dictionary of keyword arguments to be used when building the loss_function
(e.g.: {'from_logits':True, 'label_smoothing':0.5})
Returns:
built_loss:
A recipe-compatible loss function object.
Raises:
ValueError if the loss function name is not included in the valid_losses class attribute.
"""
recipe_name = loss_function['recipe_name']
kwargs = loss_function['parameters']
if recipe_name not in cls.valid_losses.keys():
raise ValueError("Invalid loss function name '{}'".format(recipe_name))
built_loss = RecipeCompat(recipe_name, cls.valid_losses[recipe_name],**kwargs)
return built_loss
@classmethod
def _loss_function_to_recipe(cls, loss_function):
""" Create a loss function dictionary from a recipe-compatible loss function object
Used when creating a ketos recipe that can be used to recreate the model.
Args:
loss_function: instance of RecipeCompat
A loss-function wrapped in a RecipeCompat object
Returns:
recipe_optimizer: dict
A dictionary with the 'name' and 'parameters' keys.
Raises:
ValueError if the loss_function name is not included in the valid_losses class attribute.
"""
recipe_name = loss_function.recipe_name
kwargs = loss_function.args
if recipe_name not in cls.valid_losses.keys():
raise ValueError("Invalid loss function name '{}'".format(recipe_name))
recipe_loss = {'recipe_name':recipe_name, 'parameters':kwargs}
return recipe_loss
@classmethod
def _metrics_from_recipe(cls, metrics):
""" Create a list of recipe-compatible metric objects from a metrics dictionary
Used when building a model from a recipe dictionary.
Args:
metrics: list of metrics dictionaries
a list of dictionaries with the following keys: {'name':..., 'parameters':{...}}.
The 'name' value must be a valid name as defined in the `valid_metrics` class attribute.
The 'parameters' value is a dictionary of keyword arguments to be used when building the metrics
(e.g.: {'from_logits':True})
Returns:
built_metrics:
A list of recipe-compatible metric objects.
Raises:
ValueError if any of the metric names is not included in the valid_metrics class attribute.
"""
built_metrics = []
for m in metrics:
recipe_name = m['recipe_name']
kwargs = m['parameters']
if recipe_name not in cls.valid_metrics.keys():
raise ValueError("Invalid metric name '{}'".format(m['recipe_name']))
built_metrics.append(RecipeCompat(recipe_name, cls.valid_metrics[recipe_name], **kwargs))
return built_metrics
@classmethod
def _metrics_to_recipe(cls, metrics):
""" Create a metrics dictionary from a list of recipe-compatible metric objects
Used when creating a ketos recipe that can be used to recreate the model
Args:
metrics: list of RecipeCompat objects
A list of RecipeCompat objects, each wrapping a metric.
Returns:
recipe_metrics: list of dicts
A list dictionaries, each with 'name' and 'parameters' keys.
Raises:
ValueError if any of the metric names is not included in the valid_metrics
class attribute.
"""
recipe_metrics = []
for m in metrics:
if m.recipe_name not in cls.valid_metrics.keys():
raise ValueError("Invalid metric name '{}'".format(m['recipe_name']))
recipe_metrics.append({'recipe_name':m.recipe_name, 'parameters':m.args})
return recipe_metrics
@classmethod
def _read_recipe_file(cls, json_file, return_recipe_compat=True):
""" Read a .json_file containing a ketos recipe and builds a recipe dictionary.
When subclassing NNInterface to create interfaces to new neural networks, this
method can be overwritten to include other recipe fields relevant to the child class.
Args:
json_file:str
Path to the .json file (e.g.: '/home/user/ketos_recupes/my_recipe.json').
return_recipe_compat:bool
If True, the returns a recipe-compatible dictionary (i.e.: where the values
are RecipeCompat objects). If false, returns a recipe dictionary (i.e.: where
the values are name+parameters dictionaries: {'name':..., 'parameters':{...}})
Returns:
recipe_dict: dict
A recipe dictionary that can be used to rebuild a model.
"""
with open(json_file, 'r') as json_recipe:
recipe_dict = json.load(json_recipe)
optimizer = cls._optimizer_from_recipe(recipe_dict['optimizer'])
loss_function = cls._loss_function_from_recipe(recipe_dict['loss_function'])
metrics = cls._metrics_from_recipe(recipe_dict['metrics'])
if return_recipe_compat == True:
recipe_dict['optimizer'] = optimizer
recipe_dict['loss_function'] = loss_function
recipe_dict['metrics'] = metrics
else:
recipe_dict['optimizer'] = cls._optimizer_to_recipe(optimizer)
recipe_dict['loss_function'] = cls._loss_function_to_recipe(loss_function)
recipe_dict['metrics'] = cls._metrics_to_recipe(metrics)
return recipe_dict
@classmethod
def _write_recipe_file(cls, json_file, recipe):
""" Write a recipe dictionary into a .json file
Args:
json_file: str
Path to the .json file (e.g.: '/home/user/ketos_recipes/my_recipe.json').
recipe:dict
A recipe dictionary containing the optimizer, loss function and metrics
in addition to other parameters necessary to build an instance of the neural network.
recipe = {"optimizer": RecipeCompat('adam',tf.keras.optimizers.Adam),
"loss_function":RecipeCompat('categorical_cross_entropy',tf.keras.losses.CategoricalCrossEntropy),
"metrics":[RecipeCompat('categorical_accuracy',tf.keras.metrics.CategoricalAccuracy)],
"another_parameter:32}
"""
with open(json_file, 'w') as json_recipe:
json.dump(recipe, json_recipe)
@classmethod
def _load_model(cls, recipe, weights_path):
""" Load a model given a recipe dictionary and the saved weights.
If multiple versions of the model are available in the folder indicated by weights_path
the latest will be selected.
Args:
recipe: dict
A dictionary containing the recipe
weights_path:str
The path to the folder containing the saved weights.
Saved weights are tensorflow chekpoint. The path should not include the checkpoint
files, only the folder containing them. (e.g.: '/home/user/my_saved_models/model_a/')
"""
instance = cls._build_from_recipe(recipe)
latest_checkpoint = tf.train.latest_checkpoint(weights_path)
instance.model.load_weights(latest_checkpoint).expect_partial()
instance.checkpoint_dir = weights_path
return instance
[docs] @classmethod
def load(cls, model_file, new_model_folder='kt-tmp', overwrite=True, load_audio_repr=False,
replace_top=False, diff_n_classes=None):
""" Load a model from a ketos (.kt) model file.
Args:
model_file:str
Path to the ketos (.kt) file
new_model_folder:str
Path to folder where files associated with the model will be stored.
By default the files will be saved to a folder named 'kt-tmp' created
in the current working directory.
overwrite: bool
If True, the 'new_model_folder' will be overwritten. Default is True.
replace_top: bool
If True, the classification top of the model will be replaced by a new,
untrained one. What is actually replaced (i.e.: what exactly is the "top")
is defined by the architecture. It is usually a block of dense layers with
the appropriate activations. Default is False.
diff_n_classes: int
Only relevant when 'replace_top' is True.
If the new model should have a different number of classes it can be specified
by this parameter. If left as None, the new model will have the same number of
classes as the original model.
load_audio_repr: bool
Depracated. Use ketos.audio.load_audio_representation_from_file instead.
If True, also return a dictionary with the audio representation.
Raises:
FileExistsError: If the 'new_model_folder' already exists and 'overwrite' is False.
Returns:
model_instance:
The loaded model
audio_repr:
Depracated. Use ketos.audio.load_audio_representation_from_file instead.
If load_audio_repr is True, also return a dictionary with the audio representation.
"""
try:
os.makedirs(new_model_folder)
except FileExistsError:
if overwrite == True:
rmtree(new_model_folder)
os.makedirs(new_model_folder)
else:
raise FileExistsError("Ketos needs a new folder for this model. Choose a folder name " \
+ "that does not exist or set 'overwrite' to True to replace the existing folder")
with ZipFile(model_file, 'r') as zip:
zip.extractall(path=new_model_folder)
recipe = cls._read_recipe_file(os.path.join(new_model_folder,"recipe.json"))
model_instance = cls._load_model(recipe, os.path.join(new_model_folder, "checkpoints"))
if replace_top == True:
new_n_classes = recipe['n_classes']
if diff_n_classes is not None:
new_n_classes = diff_n_classes
model_with_new_top = model_instance.model.clone_with_new_top(n_classes=new_n_classes)
model_instance.model = model_with_new_top
if load_audio_repr is True:
from ketos.utils import dev_format_warning
warnings.formatwarning = dev_format_warning
warnings.warn("The load_audio_repr argument is deprecated and will be removed in a future ketos version. Use ketos.audio.load_audio_representation_from_file instead.", category=FutureWarning, stacklevel=2)
from ketos.audio import load_audio_representation_from_file
audio_repr = load_audio_representation_from_file(model_file)
return model_instance, audio_repr
return model_instance
@classmethod
def _build_from_recipe(cls, recipe):
""" Build a model from a recipe dictionary
When subclassing NNInterface to create interfaces for new neural networks, the method
can be overwritten to include all the recipe fields relevant to the new class.
Args:
recipe:dict
A recipe dictionary
"""
optimizer = recipe['optimizer']
loss_function = recipe['loss_function']
metrics = recipe['metrics']
instance = cls(optimizer=optimizer, loss_function=loss_function, metrics=metrics)
return instance
[docs] @classmethod
def build(cls, recipe_file):
""" Build a model from a recipe file
Args:
recipe:str
path to .json file containing the recipe
Returns:
instance:
An instance of the neural network interface
"""
recipe = cls._read_recipe_file(recipe_file)
instance = cls._build_from_recipe(recipe)
return instance
def __init__(self, optimizer, loss_function, metrics):
self.optimizer = optimizer
self.loss_function = loss_function
self.metrics = metrics
self.model = None
self._log_dir = None
self._checkpoint_dir = None
self._tensorboard_callback = None
self._train_generator = None
self._val_generator = None
self._test_generator = None
self._train_loss = tf.keras.metrics.Mean(name='train_loss')
self._val_loss = tf.keras.metrics.Mean(name='val_loss')
self._train_metrics = []
self._val_metrics = []
for m in self.metrics:
self._train_metrics.append(m.instantiate_template(name='train_' + m.recipe_name))
self._val_metrics.append(m.instantiate_template(name='val_' + m.recipe_name))
self._early_stopping_monitor = {"metric": 'val_loss',
"decreasing": True,
"period":10,
"min_epochs": 5,
"max_epochs": None,
"delta" : 0.1,
"baseline":0.5}
def _extract_recipe_dict(self):
""" Create a recipe dictionary from a neural network instance.
The resulting recipe contains all the fields necessary to build the same network
architecture used by the instance calling this method.
When subclassing NNInterface to create interfaces for new neural networks, this method
can be overwritten to match the recipe fields expected by :func:`build_from_recipe`
Returns:
recipe:dict
A dictionary containing the recipe fields necessary to build the same network
architecture used by the instance calling this method
"""
recipe = {}
recipe['optimizer'] = self._optimizer_to_recipe(self.optimizer)
recipe['loss_function'] = self._loss_function_to_recipe(self.loss_function)
recipe['metrics'] = self._metrics_to_recipe(self.metrics)
return recipe
[docs] def save_recipe_file(self, recipe_file):
""" Creates a recipe from an existing neural network instance and save it into a .json file.
This method is a convenience method that wraps :func:`_extract_recipe_dict` and :func:`_write_recipe_file`
Args:
recipe_file:str
Path to .json file in which the recipe will be saved.
"""
recipe = self._extract_recipe_dict()
self._write_recipe_file(json_file=recipe_file, recipe=recipe)
[docs] def save(self, output_name=None, checkpoint_name=None, **kwargs):
""" Save the trained model to a file.
The model can be saved to three different formats:
* ketos (.kt)
* protobuf (.pb)
* ketos-protobuf (.ktpb), compatible with PAMGuard
The format will be inferred from the extension given to the output file, as
follows
* .kt (or any other extension): ketos
* .pb: protobuf
* .ktpb: ketos-protobuf (PAMGuard compatible)
The save method accepts different optional arguments depending on the format.
See the documentation of export functions for more information:
* ketos: :func:`export.export_to_ketos`
* protobuf: :func:`export.export_to_protobuf`
* ketos-protobuf: :func:`export.export_to_ketos_protobuf`
Args:
output_name: str
Path to the output file
checkpoint_name: str
The name of the checkpoint to be saved (e.g.:cp-0015.ckpt).
If None, will use the latest checkpoints
audio_repr: dict
Audio representation dictionary. Optional for the ketos format (.kt),
required for the ketos-protobuf (.ktpb) format, not applicable for
the protobuf (.pb) format. It is also possible to specify the path
to a json file containing the audio representation.
"""
# attempt to load weights from checkpoint
if isinstance(self.checkpoint_dir, str) and os.path.isdir(self.checkpoint_dir):
if checkpoint_name == None:
latest = tf.train.latest_checkpoint(self.checkpoint_dir)
checkpoint_name = os.path.basename(latest)
self.model.load_weights(os.path.join(self.checkpoint_dir, checkpoint_name))
else:
print(f"Warning: model.checkpoint_dir does not point to a valid directory ({self.model.checkpoint_dir})")
f = get_export_function(output_name)
f(model=self, output_name=output_name, checkpoint_name=checkpoint_name, **kwargs)
@property
def train_generator(self):
return self._train_generator
@train_generator.setter
def train_generator(self, train_generator):
""" Link a batch generator (used for training) to this instance.
Args:
train_generator: instance of BatchGenerator
A batch generator that provides training data during the training loop
"""
self._train_generator = train_generator
@property
def val_generator(self):
return self._val_generator
@val_generator.setter
def val_generator(self, val_generator):
""" Link a batch generator (used for validation) to this instance.
Args:
val_generator: instance of BatchGenerator
A batch generator that provides validation data during the training loop
"""
self._val_generator = val_generator
@property
def test_generator(self):
return self._test_generator
@test_generator.setter
def test_generator(self, test_generator):
""" Link a batch generator (used for testing) to this instance.
Args:
test_generator: instance of BatchGenerator
A batch generator that provides test data
"""
self._test_generator = test_generator
@property
def log_dir(self):
return self._log_dir
@log_dir.setter
def log_dir(self, log_dir):
""" Defines the directory where tensorboard log files and .csv log files can be stored
Note: Creates folder if it does not exist. If it already exists, this method does not delete any content.
Args:
log_dir:str
Path to the directory
"""
self._log_dir = log_dir
os.makedirs(self._log_dir, exist_ok=True)
[docs] def add_learning_rate_scheduler(self, scheduler_type="PiecewiseConstantDecay",**kwargs):
""" Add a learning rate scheduler to the current neural network interface.
Notes: - This method must be called before training and after an optimizer has been defined.
- Keep in mind that in the schedulers a 'step' corresponds to each time the optmization algorithm is called.
Normally this means that each batch is a step (i.e.: each epoch has several steps).
Args:
scheduler_type:str
One of four scheduler types:
`PiecewiseConstantDecay <https://www.tensorflow.org/api_docs/python/tf/keras/optimizers/schedules/PiecewiseConstantDecay>`_ ,
`ExponentialDecay, <https://www.tensorflow.org/api_docs/python/tf/keras/optimizers/schedules/ExponentialDecay>`_
`InverseTimeDecay <https://www.tensorflow.org/api_docs/python/tf/keras/optimizers/schedules/InverseTimeDecay>`_ , or
`PolynomialDecay <https://www.tensorflow.org/api_docs/python/tf/keras/optimizers/schedules/PolynomialDecay>`_.
Each type also requires additional arguments, as detailed below.
PiecewiseConstantDecay:
boundaries:list
A list of Tensors or ints or floats with strictly increasing entries, and with all elements
having the same type as the optimizer step.
values:list
A list of Tensors or floats or ints that specifies the values for the intervals defined by
boundaries. It should have one more element than boundaries, and all elements should have the same type.
ExponentialDecay:
initial_learning_rate:float
The initial learning rate.
decay_steps: int
The decay steps (must be positive). In each step, the learning rate is calculated as
initial_learning_rate * decay_rate ^ (step / decay_steps)
decay_rate:float
The decay rate.
staircase:bool
If True decay the learning rate at discrete intervals
InverseTimeDecay:
initial_learning_rate:float
The initial learning rate.
decay_steps:int
How often to apply decay.
decay_rate:float
The decay rate.
staircase:bool
Whether to apply decay in a discrete staircase, as opposed to continuous, fashion.
PolynomialDecay:
initial_learning_rate:float
The initial learning rate.
decay_steps:int
Must be positive. The number of steps in which the end_learning_rate should be reached.
end_learning_rate:float
The minimal end learning rate.
power:float
The power of the polynomial. Defaults to linear, 1.0.
cycle:bool
Whether or not it should cycle beyond decay_steps.
"""
scheduler_types = {"PiecewiseConstantDecay":tf.keras.optimizers.schedules.PiecewiseConstantDecay,
"ExponentialDecay":tf.keras.optimizers.schedules.ExponentialDecay,
"InverseTimeDecay":tf.keras.optimizers.schedules.InverseTimeDecay,
"PolynomialDecay":tf.keras.optimizers.schedules.PolynomialDecay}
assert scheduler_type in scheduler_types.keys(), ValueError("{0} is not a valid scheduler type. Accepted values are: {1}".format(scheduler_type, list(scheduler_types.keys())))
scheduler = scheduler_types[scheduler_type](**kwargs)
self.optimizer.instance = self.optimizer.instantiate_template(learning_rate = scheduler)
@property
def early_stopping_monitor(self):
r""" Sets an early stopping monitor.
An early stopping monitor is a dictionary specifying
how a target metric should be monitored during training.
When the conditions are met, the training loop will be stopped
and the model will keep the set of weights that resulted in the
best value for the target metric.
The following parameters are expected:
metric: str
The name of the metric to be monitored. It must be one the metrics
defined when creating a neural network interface, either through
the 'metrics' argument of the class constructor or the 'metrics' field in a recipe.
The name must be prefixed by 'train\_' or 'val\_', indicating weather the training or
validation metric should be monitored.
decreasing: bool,
If True, improvements will be indicated by a decrease in the metric value during training.
If False, improvements will be defined as an increase in the metric value.
period: int
The number of epochs the training loop will continue without any improvement before training is stopped.
Example: If period is 5, training will stop if the target metric does not improve for 5 consecutive epochs.
min_epochs: int
The number of epochs to train for before starting to monitor.
delta: float
The minimum difference between the current metric value and the best
value recorded since the monitor started. An improvement is only considered if
(current value - best value) <= delta (if decreasing is True) or
(current value - best value) >= delta (if decreasing is False)
baseline: float or None
If this value is reached, training will stop immediately.
If None, this parameter is ignored.
"""
return self._early_stopping_monitor
@early_stopping_monitor.setter
def early_stopping_monitor(self, parameters):
valid_metrics = [m.name for m in self._train_metrics] + [m.name for m in self._val_metrics] + [self._train_loss.name] + [self._val_loss.name]
assert parameters['metric'] in valid_metrics, "Invalid metric. Must be one of {}".format(str(valid_metrics))
self._early_stopping_monitor = parameters
@property
def checkpoint_dir(self):
return self._checkpoint_dir
@checkpoint_dir.setter
def checkpoint_dir(self, checkpoint_dir):
""" Defines the directory where tensorflow checkpoint files can be stored
Args:
log_dir:str
Path to the directory
"""
self._checkpoint_dir = checkpoint_dir
os.makedirs(self._checkpoint_dir, exist_ok=True)
def _set_tensorboard_callback(self):
""" Link tensorboard callback to this instances model, so that tensorboard logs can be saved
"""
self.tensorboard_callback = tf.keras.callbacks.TensorBoard(log_dir=self.log_dir, histogram_freq=1)
self.tensorboard_callback.set_model(self.model)
def _print_metrics(self, metric_values):
""" Print the metric values to the screen.
This method can be overwritten to customize the message.
Args:
metric_value:list
List of metric values. Usually returned by model.train_on_batch or generated by custom metrics.
"""
message = [self.model.metrics_names[i] + ": {:.3f} ".format(metric_values[i]) for i in range(len(self.model.metrics_names))]
print(''.join(message))
def _name_logs(self, logs, prefix="train_"):
""" Attach the prefix string to each log name.
Args:
logs:list
List of log values
prefix:str
Prefix to be added to the logged metric name
Returns:
named_log: zip
A zip iterator that yields a tuple: (prefix + log metric name, log value)
"""
named_logs = {}
for l in zip(self.metrics_names, logs):
named_logs[prefix+l[0]] = l[1]
return named_logs
@tf.function
def _train_step(self, inputs, labels):
with tf.GradientTape() as tape:
predictions = self.model(inputs, training=True)
loss = self.loss_function.instance(labels, predictions)
gradients = tape.gradient(loss, self.model.trainable_variables)
self.optimizer.instance.apply_gradients(zip(gradients, self.model.trainable_variables))
self._train_loss(loss)
for train_metric in self._train_metrics:
train_metric(labels, predictions)
@tf.function
def _val_step(self,inputs, labels):
predictions = self.model(inputs, training=False)
v_loss = self.loss_function.instance(labels, predictions)
self._val_loss(v_loss)
for val_metric in self._val_metrics:
val_metric(labels, predictions)
def _get_metric_value(self, metric_name):
if metric_name == 'train_loss':
return self._train_loss.result()
elif metric_name == "val_loss":
return self._val_loss.result()
elif metric_name.startswith("train_"):
for m in self._train_metrics:
if m.name == metric_name:
return m.result()
elif metric_name.startswith("val_"):
for m in self._val_metrics:
if m.name == metric_name:
return m.result()
[docs] def train_loop(self, n_epochs, verbose=True, validate=True, log_tensorboard=False,
tensorboard_metrics_name='tensorboard_metrics', log_csv=False, csv_name='log.csv',
checkpoint_freq=5, early_stopping=False):
""" Train the model
Typically, before starting the training loop, a few steps will already have been taken:
>>> #Set the batch generator for the training data
>>> model.train_generator = my_train_generator # doctest: +SKIP
>>> #Set the batch generator for the validation data (optional; only if the validate option is set to True)
>>> model.val_generator = my_val_generator # doctest: +SKIP
>>> # Set the log_dir
>>> model.log_dir = "./my_logs" # doctest: +SKIP
>>> # Set the checkpoint_dir
>>> model.checkpoint_dir = "./my_checkpoints" # doctest: +SKIP
>>> model.train_loop(n_epochs=50) # doctest: +SKIP
Args:
n_epochs:int
The number of epochs (i.e.: passes through the entire training dataset, as defined
by the train_generator attribute)
verbose:bool
If True, print summary metrics at the end of each epoch
validate:bool
If True, evaluate the model on the validation data (as defined by the val_generator
attribute) at the end of each epoch
log_tensorboard:bool
If True, log the training and validation (if validate is True) metrics in the tensoraboard
format.
See 'tensorboard_metrics_name' below.
tensorboard_metrics_name:string
The name of the directory where the tensorboard metrics will be saved. This directory
will be created within the path specified by the log_dir attribute.
Default is 'tensorboard_metrics'. Only relevant if log_tensorboard is True.
log_csv:bool
If True, log the training and validation (if validate is True) metrics in a csv file
(see csv_name).
The csv will have the following columns,
* epoch: the epoch number, starting from 1
* loss: the value of the loss metric
* dataset: 'train' or 'val' (only when validate is True)
In addition, each metric defined by the metrics attribute will be added as a column.
Example:
epoch,loss,dataset,CategoricalAccuracy,Precision,Recall
1,0.353,train,0.668,0.653,0.796
1,0.560,val,0.448,0.448,1.0
...
50,0.053,train,0.968,0.953,0.986
50,0.160,val,0.848,0.748,0.838
checkpoint_freq:int
The frequency (in epochs) with which checkpoints (i.e.: the model weights) will be
saved to the directory defined by the checkpoint_dir attribute.
This number should not exceed 'n_epochs'. If early_stopping (see below) is used,
this parameter is ignored and every epoch is saved as a checkpoint
so that the model state can be set to whichever chekpopint is selected by the
early stopping monitor.
If this value is <=0, no checkpoints will be saved.
early_stopping: bool
If False, train for n_epochs. If True, use the early_stop_monitor to stop training when
the conditions defined there are reached (or n_epochs is reached, whichever
happens first).
When training is stopped by the early stopping monitor, an attribute 'last_epoch_with_improvement'
will be added to the object.
This attribute holds the epoch number (starting from zero) that had the best metric value
based on the conditions set by the early_stopping_monitor.
The 'last_epoch_with_improvement' reflects the current state of the weights when trained
is stopped early.
"""
assert checkpoint_freq <= n_epochs, "The checkpoint frequency ({0}) cannot be greater than the number of epochs ({1})".format(checkpoint_freq, n_epochs)
if log_csv == True:
column_names = ['epoch', 'loss', 'dataset'] + [ m.recipe_name for m in self.metrics]
log_csv_df = pd.DataFrame(columns = column_names)
if log_tensorboard == True:
tensorboard_writer = tf.summary.create_file_writer(os.path.join(self._log_dir, tensorboard_metrics_name))
tensorboard_writer.set_as_default()
if early_stopping == True:
early_stopping_metric = []
best_metric_value = None
self.last_epoch_with_improvement = 0
epochs_without_improvement = 0
should_stop = False
checkpoint_freq = 1
for epoch in range(n_epochs):
#Reset the metric accumulators
self._train_loss.reset_states()
for train_metric in self._train_metrics:
train_metric.reset_states()
self._val_loss.reset_states()
for val_metric in self._val_metrics:
val_metric.reset_states()
for train_batch_id in range(self._train_generator.n_batches):
train_X, train_Y = next(self._train_generator)
self._train_step(train_X, train_Y)
if verbose == True:
print("\n====================================================================================")
print("Epoch: {} \ntrain_loss: {}".format(epoch + 1, self._train_loss.result()))
print("".join([m.name + ": {:.3f} ".format(m.result().numpy()) for m in self._train_metrics]))
if log_csv == True:
log_row = [epoch + 1, self._train_loss.result().numpy(), "train"]
log_row = log_row + [m.result().numpy() for m in self._train_metrics]
log_csv_df = pd.concat([log_csv_df, pd.DataFrame([log_row], columns = log_csv_df.columns)], ignore_index=True)
if log_tensorboard == True:
tf.summary.scalar('train_loss', data=self._train_loss.result().numpy(), step=epoch)
for m in self._train_metrics:
tf.summary.scalar(m.name, data=m.result().numpy(), step=epoch)
if validate == True:
for val_batch_id in range(self._val_generator.n_batches):
val_X, val_Y = next(self._val_generator)
self._val_step(val_X, val_Y)
if verbose == True:
print("val_loss: {}".format(self._val_loss.result()))
print("".join([m.name + ": {:.3f} ".format(m.result().numpy()) for m in self._val_metrics]))
if log_csv == True:
log_row = [epoch + 1, self._val_loss.result().numpy(), "val"]
log_row = log_row + [m.result().numpy() for m in self._val_metrics]
log_csv_df = pd.concat([log_csv_df, pd.DataFrame([log_row], columns = log_csv_df.columns)], ignore_index=True)
if log_tensorboard == True:
tf.summary.scalar('val_loss', data=self._val_loss.result().numpy(), step=epoch)
for m in self._val_metrics:
tf.summary.scalar(m.name, data=m.result().numpy(), step=epoch)
if verbose == True:
print("\n====================================================================================")
if (checkpoint_freq > 0) and ((epoch + 1) % checkpoint_freq == 0):
checkpoint_name = "cp-{:04d}.ckpt".format(epoch + 1)
self.model.save_weights(os.path.join(self._checkpoint_dir, checkpoint_name))
if early_stopping == True:
print("\nFocus metric", self._early_stopping_monitor['metric'])
current_early_stopping_metric = (self._get_metric_value(self._early_stopping_monitor['metric']))
if best_metric_value is None:
best_metric_value = current_early_stopping_metric
if epoch >= self._early_stopping_monitor['min_epochs']:
if self._early_stopping_monitor['decreasing'] == True:
if (self._early_stopping_monitor['baseline'] is not None) and (current_early_stopping_metric <= self._early_stopping_monitor['baseline']):
should_stop = True
self.last_epoch_with_improvement = epoch
else:
current_delta = current_early_stopping_metric - best_metric_value
if current_delta < 0 and (abs(current_delta) > self._early_stopping_monitor['delta']): #metric is decreasing = improvement
epochs_without_improvement = 0
self.last_epoch_with_improvement = epoch
best_metric_value = current_early_stopping_metric
else: # metric is not decreasing = no improvement
epochs_without_improvement += 1
elif self._early_stopping_monitor['decreasing'] == False:
if (self._early_stopping_monitor['baseline'] is not None) and (current_early_stopping_metric >= self._early_stopping_monitor['baseline']):
should_stop = True
self.last_epoch_with_improvement = epoch
else:
current_delta = current_early_stopping_metric - best_metric_value
print("abs:", abs(current_delta))
if current_delta > 0 and (abs(current_delta) > self._early_stopping_monitor['delta']): #metric is increasing = improvement
epochs_without_improvement = 0
self.last_epoch_with_improvement = epoch
best_metric_value = current_early_stopping_metric
else: # metric is not increasing = no improvement
epochs_without_improvement += 1
print("\nEpochs without improvement:", epochs_without_improvement)
print("\nCurrent value:", current_early_stopping_metric)
print("\nBest value:", best_metric_value)
if epochs_without_improvement > self._early_stopping_monitor['period']:
should_stop = True
if should_stop == True:
break
if log_csv == True:
log_csv_df.to_csv(os.path.join(self._log_dir, csv_name))
if early_stopping:
last_checkpoint_with_improvement = "cp-{:04d}.ckpt".format(self.last_epoch_with_improvement + 1)
self.model.load_weights(os.path.join(self.checkpoint_dir, last_checkpoint_with_improvement))
return {'checkpoint_name':last_checkpoint_with_improvement}
[docs] def run_on_test_generator(self, return_raw_output=False, compute_val_metrics=True, verbose=True, return_metrics=False):
""" Run the model on the test generator
Args:
return_raw_output:bool
If False, the model output will be transformed by :func:`transform_output`.
If True, the model output will be returned without any modifications.
compute_val_metrics: bool
If True, compute the same metrics used for validation when running the model on the test generator
verbose: bool
If True and compute_val_metrics is also true, print the results.
return_metrics: bool
Only relevant if 'compute_val_metrics=True'.
If True, return a dictionary with the metrics
Returns:
output
The corresponding batch of model outputs
metrics
A dictionary with the validation metrics.
Only returned if both 'compute_val_metrics' and 'return_metrics' are True.
"""
if compute_val_metrics:
self._val_loss.reset_states()
for val_metric in self._val_metrics:
val_metric.reset_states()
predictions = []
for batch_id in range(self._test_generator.n_batches):
X, Y = next(self._test_generator)
if compute_val_metrics: self._val_step(X, Y)
predictions.append(self.model(X, training=False))
predictions = np.array(predictions)
if return_raw_output == False:
reshaped_predictions = predictions.reshape(-1, predictions.shape[2])
predictions = self._transform_output(reshaped_predictions)
if compute_val_metrics == True:
if verbose == True:
print("loss: {}".format(self._val_loss.result()))
print("".join([m.name.split('val_')[1] + ": {:.3f} ".format(m.result().numpy()) for m in self._val_metrics]))
if return_metrics == True:
metrics_dict ={m.name.split('val_')[1]:m.result().numpy() for m in self._val_metrics}
metrics_dict["loss"] = self._val_loss.result()
return predictions, metrics_dict
return predictions
[docs] def run_on_instance(self, input, return_raw_output=False, transform_input=True):
""" Run the model on one input
Args:
input: numpy.array
The input in the shape expected by :func:`transform_input`
return_raw_output:bool
If False, the model output will be transformed by :func:`transform_output`.
If true, the model output will be returned without any modifications.
transform_input:bool
If True, the input is transformed by the interface's :func:`transform_input` method
Returns:
output
The model output
"""
if transform_input == True:
input = self.transform_batch(input)
output = self.model.predict(input)
if not return_raw_output:
return self._transform_output(output)
else:
return output
[docs] def run_on_batch(self, input_batch, return_raw_output=False, transform_input=True):
""" Run the model on a batch of inputs
Args:
input_batch: numpy.array
The batch of inputs
transform_input:bool
If True, the input_batch is transformed by the interface's :func:`transform_input` method
return_raw_output:bool
If False, the model output will be transformed by :func:`transform_output`.
If true, the model output will return the output given by the tensorflow predict_on_batch function.
In this case, the return is a numpy array(s) of prediction scores with shape (batch_size, n_classes).
Returns:
output
The corresponding batch of model outputs
"""
if transform_input == True:
input_batch = self.transform_batch(input_batch)
output = self.model.predict_on_batch(input_batch)
if not return_raw_output:
return self._transform_output(output)
else:
return output