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