utils.py

from datetime import datetime

import numpy as np
import pandas as pd


def generate_cyclical_features(timestamp):
    if isinstance(timestamp, str):
        timestamp = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
    hour = timestamp.hour
    dayofweek = timestamp.weekday()
    dayofyear = timestamp.timetuple().tm_yday
    month = timestamp.month

    features = {
        "hour_sin": np.sin(2 * np.pi * hour / 24),
        "hour_cos": np.cos(2 * np.pi * hour / 24),
        "dayofweek_sin": np.sin(2 * np.pi * dayofweek / 7),
        "dayofweek_cos": np.cos(2 * np.pi * dayofweek / 7),
        "dayofyear_sin": np.sin(2 * np.pi * dayofyear / 365),
        "dayofyear_cos": np.cos(2 * np.pi * dayofyear / 365),
        "month_sin": np.sin(2 * np.pi * month / 12),
        "month_cos": np.cos(2 * np.pi * month / 12),
    }

    return features


def convert_cyclical_to_original(
    hour_sin,
    hour_cos,
    dayofweek_sin,
    dayofweek_cos,
    dayofyear_sin,
    dayofyear_cos,
    month_sin,
    month_cos,
):
    hour = np.arctan2(hour_sin, hour_cos) * 24 / (2 * np.pi)
    hour = int(np.round(hour % 24))

    dayofweek = np.arctan2(dayofweek_sin, dayofweek_cos) * 7 / (2 * np.pi)
    dayofweek = int(np.round(dayofweek % 7))

    dayofyear = np.arctan2(dayofyear_sin, dayofyear_cos) * 365 / (2 * np.pi)
    dayofyear = int(np.round(dayofyear % 365))
    dayofyear = max(1, dayofyear)

    month = np.arctan2(month_sin, month_cos) * 12 / (2 * np.pi)
    month = int(np.round(month % 12))
    month = 12 if month == 0 else month

    return {
        "hour": hour,
        "dayofweek": dayofweek,
        "dayofyear": dayofyear,
        "month": month,
    }


def create_sequences(data, window_size=32):
    sequences = []
    for i in range(len(data) - window_size + 1):
        sequences.append(data[i : i + window_size])
    return np.array(sequences)


def prepare_prediction_data(sensor_data, timestamp, scaler_x, window_size=32):
    try:
        sensor_features = {
            "Air temperature [K]": sensor_data.get("air_temp", 0),
            "Process temperature [K]": sensor_data.get("process_temp", 0),
            "Rotational speed [rpm]": sensor_data.get("rotational_speed", 0),
            "Torque [Nm]": sensor_data.get("torque", 0),
            "Tool wear [min]": sensor_data.get("tool_wear", 0),
        }

        cyclical_features = generate_cyclical_features(timestamp)

        all_features = {**sensor_features, **cyclical_features}

        column_order = [
            "Air temperature [K]",
            "Process temperature [K]",
            "Rotational speed [rpm]",
            "Torque [Nm]",
            "Tool wear [min]",
            "hour_sin",
            "hour_cos",
            "dayofweek_sin",
            "dayofweek_cos",
            "dayofyear_sin",
            "dayofyear_cos",
            "month_sin",
            "month_cos",
        ]

        X_new = pd.DataFrame([all_features])
        X_new = X_new[column_order]

        if scaler_x is None:
            raise ValueError("scaler_X not loaded.")

        X_scaled = scaler_x.transform(X_new)

        X_sequence = np.repeat(X_scaled, window_size, axis=0).reshape(
            1, window_size, -1
        )

        return X_sequence

    except Exception as e:
        print(f"Error preparing prediction data: {str(e)}")
        import traceback

        traceback.print_exc()
        return None


def create_timestamp_from_predictions(predictions, sensor_timestamp=None):
    try:
        hour = predictions.get("hour", 0)
        dayofyear = predictions.get("dayofyear", 1)

        if sensor_timestamp:
            if isinstance(sensor_timestamp, str):
                input_dt = datetime.fromisoformat(
                    sensor_timestamp.replace("Z", "+00:00")
                )
            else:
                input_dt = sensor_timestamp
            year = input_dt.year
        else:
            year = datetime.now().year

        predicted_dt = datetime.strptime(f"{year}-{dayofyear}", "%Y-%j")
        predicted_dt = predicted_dt.replace(hour=hour, minute=0, second=0)

        return predicted_dt.isoformat()

    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")