Reference

Dataset

tf_autoencoder_dataset(iterator)

Create an autoencoder tf.data.Dataset from the iterator

The dataset is constructed from a generator

Parameters:

Name	Type	Description	Default
iterator	WindowedDatasetIterator	The data iterator	required

Returns:

Type	Description
Dataset	A tensorflow dataset

Source code in ceruleo/models/keras/dataset.py

def tf_autoencoder_dataset(iterator: WindowedDatasetIterator) -> tf.data.Dataset:
    """
    Create an autoencoder tf.data.Dataset from the iterator

    The dataset is constructed from a generator

    Parameters:
        iterator: The data iterator

    Returns:
        A tensorflow dataset
    """
    n_features = iterator.n_features

    def gen_train():
        for X, y, sw in iterator:
            yield X, X, sw

    a = tf.data.Dataset.from_generator(
        gen_train,
        (tf.float32, tf.float32, tf.float32),
        (
            tf.TensorShape([iterator.window_size, n_features]),
            tf.TensorShape([iterator.window_size, n_features]),
            tf.TensorShape([1]),
        ),
    )

    return a

tf_regression_dataset(iterator)

Create a forecast tf.data.Dataset from the iterator

The dataset is constructed from a generator

Parameters:

Name	Type	Description	Default
iterator	WindowedDatasetIterator	The data iterator	required

Returns:

Type	Description
Dataset	A tensorflow dataset

Source code in ceruleo/models/keras/dataset.py

def tf_regression_dataset(iterator: WindowedDatasetIterator) -> tf.data.Dataset:
    """
    Create a forecast tf.data.Dataset from the iterator

    The dataset is constructed from a generator

    Parameters:
        iterator: The data iterator

    Returns:
        A tensorflow dataset
    """
    n_features = iterator.n_features

    def generator_function():
        for X, y, sw in iterator:
            yield X, y, sw

    a = tf.data.Dataset.from_generator(
        generator_function,
        output_signature=(
            tf.TensorSpec(shape=(iterator.window_size, n_features), dtype=tf.float32),
            tf.TensorSpec(shape=(iterator.horizon, 1), dtype=tf.float32),
            tf.TensorSpec(shape=(1), dtype=tf.float32),
        ),
    )

    return a

tf_seq_to_seq_dataset(iterator)

Create a sequence to sequence tf.data.Dataset from the iterator

The dataset is constructed from a generator

Parameters:

Name	Type	Description	Default
iterator	WindowedDatasetIterator	The data iterator	required

Returns:

Type	Description
Dataset	A tensorflow dataset

Source code in ceruleo/models/keras/dataset.py

def tf_seq_to_seq_dataset(iterator: WindowedDatasetIterator) -> tf.data.Dataset:
    """
    Create a sequence to sequence tf.data.Dataset from the iterator

    The dataset is constructed from a generator

    Parameters:
        iterator: The data iterator

    Returns:
        A tensorflow dataset
    """
    n_features = iterator.n_features

    def generator_function():
        for X, y, sw in iterator:
            yield X, y, sw

    a = tf.data.Dataset.from_generator(
        generator_function,
        output_signature=(
            tf.TensorSpec(shape=(iterator.window_size, n_features), dtype=tf.float32),
            tf.TensorSpec(
                shape=(iterator.window_size, iterator.horizon), dtype=tf.float32
            ),
            tf.TensorSpec(shape=(1), dtype=tf.float32),
        ),
    )

    return a

Callbacks

PredictionCallback

Bases: Callback

Generate a plot after each epoch with the predictions

Parameters:

Name	Type	Description	Default
model		The model used predict	required
output_path	Path	Path of the output image	required
dataset	Dataset	The dataset that want to be plotted	required

Source code in ceruleo/models/keras/callbacks.py

class PredictionCallback(Callback):
    """
    Generate a plot after each epoch with the predictions

    Parameters:
        model: The model used predict
        output_path: Path of the output image
        dataset: The dataset that want to be plotted
    """

    def __init__(
        self,
        output_path: Path,
        dataset: tf.data.Dataset,
        units: str = "",
        filename_suffix: str = "",
    ):
        super().__init__()
        self.output_path = output_path
        self.dataset = dataset
        self.units = units
        self.suffix = filename_suffix
        if len(filename_suffix) > 0:
            self.output_path = self.output_path.with_stem(
                filename_suffix + "_" + self.output_path.stem
            )

    def on_epoch_end(self, epoch, logs={}):
        y_pred = self.model.predict(self.dataset)
        y_true = true_values(self.dataset)
        ax = plot_predictions(
            PredictionResult("Model", y_true, y_pred),
            figsize=(17, 5),
            units=self.units,
        )
        ax.legend()

        ax.figure.savefig(self.output_path, dpi=ax.figure.dpi)

        plt.close(ax.figure)

Losses

AsymmetricLossPM

Bases: LossFunctionWrapper

Customizable Asymmetric Loss Functions for Machine Learning-based Predictive Maintenance

Ehrig, L., Atzberger, D., Hagedorn, B., Klimke, J., & Döllner, J. (2020, October). Customizable Asymmetric Loss Functions for Machine Learning-based Predictive Maintenance. In 2020 8th International Conference on Condition Monitoring and Diagnosis (CMD) (pp. 250-253). IEEE.

Reference

Parameters:

Name	Type	Description	Default
theta_l	float	Linear to exponential change point for overpredictions	required
alpha_l	float	Quadratic term parameters for overpredictions	required
gamma_l	float	Exponential term parameters for overpredictions	required
theta_r	float	Linear to exponential change point for underpredictions	required
alpha_r	float	Quadratic term parameters for underpredictions	required
gamma_r	float	Exponential term parameters for underpredictions	required
relative_weight	bool	Wether to use weigthing relative to the RUL	True

Source code in ceruleo/models/keras/losses.py

class AsymmetricLossPM(LossFunctionWrapper):
    """
    Customizable Asymmetric Loss Functions for Machine Learning-based Predictive Maintenance


    Ehrig, L., Atzberger, D., Hagedorn, B., Klimke, J., & Döllner, J. (2020, October).
    Customizable Asymmetric Loss Functions for Machine Learning-based Predictive Maintenance.
    In 2020 8th International Conference on Condition Monitoring and Diagnosis
    (CMD) (pp. 250-253). IEEE.

    [Reference](https://ieeexplore.ieee.org/document/9287246)

    Parameters:
        theta_l: Linear to exponential change point for overpredictions
        alpha_l: Quadratic term parameters for overpredictions
        gamma_l: Exponential term parameters for overpredictions
        theta_r: Linear to exponential change point for underpredictions
        alpha_r: Quadratic term parameters for underpredictions
        gamma_r: Exponential term parameters for underpredictions
        relative_weight: Wether to use weigthing relative to the RUL
    """

    def __init__(
        self,
        *,
        theta_l: float,
        alpha_l: float,
        gamma_l: float,
        theta_r: float,
        alpha_r: float,
        gamma_r: float,
        relative_weight: bool = True,
        name="asymmetric_loss_pm",
    ):
        super().__init__(
            asymmetric_loss_pm,
            theta_l=theta_l,
            alpha_l=alpha_l,
            gamma_l=gamma_l,
            theta_r=theta_r,
            alpha_r=alpha_r,
            gamma_r=gamma_r,
            relative_weight=relative_weight,
            name=name,
        )

asymmetric_loss_pm(y_true, y_pred, *, theta_l, alpha_l, gamma_l, theta_r, alpha_r, gamma_r, relative_weight=True)

Customizable Asymmetric Loss Functions for Machine Learning-based Predictive Maintenance

Ehrig, L., Atzberger, D., Hagedorn, B., Klimke, J., & Döllner, J. (2020, October). Customizable Asymmetric Loss Functions for Machine Learning-based Predictive Maintenance. In 2020 8th International Conference on Condition Monitoring and Diagnosis (CMD) (pp. 250-253). IEEE.

Reference

Parameters:

Name	Type	Description	Default
y_true		True RUL values	required
y_pred		Predicted RUL values	required
theta_l		Linear to exponential change point for overpredictions (Positive)	required
alpha_l		Quadratic term parameters for overpredictions	required
gamma_l		Exponential term parameters for overpredictions	required
theta_r		Linear to exponential change point for underpredictions	required
alpha_r		Quadratic term parameters for underpredictions	required
gamma_r		Exponential term parameters for underpredictions	required
relative_weight	bool	Wether to use weigthing relative to the RUL	True

Returns:

Name	Type	Description
l	float	the loss computed

Source code in ceruleo/models/keras/losses.py

def asymmetric_loss_pm(
    y_true,
    y_pred,
    *,
    theta_l,
    alpha_l,
    gamma_l,
    theta_r,
    alpha_r,
    gamma_r,
    relative_weight: bool = True,
) -> float:
    """
    Customizable Asymmetric Loss Functions for Machine Learning-based Predictive Maintenance


    Ehrig, L., Atzberger, D., Hagedorn, B., Klimke, J., & Döllner, J. (2020, October).
    Customizable Asymmetric Loss Functions for Machine Learning-based Predictive Maintenance.
    In 2020 8th International Conference on Condition Monitoring and Diagnosis
    (CMD) (pp. 250-253). IEEE.

    [Reference](https://ieeexplore.ieee.org/document/9287246)

    Parameters:
        y_true: True RUL values
        y_pred: Predicted RUL values
        theta_l: Linear to exponential change point for overpredictions (Positive)
        alpha_l: Quadratic term parameters for overpredictions
        gamma_l: Exponential term parameters for overpredictions
        theta_r: Linear to exponential change point for underpredictions
        alpha_r: Quadratic term parameters for underpredictions
        gamma_r: Exponential term parameters for underpredictions
        relative_weight: Wether to use weigthing relative to the RUL

    Returns:
        l: the loss computed
    """

    errors = y_true - y_pred
    weight = tf.abs(errors) / (
        tf.clip_by_value(y_true, clip_value_min=0.9, clip_value_max=np.inf)
    )

    ll_exp = tf.cast(K.less(errors, -theta_l), errors.dtype)
    ll_quad = (1 - ll_exp) * tf.cast(K.less(errors, 0), errors.dtype)

    lr_exp = tf.cast(K.greater(errors, theta_r), errors.dtype)
    lr_quad = (1 - lr_exp) * tf.cast(K.greater(errors, 0), errors.dtype)
    ll_exp = ll_exp * (
        alpha_l
        * theta_l
        * (theta_l + 2 * gamma_l * (K.exp((K.abs(errors) - theta_l) / (gamma_l)) - 1))
    )
    ll_quad = ll_quad * alpha_l * K.pow(errors, 2)

    lr_exp = lr_exp * (
        alpha_r
        * theta_r
        * (theta_r + 2 * gamma_r * (K.exp((errors - theta_r) / (gamma_r)) - 1))
    )
    lr_quad = lr_quad * alpha_r * K.pow(errors, 2)

    if relative_weight:
        a = tf.reduce_mean(weight * (ll_exp + ll_quad + lr_exp + lr_quad))
    else:
        a = tf.reduce_mean(ll_exp + ll_quad + lr_exp + lr_quad)

    return a

relative_mae(C=0.9)

MAE weighted by the relative error

Parameters:

Name	Type	Description	Default
C	float	Minimal value for the RUL	0.9

Returns:

Type	Description
	The loss function

Source code in ceruleo/models/keras/losses.py

def relative_mae(C: float = 0.9):
    """
    MAE weighted by the relative error

    Parameters:
        C: Minimal value for the RUL

    Returns:
        The loss function
    """
    mae = tf.keras.losses.MeanAbsoluteError()

    def concrete_relative_mae(y_true, y_pred):
        errors = y_true - y_pred
        sw = tf.abs(errors) / (
            tf.clip_by_value(y_true, clip_value_min=C, clip_value_max=np.inf)
        )
        return mae(y_true, y_pred, sample_weight=sw)

    return concrete_relative_mae

relative_mse(C=0.9)

MSE weighted by the relative error

Parameters:

Name	Type	Description	Default
C	float	Minimal value for the RUL	0.9

Returns:

Type	Description
	The loss function

Source code in ceruleo/models/keras/losses.py

def relative_mse(C: float = 0.9):
    """
    MSE weighted by the relative error

    Parameters:
        C: Minimal value for the RUL

    Returns:
        The loss function
    """
    mse = tf.keras.losses.MeanSquaredError()

    def concrete_relative_mse(y_true, y_pred):
        errors = y_true - y_pred
        sw = tf.abs(errors) / (
            tf.clip_by_value(y_true, clip_value_min=C, clip_value_max=np.inf)
        )
        return mse(y_true, y_pred, sample_weight=sw)

    return concrete_relative_mse

root_mean_squared_error(y_true, y_pred)

Root mean squared error

Parameters:

Name	Type	Description	Default
y_true	array	True RUL values	required
y_pred	array	Predicted RUL values	required

Source code in ceruleo/models/keras/losses.py

def root_mean_squared_error(y_true: np.array, y_pred: np.array) -> float:
    """Root mean squared error

    Parameters:
            y_true: True RUL values
            y_pred: Predicted RUL values
    """
    return K.sqrt(K.mean(K.square(y_pred - y_true), axis=0))

Layers

ConcreteDropout

Bases: Layer

Concrete Dropout layer class from https://arxiv.org/abs/1705.07832. Dropout Feature Ranking for Deep Learning Models Chun-Hao Chang Ladislav Rampasek Anna Goldenberg

Parameters:

Name	Type	Description	Default
dropout_regularizer		Positive float, satisfying $dropout_regularizer = 2 / ( au * N)$ with model precision $ au$ (inverse observation noise) and N the number of instances in the dataset. The factor of two should be ignored for cross-entropy loss, and used only for the eucledian loss.	1e-05
init_min		Minimum value for the randomly initialized dropout rate, in [0, 1].	0.1
init_min		Maximum value for the randomly initialized dropout rate, in [0, 1], with init_min <= init_max.	0.1
name		String, name of the layer.	None

Source code in ceruleo/models/keras/layers.py

class ConcreteDropout(Layer):
    """
    Concrete Dropout layer class from https://arxiv.org/abs/1705.07832.
    Dropout Feature Ranking for Deep Learning Models
    Chun-Hao Chang
    Ladislav Rampasek
    Anna Goldenberg

    Parameters:
        dropout_regularizer: Positive float, satisfying $dropout_regularizer = 2 / (\tau * N)$
            with model precision $\tau$ (inverse observation noise) and
            N the number of instances in the dataset.
            The factor of two should be ignored for cross-entropy loss,
            and used only for the eucledian loss.
        init_min: Minimum value for the randomly initialized dropout rate, in [0, 1].
        init_min: Maximum value for the randomly initialized dropout rate, in [0, 1],
            with init_min <= init_max.
        name: String, name of the layer.

    """

    def __init__(
        self,
        dropout_regularizer=1e-5,
        init_min=0.1,
        init_max=0.9,
        name=None,
        training=True,
        **kwargs,
    ):
        super(ConcreteDropout, self).__init__(name=name, **kwargs)
        assert init_min <= init_max, "init_min must be lower or equal to init_max."

        self.dropout_regularizer = dropout_regularizer

        self.p_logit = None
        self.p = None
        self.init_min = np.log(init_min) - np.log(1.0 - init_min)
        self.init_max = np.log(init_max) - np.log(1.0 - init_max)
        self.training = training

    def build(self, input_shape):
        self.window = input_shape[-2]
        self.number_of_features = input_shape[-1]
        input_shape = tensor_shape.TensorShape(input_shape)

        self.p_logit = self.add_weight(
            name="p_logit",
            shape=[self.number_of_features],
            initializer=tf.random_uniform_initializer(self.init_min, self.init_max),
            dtype=tf.float32,
            trainable=True,
        )

    def concrete_dropout(self, p, x):
        eps = K.cast_to_floatx(K.epsilon())
        temp = 1.0 / 10.0
        unif_noise = K.random_uniform(shape=[self.number_of_features])
        drop_prob = (
            K.log(p + eps)
            - K.log(1.0 - p + eps)
            + K.log(unif_noise + eps)
            - K.log(1.0 - unif_noise + eps)
        )
        drop_prob = K.sigmoid(drop_prob / temp)
        random_tensor = 1.0 - drop_prob

        retain_prob = 1.0 - p
        x *= random_tensor
        x /= retain_prob
        return x

    def call(self, inputs, training=True):
        p = K.sigmoid(self.p_logit)

        dropout_regularizer = p * K.log(p)
        dropout_regularizer += (1.0 - p) * K.log(1.0 - p)
        dropout_regularizer *= self.dropout_regularizer * self.number_of_features
        regularizer = K.sum(dropout_regularizer)
        self.add_loss(regularizer)

        x = self.concrete_dropout(p, inputs)

        return x

LASSOLayer

Bases: Layer

LASSO Layer

Parameters:

Name	Type	Description	Default
l1	float	L1 regularization parameter	required

Source code in ceruleo/models/keras/layers.py

class LASSOLayer(Layer):
    """
    LASSO Layer

    Parameters:
        l1: L1 regularization parameter
    """

    def __init__(self, l1: float):
        super(LASSOLayer, self).__init__()
        self.l1 = l1
        self.kernel_regularizer = regularizers.L1(l1)

    def build(self, input_shape):
        W_size = np.prod(input_shape[1:])
        self.w = self.add_weight(
            shape=(W_size,),
            initializer="random_normal",
            trainable=True,
            regularizer=self.kernel_regularizer,
            constraint=ZeroWeights(self.l1),
        )

        self.input_reshape = Reshape((W_size,))
        self.output_reshape = Reshape(input_shape[1:])

    def call(self, inputs):
        x = self.input_reshape(inputs)

        x = tf.math.multiply(self.w, x)

        self.add_metric(
            tf.math.reduce_sum(tf.cast(tf.abs(self.w) > 0, tf.float32)),
            name="Number of features",
        )
        return self.output_reshape(x)

ResidualShrinkageBlock

Bases: Layer

ResidualShrinkageBlock

Source code in ceruleo/models/keras/layers.py

class ResidualShrinkageBlock(Layer):
    """
    ResidualShrinkageBlock

    """

    def build(self, input_shape):
        self.blocks = []
        for i in range(2):
            self.blocks.append(
                Sequential(
                    [
                        BatchNormalization(),
                        Activation("relu"),
                        Conv2D(1, (1, 1), padding="same"),
                    ]
                )
            )

        self.abs = Lambda(lambda x: tf.abs(x))
        self.abs_mean = Sequential(
            [
                RemoveDimension(3),
                Permute((1, 2)),
                GlobalAveragePooling1D(),
                ExpandDimension(2),
                ExpandDimension(3),
            ]
        )
        self.shrinkage = Sequential(
            [
                Flatten(),
                Dense(input_shape[-1]),
                BatchNormalization(),
                Activation("relu"),
                Dense(
                    input_shape[-1], kernel_regularizer=tf.keras.regularizers.l2(1e-4)
                ),
                Activation("sigmoid"),
                ExpandDimension(2),
                ExpandDimension(3),
            ]
        )

    def call(self, inputs):
        x = ExpandDimension()(inputs)

        x = self.blocks[0](x)

        residual = self.blocks[1](x)

        residual_abs = self.abs(residual)

        abs_mean = self.abs_mean(residual_abs)

        scales = self.shrinkage(abs_mean)
        thres = abs_mean * scales
        thres = Permute((2, 1, 3))(thres)
        sub = (residual_abs) - thres

        zeros = sub - sub

        n_sub = maximum([sub, zeros])

        residual = ops.sign(residual) * n_sub
        residual = RemoveDimension(3)(residual)
        return residual + inputs