File size: 7,432 Bytes
5769203 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 |
"""
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
|