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(' 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)