Merge pull request #44 from redBorder/add_dynamic_filters

PR-40: Add dynamic filters

Author: malvads

.github/workflows/security.yml

@@ -27,6 +27,7 @@ jobs:
 
     - name: Install safety for security checks
       run: |
+        pip install cffi
         pip install safety
 
     - name: Run safety check

resources/src/__main__.py

@@ -19,10 +19,13 @@
 
 
 import os
-from logger.logger import logger
-from server.rest import APIServer, config
-from server.production import GunicornApp
-from redborder.rq import RqManager
+import sys
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
+from resources.src.logger.logger import logger
+from resources.src.server.rest import APIServer, config
+from resources.src.server.production import GunicornApp
+from resources.src.redborder.rq import RqManager
 
 class Outliers:
     def __init__(self) -> None:

resources/src/ai/outliers.py

@@ -33,8 +33,7 @@
 import pandas as pd
 import tensorflow as tf
 
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-from logger import logger
+from resources.src.logger import logger
 
 class Autoencoder:
     """
@@ -61,8 +60,11 @@ def __init__(self, model_file, model_config_file):
                 loss_mult_minute (float): Extra penalty in the loss function for guessing wrong
                   'minute' field.
         """
-        self.check_existence(model_file, model_config_file)
-
+        try:
+            self.check_existence(model_file, model_config_file)
+        except FileNotFoundError as e:
+            logger.logger.error("Could not find the asked model files")
+            raise e
         try:
             model_config = configparser.ConfigParser()
             model_config.read(model_config_file)
@@ -77,28 +79,24 @@ def __init__(self, model_file, model_config_file):
             self.num_window = int(general_section.get('NUM_WINDOWS', 0))
             self.loss_mult_metric = float(general_section.get('LOSS_MULT_METRIC', 0))
             self.loss_mult_minute = float(general_section.get('LOSS_MULT_MINUTE', 0))
-        except FileNotFoundError:
-            logger.logger.error(f"Error: Model file '{model_config_file}' not found.")
-        except (OSError, ValueError) as e:
-            logger.logger.error(f"Error loading model conif: {e}")
+        except Exception as e:
+            logger.logger.error(f"Could not load model conif: {e}")
+            raise e
         try:
             self.model = tf.keras.models.load_model(
                 model_file,
                 compile=False
             )
-        except FileNotFoundError:
-            logger.logger.error(f"Error: Model file '{model_file}' not found.")
-        except (OSError, ValueError) as e:
-            logger.logger.error(f"Error loading the model: {e}")
+        except Exception as e:
+            logger.logger.error(f"Could not load model {e}")
+            raise e
 
     def check_existence(self, model_file, model_config_file):
         """
         Check existence of model files and copy them if missing.
 
         This function checks if the provided `model_file` and `model_config_file` exist in their
-        respective paths. If they don't exist, it renames and copies the corresponding default
-        files from the 'traffic.keras' and 'traffic.ini' files, which are expected to be located
-        in the same directory as the target files.
+        respective paths. If they don't exist, it raises an error.
 
         Args:
             model_file (str): Path to the target model file.
@@ -107,12 +105,13 @@ def check_existence(self, model_file, model_config_file):
                 - The full path to the model configuration file you want to check and potentially copy.
         """
         if not os.path.exists(model_file):
-            os.rename(f"{os.path.dirname(model_file)}/traffic.keras", model_file)
-            shutil.copy(model_file, f"{os.path.dirname(model_file)}/traffic.keras")
+            error_msg=f"Model file '{os.path.basename(model_file)}' not found"
+            logger.logger.error(error_msg)
+            raise FileNotFoundError(error_msg)
         if not os.path.exists(model_config_file):
-            os.rename(f"{os.path.dirname(model_config_file)}/traffic.ini", model_config_file)
-            shutil.copy(model_config_file, f"{os.path.dirname(model_config_file)}/traffic.ini")
-
+            error_msg=f"Model config file '{os.path.basename(model_config_file)}' not found"
+            logger.logger.error(error_msg)
+            raise FileNotFoundError(error_msg)
 
     def rescale(self, data):
         """
@@ -253,12 +252,20 @@ def compute_json(self, metric, raw_json):
 
         Args:
             metric (string): the name of field being analyzed.
-            raw_json (Json): druid Json response with the data.
+            raw_json (dict): deserialized Json druid response with the data.
 
         Returns:
-            (Json): Json with the anomalies and predictions for the data with RedBorder
+            (dict): deserialized Json with the anomalies and predictions for the data with RedBorder
               prediction Json format.
         """
+        if metric=="" or metric not in self.metrics:
+            error_msg = f"Model has not a metric called {metric}"
+            logger.logger.error(error_msg)
+            raise ValueError(error_msg)
+        if not raw_json:
+            error_msg = f"Input data is empty"
+            logger.logger.error(error_msg)
+            raise ValueError(error_msg)
         threshold = self.avg_loss+5*self.std_loss
         data, timestamps = self.input_json(raw_json)
         predicted, loss = self.calculate_predictions(data)
@@ -289,7 +296,7 @@ def input_json(self, raw_json):
         Also returns the timestamps for each entry.
 
         Args:
-            raw_json (Json): druid Json response with the data.
+            raw_json (dict): deserialized Json druid response with the data.
 
         Returns:
             data (numpy.ndarray): transformed data.
@@ -320,7 +327,7 @@ def output_json(self, metric, anomalies, predicted):
             predicted (pandas.DataFrame): predictions made by the model.
 
         Returns:
-            (Json): Json with the anomalies and predictions for the data with RedBorder prediction
+            (dict): deserialized Json with the anomalies and predictions for the data with RedBorder prediction
               Json format.
         """
         predicted = predicted.copy()
@@ -335,9 +342,18 @@ def output_json(self, metric, anomalies, predicted):
 
     @staticmethod
     def execute_prediction_model(data, metric, model_file, model_config):
-        autoencoder = Autoencoder(model_file, model_config)
-        result = autoencoder.compute_json(metric, data)
-        return result
+        try:
+            autoencoder = Autoencoder(model_file, model_config)
+            return autoencoder.compute_json(metric, data)
+        except Exception as e:
+            logger.logger.error("Couldn't execute model")
+            return Autoencoder.return_error(e)
     @staticmethod
     def return_error(error="error"):
+        """
+        Returns an adequate formatted JSON for whenever there is an error.
+
+        Args:
+            error (string): message detailing what type of error has been fired.
+        """
         return { "status": "error", "msg":error }

resources/src/ai/shallow_outliers.py

@@ -17,10 +17,14 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import os
+import sys
 import numpy as np
 import pandas as pd
 from sklearn.ensemble import IsolationForest
 
+from resources.src.logger import logger
+
 class ShallowOutliers:
     """
     Shallow AI model for outliers detection. Used whenever there is no deep learning model defined.
@@ -43,11 +47,18 @@ def predict(self, arr):
             smooth_arr (numpy.ndarray): 1D numpy array with the smoothed data. Same shape as arr.
         """
         if len(arr) == 0:
-            raise ValueError("Input array must be non empty")
+            error_msg = "Input array must be non-empty"
+            logger.logger.error(error_msg)
+            raise ValueError(error_msg)
         if arr.ndim != 1:
-            raise ValueError("Input array must be 1-dimensional")
+            error_msg = "Input array must be 1-dimensional"
+            logger.logger.error(error_msg)
+            raise ValueError(error_msg)
         if not np.issubdtype(arr.dtype, np.number):
-            raise ValueError("Input array must contain numerical data")
+            error_msg = "Input array must contain numerical data"
+            logger.logger.error(error_msg)
+            raise ValueError(error_msg)
+
         window_size = max(int(0.05 * len(arr)), min(len(arr), int(5 + np.log(len(arr)))))
         window_size += 1 if window_size % 2 == 0 else 0
         kernel = np.arange(1, window_size // 2 + 2, dtype=float)
@@ -115,10 +126,20 @@ def compute_json(self, raw_json):
 
     @staticmethod
     def execute_prediction_model(data):
-        shallow_outliers = ShallowOutliers()
-        result = shallow_outliers.compute_json(data)
-        return result
+        try:
+            result = ShallowOutliers.compute_json(data)
+        except Exception as e:
+            result = ShallowOutliers.return_error(e)
+            raise
+        finally:
+            return result
 
     @staticmethod
     def return_error(error="error"):
+        """
+        Returns an adequate formatted JSON for whenever there is an error.
+
+        Args:
+            error (string): message detailing what type of error has been fired.
+        """
         return { "status": "error", "msg":error }

resources/src/ai/traffic_filter.json

@@ -0,0 +1,5 @@
+{
+    "type": "selector",
+    "dimension": "sensor_name",
+    "value": "ASR"
+}

resources/src/ai/trainer.py

@@ -28,14 +28,13 @@
 '''
 import sys
 import datetime
+import numpy as np
 import configparser
 from datetime import datetime
 from tensorflow.keras.optimizers import AdamW
-try:
-    from ai.outliers import Autoencoder
-except:
-    from src.ai.outliers import Autoencoder
-    sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from resources.src.ai.outliers import Autoencoder
+from resources.src.logger.logger import logger
 
 """
 This module extends the Autoencoder class to allow further training of the model.
@@ -73,10 +72,14 @@ def save_model(self, save_model_file, save_config_file):
         """
         if os.path.exists(save_model_file):
             if not os.access(save_model_file, os.W_OK):
-                raise PermissionError(f"Permission denied: Cannot overwrite '{save_model_file}'")
+                error_msg = f"Permission denied: Cannot overwrite '{save_model_file}'"
+                logger.logger.error(error_msg)
+                raise PermissionError(error_msg)
         if os.path.exists(save_config_file):
             if not os.access(save_config_file, os.W_OK):
-                raise PermissionError(f"Permission denied: Cannot overwrite '{save_config_file}'")
+                error_msg = f"Permission denied: Cannot overwrite '{save_config_file}'"
+                logger.logger.error(error_msg)
+                raise PermissionError(error_msg)
         self.model.save(save_model_file)
         new_model_config = configparser.ConfigParser()
         new_model_config.add_section('Columns')
@@ -107,18 +110,18 @@ def data_augmentation(self, data):
         """
         return data
 
-    def prepare_data_for_training(self, data, augment = False):
+    def prepare_data_for_training(self, raw_data, augment = False):
         """
         Prepares data to be used for training the model.
 
         Args:
-            data (numpy ndarray): data to be used for training.
-
+            raw_data (list): list with the response form one or more druid queries.
             augment (boolean): set to True to generate more data for training.
 
         Returns:
             prep_data (numpy ndarray): transformed data for its use in the model.
         """
+        data = np.concatenate([self.input_json(query_json)[0] for query_json in raw_data], axis = 0)
         prep_data = self.rescale(data)
         if augment:
             #TODO actually augment data
@@ -132,7 +135,7 @@ def train(self, raw_data, epochs=20, batch_size=32, backup_path=None):
         Given a druid query response, it is fed to the model for training.
 
         Args:
-            raw_data (json): response form a druid query.
+            raw_data (list): list with the response form one or more druid queries.
             epochs (int): how many times should the model train on the data.
             batch_size (int): how many slices should the model take at once
             for training.
@@ -142,8 +145,7 @@ def train(self, raw_data, epochs=20, batch_size=32, backup_path=None):
             backup_path = "./backups/"
         date = datetime.now().strftime("%y-%m-%dT%H:%M")
         self.save_model(f"{backup_path}{date}.keras",f"{backup_path}{date}.ini")
-        data = self.input_json(raw_data)[0]
-        prep_data = self.prepare_data_for_training(data)
+        prep_data = self.prepare_data_for_training(raw_data)
         self.model.fit(x=prep_data, y=prep_data, epochs = epochs, batch_size = batch_size, verbose = 0)
         loss = self.model_loss(prep_data, self.model.predict(prep_data), single_value=False).numpy()
         self.avg_loss = 0.9*self.avg_loss + 0.1*loss.mean()

resources/src/config.ini

@@ -14,7 +14,8 @@ schedule_hour=* * * * *
 epochs=20
 batch_size=32
 backup_path=./backups/
-target_sensors=FlowSensor
+#target_sensors=FlowSensor
+model_names=traffic
 
 [Druid]
 druid_endpoint=http://x.x.x.x:8080/druid/v2/

resources/src/config/configmanager.py

@@ -29,7 +29,8 @@ def __init__(self, config_file):
             config_file (str): The path to the configuration file to read and write.
         """
         self.config = configparser.ConfigParser()
-        self.config.read(config_file)
+        with open(config_file, 'r') as file:
+            self.config.read_file(file)
 
     def get(self, section, option):
         """

resources/src/druid/client.py

@@ -40,7 +40,7 @@ def execute_query(self, druid_query):
 
         Returns:
             dict: The response from the Druid query in JSON format.
-        
+
         Raises:
             Exception: If the Druid query fails with a non-200 status code.
         """