"""
title: OpenStreetMap Tool
author: projectmoon
author_url: https://git.agnos.is/projectmoon/open-webui-filters
version: 0.3.0
license: AGPL-3.0+
required_open_webui_version: 0.3.9
"""
import json
import math
import requests
from typing import List, Optional
from pydantic import BaseModel, Field
VALVES_NOT_SET = """
Tell the user that the User-Agent and From headers
must be set to comply with the OSM Nominatim terms
of use: https://operations.osmfoundation.org/policies/nominatim/
""".replace("\n", " ").strip()
NO_RESULTS = ("No results found. Tell the user you found no results. "
"Do not make up answers or hallucinate. Only say you "
"found no results.")
NO_CONFUSION = ("**IMPORTANT!:** Check that the results match the location "
"the user is talking about, by analyzing the conversation history. "
"Sometimes there are places with the same "
"names, but in different cities or countries. If the results are for "
"a different city or country than the user is interested in, say so: "
"tell the user that the results are for the wrong place, and tell them "
"to be more specific in their query.")
def way_has_info(way):
"""
Determine if an OSM way entry is useful to us. This means it
has something more than just its main classification tag, and
has at least a name.
"""
return len(way['tags']) > 1 and any('name' in tag for tag in way['tags'])
def strip_nodes_from_way(way):
if 'nodes' in way:
del way['nodes']
return way
def get_bounding_box_center(bbox):
def convert(bbox, key):
return bbox[key] if isinstance(bbox[key], float) else float(bbox[key])
min_lat = convert(bbox, 'min_lat')
min_lon = convert(bbox, 'min_lon')
max_lat = convert(bbox, 'max_lat')
max_lon = convert(bbox, 'max_lon')
return {
'lon': (min_lon + max_lon) / 2,
'lat': (min_lat + max_lat) / 2
}
def haversine_distance(point1, point2):
R = 6371 # Earth radius in kilometers
lat1, lon1 = point1['lat'], point1['lon']
lat2, lon2 = point2['lat'], point2['lon']
d_lat = math.radians(lat2 - lat1)
d_lon = math.radians(lon2 - lon1)
a = (math.sin(d_lat / 2) * math.sin(d_lat / 2) +
math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) *
math.sin(d_lon / 2) * math.sin(d_lon / 2))
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
return R * c
def sort_by_closeness(origin, points):
"""
Sorts a list of { lat, lon }-like dicts by closeness to an origin point.
The origin is a dict with keys of { lat, lon }. This function adds the
distance as a dict value to the points.
"""
points_with_distance = [(point, haversine_distance(origin, point)) for point in points]
points_with_distance = sorted(points_with_distance, key=lambda pwd: pwd[1])
for point, distance in points_with_distance:
point['distance'] = distance
return [point for point, distance in points_with_distance]
class OsmSearcher:
def __init__(self, valves):
self.valves = valves
def create_headers(self) -> Optional[dict]:
if len(self.valves.user_agent) == 0 or len(self.valves.from_header) == 0:
return None
return {
'User-Agent': self.valves.user_agent,
'From': self.valves.from_header
}
@staticmethod
def group_tags(tags):
result = {}
for tag in tags:
key, value = tag.split('=')
if key not in result:
result[key] = []
result[key].append(value)
return result
@staticmethod
def fallback(nominatim_result):
"""
If we do not have Overpass Turbo results, attempt to use the
Nominatim result instead.
"""
return ([nominatim_result] if 'type' in nominatim_result
and (nominatim_result['type'] == 'amenity'
or nominatim_result['type'] == 'shop'
or nominatim_result['type'] == 'leisure'
or nominatim_result['type'] == 'tourism')
else [])
def nominatim_search(self, query, format="json", limit: int=1) -> Optional[dict]:
url = self.valves.nominatim_url
params = {
'q': query,
'format': format,
'addressdetails': 1,
'limit': limit,
}
headers = self.create_headers()
if not headers:
raise ValueError("Headers not set")
response = requests.get(url, params=params, headers=headers)
if response.status_code == 200:
data = response.json()
if not data:
raise ValueError(f"No results found for query '{query}'")
return data[:limit]
else:
print(response.text)
return None
def overpass_search(
self, place, tags, bbox, limit=5, radius=4000
) -> (List[dict], List[dict]):
headers = self.create_headers()
if not headers:
raise ValueError("Headers not set")
url = self.valves.overpass_turbo_url
center = get_bounding_box_center(bbox)
around = f"(around:{radius},{center['lat']},{center['lon']})"
tag_groups = OsmSearcher.group_tags(tags)
search_groups = [f'"{tag_type}"~"{"|".join(values)}"'
for tag_type, values in tag_groups.items()]
searches = []
for search_group in search_groups:
searches.append(
f'nwr[{search_group}]{around}'
)
search = ";\n".join(searches)
if len(search) > 0:
search += ";"
query = f"""
[out:json];
(
{search}
);
out qt;
"""
print(query)
data = { "data": query }
response = requests.get(url, params=data, headers=headers)
if response.status_code == 200:
# nodes are prioritized because they have exact GPS
# coordinates. we also include useful way entries (without
# node list) as secondary results, because there are often
# useful results that don't have a node (e.g. building or
# whole area marked for the tag type).
results = response.json()
results = results['elements'] if 'elements' in results else []
nodes = []
ways = []
for res in results:
if 'type' not in res:
continue
if res['type'] == 'node':
nodes.append(res)
elif res['type'] == 'way' and way_has_info(res):
ways.append(strip_nodes_from_way(res))
return nodes, ways
else:
print(response.text)
raise Exception(f"Error calling Overpass API: {response.text}")
def search_nearby(self, place: str, tags: List[str], limit: int=5, radius: int=4000) -> str:
headers = self.create_headers()
if not headers:
return VALVES_NOT_SET
try:
nominatim_result = self.nominatim_search(place, limit=1)
if nominatim_result:
nominatim_result = nominatim_result[0]
bbox = {
'min_lat': nominatim_result['boundingbox'][0],
'max_lat': nominatim_result['boundingbox'][1],
'min_lon': nominatim_result['boundingbox'][2],
'max_lon': nominatim_result['boundingbox'][3]
}
nodes, ways = self.overpass_search(place, tags, bbox, limit, radius)
# use results from overpass, but if they do not exist,
# fall back to the nominatim result. this may or may
# not be a good idea.
things_nearby = (nodes
if len(nodes) > 0
else OsmSearcher.fallback(nominatim_result))
origin = get_bounding_box_center(bbox)
things_nearby = sort_by_closeness(origin, things_nearby)
things_nearby = things_nearby[:limit]
other_results = ways[:(limit+5)]
if not things_nearby or len(things_nearby) == 0:
return NO_RESULTS
tag_type_str = ", ".join(tags)
example_link = "https://www.openstreetmap.org/#map=19//"
if len(other_results) > 0:
extra_info = (
"\n\n----------\n\n"
f"Additionally, here are some other results that might be useful. "
"The exact distance from the requested location is not known. "
"The seconary results are below."
"\n\n----------\n\n"
f"{str(other_results)}")
else:
extra_info = ""
return (
f"These are some of the {tag_type_str} points of interest nearby. "
"These are the results known to be closest to the requested location. "
"When telling the user about them, make sure to report "
"all the information (address, contact info, website, etc).\n\n"
"Tell the user about ALL the results, and give the CLOSEST result "
"first. The results are ordered by closeness. "
"Make friendly human-readable OpenStreetMap links when possible, "
"by using the latitude and longitude of the amenities: "
f"{example_link}\n\n"
"Give map links friendly, contextual labels. Don't just print "
f"the naked link:\n"
f' - Example: You can view it on [OpenStreetMap]({example_link})'
f' - Example: Here it is on [OpenStreetMap]({example_link})'
f' - Example: You can find it on [OpenStreetMap]({example_link})'
"\n\nAnd so on.\n\n"
"Only use relevant results. If there are no relevant results, "
"say so. Do not make up answers or hallucinate. "
f"\n\n{NO_CONFUSION}\n\n"
"The primary results are below. "
"Remember that the CLOSEST result is first, and you should use "
"that result first. "
"Prioritize OSM **nodes** over **ways** and **relations**."
"\n\n----------\n\n"
f"{str(things_nearby)}"
f"{extra_info}"
)
else:
return NO_RESULTS
except ValueError:
return NO_RESULTS
except Exception as e:
print(e)
return (f"No results were found, because of an error. "
f"Tell the user that there was an error finding results. "
f"The error was: {e}")
class Tools:
class Valves(BaseModel):
user_agent: str = Field(
default="", description="Unique user agent to identify your OSM API requests."
)
from_header: str = Field(
default="", description="Email address to identify your OSM requests."
)
nominatim_url: str = Field(
default="https://nominatim.openstreetmap.org/search",
description="URL of OSM Nominatim API for reverse geocoding (address lookup)."
)
overpass_turbo_url: str = Field(
default="https://overpass-api.de/api/interpreter",
description="URL of Overpass Turbo API for searching OpenStreetMap."
)
pass
class UserValves(BaseModel):
pass
def __init__(self):
self.valves = self.Valves()
def lookup_location(self, address_or_place: str) -> str:
"""
Looks up GPS and address details on OpenStreetMap of a given address or place.
:param address_or_place: The address or place to look up.
:return: Address details, if found. None if there's an error.
"""
searcher = OsmSearcher(self.valves)
try:
result = searcher.nominatim_search(address_or_place, limit=5)
if result:
return str(result)
else:
return NO_RESULTS
except Exception as e:
print(e)
return (f"There are no results due to an error. "
"Tell the user that there was an error. "
f"The error was: {e}. "
f"Tell the user the error message.")
def find_grocery_stores_near_place(self, place: str) -> str:
"""
Finds supermarkets, grocery stores, and other food stores on
OpenStreetMap near a given place or address. The location of the
address or place is reverse geo-coded, then nearby results
are fetched from OpenStreetMap.
:param place: The name of a place or an address. City and country must be specified, if known.
:return: A list of nearby grocery stores or supermarkets, if found.
"""
searcher = OsmSearcher(self.valves)
tags = ["shop=supermarket", "shop=grocery", "shop=convenience", "shop=greengrocer"]
return searcher.search_nearby(place, tags, limit=5)
def find_bakeries_near_place(self, place: str) -> str:
"""
Finds bakeries on OpenStreetMap near a given place or
address. The location of the address or place is reverse
geo-coded, then nearby results are fetched from OpenStreetMap.
:param place: The name of a place or an address. City and country must be specified, if known.
:return: A list of nearby bakeries, if found.
"""
searcher = OsmSearcher(self.valves)
tags = ["shop=bakery"]
return searcher.search_nearby(place, tags, limit=5)
def find_food_near_place(self, place: str) -> str:
"""
Finds restaurants, fast food, bars, breweries, pubs, etc on
OpenStreetMap near a given place or address. The location of the
address or place is reverse geo-coded, then nearby results
are fetched from OpenStreetMap.
:param place: The name of a place or an address. City and country must be specified, if known.
:return: A list of nearby restaurants, eateries, etc, if found.
"""
tags = [
"amenity=restaurant",
"amenity=fast_food",
"amenity=cafe",
"amenity=pub",
"amenity=bar",
"amenity=eatery",
"amenity=biergarten",
"amenity=canteen"
]
searcher = OsmSearcher(self.valves)
return searcher.search_nearby(place, tags, limit=5)
def find_swimming_near_place(self, place: str) -> str:
"""
Finds swimming pools, water parks, swimming areas, and other aquatic
activities on OpenStreetMap near a given place or address. The location
of the address or place is reverse geo-coded, then nearby results are fetched
from OpenStreetMap.
:param place: The name of a place or an address. City and country must be specified, if known.
:return: A list of swimming poools or places, if found.
"""
tags = [
"leisure=swimming_pool",
"leisure=swimming_area",
"leisure=water_park",
"tourism=theme_park"
]
searcher = OsmSearcher(self.valves)
return searcher.search_nearby(place, tags, limit=5, radius=10000)
def find_recreation_near_place(self, place: str) -> str:
"""
Finds playgrounds, theme parks, frisbee golf, ice skating, and other recreational
activities on OpenStreetMap near a given place or address. The location
of the address or place is reverse geo-coded, then nearby results are fetched
from OpenStreetMap.
:param place: The name of a place or an address. City and country must be specified, if known.
:return: A list of recreational places, if found.
"""
tags = [
"leisure=horse_riding",
"leisure=ice_rink",
"leisure=park",
"leisure=playground",
"leisure=disc_golf_course",
"leisure=amusement_arcade",
"tourism=theme_park"
]
searcher = OsmSearcher(self.valves)
return searcher.search_nearby(place, tags, limit=10, radius=10000)
def find_place_of_worship_near_place(self, place: str) -> str:
"""
Finds places of worship (churches, mosques, temples, etc) on
OpenStreetMap near a given place or address. The location of the
address or place is reverse geo-coded, then nearby results
are fetched from OpenStreetMap.
:param place: The name of a place or an address. City and country must be specified, if known.
:return: A list of nearby places of worship, if found.
"""
tags = ["amenity=place_of_worship"]
searcher = OsmSearcher(self.valves)
return searcher.search_nearby(place, tags, limit=5)
def find_accommodation_near_place(self, place: str) -> str:
"""
Finds accommodation (hotels, guesthouses, hostels, etc) on
OpenStreetMap near a given place or address. The location of the
address or place is reverse geo-coded, then nearby results
are fetched from OpenStreetMap.
:param place: The name of a place or an address. City and country must be specified, if known.
:return: A list of nearby accommodation, if found.
"""
tags = [
"tourism=hotel", "tourism=chalet", "tourism=guest_house", "tourism=guesthouse",
"tourism=motel", "tourism=hostel"
]
searcher = OsmSearcher(self.valves)
return searcher.search_nearby(place, tags, limit=10, radius=10000)
def find_alcohol_near_place(self, place: str) -> str:
"""
Finds beer stores, liquor stores, and similar shops on OpenStreetMap
near a given place or address. The location of the address or place is
reverse geo-coded, then nearby results
are fetched from OpenStreetMap.
:param place: The name of a place or an address. City and country must be specified, if known.
:return: A list of nearby alcohol shops, if found.
"""
searcher = OsmSearcher(self.valves)
tags = ["shop=alcohol"]
return searcher.search_nearby(place, tags, limit=5)
def find_drugs_near_place(self, place: str) -> str:
"""
Finds cannabis dispensaries, coffeeshops, smartshops, and similar stores on OpenStreetMap
near a given place or address. The location of the address or place is
reverse geo-coded, then nearby results
are fetched from OpenStreetMap.
:param place: The name of a place or an address. City and country must be specified, if known.
:return: A list of nearby cannabis and smart shops, if found.
"""
searcher = OsmSearcher(self.valves)
tags = ["shop=coffeeshop", "shop=cannabis", "shop=headshop", "shop=smartshop"]
return searcher.search_nearby(place, tags, limit=5)