""" Intelligent Route Optimizer for FleetMind Combines traffic, weather, and vehicle type for optimal routing decisions """ import logging from typing import Dict, List, Optional from chat.tools import handle_calculate_route, geocoding_service from chat.weather import weather_service logger = logging.getLogger(__name__) def calculate_intelligent_route( origin: str, destination: str, vehicle_type: str = "car", consider_weather: bool = True, consider_traffic: bool = True ) -> Dict: """ Calculate optimal route considering traffic, weather, and vehicle type Args: origin: Starting location (address or coordinates) destination: Ending location (address or coordinates) vehicle_type: Type of vehicle (motorcycle, car, van, truck) consider_weather: Whether to factor in weather conditions consider_traffic: Whether to factor in traffic conditions Returns: Comprehensive routing result with recommendations and warnings """ logger.info(f"Intelligent routing: {origin} → {destination} (vehicle: {vehicle_type})") # Step 1: Calculate base route with traffic data route_result = handle_calculate_route({ "origin": origin, "destination": destination, "vehicle_type": vehicle_type, "alternatives": True # Get alternative routes }) if not route_result.get("success"): return route_result # Return error # Step 2: Get weather data for the destination area weather_data = None weather_impact = None if consider_weather: try: # Geocode destination to get coordinates dest_geocoded = geocoding_service.geocode(destination) dest_lat = dest_geocoded["lat"] dest_lng = dest_geocoded["lng"] # Get current weather weather_data = weather_service.get_current_weather(dest_lat, dest_lng) # Assess weather impact for this vehicle type weather_impact = weather_service.assess_weather_impact(weather_data, vehicle_type) logger.info(f"Weather impact: {weather_impact['severity']} (multiplier: {weather_impact['speed_multiplier']}x)") except Exception as e: logger.warning(f"Weather data unavailable: {e}") consider_weather = False # Step 3: Calculate adjusted duration base_duration = route_result["duration"]["seconds"] traffic_duration = route_result["duration_in_traffic"]["seconds"] # Start with traffic-aware duration adjusted_duration = traffic_duration # Apply weather adjustments if available if consider_weather and weather_impact: adjusted_duration = int(adjusted_duration * weather_impact["speed_multiplier"]) # Calculate delay percentages traffic_delay_percent = 0 weather_delay_percent = 0 if consider_traffic and traffic_duration > base_duration: traffic_delay_percent = int(((traffic_duration - base_duration) / base_duration) * 100) if consider_weather and weather_impact and weather_impact["speed_multiplier"] > 1.0: weather_delay_percent = int(((weather_impact["speed_multiplier"] - 1.0) * 100)) total_delay_percent = int(((adjusted_duration - base_duration) / base_duration) * 100) if base_duration > 0 else 0 # Step 4: Generate traffic status traffic_status = "unknown" if consider_traffic: if traffic_delay_percent == 0: traffic_status = "clear" elif traffic_delay_percent < 15: traffic_status = "light" elif traffic_delay_percent < 30: traffic_status = "moderate" elif traffic_delay_percent < 50: traffic_status = "heavy" else: traffic_status = "severe" # Step 5: Generate recommendations and warnings recommendations = [] warnings = [] # Traffic recommendations if consider_traffic: if traffic_delay_percent > 30: recommendations.append(f"🚦 Heavy traffic: {traffic_delay_percent}% delay - consider alternate route or timing") elif traffic_delay_percent > 15: recommendations.append(f"🚦 Moderate traffic: {traffic_delay_percent}% delay expected") # Weather recommendations if consider_weather and weather_impact: if weather_impact["warnings"]: warnings.extend(weather_impact["warnings"]) if weather_impact["recommend_delay"]: recommendations.append("⚠️ SEVERE WEATHER: Consider delaying trip until conditions improve") if vehicle_type == "motorcycle" and not weather_impact["safe_for_motorcycle"]: warnings.append("🏍️ WARNING: Current weather not safe for motorcycle - consider alternative vehicle") # Vehicle-specific recommendations if vehicle_type == "motorcycle": if traffic_delay_percent > 40: recommendations.append("🏍️ TIP: Motorcycles can navigate heavy traffic more efficiently") # Format durations def format_duration(seconds): hours = seconds // 3600 minutes = (seconds % 3600) // 60 if hours > 0: return f"{hours}h {minutes}m" return f"{minutes}m" # Step 6: Build comprehensive response response = { "success": True, "route": { "origin": route_result["origin"], "destination": route_result["destination"], "distance": route_result["distance"], "vehicle_type": vehicle_type, "route_summary": route_result["route_summary"], "confidence": route_result["confidence"] }, "timing": { "base_duration": { "seconds": base_duration, "text": format_duration(base_duration) }, "with_traffic": { "seconds": traffic_duration, "text": format_duration(traffic_duration) }, "adjusted_duration": { "seconds": adjusted_duration, "text": format_duration(adjusted_duration) }, "traffic_delay_percent": traffic_delay_percent, "weather_delay_percent": weather_delay_percent, "total_delay_percent": total_delay_percent }, "conditions": { "traffic_status": traffic_status, "traffic_considered": consider_traffic, "weather_considered": consider_weather }, "recommendations": recommendations, "warnings": warnings } # Add weather data if available if weather_data: response["weather"] = { "conditions": weather_data["conditions"], "description": weather_data["description"], "temperature_c": round(weather_data["temperature_c"], 1), "precipitation_mm": round(weather_data["precipitation_mm"], 1), "visibility_m": weather_data["visibility_m"], "impact_severity": weather_impact["severity"] if weather_impact else "none" } # Add alternative routes if available if route_result.get("alternatives"): response["alternatives"] = route_result["alternatives"] response["alternatives_count"] = len(route_result["alternatives"]) logger.info(f"Intelligent route calculated: {format_duration(adjusted_duration)} (base: {format_duration(base_duration)})") return response