Update app.py

app.py

@@ -0,0 +1,315 @@
+from zlib import crc32
+import struct
+import gradio as gr
+import os
+import pandas as pd
+import numpy as np
+import joblib
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+
+# Define top features
+top_features = set([
+    'pm.vbatMV', 'stateEstimate.z', 'motor.m3', 'stateEstimate.yaw', 'yaw_cos',
+    'motor.m2', 'stateEstimate.y', 'stateEstimate.x', 'motor.m1', 'theta',
+    'motor.m4', 'position_magnitude', 'combined_orientation', 'pwm.m3_pwm',
+    'stateEstimate.roll', 'phi', 'pwm.m2_pwm', 'roll_cos', 'vx_cosine',
+    'stateEstimate.vx', 'velocity_magnitude', 'stateEstimate.vy', 'pwm.m4_pwm',
+    'stateEstimate.vz', 'pwm.m1_pwm'
+])
+
+# Load the median values from the CSV once
+feature_medians = pd.read_csv("model/feature_medians.csv")
+medians_dict = feature_medians.set_index('Feature')['Median'].to_dict()
+
+# Load the label encoder, scaler, and saved feature names
+label_encoder = joblib.load('model/label_encoder.pkl')
+scaler = joblib.load('model/scaler.pkl')
+saved_feature_names = joblib.load('model/feature_names.pkl')
+
+# Define the EnhancedFaultDetectionNN model
+class EnhancedFaultDetectionNN(nn.Module):
+    def __init__(self, input_size, output_size, dropout_prob=0.08):
+        super(EnhancedFaultDetectionNN, self).__init__()
+
+        self.fc1 = nn.Linear(input_size, 1024)
+        self.bn1 = nn.BatchNorm1d(1024)
+        self.fc2 = nn.Linear(1024, 512)
+        self.bn2 = nn.BatchNorm1d(512)
+        self.fc3 = nn.Linear(512, 256)
+        self.bn3 = nn.BatchNorm1d(256)
+        self.fc4 = nn.Linear(256, output_size)
+        self.dropout = nn.Dropout(dropout_prob)
+
+    def forward(self, x):
+        x = F.relu(self.bn1(self.fc1(x)))
+        x = self.dropout(x)
+        x = F.relu(self.bn2(self.fc2(x)))
+        x = self.dropout(x)
+        x = F.relu(self.bn3(self.fc3(x)))
+        x = self.dropout(x)
+        x = self.fc4(x)
+        return x
+
+# Load the PyTorch model
+model_path = 'model/best_model_without_oversampling128.pth'
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+input_size = len(saved_feature_names)
+output_size = len(label_encoder.classes_)
+
+model = EnhancedFaultDetectionNN(input_size, output_size).to(device)
+model.load_state_dict(torch.load(model_path, map_location=device))
+model.eval()
+
+# Mapping of fault types to corresponding images and comments
+defect_image_map = {
+
+    "Extra Weight": {
+        "image": "images/weight.png",
+        "comment": "A weight added near the M3 motor causes lift imbalance."
+    },
+    "Propeller Cut": {
+        "image": "images/propeller_cut.png",
+        "comment": "A cut on the M2 propeller reduces thrust and causes instability."
+    },
+    "Tape on Propeller": {
+        "image": "images/tape.png",
+        "comment": "Tape on the M3 propeller leads to imbalance, drag, and vibrations, reducing stability."
+    },
+    "Normal Flight": {
+        "image": "images/normal_flight.png",
+        "comment": "The quadcopter operates normally with balanced thrust and stability."
+    },
+}
+
+# List of log files corresponding to the fault types
+log_files = [
+    "Logs_Samples/add_weight_W1_near_M3_E9_log04",
+    "Logs_Samples/cut_M2_0.5мм_46.5мм_E9_log02",
+    "Logs_Samples/tape_on_propeller_M3_E9_log01",
+    "Logs_Samples/normal_flight_E8_log03"
+]
+
+# Mapping simplified labels to their corresponding folder names
+LabelsMap = {
+    "Extra Weight": "add_weight_W1_near_M3",
+    "Propeller Cut": "cut_M2_0.5мм_46.5мм",
+    "Tape on Propeller": "tape_on_propeller_M3",
+    "Normal Flight": "normal_flight"
+}
+
+# Function to retrieve the log file path using LabelsMap and log_files
+def get_log_file_path(label_key):
+    label_value = LabelsMap[label_key]
+    for log_file in log_files:
+        if label_value in log_file:
+            return log_file
+    return None  # Return None if no matching file is found
+
+def get_name(data, idx):
+    end_idx = idx
+    while data[end_idx] != 0:
+        end_idx += 1
+    return data[idx:end_idx].decode("utf-8"), end_idx + 1
+
+def cfusdlog_decode(file):
+    data = file.read()
+
+    if data[0] != 0xBC:
+        raise gr.Error("Invalid file format: Magic header not found.")
+
+    crc = crc32(data[0:-4])
+    expected_crc, = struct.unpack('I', data[-4:])
+    if crc != expected_crc:
+        raise gr.Error("File integrity check failed: CRC mismatch.")
+
+    version, num_event_types = struct.unpack('HH', data[1:5])
+    if version not in [1, 2]:
+        raise gr.Error(f"Unsupported log file version: {version}")
+
+    result = {}
+    event_by_id = {}
+    idx = 5
+
+    for _ in range(num_event_types):
+        event_id, = struct.unpack('H', data[idx:idx+2])
+        idx += 2
+        event_name, idx = get_name(data, idx)
+        result[event_name] = {'timestamp': []}
+        num_variables, = struct.unpack('H', data[idx:idx+2])
+        idx += 2
+        fmt_str = "<"
+        variables = []
+        for _ in range(num_variables):
+            var_name_and_type, idx = get_name(data, idx)
+            var_name = var_name_and_type[:-3]
+            var_type = var_name_and_type[-2]
+            result[event_name][var_name] = []
+            fmt_str += var_type
+            variables.append(var_name)
+        event_by_id[event_id] = {
+            'name': event_name,
+            'fmt_str': fmt_str,
+            'num_bytes': struct.calcsize(fmt_str),
+            'variables': variables,
+        }
+
+    while idx < len(data) - 4:
+        if version == 1:
+            event_id, timestamp = struct.unpack('<HI', data[idx:idx+6])
+            idx += 6
+        elif version == 2:
+            event_id, timestamp = struct.unpack('<HQ', data[idx:idx+10])
+            timestamp /= 1000.0
+            idx += 10
+        event = event_by_id[event_id]
+        event_data = struct.unpack(event['fmt_str'], data[idx:idx+event['num_bytes']])
+        idx += event['num_bytes']
+        for var, value in zip(event['variables'], event_data):
+            result[event['name']][var].append(value)
+        result[event['name']]['timestamp'].append(timestamp)
+
+    for event_name, event_data in result.items():
+        for var_name, var_data in event_data.items():
+            result[event_name][var_name] = np.array(var_data)
+
+    return {k: v for k, v in result.items() if len(v['timestamp']) > 0}  # Ensure that only non-empty timestamps are kept
+
+def fix_time(log_data):
+    try:
+        timestamps = log_data["timestamp"]
+        if len(timestamps) == 0:
+            raise gr.Error("Timestamp data is empty.")
+        first_value = timestamps[0]
+        log_data["timestamp"] = [t - first_value for t in timestamps]
+    except KeyError:
+        raise gr.Error("Timestamp key not found in the log data.")
+    except Exception as e:
+        raise gr.Error(f"Failed to adjust timestamps: {e}")
+
+def process_log_file(file):
+    try:
+        log_data = cfusdlog_decode(file)
+        log_data = log_data.get('fixedFrequency', {})
+        if not log_data:
+            raise gr.Warning(f"No 'fixedFrequency' data found in the log file")
+
+        fix_time(log_data)
+        parent_dir_name = os.path.basename(os.path.dirname(file.name))
+        log_data["true_label"] = [parent_dir_name] * len(log_data.get("timestamp", []))
+
+        df = pd.DataFrame(log_data)
+        return df
+    except Exception as e:
+        raise gr.Error(f"Failed to process log file: {e}")
+
+def preprocess_single_data_point(single_data_point):
+    try:
+        if 'timestamp' in single_data_point.columns:
+            single_data_point.drop(columns=["timestamp"], inplace=True)
+
+        single_data_point.fillna(medians_dict, inplace=True)
+
+        state_x, state_y, state_z = single_data_point[['stateEstimate.x', 'stateEstimate.y', 'stateEstimate.z']].values.T
+        single_data_point['r'] = np.sqrt(state_x**2 + state_y**2 + state_z**2)
+        single_data_point['theta'] = np.arccos(np.clip(single_data_point['stateEstimate.z'] / single_data_point['r'], -1.0, 1.0))  # Clip to avoid invalid values
+        single_data_point['phi'] = np.arctan2(single_data_point['stateEstimate.y'], single_data_point['stateEstimate.x'])
+        single_data_point['position_magnitude'] = single_data_point['r']
+
+        velocity_x, velocity_y, velocity_z = single_data_point[['stateEstimate.vx', 'stateEstimate.vy', 'stateEstimate.vz']].values.T
+        single_data_point['velocity_magnitude'] = np.sqrt(velocity_x**2 + velocity_y**2 + velocity_z**2)
+        single_data_point['vx_cosine'] = np.divide(velocity_x, single_data_point['velocity_magnitude'], out=np.zeros_like(velocity_x), where=single_data_point['velocity_magnitude']!=0)
+        single_data_point['vy_cosine'] = np.divide(velocity_y, single_data_point['velocity_magnitude'], out=np.zeros_like(velocity_y), where=single_data_point['velocity_magnitude']!=0)
+        single_data_point['vz_cosine'] = np.divide(velocity_z, single_data_point['velocity_magnitude'], out=np.zeros_like(velocity_z), where=single_data_point['velocity_magnitude']!=0)
+
+        roll, yaw = single_data_point[['stateEstimate.roll', 'stateEstimate.yaw']].values.T
+        single_data_point['combined_orientation'] = roll + yaw
+        single_data_point['roll_sin'] = np.sin(np.radians(roll))
+        single_data_point['roll_cos'] = np.cos(np.radians(roll))
+        single_data_point['yaw_sin'] = np.sin(np.radians(yaw))
+        single_data_point['yaw_cos'] = np.cos(np.radians(yaw))
+
+        features_to_keep = list(top_features.intersection(single_data_point.columns))
+        return single_data_point[features_to_keep + ['true_label']]
+    except Exception as e:
+        raise gr.Error(f"Failed to preprocess single data point: {e}")
+
+def predict(file_path):
+    try:
+        with open(file_path, 'rb') as file:
+            log_df = process_log_file(file)
+            if log_df is not None:
+                single_data_point = log_df.sample(1)
+                preprocessed_data_point = preprocess_single_data_point(single_data_point)
+                if preprocessed_data_point is not None:
+                    X = preprocessed_data_point.drop(columns=['true_label'])
+                    y = preprocessed_data_point['true_label']
+
+                    X_ordered = X[saved_feature_names]
+                    X_scaled = scaler.transform(X_ordered)
+                    X_tensor = torch.tensor(X_scaled, dtype=torch.float32).to(device)
+
+                    with torch.no_grad():
+                        logits = model(X_tensor)
+                        probabilities = F.softmax(logits, dim=1)
+                        confidence_scores, predicted_classes = torch.max(probabilities, dim=1)
+
+                    predicted_labels = label_encoder.inverse_transform(predicted_classes.cpu().numpy())
+                    confidence_scores = confidence_scores.cpu().numpy()
+
+                    predicted_label_value = predicted_labels[0]
+                    predicted_label_key = [k for k, v in LabelsMap.items() if v == predicted_label_value][0]
+                    label_confidence_pairs = f"{predicted_label_key}: {predicted_label_value} (Confidence: {confidence_scores[0]:.4f})"
+
+                    # Retrieve the corresponding image and comment using the key name
+                    defect_info = defect_image_map.get(predicted_label_key, {"image": "images/Placeholder.png", "comment": "No information available."})
+                    image_path = defect_info["image"]
+                    comment = defect_info["comment"]
+
+                    return image_path, f"{label_confidence_pairs}\n\nComment: {comment}"
+            else:
+                raise gr.Warning("Log file processing returned no data.")
+    except Exception as e:
+        raise gr.Error(f"Failed to process file: {e}")
+
+    return None, "Failed to process file"
+
+# Gradio interface
+with gr.Blocks() as demo:
+    gr.Markdown("## Fault Detection in Nano-Quadcopter")
+    gr.Markdown("This interface classifies faults in a nano-quadcopter using a deep neural network model.")
+
+    with gr.Row():
+        with gr.Column():
+            example_dropdown = gr.Dropdown(
+                choices=["Extra Weight", "Propeller Cut", "Tape on Propeller", "Normal Flight"],
+                label="Select Fault Type"
+            )
+            submit_btn = gr.Button("Classify")
+
+        with gr.Column():
+            image_output = gr.Image(type="filepath", label="Corresponding Image")
+            label_output = gr.Textbox(label="Predicted Label and Confidence Score")
+
+    def classify_example(example):
+        try:
+            file_path = get_log_file_path(example)
+            if file_path:
+                file_path = file_path
+                image_path, label_and_comment = predict(file_path)
+                return image_path, label_and_comment
+            else:
+                raise gr.Error("No matching log file found.")
+        except KeyError as e:
+            raise gr.Error(f"Error: {e}")
+
+    submit_btn.click(
+        fn=classify_example,
+        inputs=[example_dropdown],
+        outputs=[image_output, label_output],
+    )
+
+# Launch the app
+if __name__ == "__main__":
+    demo.launch(share=True, debug=True)