app.py

import gradio as gr
import numpy as np
import json
import joblib
import tensorflow as tf
import pandas as pd
from joblib import load
from tensorflow.keras.models import load_model
from sklearn.preprocessing import MinMaxScaler
import matplotlib.pyplot as plt
import os
import sklearn

# Assuming the previous setup and imports have been done

def predict_and_plot(velocity, temperature, precipitation, humidity):
    try:
        # Prepare the example data
        example_data = pd.DataFrame({
            'Velocity(mph)': [velocity],
            'Temperature': [temperature],
            'Precipitation': [precipitation],
            'Humidity': [humidity]
        })

        # Scale the example data
        example_data_scaled = scaler_X.transform(example_data)

        # Function to predict contamination levels
        def predict_contamination(example_data_scaled):
            # Predict using MLP model
            mlp_predictions_contamination, mlp_predictions_gradients = loaded_mlp_model.predict(example_data_scaled)

            # Predict using RF model
            rf_predictions = loaded_rf_model.predict(example_data_scaled)

            # Combine predictions for meta model
            combined_features = np.concatenate([np.concatenate([mlp_predictions_contamination, mlp_predictions_gradients], axis=1), rf_predictions], axis=1)

            # Predict using meta model
            meta_predictions = loaded_meta_model.predict(combined_features)

            return meta_predictions[:, :6]  # Assuming the first 6 columns are contamination predictions

        # Predict contamination levels for the single example
        contamination_levels = predict_contamination(example_data_scaled)

        # Simulate contamination levels at multiple time intervals
        time_intervals = np.arange(0, 601, 60)  # Simulating time intervals from 0 to 600 seconds

        # Generate simulated contamination levels (linear interpolation between predicted values)
        simulated_contamination_levels = np.array([
            np.linspace(contamination_levels[0][i], contamination_levels[0][i] * 2, len(time_intervals))
            for i in range(contamination_levels.shape[1])
        ]).T

        # Function to calculate cleaning time using linear interpolation
        def calculate_cleaning_time(time_intervals, contamination_levels, threshold=0.4):
            cleaning_times = []
            for i in range(contamination_levels.shape[1]):
                levels = contamination_levels[:, i]
                for j in range(1, len(levels)):
                    if levels[j-1] <= threshold <= levels[j]:
                        # Linear interpolation
                        t1, t2 = time_intervals[j-1], time_intervals[j]
                        c1, c2 = levels[j-1], levels[j]
                        cleaning_time = t1 + (threshold - c1) * (t2 - t1) / (c2 - c1)
                        cleaning_times.append(cleaning_time)
                        break
            return cleaning_times

        # Calculate cleaning times for all 6 lidars
        cleaning_times = calculate_cleaning_time(time_intervals, simulated_contamination_levels)

        # Lidar names
        lidar_names = ['F/L', 'F/R', 'Left', 'Right', 'Roof', 'Rear']

        # Plot the graph
        plt.figure(figsize=(12, 8))

        for i in range(simulated_contamination_levels.shape[1]):
            plt.plot(time_intervals, simulated_contamination_levels[:, i], label=f'{lidar_names[i]}')
            plt.axhline(y=0.4, color='r', linestyle='--', label='Contamination Threshold' if i == 0 else "")
            if i < len(cleaning_times):
                plt.scatter(cleaning_times[i], 0.4, color='k')  # Mark the cleaning time point

        plt.title('Contamination Levels Over Time for Each Lidar')
        plt.xlabel('Time (seconds)')
        plt.ylabel('Contamination Level')
        plt.legend()
        plt.grid(True)
        
        # Save the plot to a file
        plt.savefig('plot.png')
        
        # Return the plot and predictions
        return plt, [f"{val * 100:.2f}%" for val in contamination_levels[0]], [f"{val:.2f}" for val in cleaning_times]

    except Exception as e:
        print(f"Error in Gradio interface: {e}")
        return plt.figure(), ["Error"] * 6, ["Error"] * 6

inputs = [
    gr.Slider(minimum=0, maximum=100, value=50, step=0.05, label="Velocity (mph)"),
    gr.Slider(minimum=-2, maximum=30, value=0, step=0.5, label="Temperature (°C)"),
    gr.Slider(minimum=0, maximum=1, value=0, step=0.01, label="Precipitation (inch)"),
    gr.Slider(minimum=0, maximum=100, value=50, label="Humidity (%)")
]

contamination_outputs = [
    gr.Textbox(label="Front Left Contamination"),
    gr.Textbox(label="Front Right Contamination"),
    gr.Textbox(label="Left Contamination"),
    gr.Textbox(label="Right Contamination"),
    gr.Textbox(label="Roof Contamination"),
    gr.Textbox(label="Rear Contamination")
]

cleaning_time_outputs = [
    gr.Textbox(label="Front Left Cleaning Time"),
    gr.Textbox(label="Front Right Cleaning Time"),
    gr.Textbox(label="Left Cleaning Time"),
    gr.Textbox(label="Right Cleaning Time"),
    gr.Textbox(label="Roof Cleaning Time"),
    gr.Textbox(label="Rear Cleaning Time")
]

with gr.Blocks() as demo:
    gr.Markdown("<h1 style='text-align: center;'>Environmental Factor-Based Contamination & Cleaning Time Prediction</h1>")
    gr.Markdown("This application predicts the contamination levels, corresponding gradients, and cleaning times for different parts of a car's LiDAR system based on environmental factors such as velocity, temperature, precipitation, and humidity.")
    
    with gr.Row():
        with gr.Column():
            gr.Markdown("### Input Parameters")
            for inp in inputs:
                inp.render()

            # Centered image display
            with gr.Row():
                with gr.Column(scale=1, min_width=0):
                    gr.Image(image_path)  # Ensure the image is centered
            
            gr.Button(value="Submit", variant="primary").click(
                fn=predict_and_plot, 
                inputs=inputs, 
                outputs=[gr.Plot(label="Contamination Levels Over Time")] + contamination_outputs + cleaning_time_outputs
            )
            gr.Button(value="Clear").click(fn=lambda: None)
            
        with gr.Column():
            gr.Markdown("### Contamination Predictions")
            for out in contamination_outputs:
                out.render()
                
        with gr.Column():
            gr.Markdown("### Cleaning Time Predictions")
            for out in cleaning_time_outputs:
                out.render()

demo.launch()