File size: 8,288 Bytes
57517e4 0781dee 57517e4 0781dee 1770619 0781dee 57517e4 0781dee 57517e4 0781dee 57517e4 0781dee 57517e4 0781dee dfaeaf8 0781dee 57517e4 0781dee 57517e4 0781dee 57517e4 0781dee 57517e4 0781dee 57517e4 0781dee 57517e4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 |
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
# Display library versions
print(f"Gradio version: {gr.__version__}")
print(f"NumPy version: {np.__version__}")
print(f"Scikit-learn version: {sklearn.__version__}")
print(f"Joblib version: {joblib.__version__}")
print(f"TensorFlow version: {tf.__version__}")
print(f"Pandas version: {pd.__version__}")
# Directory paths for the saved models
script_dir = os.path.dirname(os.path.abspath(__file__))
scaler_path = os.path.join(script_dir, 'toolkit', 'scaler_X.json')
rf_model_path = os.path.join(script_dir, 'toolkit', 'rf_model.joblib')
mlp_model_path = os.path.join(script_dir, 'toolkit', 'mlp_model.keras')
meta_model_path = os.path.join(script_dir, 'toolkit', 'meta_model.joblib')
image_path = os.path.join(script_dir, 'toolkit', 'car.png')
# Load the scaler and models
try:
# Load the scaler
with open(scaler_path, 'r') as f:
scaler_params = json.load(f)
scaler_X = MinMaxScaler()
scaler_X.scale_ = np.array(scaler_params["scale_"])
scaler_X.min_ = np.array(scaler_params["min_"])
scaler_X.data_min_ = np.array(scaler_params["data_min_"])
scaler_X.data_max_ = np.array(scaler_params["data_max_"])
scaler_X.data_range_ = np.array(scaler_params["data_range_"])
scaler_X.n_features_in_ = scaler_params["n_features_in_"]
scaler_X.feature_names_in_ = np.array(scaler_params["feature_names_in_"])
# Load the models
loaded_rf_model = load(rf_model_path)
print("Random Forest model loaded successfully.")
loaded_mlp_model = load_model(mlp_model_path)
print("MLP model loaded successfully.")
loaded_meta_model = load(meta_model_path)
print("Meta model loaded successfully.")
except Exception as e:
print(f"Error loading models or scaler: {e}")
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)
# 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()
|