| from typing import Dict, Tuple, Optional |
| from dotenv import load_dotenv |
| import pandas as pd |
| import requests |
| import json |
| import time |
| import os |
|
|
| load_dotenv() |
|
|
| TOMTOM_API_KEY = os.getenv("TOMTOM_API_KEY") |
| OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY") |
|
|
|
|
|
|
| def get_historical_weather(lat: float, lon: float, start_date: str, end_date: str): |
| """Fetch hourly historical weather directly from Open-Meteo API (no SDK).""" |
| url = "https://archive-api.open-meteo.com/v1/archive" |
| params = { |
| "latitude": lat, |
| "longitude": lon, |
| "start_date": start_date, |
| "end_date": end_date, |
| "hourly": "temperature_2m,relative_humidity_2m,rain,snowfall,wind_speed_10m", |
| "timezone": "auto" |
| } |
| try: |
| response = requests.get(url, params=params, timeout=10) |
| response.raise_for_status() |
| data = response.json() |
|
|
| hourly = data.get("hourly", {}) |
| df = pd.DataFrame({ |
| "time": hourly.get("time", []), |
| "temperature": hourly.get("temperature_2m", []), |
| "humidity": hourly.get("relative_humidity_2m", []), |
| "rain": hourly.get("rain", []), |
| "snow": hourly.get("snowfall", []), |
| "wind_speed": hourly.get("wind_speed_10m", []), |
| }) |
|
|
| df["time"] = pd.to_datetime(df["time"]) |
| return df |
|
|
| except Exception as e: |
| print(f"Historical weather error: {e}") |
| return None |
|
|
|
|
| def get_realtime_traffic(origin_coords: Tuple[float, float], dest_coords: Tuple[float, float]) -> Optional[Dict]: |
| """Get real-time traffic using TomTom API""" |
| base_url = "https://api.tomtom.com/routing/1/calculateRoute" |
| origin = f"{origin_coords[0]},{origin_coords[1]}" |
| destination = f"{dest_coords[0]},{dest_coords[1]}" |
| url = f"{base_url}/{origin}:{destination}/json" |
| |
| headers = {"Accept": "application/json"} |
| params = { |
| "key": TOMTOM_API_KEY, |
| "traffic": "true", |
| "computeTravelTimeFor": "all", |
| "routeType": "fastest", |
| } |
|
|
| try: |
| response = requests.get(url, headers=headers, params=params, timeout=10) |
| response.raise_for_status() |
| data = response.json() |
| return data["routes"][0]["summary"] |
| except Exception as e: |
| print(f"Error fetching traffic data: {e}") |
| return None |
|
|
|
|
| def get_weather_data(lat: float, lon: float) -> Optional[Dict]: |
| """ |
| Get weather data using OpenWeatherMap API |
| Returns: temperature, condition, precipitation, wind speed |
| """ |
| if not OPENWEATHER_API_KEY: |
| return None |
| |
| url = "https://api.openweathermap.org/data/2.5/weather" |
| params = { |
| "lat": lat, |
| "lon": lon, |
| "appid": OPENWEATHER_API_KEY, |
| "units": "metric" |
| } |
| |
| try: |
| response = requests.get(url, params=params, timeout=10) |
| response.raise_for_status() |
| data = response.json() |
| |
| return { |
| "temperature": data["main"]["temp"], |
| "feels_like": data["main"]["feels_like"], |
| "condition": data["weather"][0]["main"], |
| "description": data["weather"][0]["description"], |
| "humidity": data["main"]["humidity"], |
| "wind_speed": data["wind"]["speed"], |
| "rain": data.get("rain", {}).get("1h", 0), |
| "visibility": data.get("visibility", 10000) / 1000 |
| } |
| except Exception as e: |
| print(f"Weather API error: {e}") |
| return None |
|
|
|
|
| def get_weather_along_route(origin_coords: Tuple[float, float], |
| dest_coords: Tuple[float, float]) -> Dict: |
| """ |
| Get weather for origin and destination, calculate weather impact |
| """ |
| origin_weather = get_weather_data(origin_coords[0], origin_coords[1]) |
| dest_weather = get_weather_data(dest_coords[0], dest_coords[1]) |
| |
| weather_factor = 1.0 |
| warnings = [] |
| |
| if origin_weather: |
| if origin_weather["rain"] > 2.5: |
| weather_factor += 0.3 |
| warnings.append(f"Heavy rain at origin ({origin_weather['rain']:.1f}mm/h)") |
| elif origin_weather["rain"] > 0.5: |
| weather_factor += 0.15 |
| warnings.append(f"Light rain at origin ({origin_weather['rain']:.1f}mm/h)") |
| |
| |
| if origin_weather["visibility"] < 1: |
| weather_factor += 0.2 |
| warnings.append(f"Low visibility at origin ({origin_weather['visibility']:.1f}km)") |
| |
| |
| if origin_weather["wind_speed"] > 15: |
| weather_factor += 0.1 |
| warnings.append(f"Strong winds at origin ({origin_weather['wind_speed']:.1f}m/s)") |
| |
| if dest_weather: |
| if dest_weather["rain"] > 2.5: |
| weather_factor += 0.2 |
| warnings.append(f"Heavy rain at destination ({dest_weather['rain']:.1f}mm/h)") |
| elif dest_weather["rain"] > 0.5: |
| weather_factor += 0.1 |
| |
| return { |
| "origin": origin_weather, |
| "destination": dest_weather, |
| "weather_factor": min(weather_factor, 1.8), |
| "warnings": warnings |
| } |
|
|
|
|
| def geocode_address_nominatim(address: str) -> Optional[Tuple[float, float]]: |
| """Geocoding using Nominatim (OpenStreetMap)""" |
| url = "https://nominatim.openstreetmap.org/search" |
| params = { |
| "q": address, |
| "format": "json", |
| "limit": 1 |
| } |
| headers = { |
| "User-Agent": "DeliveryOptimizationSystem/1.0" |
| } |
| |
| try: |
| time.sleep(1) |
| response = requests.get(url, params=params, headers=headers, timeout=30) |
| response.raise_for_status() |
| data = response.json() |
| |
| if data: |
| return (float(data[0]["lat"]), float(data[0]["lon"])) |
| return None |
| except Exception as e: |
| print(f"Geocoding error: {e}") |
| return None |
|
|
|
|
| def get_route_osrm(origin_coords: Tuple[float, float], |
| dest_coords: Tuple[float, float]) -> Optional[Dict]: |
| """Routing using OSRM (OpenStreetMap Routing Machine)""" |
| lat1, lon1 = origin_coords |
| lat2, lon2 = dest_coords |
| |
| url = f"http://router.project-osrm.org/route/v1/driving/{lon1},{lat1};{lon2},{lat2}" |
| params = { |
| "overview": "full", |
| "geometries": "geojson", |
| "steps": "true", |
| "annotations": "true" |
| } |
| |
| try: |
| response = requests.get(url, params=params, timeout=15) |
| response.raise_for_status() |
| data = response.json() |
| |
| if data.get("code") == "Ok" and data.get("routes"): |
| route = data["routes"][0] |
| return { |
| "distance_km": route["distance"] / 1000, |
| "duration_min": route["duration"] / 60, |
| "geometry": route["geometry"], |
| "steps": route["legs"][0].get("steps", []) |
| } |
| return None |
| except Exception as e: |
| print(f"Routing error: {e}") |
| return None |
|
|
|
|
| def get_alternative_routes_osrm(origin_coords: Tuple[float, float], |
| dest_coords: Tuple[float, float]) -> list: |
| """Get multiple route options""" |
| lat1, lon1 = origin_coords |
| lat2, lon2 = dest_coords |
| |
| url = f"http://router.project-osrm.org/route/v1/driving/{lon1},{lat1};{lon2},{lat2}" |
| params = { |
| "alternatives": "true", |
| "steps": "true", |
| "overview": "full" |
| } |
| |
| try: |
| response = requests.get(url, params=params, timeout=15) |
| data = response.json() |
| |
| if data.get("code") == "Ok": |
| routes = [] |
| for route in data.get("routes", []): |
| routes.append({ |
| "distance_km": route["distance"] / 1000, |
| "duration_min": route["duration"] / 60 |
| }) |
| return routes |
| return [] |
| except: |
| return [] |
|
|
|
|
| def get_detailed_route_with_instructions(origin_coords: Tuple[float, float], |
| dest_coords: Tuple[float, float]) -> Optional[Dict]: |
| """Get detailed route with turn-by-turn instructions and road names""" |
| lat1, lon1 = origin_coords |
| lat2, lon2 = dest_coords |
| |
| url = f"http://router.project-osrm.org/route/v1/driving/{lon1},{lat1};{lon2},{lat2}" |
| params = { |
| "alternatives": "true", |
| "steps": "true", |
| "overview": "full", |
| "geometries": "geojson", |
| "annotations": "true" |
| } |
| |
| try: |
| response = requests.get(url, params=params, timeout=15) |
| response.raise_for_status() |
| data = response.json() |
| |
| if data.get("code") == "Ok" and data.get("routes"): |
| all_routes = [] |
| |
| for route_idx, route in enumerate(data.get("routes", [])): |
| route_info = { |
| "route_number": route_idx + 1, |
| "distance_km": route["distance"] / 1000, |
| "duration_min": route["duration"] / 60, |
| "instructions": [] |
| } |
| |
| for leg in route.get("legs", []): |
| for step in leg.get("steps", []): |
| instruction = { |
| "distance_km": step["distance"] / 1000, |
| "duration_min": step["duration"] / 60, |
| "instruction": step.get("maneuver", {}).get("type", "continue"), |
| "road_name": step.get("name", "Unnamed road"), |
| "direction": step.get("maneuver", {}).get("modifier", "") |
| } |
| route_info["instructions"].append(instruction) |
| |
| all_routes.append(route_info) |
| |
| return all_routes |
| return None |
| except Exception as e: |
| print(f"Detailed routing error: {e}") |
| return None |
|
|
|
|
