"""
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
