# 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) 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(""" **Welcome to the Route Planning Assistant!** This professional application leverages advanced geospatial routing and AI-driven processing to help you plan optimized travel routes across multiple cities. It integrates with the OSRM routing engine for real-time route data and uses interactive mapping technology to visualize your journey. **Features include:** - **Dynamic Route Planning:** Automatically calculates the most efficient travel path based on current road networks. - **Interactive Map Visualization:** View your planned route on a responsive map, complete with distance and duration details. - **Conversational Interface:** Communicate with our AI assistant to specify your travel preferences and receive detailed route analysis. **Usage Examples:** - "Show me a route from Paris to Berlin via Amsterdam." - "Plan a trip from London to Rome through Paris and Vienna." **Supported 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)