api.py

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  # km
        }
    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)")
        
        # Visibility impact
        if origin_weather["visibility"] < 1:
            weather_factor += 0.2
            warnings.append(f"Low visibility at origin ({origin_weather['visibility']:.1f}km)")
        
        # Wind impact
        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