src/vav/VAVAnomalizer.py

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


class VAVAnomalizer:
    def __init__(
        self,
        rtu_id,
        prediction_model_path,
        clustering_model_path,
        num_inputs,
        num_outputs,
    ):
        """
        Initializes a VAVAnomalizer object.

        Args:
            rtu_id (int): The ID of the RTU (Roof Top Unit) associated with the VAV (Variable Air Volume) system.
            prediction_model_path (str): The file path to the prediction model.
            clustering_model_path (str): The file path to the clustering model.
            num_inputs (int): The number of input features for the prediction model.
            num_outputs (int): The number of output features for the prediction model.
        """
        self.rtu_id = rtu_id
        self.num_inputs = num_inputs
        self.num_outputs = num_outputs
        self.load_models(prediction_model_path, clustering_model_path)

    def load_models(self, prediction_model_path, clustering_model_path):
        """
        Loads the prediction model and clustering model.

        Args:
            prediction_model_path (str): The file path to the prediction model.
            clustering_model_path (str): The file path to the clustering model.
        """
        self.model = load_model(prediction_model_path)
        self.kmeans_model = joblib.load(clustering_model_path)

    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 predict(self, df_new):
        """
        Makes predictions using the prediction model.

        Args:
            df_new (numpy.ndarray): The new data for prediction.

        Returns:
            numpy.ndarray: The predicted values.
        """
        return self.model.predict(df_new)

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

        Args:
            df_trans (numpy.ndarray): The transformed data.
            pred (numpy.ndarray): The predicted values.

        Returns:
            numpy.ndarray: The actual values.
            numpy.ndarray: The residuals.
        """
        actual = df_trans[30, : self.num_outputs]
        resid = actual - pred
        return actual, resid

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

        Args:
            resid (array): Residual values.

        Returns:
            array: Array of distances.
        """
        dist = []
        dist.append(np.linalg.norm(resid - self.kmeans_model.cluster_centers_[0]))

        return np.array(dist)

    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 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