Spaces:
Sleeping
Sleeping
# 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) | |