src/rtu/RTUAnomalizer.py

import numpy as np
from tensorflow.keras.models import load_model
import joblib


class RTUAnomalizer:
    """
    Class for performing anomaly detection on RTU (Roof Top Unit) data.
    """

    model = None
    kmeans_models = []

    def __init__(
        self,
        prediction_model_path=None,
        clustering_model_paths=None,
        num_inputs=None,
        num_outputs=None,
    ):
        """
        Initialize the RTUAnomalizer object.

        Args:
            prediction_model_path (str): Path to the prediction model file.
            clustering_model_paths (list): List of paths to the clustering model files.
            num_inputs (int): Number of input features.
            num_outputs (int): Number of output features.
        """
        self.num_inputs = num_inputs
        self.num_outputs = num_outputs
        if prediction_model_path is not None and clustering_model_paths is not None:
            self.load_models(prediction_model_path, clustering_model_paths)

    def initialize_lists(self, size=30):
        """
        Initialize lists for storing actual, predicted, and residual values.

        Args:
            size (int): Size of the lists.

        Returns:
            tuple: A tuple containing three lists initialized with zeros.
        """
        initial_values = [0] * size
        return initial_values.copy(), initial_values.copy(), initial_values.copy()

    def load_models(self, prediction_model_path, clustering_model_paths):
        """
        Load the prediction and clustering models.

        Args:
            prediction_model_path (str): Path to the prediction model file.
            clustering_model_paths (list): List of paths to the clustering model files.
        """
        self.model = load_model(prediction_model_path)

        for path in clustering_model_paths:
            self.kmeans_models.append(joblib.load(path))

    def predict(self, df_new):
        """
        Make predictions using the prediction model.

        Args:
            df_new (DataFrame): Input data for prediction.

        Returns:
            array: Predicted values.
        """
        return self.model.predict(df_new)

    def calculate_residuals(self, df_trans, pred):
        """
        Calculate the residuals between actual and predicted values.

        Args:
            df_trans (DataFrame): Transformed input data.
            pred (array): Predicted values.

        Returns:
            tuple: A tuple containing the actual values and residuals.
        """
        actual = df_trans[30, : self.num_outputs]
        resid = actual - pred
        return actual, resid

    def resize_prediction(self, pred, df_trans):
        """
        Resize the predicted values to match the shape of the transformed input data.

        Args:
            pred (array): Predicted values.
            df_trans (DataFrame): Transformed input data.

        Returns:
            array: Resized predicted values.
        """
        pred = np.resize(
            pred, (pred.shape[0], pred.shape[1] + len(df_trans[30, self.num_outputs :]))
        )
        pred[:, -len(df_trans[30, self.num_outputs :]) :] = df_trans[
            30, self.num_outputs :
        ]
        return pred

    def inverse_transform(self, scaler, pred, df_trans):
        """
        Inverse transform the predicted and actual values.

        Args:
            scaler (object): Scaler object for inverse transformation.
            pred (array): Predicted values.
            df_trans (DataFrame): Transformed input data.

        Returns:
            tuple: A tuple containing the actual and predicted values after inverse transformation.
        """
        pred = scaler.inverse_transform(np.array(pred))
        actual = scaler.inverse_transform(np.array([df_trans[30, :]]))
        return actual, pred

    def update_lists(self, actual_list, pred_list, resid_list, actual, pred, resid):
        """
        Update the lists of actual, predicted, and residual values.

        Args:
            actual_list (list): List of actual values.
            pred_list (list): List of predicted values.
            resid_list (list): List of residual values.
            actual (array): Actual values.
            pred (array): Predicted values.
            resid (array): Residual values.

        Returns:
            tuple: A tuple containing the updated lists of actual, predicted, and residual values.
        """
        actual_list.pop(0)
        pred_list.pop(0)
        resid_list.pop(0)
        actual_list.append(actual[0, 1])
        pred_list.append(pred[0, 1])
        resid_list.append(resid[0, 1])
        return actual_list, pred_list, resid_list

    def calculate_distances(self, resid):
        """
        Calculate the distances between residuals and cluster centers.

        Args:
            resid (array): Residual values.

        Returns:
            array: Array of distances.
        """
        dist = []
        for i, model in enumerate(self.kmeans_models):
            dist.append(
                np.linalg.norm(
                    resid[:, (i * 7) + 1 : (i * 7) + 8] - model.cluster_centers_[0],
                    ord=2,
                    axis=1,
                )
            )

        return np.array(dist)

    def pipeline(self, df_new, df_trans, scaler):
        """
        Perform the anomaly detection pipeline.

        Args:
            df_new (DataFrame): Input data for prediction.
            df_trans (DataFrame): Transformed input data.
            scaler (object): Scaler object for inverse transformation.

        Returns:
            tuple: A tuple containing the lists of actual, predicted, and residual values, and the distances.
        """
        actual_list, pred_list, resid_list = self.initialize_lists()
        pred = self.predict(df_new)
        actual, resid = self.calculate_residuals(df_trans, pred)
        pred = self.resize_prediction(pred, df_trans)
        actual, pred = self.inverse_transform(scaler, pred, df_trans)
        actual_list, pred_list, resid_list = self.update_lists(
            actual_list, pred_list, resid_list, actual, pred, resid
        )
        dist = self.calculate_distances(resid)
        return actual_list, pred_list, resid_list, dist