"""
Weather service for FleetMind
Provides weather data for intelligent routing decisions
"""
import os
import logging
import requests
from typing import Dict, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
class WeatherService:
"""Handle weather data fetching with OpenWeatherMap API and mock fallback"""
def __init__(self):
self.api_key = os.getenv("OPENWEATHERMAP_API_KEY", "")
self.use_mock = not self.api_key or self.api_key.startswith("your_")
self.base_url = "https://api.openweathermap.org/data/2.5/weather"
if self.use_mock:
logger.info("Weather Service: Using mock (OPENWEATHERMAP_API_KEY not configured)")
else:
logger.info("Weather Service: Using OpenWeatherMap API")
def get_current_weather(self, lat: float, lng: float) -> Dict:
"""
Get current weather conditions at specified coordinates
Args:
lat: Latitude
lng: Longitude
Returns:
Dict with weather data: temp, conditions, precipitation, visibility, wind
"""
if self.use_mock:
return self._get_weather_mock(lat, lng)
else:
try:
return self._get_weather_openweathermap(lat, lng)
except Exception as e:
logger.error(f"OpenWeatherMap API failed: {e}, falling back to mock")
return self._get_weather_mock(lat, lng)
def _get_weather_openweathermap(self, lat: float, lng: float) -> Dict:
"""Fetch weather from OpenWeatherMap API"""
try:
params = {
"lat": lat,
"lon": lng,
"appid": self.api_key,
"units": "metric" # Celsius, km/h
}
response = requests.get(self.base_url, params=params, timeout=5)
response.raise_for_status()
data = response.json()
# Extract weather information
main = data.get("main", {})
weather = data.get("weather", [{}])[0]
wind = data.get("wind", {})
rain = data.get("rain", {})
snow = data.get("snow", {})
visibility = data.get("visibility", 10000) # Default 10km
# Calculate precipitation (rain + snow in last hour)
precipitation_mm = rain.get("1h", 0) + snow.get("1h", 0)
weather_data = {
"temperature_c": main.get("temp", 20),
"feels_like_c": main.get("feels_like", 20),
"humidity_percent": main.get("humidity", 50),
"conditions": weather.get("main", "Clear"),
"description": weather.get("description", "clear sky"),
"precipitation_mm": precipitation_mm,
"visibility_m": visibility,
"wind_speed_mps": wind.get("speed", 0),
"wind_gust_mps": wind.get("gust", 0),
"timestamp": datetime.now().isoformat(),
"source": "OpenWeatherMap API"
}
logger.info(f"Weather fetched: {weather_data['conditions']}, {weather_data['temperature_c']}°C")
return weather_data
except Exception as e:
logger.error(f"OpenWeatherMap API error: {e}")
raise
def _get_weather_mock(self, lat: float, lng: float) -> Dict:
"""Mock weather data for testing"""
# Generate pseudo-random but realistic weather based on coordinates
import random
random.seed(int(lat * 1000) + int(lng * 1000))
conditions_options = ["Clear", "Clouds", "Rain", "Drizzle", "Fog"]
weights = [0.5, 0.3, 0.15, 0.03, 0.02] # Mostly clear/cloudy
conditions = random.choices(conditions_options, weights=weights)[0]
if conditions == "Clear":
precipitation_mm = 0
visibility_m = 10000
description = "clear sky"
elif conditions == "Clouds":
precipitation_mm = 0
visibility_m = 8000
description = "scattered clouds"
elif conditions == "Rain":
precipitation_mm = random.uniform(2, 10)
visibility_m = random.randint(5000, 8000)
description = "moderate rain"
elif conditions == "Drizzle":
precipitation_mm = random.uniform(0.5, 2)
visibility_m = random.randint(6000, 9000)
description = "light rain"
else: # Fog
precipitation_mm = 0
visibility_m = random.randint(500, 2000)
description = "foggy"
weather_data = {
"temperature_c": random.uniform(10, 25),
"feels_like_c": random.uniform(8, 23),
"humidity_percent": random.randint(40, 80),
"conditions": conditions,
"description": description,
"precipitation_mm": precipitation_mm,
"visibility_m": visibility_m,
"wind_speed_mps": random.uniform(0, 8),
"wind_gust_mps": random.uniform(0, 12),
"timestamp": datetime.now().isoformat(),
"source": "Mock (testing)"
}
logger.info(f"Mock weather: {conditions}, {weather_data['temperature_c']:.1f}°C")
return weather_data
def assess_weather_impact(self, weather_data: Dict, vehicle_type: str = "car") -> Dict:
"""
Assess how weather affects routing for a given vehicle type
Args:
weather_data: Weather data from get_current_weather()
vehicle_type: Type of vehicle (car, van, truck, motorcycle)
Returns:
Dict with impact assessment and warnings
"""
precipitation = weather_data.get("precipitation_mm", 0)
visibility = weather_data.get("visibility_m", 10000)
wind_speed = weather_data.get("wind_speed_mps", 0)
conditions = weather_data.get("conditions", "Clear")
impact_score = 0 # 0 = no impact, 1 = severe impact
speed_multiplier = 1.0 # 1.0 = no change, 1.5 = 50% slower
warnings = []
severity = "none"
# Precipitation impact
if precipitation > 10: # Heavy rain (>10mm/h)
impact_score += 0.6
speed_multiplier *= 1.4
warnings.append("⚠️ Heavy rain - significantly reduced speeds")
severity = "severe"
if vehicle_type == "motorcycle":
speed_multiplier *= 1.2 # Additional 20% slower for motorcycles
warnings.append("🏍️ DANGER: Motorcycle in heavy rain - consider rescheduling")
elif precipitation > 5: # Moderate rain (5-10mm/h)
impact_score += 0.3
speed_multiplier *= 1.2
warnings.append("🌧️ Moderate rain - reduced speeds")
severity = "moderate"
if vehicle_type == "motorcycle":
speed_multiplier *= 1.15
warnings.append("🏍️ Caution: Wet roads for motorcycle")
elif precipitation > 0: # Light rain
impact_score += 0.1
speed_multiplier *= 1.1
if vehicle_type == "motorcycle":
warnings.append("🏍️ Light rain - exercise caution")
severity = "minor"
# Visibility impact
if visibility < 1000: # Poor visibility (<1km)
impact_score += 0.5
speed_multiplier *= 1.3
warnings.append("🌫️ Poor visibility - drive carefully")
if severity == "none":
severity = "moderate"
elif visibility < 5000: # Reduced visibility
impact_score += 0.2
speed_multiplier *= 1.1
if severity == "none":
severity = "minor"
# Wind impact (mainly for motorcycles and high-profile vehicles)
if wind_speed > 15: # Strong wind (>54 km/h)
if vehicle_type in ["motorcycle", "truck"]:
impact_score += 0.3
speed_multiplier *= 1.15
warnings.append(f"💨 Strong winds - affects {vehicle_type}")
if severity == "none":
severity = "moderate"
# Extreme conditions
if conditions == "Thunderstorm":
impact_score += 0.8
speed_multiplier *= 1.6
warnings.append("⛈️ SEVERE: Thunderstorm - consider delaying trip")
severity = "severe"
if conditions == "Snow":
impact_score += 0.7
speed_multiplier *= 1.5
warnings.append("❄️ Snow conditions - significantly reduced speeds")
severity = "severe"
# Cap impact score at 1.0
impact_score = min(impact_score, 1.0)
return {
"impact_score": round(impact_score, 2),
"speed_multiplier": round(speed_multiplier, 2),
"severity": severity,
"warnings": warnings,
"safe_for_motorcycle": precipitation < 5 and visibility > 3000 and wind_speed < 12,
"recommend_delay": severity == "severe"
}
def get_status(self) -> str:
"""Get weather service status"""
if self.use_mock:
return "⚠️ Mock weather service (configure OPENWEATHERMAP_API_KEY)"
else:
return "✅ OpenWeatherMap API connected"
# Global weather service instance
weather_service = WeatherService()
