feat(controller): implement binary and classification prediction methods with error handling

controller.py

@@ -4,22 +4,154 @@ from utils import (
     convert_cyclical_to_original,
     create_timestamp_from_predictions,
     prepare_prediction_data,
+    prepare_sensor_data_for_anomaly,
+    calculate_risk_score,
+    get_risk_level,
+    get_failure_severity,
+    get_failure_type_name,
 )
 
+from datetime import datetime
 
 class Controller:
     def __init__(self):
         self.__database = Database()
         self.__sensor = self.__database.get_sensor_readings()
         self.__model = Model()
-
+        
+    def _get_hardcoded_error_sensor(self):
+        """Hardcoded sensor data yang pasti error untuk testing"""
+        return {
+            "id": 9999,
+            "air_temp": 298.9,
+            "process_temp": 309.1,
+            "rotational_speed": 2861,
+            "torque": 4.6,
+            "tool_wear": 143,
+            "timestamp": datetime.now(),
+            "created_at": datetime.now()
+        }
     # Binary method
     def predict_binary(self):
-        pass
+        if self.__sensor is None:
+            return {
+                "success": False,
+                "error": "No sensor data available from database.",
+            }
+
+        if (
+            self.__model.model_binary is None
+            or self.__model.preprocessor_anomaly is None
+        ):
+            return {
+                "success": False,
+                "error": "Binary model or preprocessor not loaded.",
+            }
+
+        X_scaled = prepare_sensor_data_for_anomaly(
+            self.__sensor, self.__model.preprocessor_anomaly
+        )
+
+        if X_scaled is None:
+            return {"success": False, "error": "Failed to prepare sensor data."}
+
+        if hasattr(self.__model.model_binary, "predict_proba"):
+            probabilities = self.__model.model_binary.predict_proba(X_scaled)[0]
+            confidence_normal = float(probabilities[0])
+            confidence_error = float(probabilities[1])
+            
+            is_error = int(confidence_error > 0.5)
+            confidence = confidence_error if is_error else confidence_normal
+        else:
+            prediction = self.__model.model_binary.predict(X_scaled)
+            is_error = int(prediction[0])
+            confidence = 1.0 if is_error else 0.0
+
+        risk_score = confidence * 100 if is_error else (1 - confidence) * 100
+
+        result = {
+            "success": True,
+            "failure_predicted": bool(is_error),
+            "confidence": float(f"{confidence}"),
+            "risk_score": float(f"{risk_score}"),
+            "sensor_data": {
+                "air_temp": self.__sensor.get("air_temp"),
+                "process_temp": self.__sensor.get("process_temp"),
+                "rotational_speed": self.__sensor.get("rotational_speed"),
+                "torque": self.__sensor.get("torque"),
+                "tool_wear": self.__sensor.get("tool_wear"),
+            },
+        }
+
+        if is_error:
+            multiclass_result = self.predict_classification()
+            if multiclass_result.get("success"):
+                result["failure_analysis"] = {
+                    "failure_type": multiclass_result.get("failure_type"),
+                    "confidence": multiclass_result.get("confidence"),
+                    "risk_level": multiclass_result.get("risk_level"),
+                    "risk_score": multiclass_result.get("risk_score"),
+                    "all_probabilities": multiclass_result.get("all_probabilities"),
+                }
+        else:
+            result["failure_analysis"] = {
+                "failure_type": "No Failure",
+                "risk_level": "Very Low",
+            }
+
+        return result
 
     # Classification Method
     def predict_classification(self):
-        pass
+        if self.__sensor is None:
+            return {
+                "success": False,
+                "error": "No sensor data available from database.",
+            }
+
+        if (
+            self.__model.model_multiclass is None
+            or self.__model.preprocessor_anomaly is None
+        ):
+            return {"success": False, "error": "Model or scalers not loaded."}
+
+        X_scaled = prepare_sensor_data_for_anomaly(
+            self.__sensor, self.__model.preprocessor_anomaly
+        )
+
+        if X_scaled is None:
+            return {"success": False, "error": "Failed to prepare sensor data."}
+
+        prediction = self.__model.model_multiclass.predict(X_scaled)
+
+        if hasattr(self.__model.model_multiclass, "predict_proba"):
+            probabilities = self.__model.model_multiclass.predict_proba(X_scaled)[0]
+            failure_index = int(probabilities.argmax())
+            confidence = float(probabilities[failure_index])
+
+            all_probs = {
+                get_failure_type_name(i): float(prob)
+                for i, prob in enumerate(probabilities)
+            }
+        else:
+            failure_index = int(prediction[0])
+            confidence = 1.0
+            all_probs = None
+
+        failure_type = get_failure_type_name(failure_index)
+
+        severity = get_failure_severity(failure_index)
+        risk_score = calculate_risk_score(confidence, severity)
+        risk_level = get_risk_level(risk_score)
+
+        return {
+            "success": True,
+            "failure_type": failure_type,
+            "confidence": float(f"{confidence}"),
+            "risk_score": float(f"{risk_score}"),
+            "risk_level": risk_level,
+            "all_probabilities": all_probs,
+        }
 
     # Time series method
     def predict_time_series(self):
@@ -44,7 +176,7 @@ class Controller:
 
         if X_sequence is None:
             return {"success": False, "error": "Failed to prepare prediction data."}
-            
+
         scaled_prediction = self.__model.model_lstm.predict(X_sequence, verbose=0)
 
         prediction = self.__model.scaler_y.inverse_transform(scaled_prediction)

requirements.txt

@@ -6,4 +6,5 @@ joblib
 keras
 numpy
 pandas
-scikit-learn
+scikit-learn
+xgboost