feat(anomaly): add functions for preparing sensor data and calculating risk scores

utils.py

@@ -141,3 +141,76 @@ def create_timestamp_from_predictions(predictions, sensor_timestamp=None):
     except Exception as e:
         print(f"Error creating timestamp: {str(e)}")
         return None
+
+
+def prepare_sensor_data_for_anomaly(sensor_data, preprocessor):
+    try:
+        sensor_features = {
+            "Air temperature [K]": sensor_data.get("air_temp"),
+            "Process temperature [K]": sensor_data.get("process_temp"),
+            "Rotational speed [rpm]": sensor_data.get("rotational_speed"),
+            "Torque [Nm]": sensor_data.get("torque"),
+            "Tool wear [min]": sensor_data.get("tool_wear"),
+        }
+        if None in sensor_features.values():
+            print("Error: Missing sensor data!")
+            return None
+
+        X = pd.DataFrame([sensor_features])
+
+        if preprocessor is None:
+            raise ValueError("preprocessor not loaded.")
+
+        X_transormed = preprocessor.transform(X)
+
+        return X_transormed
+
+    except Exception as e:
+        print(f"Error preparing prediction data: {str(e)}")
+        import traceback
+
+        traceback.print_exc()
+        return None
+
+
+def calculate_risk_score(confidence, severity):
+    severity_weight = severity / 5.0
+    risk_score = confidence * severity_weight * 100
+
+    return risk_score
+
+
+def get_risk_level(risk_score):
+    if risk_score >= 80:
+        return "Critical"
+    elif risk_score >= 60:
+        return "High"
+    elif risk_score >= 40:
+        return "Medium"
+    elif risk_score >= 20:
+        return "Low"
+    else:
+        return "Very Low"
+
+
+def get_failure_severity(prediction_index):
+    severity_mapping = {
+        0: 4,  # Heat Dissipation Failure - High
+        1: 4,  # Overstrain Failure - High
+        2: 5,  # Power Failure - Critical
+        3: 2,  # Random Failures - Low-Medium
+        4: 3,  # Tool Wear Failure - Medium
+    }
+
+    return severity_mapping.get(prediction_index, 3)
+
+
+def get_failure_type_name(prediction_index):
+    failure_types = {
+        0: "Heat Dissipation Failure",
+        1: "Overstrain Failure",
+        2: "Power Failure",
+        3: "Random Failures",
+        4: "Tool Wear Failure",
+    }
+    return failure_types.get(prediction_index, "Unknown Failure")