| import logging |
|
|
| from model import Model |
| from utils import ( |
| calculate_risk_score, |
| convert_cyclical_to_original, |
| create_timestamp_from_predictions, |
| get_failure_severity, |
| get_failure_type_name, |
| get_risk_level, |
| prepare_prediction_data, |
| prepare_sensor_data_for_anomaly, |
| ) |
| import numpy as np |
| from datetime import datetime, timedelta |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class Controller: |
| def __init__(self, database_sensor): |
| self.__sensor = database_sensor |
| self.__model = Model() |
| logger.info("Controller initialized with Model") |
| |
| def set_sensor_data(self, sensor): |
| self.__sensor = sensor |
| if isinstance(sensor, list): |
| logger.debug(f"Sensor data updated with {len(sensor)} readings") |
| else: |
| logger.debug(f"Sensor data updated for machine_id: {sensor.get('machine_id')}") |
| |
| |
| def predict_binary(self): |
| if self.__sensor is None: |
| logger.warning("Binary prediction attempted with no sensor data") |
| return { |
| "success": False, |
| "error": "No sensor data available from database.", |
| } |
| |
| |
| if isinstance(self.__sensor, dict) and self.__sensor.get("message") == "Data already predicted": |
| logger.info(f"Skipping binary prediction - data already predicted for UDI: {self.__sensor.get('udi')}") |
| return { |
| "success": False, |
| "error": "Data already predicted", |
| } |
| |
| if ( |
| self.__model.model_binary is None |
| or self.__model.preprocessor_anomaly is None |
| ): |
| logger.error("Binary model or preprocessor not loaded - cannot perform prediction") |
| return { |
| "success": False, |
| "error": "Binary model or preprocessor not loaded.", |
| } |
| |
| try: |
| |
| sensor_dict = self.__sensor if isinstance(self.__sensor, dict) else self.__sensor[0] |
| |
| X_scaled = prepare_sensor_data_for_anomaly( |
| sensor_dict, self.__model.preprocessor_anomaly |
| ) |
| |
| if X_scaled is None: |
| logger.error("Failed to prepare sensor data for binary prediction") |
| 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 |
| |
| logger.info(f"Binary prediction complete - Failure: {bool(is_error)}, Confidence: {confidence:.2f}, Risk: {risk_score:.2f}") |
| |
| result = { |
| "success": True, |
| "failure_predicted": bool(is_error), |
| "confidence": float(f"{confidence}"), |
| "risk_score": float(f"{risk_score}"), |
| } |
| return result |
| except Exception as e: |
| logger.error(f"Exception in binary prediction: {str(e)}") |
| return { |
| "success": False, |
| "message": f"Failed to predict binary data: {str(e)}" |
| } |
|
|
| |
| def predict_classification(self): |
| if self.__sensor is None: |
| logger.warning("Classification prediction attempted with no sensor data") |
| return { |
| "success": False, |
| "error": "No sensor data available from database.", |
| } |
| |
| |
| if isinstance(self.__sensor, dict) and self.__sensor.get("message") == "Data already predicted": |
| logger.info(f"Skipping classification prediction - data already predicted for UDI: {self.__sensor.get('udi')}") |
| return { |
| "success": False, |
| "error": "Data already predicted", |
| } |
|
|
| if ( |
| self.__model.model_multiclass is None |
| or self.__model.preprocessor_anomaly is None |
| ): |
| logger.error("Multiclass model or preprocessor not loaded - cannot perform prediction") |
| return {"success": False, "error": "Model or scalers not loaded."} |
| |
| try: |
| |
| sensor_dict = self.__sensor if isinstance(self.__sensor, dict) else self.__sensor[0] |
| |
| X_scaled = prepare_sensor_data_for_anomaly( |
| sensor_dict, self.__model.preprocessor_anomaly |
| ) |
| |
| if X_scaled is None: |
| logger.error("Failed to prepare sensor data for classification prediction") |
| 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) |
| |
| logger.info(f"Classification prediction complete - Failure Type: {failure_type}, Confidence: {confidence:.2f}, Risk Level: {risk_level}, Risk Score: {risk_score:.2f}") |
| |
| 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, |
| } |
| except Exception as e: |
| logger.error(f"Exception in classification prediction: {str(e)}") |
| return { |
| "success": False, |
| "message": f"Failed to predict classification data: {str(e)}" |
| } |
|
|
| |
| def predict_time_series(self): |
| |
| if self.__sensor is None: |
| logger.warning("Time series prediction attempted with no sensor data") |
| return {"success": False, "error": "No sensor data available."} |
| |
| if isinstance(self.__sensor, dict) and self.__sensor.get("message") == "Data already predicted": |
| logger.info(f"Skipping time series - data already predicted for UDI: {self.__sensor.get('udi')}") |
| return {"success": False, "error": "Data already predicted"} |
|
|
| if self.__model.model_lstm is None or self.__model.scaler_lstm is None: |
| logger.error("LSTM model or scaler_x not loaded") |
| return {"success": False, "error": "Model or scalers not loaded."} |
| |
| try: |
| sensor_data = self.__sensor if isinstance(self.__sensor, list) else [self.__sensor] |
| |
| input_data = [] |
| for row in sensor_data: |
| input_data.append([ |
| row.get("air_temp"), |
| row.get("process_temp"), |
| row.get("rotational_speed"), |
| row.get("torque"), |
| row.get("tool_wear", 0) |
| ]) |
| |
| input_array = np.array(input_data) |
| if len(input_array) < 30: |
| missing = 30 - len(input_array) |
| padding = np.tile(input_array[0], (missing, 1)) |
| input_array = np.vstack([padding, input_array]) |
| elif len(input_array) > 30: |
| input_array = input_array[-30:] |
| |
| input_scaled = self.__model.scaler_lstm.transform(input_array) |
| input_reshaped = input_scaled.reshape(1, 30, 5) |
| health_score = float(self.__model.model_lstm.predict(input_reshaped, verbose=0)[0][0]) |
| |
| |
| MAX_LIFE_DAYS = 7.0 |
| MAX_LIFE_MINUTES = MAX_LIFE_DAYS * 24 * 60 |
| |
| rul_minutes_left = health_score * MAX_LIFE_MINUTES |
| days_remaining = health_score * MAX_LIFE_DAYS |
| |
| now_str = sensor_data[0].get("timestamp") |
| if isinstance(now_str, str): |
| now = datetime.fromisoformat(now_str.replace('Z', '+00:00')) |
| else: |
| now = datetime.now() |
| |
| failure_date = now + timedelta(minutes=rul_minutes_left) |
| |
| if days_remaining < 1.0: |
| status = "Critical" |
| elif days_remaining < 3.0: |
| status = "Warning" |
| else: |
| status = "Good" |
|
|
| logger.info(f"Health: {health_score*100:.1f}% -> {days_remaining:.2f} Days Left") |
| |
| return { |
| "success": True, |
| "predictions": { |
| "health_score": round(health_score, 4), |
| "health_percentage": round(health_score * 100, 2), |
| "rul_minutes": round(rul_minutes_left, 2), |
| "days_remaining": round(days_remaining, 2), |
| "predicted_failure_date": failure_date.strftime('%Y-%m-%d %H:%M:%S'), |
| "status": status |
| }, |
| "input_timestamp": str(now_str), |
| } |
| except Exception as e: |
| logger.error(f"Exception in time series prediction: {str(e)}") |
| return { |
| "success": False, |
| "message": f"Failed to predict time-series data: {str(e)}" |
| } |
