import gradio as gr import pandas as pd import numpy as np from sklearn.preprocessing import LabelEncoder from sklearn.ensemble import RandomForestRegressor import matplotlib.pyplot as plt import io # -------------------------- # Global variables # -------------------------- DATA = None ENCODERS = {} MODEL = None # -------------------------- # Load and preprocess dataset # -------------------------- def load_dataset(file_obj): global DATA, ENCODERS try: df = pd.read_csv(file_obj.name) except: df = pd.read_csv(file_obj) preview, msg = preprocess_data(df) return preview, msg def preprocess_data(df): global DATA, ENCENCODERS required_cols = [ "date", "Usage_kWh", "Lagging_Current_Reactive.Power_kVarh", "Leading_Current_Reactive_Power_kVarh", "CO2(tCO2)", "Lagging_Current_Power_Factor", "Leading_Current_Power_Factor", "NSM", "WeekStatus", "Day_of_week", "Load_Type" ] missing = [c for c in required_cols if c not in df.columns] if missing: return pd.DataFrame(), f"❌ Missing columns: {missing}" df['date'] = pd.to_datetime(df['date'], errors='coerce') df['month'] = df['date'].dt.month cat_cols = ["WeekStatus", "Day_of_week", "Load_Type"] global ENCODERS ENCODERS = {} for col in cat_cols: enc = LabelEncoder() df[col] = enc.fit_transform(df[col].astype(str)) ENCODERS[col] = enc global DATA DATA = df return df.head(10), "✅ Dataset uploaded and processed successfully!" # -------------------------- # Train model # -------------------------- def train_model(): global DATA, MODEL if DATA is None: return "❌ Upload dataset first!" features = [ "month", "Lagging_Current_Reactive.Power_kVarh", "Leading_Current_Reactive_Power_kVarh", "CO2(tCO2)", "Lagging_Current_Power_Factor", "Leading_Current_Power_Factor", "NSM", "WeekStatus", "Day_of_week", "Load_Type" ] X = DATA[features] y = DATA["Usage_kWh"] MODEL = RandomForestRegressor(n_estimators=100, random_state=42) MODEL.fit(X, y) return "✅ Model trained successfully!" # -------------------------- # Generate prediction # -------------------------- def generate_prediction(load, week, day, co2, lag, lead, usage, nsm): global MODEL, ENCODERS if MODEL is None: return "❌ Model not trained yet!" try: load_enc = ENCODERS["Load_Type"].transform([load])[0] week_enc = ENCODERS["WeekStatus"].transform([week])[0] day_enc = ENCODERS["Day_of_week"].transform([day])[0] X_input = np.array([[lag, lead, co2, usage, nsm, load_enc, week_enc, day_enc, 0, 0]]) X_input = X_input[:, :10] pred = MODEL.predict(X_input)[0] return round(pred, 2) except Exception as e: return f"❌ Prediction error: {e}" # -------------------------- # Generate plot # -------------------------- def generate_plot(): global DATA if DATA is None or DATA.empty: return None fig, ax = plt.subplots(figsize=(10,5)) ax.plot(DATA['date'], DATA['Usage_kWh'], color='#2563EB') ax.set_title("Energy Usage Over Time", color='#0F172A', fontsize=16) ax.set_xlabel("Date") ax.set_ylabel("Usage (kWh)") ax.grid(True, linestyle='--', alpha=0.5) buf = io.BytesIO() fig.savefig(buf, format='png', bbox_inches='tight') buf.seek(0) plt.close(fig) return buf # -------------------------- # Chatbot function # -------------------------- def on_user_message(msg, load, week, day, co2, lag, lead, usage, nsm): if msg.strip() == "": return [], msg advice = [] if DATA is not None: mean_usage = DATA['Usage_kWh'].mean() if usage > mean_usage: advice.append("⚠️ Your usage is above average. Check machinery.") if lag > 50: advice.append("Consider power factor correction for lagging kVarh.") if lead > 50: advice.append("Check capacitive reactive power usage.") if DATA is not None: co2_75 = np.percentile(DATA['CO2(tCO2)'], 75) if co2 > co2_75: advice.append("⚠️ High CO₂ detected. Improve ventilation or reduce production temporarily.") if not advice: advice.append("✅ All systems normal. No major issues detected.") # Return messages as dictionaries (Gradio v3.43+) return [ {"role": "user", "content": msg}, {"role": "assistant", "content": "\n".join(advice)} ], "" # -------------------------- # Gradio UI # -------------------------- with gr.Blocks() as app: gr.Markdown("## ⚡ Smart Factory Energy Advisor") gr.Markdown("Analyze energy usage, detect inefficiencies, and receive actionable recommendations.") # Dataset upload file = gr.File(label="Upload CSV") upload_btn = gr.Button("Upload Dataset") preview = gr.Dataframe(label="Dataset Preview (First 10 Rows)", interactive=False) status_box = gr.Textbox(label="Status Message", interactive=False) upload_btn.click( fn=load_dataset, inputs=file, outputs=[preview, status_box] ) # User input panel gr.Markdown("### ⚙️ User Input Panel") with gr.Row(): with gr.Column(scale=1): load = gr.Dropdown(["Light","Medium","Maximum"], label="Load Type") week = gr.Dropdown(["Weekday","Weekend"], label="Week Status") day = gr.Dropdown( ["Sunday","Monday","Tuesday","Wednesday","Thursday","Friday","Saturday"], label="Day of Week" ) co2 = gr.Number(label="CO₂ Level (tCO2)") lag = gr.Number(label="Lagging Reactive Power (kVarh)") lead = gr.Number(label="Leading Reactive Power (kVarh)") usage = gr.Number(label="Current Usage (kWh)") nsm = gr.Number(label="NSM (Seconds from Midnight)") # Chatbot gr.Markdown("### 🤖 Chatbot") chatbot = gr.Chatbot() msg = gr.Textbox(label="Ask a question or request advice/prediction") send_btn = gr.Button("Send") send_btn.click( fn=on_user_message, inputs=[msg, load, week, day, co2, lag, lead, usage, nsm], outputs=[chatbot, msg] ) # Train & Predict gr.Markdown("### 🔮 Energy Prediction") train_btn = gr.Button("Train Model") train_out = gr.Textbox(label="Training Status", interactive=False) train_btn.click(fn=train_model, inputs=[], outputs=train_out) pred_btn = gr.Button("Predict Energy Usage") pred_out = gr.Textbox(label="Predicted Usage (kWh)", interactive=False) pred_btn.click( fn=generate_prediction, inputs=[load, week, day, co2, lag, lead, usage, nsm], outputs=pred_out ) # Plot chart gr.Markdown("### 📈 Consumption Chart") plot_btn = gr.Button("Show Consumption Chart") plot_img = gr.Image() plot_btn.click(fn=generate_plot, inputs=[], outputs=plot_img) app.launch()