Create app.py

app.py

@@ -0,0 +1,254 @@
+import os
+import json
+from typing import Dict, List, Tuple, Optional
+from dotenv import load_dotenv
+import requests
+import folium
+from folium import plugins
+import polyline
+from openai import OpenAI
+import gradio as gr
+
+# Load environment variables
+load_dotenv()
+OPENAI_API_KEY = os.getenv('OPENAI_API_KEY', 'your-key-here')
+OSRM_SERVER = "http://router.project-osrm.org"  # Public OSRM demo server
+
+# Initialize OpenAI client
+client = OpenAI(api_key=OPENAI_API_KEY)
+
+# City database with coordinates
+CITY_INFO = {
+    "New York": {"coordinates": (40.7128, -74.0060)},
+    "London": {"coordinates": (51.5074, -0.1278)},
+    "Paris": {"coordinates": (48.8566, 2.3522)},
+    "Berlin": {"coordinates": (52.5200, 13.4050)},
+    "Madrid": {"coordinates": (40.4168, -3.7038)},
+    "Rome": {"coordinates": (41.9028, 12.4964)},
+    "Amsterdam": {"coordinates": (52.3676, 4.9041)},
+    "Brussels": {"coordinates": (50.8503, 4.3517)},
+    "Vienna": {"coordinates": (48.2082, 16.3738)},
+    "Prague": {"coordinates": (50.0755, 14.4378)}
+}
+
+def get_route(start_coords: Tuple[float, float], end_coords: Tuple[float, float]) -> Dict:
+    """
+    Get route information between two points using OSRM.
+    """
+    # Format coordinates for OSRM API
+    coords = f"{start_coords[1]},{start_coords[0]};{end_coords[1]},{end_coords[0]}"
+    url = f"{OSRM_SERVER}/route/v1/driving/{coords}?overview=full&geometries=polyline"
+    
+    try:
+        response = requests.get(url)
+        response.raise_for_status()
+        data = response.json()
+        
+        if data["code"] != "Ok":
+            raise Exception("Route not found")
+        
+        route = data["routes"][0]
+        return {
+            "distance": route["distance"],  # meters
+            "duration": route["duration"],  # seconds
+            "geometry": route["geometry"]   # encoded polyline
+        }
+    except Exception as e:
+        raise Exception(f"Error getting route: {str(e)}")
+
+def create_route_map(routes: List[Dict], cities: List[str], coords: List[Tuple[float, float]]) -> str:
+    """
+    Create an interactive map with the route visualization.
+    """
+    # Calculate map center
+    center_lat = sum(lat for lat, _ in coords) / len(coords)
+    center_lon = sum(lon for _, lon in coords) / len(coords)
+    
+    # Create the map
+    m = folium.Map(location=[center_lat, center_lon], zoom_start=4)
+    
+    # Colors for different route segments
+    colors = ['blue', 'red', 'green', 'purple', 'orange']
+    
+    # Add route segments
+    for i, route in enumerate(routes):
+        # Decode the polyline
+        route_coords = polyline.decode(route["geometry"])
+        
+        # Add the route line
+        color = colors[i % len(colors)]
+        folium.PolyLine(
+            route_coords,
+            weight=3,
+            color=color,
+            opacity=0.8,
+            popup=f"Distance: {route['distance']/1000:.1f}km\nDuration: {route['duration']/3600:.1f}h"
+        ).add_to(m)
+    
+    # Add markers for cities
+    for i, (city, coord) in enumerate(zip(cities, coords)):
+        folium.Marker(
+            location=[coord[0], coord[1]],
+            popup=f"{city} (Stop {i+1})",
+            icon=folium.Icon(color='red', icon='info-sign')
+        ).add_to(m)
+    
+    # Add automatic bounds
+    m.fit_bounds([[coord[0], coord[1]] for coord in coords])
+    
+    return m._repr_html_()
+
+def plan_route(cities: List[str]) -> Dict:
+    """
+    Plan a route through the given cities.
+    """
+    if len(cities) < 2:
+        return {"error": "Need at least 2 cities to plan a route"}
+    
+    # Get coordinates for valid cities
+    coords = []
+    valid_cities = []
+    for city in cities:
+        if city in CITY_INFO:
+            coords.append(CITY_INFO[city]["coordinates"])
+            valid_cities.append(city)
+    
+    if len(coords) < 2:
+        return {"error": "Not enough valid cities provided"}
+    
+    try:
+        # Get routes between consecutive cities
+        routes = []
+        total_distance = 0
+        total_duration = 0
+        
+        for i in range(len(coords) - 1):
+            route = get_route(coords[i], coords[i + 1])
+            routes.append(route)
+            total_distance += route["distance"]
+            total_duration += route["duration"]
+        
+        # Create map visualization
+        map_html = create_route_map(routes, valid_cities, coords)
+        
+        return {
+            "map_html": map_html,
+            "total_distance": total_distance,
+            "total_duration": total_duration,
+            "cities": valid_cities
+        }
+    
+    except Exception as e:
+        return {"error": str(e)}
+
+# OpenAI function definition
+tools = [
+    {
+        "name": "plan_route",
+        "description": "Plans a route through multiple European cities and returns an interactive map.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "cities": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "List of city names to visit in order"
+                }
+            },
+            "required": ["cities"]
+        }
+    }
+]
+
+system_message = """You are a Route Planning Assistant. When users request a route through cities:
+1. Extract the city names from their request
+2. Call plan_route with these cities in the order specified
+3. Explain the route details, including total distance and duration
+Example: For 'Show me a route from Paris to Berlin via Amsterdam', call plan_route with cities=['Paris', 'Amsterdam', 'Berlin']."""
+
+def chat_with_openai(message: str, history: List) -> tuple[str, Optional[str]]:
+    """Process chat messages and handle function calling."""
+    messages = [{"role": "system", "content": system_message}]
+    
+    # Add conversation history
+    for human, assistant in history:
+        messages.extend([
+            {"role": "user", "content": human},
+            {"role": "assistant", "content": assistant}
+        ])
+    
+    messages.append({"role": "user", "content": message})
+    
+    try:
+        # Get initial response
+        response = client.chat.completions.create(
+            model="gpt-4",
+            messages=messages,
+            functions=tools,
+            function_call="auto"
+        )
+        
+        # Handle function calling
+        if response.choices[0].message.function_call:
+            function_call = response.choices[0].message.function_call
+            
+            # Parse and execute function
+            args = json.loads(function_call.arguments)
+            result = plan_route(args["cities"])
+            
+            if "error" in result:
+                return f"Error: {result['error']}", None
+            
+            # Format the response message
+            response_message = (
+                f"I've planned your route through {', '.join(result['cities'])}.\n"
+                f"Total distance: {result['total_distance']/1000:.1f} km\n"
+                f"Total duration: {result['total_duration']/3600:.1f} hours"
+            )
+            
+            return response_message, result.get("map_html")
+        
+        return response.choices[0].message.content, None
+    
+    except Exception as e:
+        return f"An error occurred: {str(e)}", None
+
+# Gradio interface
+def create_gradio_interface():
+    """Creates and returns the Gradio interface."""
+    with gr.Blocks() as demo:
+        gr.Markdown("# Route Planning Assistant")
+        gr.Markdown("""
+        Enter a routing request. Examples:
+        - Show me a route from Paris to Berlin via Amsterdam
+        - Plan a trip from London to Rome through Paris and Vienna
+        
+        Available cities: London, Paris, Berlin, Madrid, Rome, Amsterdam, Brussels, Vienna, Prague
+        """)
+        
+        with gr.Row():
+            text_input = gr.Textbox(
+                label="Your Request",
+                placeholder="Enter your routing request..."
+            )
+        
+        with gr.Row():
+            chatbot = gr.Chatbot(label="Conversation")
+            map_output = gr.HTML(label="Route Map")
+        
+        def process_message(message, history):
+            answer, map_html = chat_with_openai(message, history)
+            history.append((message, answer))
+            return "", history, map_html or ""
+        
+        text_input.submit(
+            process_message,
+            [text_input, chatbot],
+            [text_input, chatbot, map_output]
+        )
+    
+    return demo
+
+if __name__ == "__main__":
+    demo = create_gradio_interface()
+    demo.launch(share=True)