Update model.py

model.py

@@ -1,12 +1,5 @@
-from transformers import AutoModelForSeq2SeqLM, AutoTokenizer
-import torch
 import logging
 from typing import Dict, List
-import time
-try:
-    import psutil
-except ImportError:
-    psutil = None
 
 # Configure logging
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
@@ -29,75 +22,124 @@ def get_weather_condition(score: int) -> str:
 
 def call_ai_model_for_insights(input_data: Dict, delay_risk: float) -> List[str]:
     """
-    Use T5-Small in Hugging Face Space (CPU) to generate insights based on input data and delay risk.
+    Generate detailed hardcoded insights based on input data and delay risk, bypassing T5-Small.
+    Returns 3-5 prioritized, phase/task-specific insights tailored to conditions.
     """
-    model_name = "t5-small"
-    max_retries = 3
-    retry_delay = 30  # Increased for network stability
-    
-    # Log system resources if psutil is available
-    if psutil:
-        try:
-            memory = psutil.virtual_memory()
-            logger.info(f"System memory - Total: {memory.total / 1e9:.2f} GB, Available: {memory.available / 1e9:.2f} GB, Used: {memory.percent}%")
-        except Exception as e:
-            logger.warning(f"Failed to log system memory: {str(e)}")
+    logger.info("Generating detailed hardcoded AI insights")
+    phase = input_data.get("phase", "")
+    task = input_data.get("task", "")
+    current_progress = input_data.get("current_progress", 0)
+    expected_duration = input_data.get("task_expected_duration", 0)
+    actual_duration = input_data.get("task_actual_duration", 0)
+    workforce_gap = input_data.get("workforce_gap", 0)
+    skill_level = input_data.get("workforce_skill_level", "").lower()
+    shift_hours = input_data.get("workforce_shift_hours", 0)
+    weather_score = input_data.get("weather_impact_score", 0)
+    weather_condition = input_data.get("weather_condition", get_weather_condition(weather_score))
+
+    # Initialize insights with scores for prioritization
+    insights = []
+
+    # Helper function to add insight with priority score
+    def add_insight(message: str, priority: float):
+        insights.append((message, priority))
+
+    # Delay risk-based insights
+    if delay_risk > 75:
+        add_insight(f"Urgent: Accelerate {phase}: {task} to mitigate high delay risk", 1.0)
+    elif delay_risk > 50:
+        add_insight(f"Monitor {phase}: {task} closely to prevent delays", 0.9)
+    elif delay_risk > 25:
+        add_insight(f"Maintain current pace for {phase}: {task}", 0.7)
     else:
-        logger.warning("psutil not available; cannot log system memory usage")
-    
-    for attempt in range(max_retries):
-        try:
-            logger.info(f"Attempt {attempt + 1}/{max_retries} - Loading model: {model_name}")
-            tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=False, use_fast=True)
-            model = AutoModelForSeq2SeqLM.from_pretrained(
-                model_name,
-                torch_dtype=torch.float32,
-                use_safetensors=True,
-                trust_remote_code=False,
-                low_cpu_mem_usage=True
-            )
-            
-            logger.info("Model loaded successfully. Generating insights...")
-            prompt = f"""
-            Summarize risk:
-            Risk: {delay_risk:.1f}%
-
-            Format: ["Insight 1"].
-            """
-            
-            with torch.no_grad():
-                inputs = tokenizer(prompt, return_tensors="pt", max_length=8, truncation=True).to("cpu")
-                outputs = model.generate(
-                    **inputs,
-                    max_new_tokens=5,
-                    num_beams=1,
-                    temperature=0.7,
-                    do_sample=True
-                )
-            response = tokenizer.decode(outputs[0], skip_special_tokens=True)
-            
-            insights = [line.strip() for line in response.split("\n") if line.strip() and line.strip() not in [prompt]]
-            logger.info(f"Generated insights: {insights}")
-            return insights[:1] or ["No insights generated."]
-        except Exception as e:
-            logger.error(f"Attempt {attempt + 1}/{max_retries} - Model inference failed: {str(e)}")
-            if attempt < max_retries - 1:
-                logger.info(f"Retrying in {retry_delay} seconds...")
-                time.sleep(retry_delay)
-            else:
-                logger.error("Max retries reached. Using fallback insights.")
-                fallback_insights = []
-                if delay_risk > 75:
-                    fallback_insights.append("High risk detected.")
-                elif delay_risk > 50:
-                    fallback_insights.append("Moderate risk found.")
-                return fallback_insights or ["AI model failed to generate insights; check system resources."]
+        add_insight(f"Optimize resource allocation for {phase}: {task}", 0.6)
+
+    # Phase/task-specific insights
+    task_specific = {
+        "Planning": {
+            "Define Scope": "Ensure clear stakeholder alignment for Planning: Define Scope",
+            "Resource Allocation": "Secure budget approvals early for Planning: Resource Allocation",
+            "Permit Acquisition": "Secure permits early for Planning: Permit Acquisition"
+        },
+        "Design": {
+            "Architectural Drafting": "Engage architects early for Design: Architectural Drafting",
+            "Engineering Analysis": "Hire additional engineers for Design: Engineering Analysis",
+            "Design Review": "Schedule thorough reviews for Design: Design Review"
+        },
+        "Construction": {
+            "Foundation Work": "Optimize material delivery for Construction: Foundation Work",
+            "Structural Build": "Ensure equipment availability for Construction: Structural Build",
+            "Utility Installation": "Coordinate subcontractors for Construction: Utility Installation"
+        }
+    }
+    if phase in task_specific and task in task_specific[phase]:
+        add_insight(task_specific[phase][task], 0.8)
+
+    # Workforce-based insights
+    if workforce_gap > 30:
+        add_insight(f"Hire subcontractors immediately to address {workforce_gap}% workforce shortage", 1.0)
+    elif workforce_gap > 15:
+        add_insight(f"Hire additional workers to reduce {workforce_gap}% workforce gap", 0.9)
+    elif workforce_gap > 5:
+        add_insight("Consider temporary staff to address minor workforce gap", 0.7)
+
+    if skill_level == "low":
+        add_insight(f"Provide training to improve low skill levels for {phase}: {task}", 0.9)
+    elif skill_level == "medium" and delay_risk > 50:
+        add_insight(f"Upskill workforce for efficiency in {phase}: {task}", 0.8)
+    elif skill_level == "high" and delay_risk < 25:
+        add_insight("Leverage high skill levels to maintain progress", 0.6)
+
+    if shift_hours < 6:
+        add_insight(f"Extend shift hours beyond {shift_hours} to meet {phase}: {task} deadlines", 0.9)
+    elif shift_hours < 8 and delay_risk > 50:
+        add_insight(f"Increase shift hours to {8} for {phase}: {task}", 0.8)
+    elif shift_hours > 10:
+        add_insight("Balance shifts to prevent burnout", 0.7)
+
+    # Weather-based insights
+    if weather_score > 70:
+        add_insight(f"Reschedule {phase}: {task} to avoid {weather_condition.lower()}", 1.0)
+    elif weather_score > 50:
+        add_insight(f"Shift to indoor tasks during {weather_condition.lower()} for {phase}: {task}", 0.9)
+    elif weather_score > 30:
+        add_insight(f"Continue monitoring {weather_condition.lower()} for {phase}: {task}", 0.7)
+
+    # Progress and duration-based insights
+    if expected_duration > 0 and actual_duration > expected_duration:
+        overrun_pct = ((actual_duration - expected_duration) / expected_duration) * 100
+        if overrun_pct > 20:
+            add_insight(f"Address duration overrun urgently for {phase}: {task}", 1.0)
+        elif overrun_pct > 10:
+            add_insight(f"Review scheduling to address {overrun_pct:.1f}% overrun in {phase}: {task}", 0.8)
+
+    if expected_duration > 0:
+        expected_progress = min((actual_duration / expected_duration) * 100, 100)
+        if current_progress < expected_progress * 0.8:
+            add_insight(f"Accelerate task progress for {phase}: {task} to align with schedule", 0.9)
+        elif current_progress < 50 and delay_risk > 50:
+            add_insight(f"Increase resources to boost {current_progress}% progress in {phase}: {task}", 0.8)
+
+    # Edge cases
+    if workforce_gap >= 90:
+        add_insight("Critical: Halt non-essential tasks until [Phase: Task] until workforce gap is resolved", 1.1)
+    if current_progress == 0 and delay_risk > 50:
+        add_insight(f"Initiate {phase}: {task} immediately to avoid further delays", 1.0)
+    if expected_duration == 0 or actual_duration == 0:
+        add_insight("Provide accurate duration estimates for reliable predictions", 0.7)
+
+    # Sort insights by priority and select top 3-5
+    insights.sort(key=lambda x: x[1], reverse=True)
+    selected_insights = [insight[0] for insight in insights[:5]]
+
+    logger.info(f"Generated insights: {selected_insights}")
+    return selected_insights or [f"No significant delay factors detected for {phase}: {task}"]
 
 def predict_delay(input_data: Dict) -> Dict:
     """
     Predict delay probability based on project task data.
     Uses task duration, progress, workforce info, and weather impact.
-    Insights are generated by T5-Small (CPU).
+    Insights are generated using detailed hardcoded rules.
     """
     logger.info("Starting delay prediction")
     phase = input_data.get("phase", "")
@@ -173,7 +215,7 @@ def predict_delay(input_data: Dict) -> Dict:
                 "risk": round(task_risk, 1)
             })
 
-    # Generate AI-driven insights
+    # Generate hardcoded insights
     insights = call_ai_model_for_insights(input_data, delay_risk)
 
     logger.info(f"Prediction completed: Delay risk = {delay_risk:.1f}%")