Update app.py

app.py

@@ -1,20 +1,26 @@
-import pandas as pd
+import datetime
+import logging
+import sys
 import requests
 import gradio as gr
-import logging
-from uuid import uuid4
 from simple_salesforce import Salesforce
 
-# Salesforce credentials - replace with yours securely
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    handlers=[logging.StreamHandler(sys.stdout), logging.FileHandler("app_log.txt")]
+)
+logger = logging.getLogger(__name__)
+
+# Salesforce credentials — replace with your actual Salesforce credentials
 SALESFORCE_USERNAME = "vaneshdevarapalli866@agentforce.com"
 SALESFORCE_PASSWORD = "vanesh@331"
 SALESFORCE_SECURITY_TOKEN = "VRUVbBOdG0s9Q4xy0W6DB1Y6b"
 
-# Configure logging
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
-logger = logging.getLogger(__name__)
+# Hugging Face model endpoint URL — replace with your real endpoint URL
+HUGGINGFACE_API_URL = "https://your-huggingface-model-endpoint"
 
-# Connect to Salesforce
 def connect_to_salesforce():
     try:
         sf_instance = Salesforce(
@@ -26,136 +32,115 @@ def connect_to_salesforce():
         logger.info("Connected to Salesforce successfully.")
         return sf_instance
     except Exception as e:
-        logger.error(f"Salesforce connection failed: {e}")
+        logger.error(f"Failed to connect to Salesforce: {e}")
         raise
 
 sf = connect_to_salesforce()
 
-# Load dataset
-def load_dataset(file_path="equipment_data.csv"):
+def get_ai_suggestion(usage_hours, idle_hours, movement_frequency, cost_per_hour, last_maintenance_str):
     try:
-        df = pd.read_csv(file_path)
-        required_columns = ["Equipment_ID__c", "Equipment_Type__c", "Usage_Hours__c", "Idle_Hours__c"]
-        optional_columns = ["Movement_Frequency__c", "Cost_Per_Hour__c"]
-        missing_required = [col for col in required_columns if col not in df.columns]
-        if missing_required:
-            logger.error(f"Missing required columns: {missing_required}")
-            return None
-
-        # Convert numeric columns properly
-        numeric_cols = [col for col in required_columns + optional_columns if col in df.columns and col not in ["Equipment_ID__c", "Equipment_Type__c"]]
-        for col in numeric_cols:
-            df[col] = pd.to_numeric(df[col], errors='coerce')
-
-        # Fill NaNs with 0 for numeric columns
-        if df[numeric_cols].isnull().any().any():
-            logger.warning("NaN detected in numeric columns, filling with 0")
-            df[numeric_cols] = df[numeric_cols].fillna(0)
-
-        # Add missing optional columns
-        for col in optional_columns:
-            if col not in df.columns:
-                df[col] = 0.0
-        return df
+        payload = {
+            "usage_hours": usage_hours,
+            "idle_hours": idle_hours,
+            "movement_frequency": movement_frequency,
+            "cost_per_hour": cost_per_hour,
+            "last_maintenance": last_maintenance_str or ""
+        }
+        logger.info(f"Calling AI model with payload: {payload}")
+        response = requests.post(HUGGINGFACE_API_URL, json=payload, timeout=15)
+        response.raise_for_status()
+        data = response.json()
+        logger.info(f"Received AI response: {data}")
+
+        suggestion = data.get("suggestion", "No Action")
+        confidence = float(data.get("confidence", 0))
+        utilization_score = float(data.get("utilization_score", 0))
+        return suggestion, confidence, utilization_score
     except Exception as e:
-        logger.error(f"Failed to load dataset: {e}")
-        return None
-
-df = load_dataset()
-equipment_types = sorted(df["Equipment_Type__c"].dropna().unique().tolist()) if df is not None and not df.empty else ["No Equipment Types"]
-suggestion_types = ["Move", "Pause Rent", "Repair", "Replace"]
-
-# Dummy AI suggestion function (Replace with your actual AI integration)
-def call_model(row):
-    # Example logic for AI suggestion (random or based on usage hours)
-    usage = float(row.get("Usage_Hours__c", 0))
-    if usage > 8:
-        return "Pause Rent", 0.9
-    elif usage > 4:
-        return "Move", 0.8
-    else:
-        return "Repair", 0.7
-
-# Filter equipment by type and suggestion
-def filter_equipment(equipment_type, suggestion):
-    if not equipment_type or not suggestion or df is None or df.empty:
-        return "No data available or invalid filters selected.", ""
+        logger.error(f"Error while calling AI model: {e}")
+        # Return fallback default values
+        return "No Action", 0, 0
+
+def process_equipment_utilization(equipment_id, project_id, usage_hours, idle_hours,
+                                  movement_frequency, cost_per_hour, report_link, last_maintenance):
+    last_maintenance_str = last_maintenance.strftime("%Y-%m-%d") if last_maintenance else None
+
+    # Get AI outputs
+    ai_suggestion, suggestion_confidence, utilization_score = get_ai_suggestion(
+        usage_hours, idle_hours, movement_frequency, cost_per_hour, last_maintenance_str
+    )
+
+    salesforce_record = {
+        "Equipment_ID__c": equipment_id,
+        "Project_ID__c": project_id,
+        "Usage_Hours__c": usage_hours,
+        "Idle_Hours__c": idle_hours,
+        "AI_Suggestion__c": ai_suggestion,
+        "Suggestion_Confidence__c": suggestion_confidence,
+        "Utilization_Score__c": utilization_score,
+        "Report_Link__c": report_link,
+        "Last_Maintenance__c": last_maintenance_str,
+        "Dashboard_Flag__c": False
+    }
+
     try:
-        filtered = df[df["Equipment_Type__c"].str.lower() == equipment_type.lower()].copy()
-        if filtered.empty:
-            return f"No equipment found for type: {equipment_type}.", ""
-
-        # Apply dummy AI model predictions
-        filtered["AI_Suggestion__c"] = None
-        filtered["Suggestion_Confidence__c"] = 0.0
-        for idx, row in filtered.iterrows():
-            s, conf = call_model(row)
-            filtered.at[idx, "AI_Suggestion__c"] = s
-            filtered.at[idx, "Suggestion_Confidence__c"] = conf
-
-        # Filter by suggestion
-        filtered = filtered[filtered["AI_Suggestion__c"].str.lower() == suggestion.lower()]
-        if filtered.empty:
-            return f"No equipment with suggestion '{suggestion}' for type '{equipment_type}'.", ""
-
-        # Generate display text for equipment and confidence
-        cards = [
-            f"ID: {row['Equipment_ID__c']} | Usage: {row['Usage_Hours__c']:.2f} hrs | "
-            f"Idle: {row['Idle_Hours__c']:.2f} hrs | AI Suggestion: {row['AI_Suggestion__c']} "
-            f"({row['Suggestion_Confidence__c']:.2%})"
-            for _, row in filtered.iterrows()
-        ]
-        confidences = [f"{row['Equipment_ID__c']}: {row['Suggestion_Confidence__c']:.2%}" for _, row in filtered.iterrows()]
-        return "\n\n".join(cards), "\n".join(confidences)
+        response = sf.Equipment_Utilization_Record__c.create(salesforce_record)
+        logger.info(f"Record created in Salesforce: {response}")
+        return {
+            "Salesforce_Record_Id": response.get("id"),
+            "status": "Success",
+            "AI_Suggestion": ai_suggestion,
+            "Suggestion_Confidence": suggestion_confidence,
+            "Utilization_Score": utilization_score
+        }
     except Exception as e:
-        logger.error(f"Error filtering equipment: {e}")
-        return "An error occurred while filtering equipment.", ""
+        logger.error(f"Salesforce create record error: {e}")
+        return {"error": str(e)}
 
-# Export filtered results to CSV file
-def export_csv(equipment_type, suggestion):
-    if not equipment_type or not suggestion or df is None or df.empty:
-        return None
+def gradio_interface(equipment_id, project_id, usage_hours, idle_hours,
+                     movement_frequency, cost_per_hour, report_link, last_maintenance):
+    # Validate numeric inputs
     try:
-        filtered = df[df["Equipment_Type__c"].str.lower() == equipment_type.lower()].copy()
-        if filtered.empty:
-            return None
-        filtered["AI_Suggestion__c"] = None
-        filtered["Suggestion_Confidence__c"] = 0.0
-        for idx, row in filtered.iterrows():
-            s, conf = call_model(row)
-            filtered.at[idx, "AI_Suggestion__c"] = s
-            filtered.at[idx, "Suggestion_Confidence__c"] = conf
-        filtered = filtered[filtered["AI_Suggestion__c"].str.lower() == suggestion.lower()]
-        if filtered.empty:
-            return None
-        filename = f"filtered_equipment_{uuid4().hex[:8]}.csv"
-        filtered.to_csv(filename, index=False)
-        return filename
+        usage_hours = float(usage_hours)
+        idle_hours = float(idle_hours)
+        movement_frequency = float(movement_frequency)
+        cost_per_hour = float(cost_per_hour)
     except Exception as e:
-        logger.error(f"Error exporting CSV: {e}")
-        return None
+        return {"error": f"Invalid numeric input: {e}"}
 
-# Build Gradio UI without theme to avoid errors
-with gr.Blocks() as app:
-    gr.Markdown("# Equipment Utilization Dashboard")
-    gr.Markdown("Filter equipment by type and AI suggestion to optimize utilization.")
-
-    with gr.Row():
-        etype = gr.Dropdown(choices=equipment_types, label="Equipment Type",
-                           value=equipment_types[0] if equipment_types else None)
-        suggestion = gr.Dropdown(choices=suggestion_types, label="Suggestion Type",
-                                value=suggestion_types[0])
-
-    details = gr.Textbox(label="Equipment Details", lines=8, placeholder="Select equipment type and suggestion...")
-    confidence = gr.Textbox(label="Confidence Scores", lines=5, placeholder="Confidence scores will appear here...")
-
-    export_btn = gr.Button("Export to CSV", variant="primary")
-    csv_file = gr.File(label="Download CSV")
-
-    # Define interactions
-    etype.change(fn=filter_equipment, inputs=[etype, suggestion], outputs=[details, confidence])
-    suggestion.change(fn=filter_equipment, inputs=[etype, suggestion], outputs=[details, confidence])
-    export_btn.click(fn=export_csv, inputs=[etype, suggestion], outputs=csv_file)
+    # Validate last maintenance date format
+    try:
+        last_maintenance_dt = datetime.datetime.strptime(last_maintenance, "%Y-%m-%d") if last_maintenance else None
+    except Exception as e:
+        return {"error": f"Invalid date format for Last Maintenance (expected YYYY-MM-DD): {e}"}
+
+    return process_equipment_utilization(
+        equipment_id,
+        project_id,
+        usage_hours,
+        idle_hours,
+        movement_frequency,
+        cost_per_hour,
+        report_link,
+        last_maintenance_dt
+    )
+
+iface = gr.Interface(
+    fn=gradio_interface,
+    inputs=[
+        gr.Textbox(label="Equipment ID (Salesforce Record ID)"),
+        gr.Textbox(label="Project ID (Salesforce Record ID)"),
+        gr.Number(label="Usage Hours", value=0, minimum=0),
+        gr.Number(label="Idle Hours", value=0, minimum=0),
+        gr.Number(label="Movement Frequency", value=0, minimum=0),
+        gr.Number(label="Cost per Hour", value=0, minimum=0),
+        gr.Textbox(label="Report Link (URL)"),
+        gr.Textbox(label="Last Maintenance Date (YYYY-MM-DD)")
+    ],
+    outputs=gr.JSON(label="Salesforce Record Creation Result"),
+    title="Equipment Utilization AI Integration",
+    description="Input equipment data to get AI suggestions and save the record to Salesforce."
+)
 
 if __name__ == "__main__":
-    app.launch(share=False)
+    iface.launch(share=False)