Update app.py

app.py

@@ -1,116 +1,120 @@
 import gradio as gr
 import pandas as pd
-import random
-from datetime import datetime, timedelta
+from io import BytesIO
+from fpdf import FPDF
+import matplotlib.pyplot as plt
 from groq import Groq
 
 # Initialize Groq API
 client = Groq(api_key="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")  # Replace with your Groq API key
 
-# Predefined resource inference logic
-def infer_resources(schedule):
-    resource_map = {
-        "Excavation": {"labor": 10, "equipment": "Excavator", "material": "Soil"},
-        "Foundation": {"labor": 15, "equipment": "Concrete Mixer", "material": "Concrete"},
-        "Framing": {"labor": 20, "equipment": "Cranes", "material": "Steel"},
-        "Finishing": {"labor": 5, "equipment": "Hand Tools", "material": "Paint"}
-    }
-
-    inferred_resources = []
-    for _, row in schedule.iterrows():
-        task = row["task"]
-        resources = resource_map.get(task, {"labor": 5, "equipment": "General", "material": "Standard"})
-        inferred_resources.append({
-            "task": task,
-            "labor": resources["labor"],
-            "equipment": resources["equipment"],
-            "material": resources["material"]
-        })
-
-    return pd.DataFrame(inferred_resources)
-
-# Fill missing columns
-def fill_missing_columns(schedule):
-    # Generate random dates if missing
-    if "start_date" not in schedule.columns:
-        schedule["start_date"] = [
-            (datetime.now() + timedelta(days=random.randint(1, 30))).strftime("%Y-%m-%d")
-            for _ in range(len(schedule))
-        ]
-    if "end_date" not in schedule.columns:
-        schedule["end_date"] = [
-            (datetime.strptime(start, "%Y-%m-%d") + timedelta(days=random.randint(5, 15))).strftime("%Y-%m-%d")
-            for start in schedule["start_date"]
-        ]
-    return schedule
-
-# Mock optimization logic
-def mock_optimize_schedule(schedule_with_resources):
-    optimized_schedule = []
-    conflicts = []
-
-    for _, row in schedule_with_resources.iterrows():
-        task = row["task"]
-        start_date = row["start_date"]
-        end_date = row["end_date"]
-        labor = row["labor"]
-        equipment = row["equipment"]
-        material = row["material"]
-
-        # Check for conflicts (mock logic)
-        if labor > 20:  # Example conflict condition
-            conflicts.append(f"Task '{task}' exceeds labor capacity.")
-        
-        optimized_schedule.append({
-            "task": task,
-            "start_date": start_date,
-            "end_date": end_date,
-            "labor": labor,
-            "equipment": equipment,
-            "material": material,
-            "conflict": "Yes" if f"Task '{task}' exceeds labor capacity." in conflicts else "No"
-        })
-
-    return pd.DataFrame(optimized_schedule), conflicts
-
-# Main function for resource optimization
-def optimize_resources(schedule_file):
+# Function to interact with GROQ API for optimization
+def optimize_schedule_with_groq(schedule):
     try:
-        # Load schedule file
-        schedule = pd.read_csv(schedule_file.name)
-
-        # Ensure the 'task' column exists
-        if "task" not in schedule.columns:
-            raise ValueError("The uploaded schedule must contain a 'task' column.")
-
-        # Fill missing columns
-        schedule = fill_missing_columns(schedule)
-
-        # Infer resources
-        inferred_resources = infer_resources(schedule)
-        schedule_with_resources = pd.concat([schedule, inferred_resources], axis=1)
-
-        # Perform optimization (mocked for now)
-        optimized_schedule_df, conflicts = mock_optimize_schedule(schedule_with_resources)
-
-        return optimized_schedule_df, "\n".join(conflicts) if conflicts else "No conflicts detected."
+        optimized_schedule = client.optimize_schedule(schedule.to_dict(orient="records"))
+        return pd.DataFrame(optimized_schedule)
     except Exception as e:
-        return f"Error: {e}"
-
-# Define Gradio interface
-interface = gr.Interface(
-    fn=optimize_resources,
-    inputs=[
-        gr.File(label="Upload Schedule File (CSV)")
-    ],
+        print(f"Error using GROQ API: {e}")
+        raise e
+
+# Function to generate PDF report
+def generate_pdf_report(schedule, conflicts, cost_estimates):
+    pdf = FPDF()
+    pdf.set_auto_page_break(auto=True, margin=15)
+    pdf.add_page()
+
+    # Title
+    pdf.set_font("Arial", "B", 16)
+    pdf.cell(200, 10, txt="Project Schedule Optimization Report", ln=True, align="C")
+
+    # Schedule Table
+    pdf.set_font("Arial", "B", 12)
+    pdf.cell(200, 10, txt="Optimized Schedule", ln=True, align="L")
+    pdf.set_font("Arial", size=10)
+    for index, row in schedule.iterrows():
+        pdf.cell(0, 10, txt=f"{row['tasks']}: {row['start_date']} - {row['end_date']}", ln=True)
+
+    # Conflict Details
+    pdf.set_font("Arial", "B", 12)
+    pdf.cell(200, 10, txt="\nConflict Details", ln=True, align="L")
+    pdf.set_font("Arial", size=10)
+    for conflict in conflicts:
+        pdf.cell(0, 10, txt=f"{conflict}", ln=True)
+
+    # Cost Estimates
+    pdf.set_font("Arial", "B", 12)
+    pdf.cell(200, 10, txt="\nCost Estimates", ln=True, align="L")
+    pdf.set_font("Arial", size=10)
+    for resource, cost in cost_estimates.items():
+        pdf.cell(0, 10, txt=f"{resource}: ${cost:.2f}", ln=True)
+
+    # Save PDF to a buffer
+    buffer = BytesIO()
+    pdf.output(buffer)
+    buffer.seek(0)
+    return buffer
+
+# Generate a sample visualization
+def generate_visualization(schedule):
+    fig, ax = plt.subplots(figsize=(10, 6))
+    tasks = schedule['tasks']
+    start_dates = pd.to_datetime(schedule['start_date'])
+    end_dates = pd.to_datetime(schedule['end_date'])
+    durations = (end_dates - start_dates).dt.days
+
+    ax.barh(tasks, durations, left=start_dates.map(lambda x: x.toordinal()), color='skyblue')
+    ax.set_xlabel("Dates")
+    ax.set_ylabel("Tasks")
+    ax.set_title("Gantt Chart (Simplified)")
+
+    # Convert figure to BytesIO object
+    buf = BytesIO()
+    plt.savefig(buf, format='png')
+    buf.seek(0)
+    return buf
+
+# Main function
+def process_schedule(file):
+    schedule = pd.read_csv(file)
+
+    # Basic checks
+    if 'tasks' not in schedule.columns:
+        return "Error: 'tasks' column is mandatory.", None
+
+    # Infer missing columns (dummy inference for demonstration)
+    if 'start_date' not in schedule.columns:
+        schedule['start_date'] = pd.date_range("2025-01-01", periods=len(schedule))
+    if 'end_date' not in schedule.columns:
+        schedule['end_date'] = schedule['start_date'] + pd.to_timedelta(7, unit='d')
+    if 'required_resources' not in schedule.columns:
+        schedule['required_resources'] = ["Labor"] * len(schedule)
+
+    # Use GROQ API to optimize the schedule
+    optimized_schedule = optimize_schedule_with_groq(schedule)
+
+    # Placeholder conflict and cost calculations
+    conflicts = ["Task 1 and Task 2 overlap.", "Resource 'Crane' exceeds availability."]
+    cost_estimates = {"Labor": 5000, "Equipment": 2000}
+
+    # Generate PDF Report
+    pdf_report = generate_pdf_report(optimized_schedule, conflicts, cost_estimates)
+
+    # Generate Visualization
+    gantt_chart = generate_visualization(optimized_schedule)
+
+    return pdf_report, gantt_chart
+
+# Gradio interface
+iface = gr.Interface(
+    fn=process_schedule,
+    inputs=gr.File(label="Upload Schedule File (CSV)"),
     outputs=[
-        gr.Dataframe(label="Optimized Schedule"),  # Tabular output
-        gr.Textbox(label="Conflicts")  # Text output for conflict details
+        gr.File(label="Download PDF Report"),
+        gr.Image(label="Visualization (Gantt Chart)")
     ],
-    title="Dynamic Intelligent Resource Loading",
-    description="Upload a construction schedule with at least a 'task' column. The app will dynamically infer other details and optimize the schedule."
+    title="Intelligent Resource Loading in Construction Schedule",
+    description="Upload a schedule file to generate a PDF report with optimized schedule, conflict details, cost estimates, and visualizations."
 )
 
-# Launch the app
 if __name__ == "__main__":
-    interface.launch()
+    iface.launch()