Spaces:

Rudraaaa76
/

Driving-Behavior

Sleeping

File size: 8,470 Bytes

# from flask import Flask, request, jsonify
# from flask_cors import CORS
# import numpy as np
# import pandas as pd
# import joblib
# import requests
# import os
# import tensorflow as tf  # Required for TFLite interpreter

# # URLs for hosted files
# # tflite_model_url = "https://drive.google.com/uc?id=1j5JU2xD2iwi5STzjH2gKILAbekiKWBkp&export=download"
# # scaler_url = "https://drive.google.com/uc?id=1Qu2ogpNw8MqPbstNpcEbO0oRDQ56qX5X&export=download"
# # label_encoder_url = "https://drive.google.com/uc?id=1qYi5agK5vDKc-k6UcaRt7yL9AFJ3O_7-&export=download"

# # Local file paths
# tflite_model_path = "motion_classification_model.tflite"
# scaler_path = "scaler.pkl"
# label_encoder_path = "label_encoder.pkl"

# # Function to download files if they don't exist locally
# # def download_file(url, local_path):
# #     if not os.path.exists(local_path):
# #         print(f"Downloading {local_path} from {url}...")
# #         response = requests.get(url)
# #         with open(local_path, 'wb') as file:
# #             file.write(response.content)
# #         print(f"Downloaded {local_path}")

# # Download required files
# # download_file(tflite_model_url, tflite_model_path)
# # download_file(scaler_url, scaler_path)
# # download_file(label_encoder_url, label_encoder_path)

# # Load the scaler and label encoder
# scaler = joblib.load(scaler_path)
# label_encoder = joblib.load(label_encoder_path)

# # Initialize the TFLite interpreter
# interpreter = tf.lite.Interpreter(model_path=tflite_model_path)
# interpreter.allocate_tensors()

# # Get input and output tensor details
# input_details = interpreter.get_input_details()
# output_details = interpreter.get_output_details()

# # Feature Columns
# feature_columns = ['AccX', 'AccY', 'AccZ', 'GyroX', 'GyroY', 'GyroZ']
# sequence_length = 50

# # Initialize Flask App
# app = Flask(__name__)
# CORS(app)


# @app.route('/')
# def home():
#     return "Welcome to driving behavior analysis"

# # Define a route for the prediction function
# @app.route('/predict', methods=['POST'])
# def predict_behavior():
#     try:
#         # Get the data from the request
#         data = request.json

#         # Convert the data to a DataFrame
#         df = pd.DataFrame(data)

#         # Validate required columns
#         if not all(col in df.columns for col in feature_columns):
#             return jsonify({'error': 'Missing columns'}), 400

#         # Scale the data
#         df[feature_columns] = scaler.transform(df[feature_columns])

#         # Create sequences
#         sequences = []
#         if len(df) >= sequence_length:
#             # If enough data for full sequences
#             for i in range(len(df) - sequence_length + 1):
#                 seq = df.iloc[i:i+sequence_length][feature_columns].values
#                 sequences.append(seq)
#         else:
#             # Pad the data if it's smaller than the sequence length
#             padded_data = np.pad(
#                 df[feature_columns].values,
#                 ((sequence_length - len(df), 0), (0, 0)),  # Pad missing rows
#                 mode='constant',
#                 constant_values=0
#             )
#             sequences.append(padded_data)

#         # Convert to NumPy array
#         X_input = np.array(sequences, dtype=np.float32)

#         # Make predictions using TFLite model
#         predictions = []
#         for seq in X_input:
#             # Prepare the input tensor
#             interpreter.set_tensor(input_details[0]['index'], [seq])
#             interpreter.invoke()

#             # Get the output tensor
#             output_data = interpreter.get_tensor(output_details[0]['index'])
#             predictions.append(output_data)

#         # Get predicted classes
#         predicted_classes = np.argmax(predictions, axis=2).flatten()

#         # Convert integers to class labels
#         class_labels = label_encoder.inverse_transform(predicted_classes)

#         # Calculate class frequencies
#         unique_classes, counts = np.unique(class_labels, return_counts=True)
#         max_count = np.max(counts)
#         most_frequent_classes = unique_classes[counts == max_count]

#         # Select the first class in case of ties
#         most_frequent_class = most_frequent_classes[0]  # Select the first class alphabetically

#         # Return the predicted class labels and the most frequent class
#         return jsonify({
#             "predicted_classes": list(class_labels),  # Full list of predictions
#             "most_frequent_class": most_frequent_class
#         })

#     except Exception as e:
#         return jsonify({'error': str(e)}), 500

# if __name__ == '__main__':
#     app.run(host="0.0.0.0", port=7860)


from flask import Flask, request, jsonify
from flask_cors import CORS
import numpy as np
import pandas as pd
import joblib
import tensorflow as tf  # Required for TFLite interpreter

# Local file paths
tflite_model_path = "motion_classification_model.tflite"
scaler_path = "scaler.pkl"
label_encoder_path = "label_encoder.pkl"

# Load the scaler and label encoder
scaler = joblib.load(scaler_path)
label_encoder = joblib.load(label_encoder_path)

# Initialize the TFLite interpreter
interpreter = tf.lite.Interpreter(model_path=tflite_model_path)
interpreter.allocate_tensors()

# Get input and output tensor details
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

# Feature Columns
feature_columns = ['AccX', 'AccY', 'AccZ', 'GyroX', 'GyroY', 'GyroZ']
sequence_length = 50

# Initialize Flask App
app = Flask(__name__)
CORS(app)

@app.route('/')
def home():
    return "Welcome to driving behavior analysis API"

@app.route('/predict', methods=['POST'])
def predict_behavior():
    try:
        # Get data from the request
        data = request.json
        df = pd.DataFrame(data)

        # Validate required columns
        if not all(col in df.columns for col in feature_columns):
            return jsonify({'error': 'Missing required sensor columns'}), 400

        # Scale the data
        df[feature_columns] = scaler.transform(df[feature_columns])

        # Compute Jerk (Rate of Change of Acceleration)
        df['JerkX'] = df['AccX'].diff().fillna(0)
        df['JerkY'] = df['AccY'].diff().fillna(0)
        df['JerkZ'] = df['AccZ'].diff().fillna(0)

        # Identify Harsh Braking (Sudden drop in AccX)
        harsh_braking_events = (df['JerkX'] < -3).sum()  # Adjusted threshold from -5 to -3

        # Identify Harsh Cornering (Sharp change in GyroZ)
        harsh_cornering_events = (df['GyroZ'].diff().abs() > 1.8).sum()  # Adjusted threshold from 30 to 1.5

        # Create sequences for model input
        sequences = []
        if len(df) >= sequence_length:
            for i in range(len(df) - sequence_length + 1):
                seq = df.iloc[i:i+sequence_length][feature_columns].values
                sequences.append(seq)
        else:
            # Pad if data is smaller than sequence length
            padded_data = np.pad(
                df[feature_columns].values,
                ((sequence_length - len(df), 0), (0, 0)),
                mode='constant',
                constant_values=0
            )
            sequences.append(padded_data)

        # Convert to NumPy array
        X_input = np.array(sequences, dtype=np.float32)

        # Predict using TFLite model
        predictions = []
        for seq in X_input:
            interpreter.set_tensor(input_details[0]['index'], [seq])
            interpreter.invoke()
            output_data = interpreter.get_tensor(output_details[0]['index'])
            predictions.append(output_data)

        # Get predicted classes
        predicted_classes = np.argmax(predictions, axis=2).flatten()
        class_labels = label_encoder.inverse_transform(predicted_classes)

        # Find most frequent class
        unique_classes, counts = np.unique(class_labels, return_counts=True)
        most_frequent_class = unique_classes[np.argmax(counts)]

        # Return results
        return jsonify({
            "predicted_classes": list(class_labels),
            "most_frequent_class": most_frequent_class,
            "harsh_braking_count": int(harsh_braking_events),
            "harsh_cornering_count": int(harsh_cornering_events)
        })

    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(host="0.0.0.0", port=7860)