main works

mqttpublisher.ipynb

@@ -2,14 +2,14 @@
  "cells": [
   {
    "cell_type": "code",
-   "execution_count": 10,
+   "execution_count": 4,
    "metadata": {},
    "outputs": [
     {
      "name": "stderr",
      "output_type": "stream",
      "text": [
-      "C:\\Users\\jerin\\AppData\\Local\\Temp\\ipykernel_4616\\2478473330.py:13: DeprecationWarning: Callback API version 1 is deprecated, update to latest version\n",
+      "C:\\Users\\arbal\\AppData\\Local\\Temp\\ipykernel_20872\\3033510885.py:13: DeprecationWarning: Callback API version 1 is deprecated, update to latest version\n",
       "  client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, clientId)\n"
      ]
     },
@@ -171,221 +171,8 @@
       "published!\n",
       "published!\n",
       "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
-      "published!\n",
       "published!\n"
      ]
-    },
-    {
-     "ename": "KeyboardInterrupt",
-     "evalue": "",
-     "output_type": "error",
-     "traceback": [
-      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
-      "\u001b[1;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
-      "Cell \u001b[1;32mIn[10], line 94\u001b[0m\n\u001b[0;32m     90\u001b[0m         time\u001b[38;5;241m.\u001b[39msleep(\u001b[38;5;241m2\u001b[39m)\n\u001b[0;32m     93\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28;01mTrue\u001b[39;00m:\n\u001b[1;32m---> 94\u001b[0m     \u001b[43mpublish_sensor_data\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m     95\u001b[0m     \u001b[38;5;66;03m# time.sleep(0.1)\u001b[39;00m\n\u001b[0;32m     96\u001b[0m client\u001b[38;5;241m.\u001b[39mdisconnect()\n",
-      "Cell \u001b[1;32mIn[10], line 90\u001b[0m, in \u001b[0;36mpublish_sensor_data\u001b[1;34m()\u001b[0m\n\u001b[0;32m     55\u001b[0m client\u001b[38;5;241m.\u001b[39mpublish(topic, payload\u001b[38;5;241m=\u001b[39mjson\u001b[38;5;241m.\u001b[39mdumps({\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mhp_hws_temp\u001b[39m\u001b[38;5;124m'\u001b[39m:hp_hws_temp,\n\u001b[0;32m     56\u001b[0m \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mrtu_003_sa_temp\u001b[39m\u001b[38;5;124m'\u001b[39m:rtu_003_sa_temp,\n\u001b[0;32m     57\u001b[0m \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mrtu_003_oadmpr_pct\u001b[39m\u001b[38;5;124m'\u001b[39m: rtu_003_oadmpr_pct,\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m     87\u001b[0m \u001b[38;5;124m'\u001b[39m\u001b[38;5;124mrelative_humidity_set_1\u001b[39m\u001b[38;5;124m'\u001b[39m:relative_humidity_set_1,\n\u001b[0;32m     88\u001b[0m \u001b[38;5;124m'\u001b[39m\u001b[38;5;124msolar_radiation_set_1\u001b[39m\u001b[38;5;124m'\u001b[39m:solar_radiation_set_1}))\n\u001b[0;32m     89\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mpublished!\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m---> 90\u001b[0m time\u001b[38;5;241m.\u001b[39msleep(\u001b[38;5;241m2\u001b[39m)\n",
-      "\u001b[1;31mKeyboardInterrupt\u001b[0m: "
-     ]
     }
    ],
    "source": [
@@ -399,7 +186,7 @@
     "broker_address = \"localhost\"\n",
     "broker_port = 1883\n",
     "\n",
-    "df = pd.read_csv(\"sample_data_smooth_01.csv\")\n",
+    "df = pd.read_csv(\"data/sample_data_smooth_01.csv\")\n",
     "\n",
     "client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, clientId)\n",
     "client.connect(broker_address, broker_port)\n",
@@ -436,10 +223,10 @@
     "        rtu_002_ma_temp = row['rtu_002_ma_temp']\n",
     "        rtu_002_sf_vfd_spd_fbk_tn = row['rtu_002_sf_vfd_spd_fbk_tn']\n",
     "        rtu_002_rf_vfd_spd_fbk_tn = row['rtu_002_rf_vfd_spd_fbk_tn']\n",
-    "        rtu_004_sat_sp_tn = row['rtu_004_sat_sp_tn']\n",
-    "        rtu_003_sat_sp_tn = row['rtu_003_sat_sp_tn']\n",
-    "        rtu_001_sat_sp_tn = row['rtu_001_sat_sp_tn']\n",
-    "        rtu_002_sat_sp_tn = row['rtu_002_sat_sp_tn']\n",
+    "        # rtu_004_sat_sp_tn = row['rtu_004_sat_sp_tn']\n",
+    "        # rtu_003_sat_sp_tn = row['rtu_003_sat_sp_tn']\n",
+    "        # rtu_001_sat_sp_tn = row['rtu_001_sat_sp_tn']\n",
+    "        # rtu_002_sat_sp_tn = row['rtu_002_sat_sp_tn']\n",
     "        air_temp_set_1 = row['air_temp_set_1']\n",
     "        air_temp_set_2 = row['air_temp_set_2']\n",
     "        dew_point_temperature_set_1d = row['dew_point_temperature_set_1d']\n",
@@ -476,17 +263,17 @@
     "        'rtu_002_ma_temp':rtu_002_ma_temp,\n",
     "        'rtu_002_sf_vfd_spd_fbk_tn':rtu_002_sf_vfd_spd_fbk_tn,\n",
     "        'rtu_002_rf_vfd_spd_fbk_tn':rtu_002_rf_vfd_spd_fbk_tn,\n",
-    "        'rtu_004_sat_sp_tn':rtu_004_sat_sp_tn,\n",
-    "        'rtu_003_sat_sp_tn' :rtu_003_sat_sp_tn,\n",
-    "        'rtu_001_sat_sp_tn':rtu_001_sat_sp_tn,\n",
-    "        'rtu_002_sat_sp_tn':rtu_002_sat_sp_tn,\n",
+    "        # 'rtu_004_sat_sp_tn':rtu_004_sat_sp_tn,\n",
+    "        # 'rtu_003_sat_sp_tn' :rtu_003_sat_sp_tn,\n",
+    "        # 'rtu_001_sat_sp_tn':rtu_001_sat_sp_tn,\n",
+    "        # 'rtu_002_sat_sp_tn':rtu_002_sat_sp_tn,\n",
     "        'air_temp_set_1':air_temp_set_1,\n",
     "        'air_temp_set_2':air_temp_set_2,\n",
     "        'dew_point_temperature_set_1d':dew_point_temperature_set_1d,\n",
     "        'relative_humidity_set_1':relative_humidity_set_1,\n",
     "        'solar_radiation_set_1':solar_radiation_set_1}))\n",
     "        print(\"published!\")\n",
-    "        time.sleep(2)\n",
+    "        time.sleep(0.2)\n",
     "\n",
     "\n",
     "while True:\n",

src/main.py

@@ -4,28 +4,27 @@ from rtu.RTUPipeline import RTUPipeline
 import paho.mqtt.client as mqtt
 
 
-        
 def main():
     rtu_data_pipeline = RTUPipeline(scaler_path="src/rtu/models/scaler_1.pkl")
     print(rtu_data_pipeline.scaler)
     rtu_anomalizer = RTUAnomalizer(
         prediction_model_path="src/rtu/models/lstm_4rtu_smooth_02.keras",
         clustering_model_paths=[
-            "rtu/models/kmeans_model1.pkl",
-            "rtu/models/kmeans_model2.pkl",
-            "rtu/models/kmeans_model3.pkl",
-            "rtu/models/kmeans_model4.pkl",
+            "src/rtu/models/kmeans_model1.pkl",
+            "src/rtu/models/kmeans_model2.pkl",
+            "src/rtu/models/kmeans_model3.pkl",
+            "src/rtu/models/kmeans_model4.pkl",
         ],
         num_inputs=rtu_data_pipeline.num_inputs,
-        num_outputs=rtu_data_pipeline.num_outputs
+        num_outputs=rtu_data_pipeline.num_outputs,
     )
-    
+
     def on_message(client, userdata, message):
-        print(json.loads(message.payload.decode()))
+        # print(json.loads(message.payload.decode()))
         df_new, df_trans = rtu_data_pipeline.fit(message)
-        out = rtu_anomalizer.predict(df_new, df_trans, rtu_data_pipeline.scaler)
-        print(out)
-    
+        if not df_new is None and not df_trans is None:
+            out = rtu_anomalizer.pipeline(df_new, df_trans, rtu_data_pipeline.scaler)
+
     broker_address = "localhost"
     broker_port = 1883
     topic = "sensor_data"
@@ -34,9 +33,7 @@ def main():
     client.connect(broker_address, broker_port)
     client.subscribe(topic)
     client.loop_forever()
-    
-    
-if __name__=='__main__':
+
+
+if __name__ == "__main__":
     main()
-   
-    

src/rtu/RTUAnomalizer.py

@@ -2,59 +2,76 @@ import numpy as np
 from tensorflow.keras.models import load_model
 import joblib
 
+
 class RTUAnomalizer:
     model = None
     kmeans_models = []
 
-    def __init__(self, prediction_model_path = None, clustering_model_paths= None, num_inputs = None, num_outputs = None):
-        
+    def __init__(
+        self,
+        prediction_model_path=None,
+        clustering_model_paths=None,
+        num_inputs=None,
+        num_outputs=None,
+    ):
+
         self.num_inputs = num_inputs
         self.num_outputs = num_outputs
         if not prediction_model_path is None and not clustering_model_paths is None:
             self.load_models(prediction_model_path, clustering_model_paths)
 
-    def initialize_lists(size=30):
+    def initialize_lists(self, size=30):
         initial_values = [0] * size
         return initial_values.copy(), initial_values.copy(), initial_values.copy()
 
     def load_models(self, prediction_model_path, clustering_model_paths):
         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):
         return self.model.predict(df_new)
 
-    def calculate_residuals(self,df_trans, pred):
-        actual = df_trans[30,:self.num_outputs+1]
+    def calculate_residuals(self, df_trans, pred):
+        actual = df_trans[30, : self.num_outputs]
         resid = actual - pred
         return actual, resid
 
-    def resize_prediction(self,pred, df_trans):
-        pred.resize((pred.shape[0], pred.shape[1] + len(df_trans[30,self.num_outputs+1:])))
-        pred[:, -len(df_trans[30,self.num_outputs+1:]):] = df_trans[30,self.num_outputs+1:]
+    def resize_prediction(self, pred, df_trans):
+        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(scaler, pred, df_trans):
+    def inverse_transform(self, scaler, pred, df_trans):
         pred = scaler.inverse_transform(np.array(pred))
-        actual = scaler.inverse_transform(np.array([df_trans[30,:]]))
+        actual = scaler.inverse_transform(np.array([df_trans[30, :]]))
         return actual, pred
 
-    def update_lists(actual_list, pred_list, resid_list, actual, pred, resid):
+    def update_lists(self, actual_list, pred_list, resid_list, actual, pred, resid):
         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])
+        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):
+    def calculate_distances(self, resid):
         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))
-            
+            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):
@@ -63,6 +80,8 @@ class RTUAnomalizer:
         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)
+        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

src/rtu/RTUPipeline.py

@@ -50,8 +50,9 @@ class RTUPipeline:
             # "relative_humidity_set_1",
             # "solar_radiation_set_1",
         ]
-        
-        self.input_col_names  =    ["air_temp_set_1",
+
+        self.input_col_names = [
+            "air_temp_set_1",
             "air_temp_set_2",
             "dew_point_temperature_set_1d",
             "relative_humidity_set_1",
@@ -59,17 +60,17 @@ class RTUPipeline:
         ]
         self.num_inputs = len(self.input_col_names)
         self.num_outputs = len(self.output_col_names)
-        self.column_names = self.output_col_names+self.input_col_names
+        self.column_names = self.output_col_names + self.input_col_names
 
         if scaler_path:
             self.scaler = self.get_scaler(scaler_path)
-        self.df = pd.DataFrame(columns = self.column_names)
+        self.df = pd.DataFrame(columns=self.column_names)
 
     def get_scaler(self, scaler_path):
         return joblib.load(scaler_path)
 
     def get_window(self, df):
-        len_df = np.len(df)
+        len_df = len(df)
         if len_df > 30:
             return df[len_df - 31 : len_df].astype("float32")
         else:
@@ -78,12 +79,12 @@ class RTUPipeline:
     def transform_window(self, df_window):
         return self.scaler.transform(df_window)
 
-    def prepare_input(self,df_trans):
+    def prepare_input(self, df_trans):
         return df_trans[:30, :].reshape((1, 30, len(self.column_names)))
-    
+
     def extract_data_from_message(self, message):
         payload = json.loads(message.payload.decode())
-     
+
         len_df = len(self.df)
         # self.df.loc[len_df] = {'hp_hws_temp':payload['hp_hws_temp'],
         # 'rtu_003_sa_temp':payload['rtu_003_sa_temp'],
@@ -123,18 +124,20 @@ class RTUPipeline:
         # 'dew_point_temperature_set_1d':payload["dew_point_temperature_set_1d"],
         # 'relative_humidity_set_1':payload["relative_humidity_set_1"],
         # 'solar_radiation_set_1':payload["solar_radiation_set_1"]}
-        
-        self.df.loc[len_df] = {}
+
+        k = {}
         for col in self.column_names:
-            self.df.loc[len_df][col] = payload[col]
+            k[col] = payload[col]
+        self.df.loc[len_df] = k
         return self.df
-            
-    
-    def fit(self,message):
-        len_df = np.len(df)
+
+    def fit(self, message):
         df = self.extract_data_from_message(message)
-        df_window = self.get_window(df, len_df)
+        df_window = self.get_window(df)
         if df_window is not None:
-            df_trans = self.transform_window(df_window, self.scaler)
+            df_trans = self.transform_window(df_window)
             df_new = self.prepare_input(df_trans)
-        return df_new,df_trans
+        else:
+            df_new = None
+            df_trans = None
+        return df_new, df_trans