[ { "id": "bcf0f5f0-267d-4580-be8a-b89be07379dc", "userId": "c8801c6a-9835-41d0-8a71-1f9b1f475770", "tool": { "id": "openstreetmap", "name": "OpenStreetMap", "meta": { "description": "Allows models to automatically query OpenStreetMap for information and nearby points of interest.", "manifest": { "title": "OpenStreetMap Tool", "author": "projectmoon", "author_url": "https://git.agnos.is/projectmoon/open-webui-filters", "version": "2.2.1", "license": "AGPL-3.0+", "required_open_webui_version": "0.4.3", "requirements": "openrouteservice, pygments" } }, "content": "\"\"\"\ntitle: OpenStreetMap Tool\nauthor: projectmoon\nauthor_url: https://git.agnos.is/projectmoon/open-webui-filters\nversion: 2.2.1\nlicense: AGPL-3.0+\nrequired_open_webui_version: 0.4.3\nrequirements: openrouteservice, pygments\n\"\"\"\nimport itertools\nimport json\nimport math\nimport requests\n\nfrom pygments import highlight\nfrom pygments.lexers import JsonLexer\nfrom pygments.formatters import HtmlFormatter\n\nimport openrouteservice\nfrom openrouteservice.directions import directions as ors_directions\n\nfrom urllib.parse import urljoin\nfrom operator import itemgetter\nfrom typing import List, Optional\nfrom pydantic import BaseModel, Field\n\n# Yoinked from the OpenWebUI CSS\nFONTS = \",\".join([\n \"-apple-system\", \"BlinkMacSystemFont\", \"Inter\",\n \"ui-sans-serif\", \"system-ui\", \"Segoe UI\",\n \"Roboto\", \"Ubuntu\", \"Cantarell\", \"Noto Sans\",\n \"sans-serif\", \"Helvetica Neue\", \"Arial\",\n \"\\\"Apple Color Emoji\\\"\", \"\\\"Segoe UI Emoji\\\"\",\n \"Segoe UI Symbol\", \"\\\"Noto Color Emoji\\\"\"\n])\n\nFONT_CSS = f\"\"\"\nhtml {{ font-family: {FONTS}; }}\n\n@media (prefers-color-scheme: dark) {{\n html {{\n --tw-text-opacity: 1;\n color: rgb(227 227 227 / var(--tw-text-opacity));\n }}\n}}\n\"\"\"\n\nHIGHLIGHT_CSS = HtmlFormatter().get_style_defs('.highlight')\n\nNOMINATIM_LOOKUP_TYPES = {\n \"node\": \"N\",\n \"route\": \"R\",\n \"way\": \"W\"\n}\n\nOLD_VALVE_SETTING = \"\"\" Tell the user that you cannot search\nOpenStreetMap until the configuration is fixed. The Nominatim URL\nvalve setting needs to be updated. There has been a breaking change in\n1.0 of the OpenStreetMap tool. The valve setting is currently set to:\n`{OLD}`.\n\nIt shoule be set to the root URL of the Nominatim endpoint, for\nexample:\n\n`https://nominatim.openstreetmap.org/`\n\nInform the user they need to fix this configuration setting.\n\"\"\".replace(\"\\n\", \" \").strip()\n\nVALVES_NOT_SET = \"\"\"\nTell the user that the User-Agent and From headers\nmust be set to comply with the OSM Nominatim terms\nof use: https://operations.osmfoundation.org/policies/nominatim/\n\"\"\".replace(\"\\n\", \" \").strip()\n\nNO_RESULTS = (\"No results found. Tell the user you found no results. \"\n \"Do not make up answers or hallucinate. Only say you \"\n \"found no results.\")\n\nNO_RESULTS_BAD_ADDRESS = (\"No results found. Tell the user you found no results because \"\n \"OpenStreetMap could not resolve the address. \"\n \"Print the exact address or location you searched for. \"\n \"Suggest to the user that they refine their \"\n \"question, for example removing the apartment number, sub-unit, \"\n \"etc. Example: If `123 Main Street, Apt 4` returned no results, \"\n \"suggest that the user searc for `123 Main Street` instead. \"\n \"Use the address the user searched for in your example.\")\n\nNO_CONFUSION = (\"**IMPORTANT!:** Check that the results match the location \"\n \"the user is talking about, by analyzing the conversation history. \"\n \"Sometimes there are places with the same \"\n \"names, but in different cities or countries. If the results are for \"\n \"a different city or country than the user is interested in, say so: \"\n \"tell the user that the results are for the wrong place, and tell them \"\n \"to be more specific in their query.\")\n\n# Give examples of OSM links to help prevent wonky generated links\n# with correct GPS coords but incorrect URLs.\nEXAMPLE_OSM_LINK = \"https://www.openstreetmap.org/#map=19/\u003Clat\u003E/\u003Clon\u003E\"\nOSM_LINK_INSTRUCTIONS = (\n \"Make friendly human-readable OpenStreetMap links when possible, \"\n \"by using the latitude and longitude of the amenities: \"\n f\"{EXAMPLE_OSM_LINK}\\n\\n\"\n)\n\ndef chunk_list(input_list, chunk_size):\n it = iter(input_list)\n return list(\n itertools.zip_longest(*[iter(it)] * chunk_size, fillvalue=None)\n )\n\ndef to_lookup(thing) -\u003E Optional[str]:\n lookup_type = NOMINATIM_LOOKUP_TYPES.get(thing['type'])\n if lookup_type is not None:\n return f\"{lookup_type}{thing['id']}\"\n\ndef specific_place_instructions() -\u003E str:\n return (\n \"# Result Instructions\\n\"\n \"These are search results ordered by relevance for the \"\n \"address, place, landmark, or location the user is asking \"\n \"about. **IMPORTANT!:** Tell the user all relevant information, \"\n \"including address, contact information, and the OpenStreetMap link. \"\n \"Make the map link into a nice human-readable markdown link.\"\n )\n\ndef navigation_instructions(travel_type) -\u003E str:\n return (\n \"# Result Instructions\\n\"\n \"This is the navigation route that the user has requested. \"\n f\"These instructions are for travel by {travel_type}. \"\n \"Tell the user the total distance, \"\n \"and estimated travel time. \"\n \"If the user **specifically asked for it**, also tell \"\n \"them the route itself. When telling the route, you must tell \"\n f\"the user that it's a **{travel_type}** route.\"\n )\n\ndef detailed_instructions(tag_type_str: str) -\u003E str:\n \"\"\"\n Produce detailed instructions for models good at following\n detailed instructions.\n \"\"\"\n return (\n \"# Detailed Search Result Instructions\\n\"\n f\"These are some of the {tag_type_str} points of interest nearby. \"\n \"These are the results known to be closest to the requested location. \"\n \"When telling the user about them, make sure to report \"\n \"all the information (address, contact info, website, etc).\\n\\n\"\n \"Tell the user about ALL the results, and give closer results \"\n \"first. Closer results are higher in the list. When telling the \"\n \"user the distance, use the TRAVEL DISTANCE. Do not say one \"\n \"distance is farther away than another. Just say what the \"\n \"distances are. \"\n f\"{OSM_LINK_INSTRUCTIONS}\"\n \"Give map links friendly, contextual labels. Don't just print \"\n f\"the naked link:\\n\"\n f' - Example: You can view it on [OpenStreetMap]({EXAMPLE_OSM_LINK})\\n'\n f' - Example: Here it is on [OpenStreetMap]({EXAMPLE_OSM_LINK})\\n'\n f' - Example: You can find it on [OpenStreetMap]({EXAMPLE_OSM_LINK})\\n'\n \"\\n\\nAnd so on.\\n\\n\"\n \"Only use relevant results. If there are no relevant results, \"\n \"say so. Do not make up answers or hallucinate. \"\n f\"\\n\\n{NO_CONFUSION}\\n\\n\"\n \"Remember that the CLOSEST result is first, and you should use \"\n \"that result first.\\n\\n\"\n \"The results (if present) are below, in Markdown format.\\n\\n\"\n \"**ALWAYS SAY THE CLOSEST RESULT FIRST!**\"\n )\n\ndef simple_instructions(tag_type_str: str) -\u003E str:\n \"\"\"\n Produce simpler markdown-oriented instructions for models that do\n better with that.\n \"\"\"\n return (\n \"# OpenStreetMap Result Instructions\\n\"\n f\"These are some of the {tag_type_str} points of interest nearby. \"\n \"These are the results known to be closest to the requested location. \"\n \"For each result, report the following information: \\n\"\n \" - Name\\n\"\n \" - Address\\n\"\n \" - OpenStreetMap Link (make it a human readable link like 'View on OpenStreetMap')\\n\"\n \" - Contact information (address, phone, website, email, etc)\\n\\n\"\n \"Tell the user about ALL the results, and give the CLOSEST result \"\n \"first. The results are ordered by closeness as the crow flies. \"\n \"When telling the user about distances, use the TRAVEL DISTANCE only. \"\n \"Only use relevant results. If there are no relevant results, \"\n \"say so. Do not make up answers or hallucinate. \"\n \"Make sure that your results are in the actual location the user is talking about, \"\n \"and not a place of the same name in a different country.\"\n \"The search results are below.\"\n )\n\ndef merge_from_nominatim(thing, nominatim_result) -\u003E Optional[dict]:\n \"\"\"Merge information into object missing all or some of it.\"\"\"\n if thing is None:\n return None\n\n if 'address' not in nominatim_result:\n return None\n\n nominatim_address = nominatim_result['address']\n\n # prioritize actual name, road name, then display name. display\n # name is often the full address, which is a bit much.\n nominatim_name = nominatim_result.get('name')\n nominatim_road = nominatim_address.get('road')\n nominatim_display_name = nominatim_result.get('display_name')\n thing_name = thing.get('name')\n\n if nominatim_name and not thing_name:\n thing['name'] = nominatim_name.strip()\n elif nominatim_road and not thing_name:\n thing['name'] = nominatim_road.strip()\n elif nominatim_display_name and not thing_name:\n thing['name'] = nominatim_display_name.strip()\n\n tags = thing.get('tags', {})\n\n for key in nominatim_address:\n obj_key = f\"addr:{key}\"\n if obj_key not in tags:\n tags[obj_key] = nominatim_address[key]\n\n thing['tags'] = tags\n return thing\n\ndef pretty_print_thing_json(thing):\n \"\"\"Converts an OSM thing to nice JSON HTML.\"\"\"\n formatted_json_str = json.dumps(thing, indent=2)\n lexer = JsonLexer()\n formatter = HtmlFormatter(style='colorful')\n return highlight(formatted_json_str, lexer, formatter)\n\n\ndef thing_is_useful(thing):\n \"\"\"\n Determine if an OSM way entry is useful to us. This means it\n has something more than just its main classification tag, and\n (usually) has at least a name. Some exceptions are made for ways\n that do not have names.\n \"\"\"\n tags = thing.get('tags', {})\n has_tags = len(tags) \u003E 1\n has_useful_tags = (\n 'leisure' in tags or\n 'shop' in tags or\n 'amenity' in tags or\n 'car:rental' in tags or\n 'rental' in tags or\n 'car_rental' in tags or\n 'service:bicycle:rental' in tags or\n 'tourism' in tags\n )\n\n # there can be a lot of artwork in city centers. drop ones that\n # aren't as notable. we define notable by the thing having wiki\n # entries, or by being tagged as historical.\n if tags.get('tourism', '') == 'artwork':\n notable = (\n 'wikipedia' in tags or\n 'wikimedia_commons' in tags\n )\n else:\n notable = True\n\n return has_tags and has_useful_tags and notable\n\ndef thing_has_info(thing):\n has_name = any('name' in tag for tag in thing['tags'])\n return thing_is_useful(thing) and has_name\n # is_exception = way['tags'].get('leisure', None) is not None\n # return has_tags and (has_name or is_exception)\n\ndef process_way_result(way) -\u003E Optional[dict]:\n \"\"\"\n Post-process an OSM Way dict to remove the geometry and node\n info, and calculate a single GPS coordinate from its bounding\n box.\n \"\"\"\n if 'nodes' in way:\n del way['nodes']\n\n if 'geometry' in way:\n del way['geometry']\n\n if 'bounds' in way:\n way_center = get_bounding_box_center(way['bounds'])\n way['lat'] = way_center['lat']\n way['lon'] = way_center['lon']\n del way['bounds']\n return way\n\n return None\n\ndef get_bounding_box_center(bbox):\n def convert(bbox, key):\n return bbox[key] if isinstance(bbox[key], float) else float(bbox[key])\n\n min_lat = convert(bbox, 'minlat')\n min_lon = convert(bbox, 'minlon')\n max_lat = convert(bbox, 'maxlat')\n max_lon = convert(bbox, 'maxlon')\n\n return {\n 'lon': (min_lon + max_lon) / 2,\n 'lat': (min_lat + max_lat) / 2\n }\n\ndef haversine_distance(point1, point2):\n R = 6371 # Earth radius in kilometers\n\n lat1, lon1 = point1['lat'], point1['lon']\n lat2, lon2 = point2['lat'], point2['lon']\n\n d_lat = math.radians(lat2 - lat1)\n d_lon = math.radians(lon2 - lon1)\n a = (math.sin(d_lat / 2) * math.sin(d_lat / 2) +\n math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) *\n math.sin(d_lon / 2) * math.sin(d_lon / 2))\n c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))\n\n return R * c\n\ndef sort_by_closeness(origin, points, *keys: str):\n \"\"\"\n Sorts a list of { lat, lon }-like dicts by closeness to an origin point.\n The origin is a dict with keys of { lat, lon }.\n \"\"\"\n return sorted(points, key=itemgetter(*keys))\n\ndef sort_by_rank(things):\n \"\"\"\n Calculate a rank for a list of things. More important ones are\n pushed towards the top. Currently only for tourism tags.\n \"\"\"\n def rank_thing(thing: dict) -\u003E int:\n tags = thing.get('tags', {})\n if not 'tourism' in tags:\n return 0\n\n rank = len([name for name in tags.keys()\n if name.startswith(\"name\")])\n rank += 5 if 'historic' in tags else 0\n rank += 5 if 'wikipedia' in tags else 0\n rank += 1 if 'wikimedia_commons' in tags else 0\n rank += 5 if tags.get('tourism', '') == 'museum' else 0\n rank += 5 if tags.get('tourism', '') == 'aquarium' else 0\n rank += 5 if tags.get('tourism', '') == 'zoo' else 0\n return rank\n\n return sorted(things, reverse=True, key=lambda t: (rank_thing(t), -t['distance']))\n\ndef get_or_none(tags: dict, *keys: str) -\u003E Optional[str]:\n \"\"\"\n Try to extract a value from a dict by trying keys in order, or\n return None if none of the keys were found.\n \"\"\"\n for key in keys:\n if key in tags:\n return tags[key]\n\n return None\n\ndef all_are_none(*args) -\u003E bool:\n for arg in args:\n if arg is not None:\n return False\n\n return True\n\ndef friendly_shop_name(shop_type: str) -\u003E str:\n \"\"\"\n Make certain shop types more friendly for LLM interpretation.\n \"\"\"\n if shop_type == \"doityourself\":\n return \"hardware\"\n else:\n return shop_type\n\ndef parse_thing_address(thing: dict) -\u003E Optional[str]:\n \"\"\"\n Parse address from either an Overpass result or Nominatim\n result.\n \"\"\"\n if 'address' in thing:\n # nominatim result\n return parse_address_from_address_obj(thing['address'])\n else:\n return parse_address_from_tags(thing['tags'])\n\ndef parse_address_from_address_obj(address) -\u003E Optional[str]:\n \"\"\"Parse address from Nominatim address object.\"\"\"\n house_number = get_or_none(address, \"house_number\")\n street = get_or_none(address, \"road\")\n city = get_or_none(address, \"city\")\n state = get_or_none(address, \"state\")\n postal_code = get_or_none(address, \"postcode\")\n\n # if all are none, that means we don't know the address at all.\n if all_are_none(house_number, street, city, state, postal_code):\n return None\n\n # Handle missing values to create complete-ish addresses, even if\n # we have missing data. We will get either a partly complete\n # address, or None if all the values are missing.\n line1 = filter(None, [street, house_number])\n line2 = filter(None, [city, state, postal_code])\n line1 = \" \".join(line1).strip()\n line2 = \" \".join(line2).strip()\n full_address = filter(None, [line1, line2])\n full_address = \", \".join(full_address).strip()\n return full_address if len(full_address) \u003E 0 else None\n\ndef parse_address_from_tags(tags: dict) -\u003E Optional[str]:\n \"\"\"Parse address from Overpass tags object.\"\"\"\n house_number = get_or_none(tags, \"addr:housenumber\", \"addr:house_number\")\n street = get_or_none(tags, \"addr:street\")\n city = get_or_none(tags, \"addr:city\")\n state = get_or_none(tags, \"addr:state\", \"addr:province\")\n postal_code = get_or_none(\n tags,\n \"addr:postcode\", \"addr:post_code\", \"addr:postal_code\",\n \"addr:zipcode\", \"addr:zip_code\"\n )\n\n # if all are none, that means we don't know the address at all.\n if all_are_none(house_number, street, city, state, postal_code):\n return None\n\n # Handle missing values to create complete-ish addresses, even if\n # we have missing data. We will get either a partly complete\n # address, or None if all the values are missing.\n line1 = filter(None, [street, house_number])\n line2 = filter(None, [city, state, postal_code])\n line1 = \" \".join(line1).strip()\n line2 = \" \".join(line2).strip()\n full_address = filter(None, [line1, line2])\n full_address = \", \".join(full_address).strip()\n return full_address if len(full_address) \u003E 0 else None\n\ndef parse_thing_amenity_type(thing: dict, tags: dict) -\u003E Optional[dict]:\n \"\"\"\n Extract amenity type or other identifying category from\n Nominatim or Overpass result object.\n \"\"\"\n if 'amenity' in tags:\n return tags['amenity']\n\n if thing.get('class') == 'amenity' or thing.get('class') == 'shop':\n return thing.get('type')\n\n # fall back to tag categories, like shop=*\n if 'shop' in tags:\n return friendly_shop_name(tags['shop'])\n if 'leisure' in tags:\n return friendly_shop_name(tags['leisure'])\n\n return None\n\ndef parse_and_validate_thing(thing: dict) -\u003E Optional[dict]:\n \"\"\"\n Parse an OSM result (node or post-processed way) and make it\n more friendly to work with. Helps remove ambiguity of the LLM\n interpreting the raw JSON data. If there is not enough data,\n discard the result.\n \"\"\"\n tags: dict = thing['tags'] if 'tags' in thing else {}\n\n # Currently we define \"enough data\" as at least having lat, lon,\n # and a name. nameless things are allowed if they are in a certain\n # class of POIs (leisure).\n has_name = 'name' in tags or 'name' in thing\n is_leisure = 'leisure' in tags or 'leisure' in thing\n if 'lat' not in thing or 'lon' not in thing:\n return None\n\n if not has_name and not is_leisure:\n return None\n\n friendly_thing = {}\n name: str = (tags['name'] if 'name' in tags\n else thing['name'] if 'name' in thing\n else str(thing['id']) if 'id' in thing\n else str(thing['osm_id']) if 'osm_id' in thing\n else \"unknown\")\n\n address: str = parse_thing_address(thing)\n distance: Optional[float] = thing.get('distance', None)\n nav_distance: Optional[float] = thing.get('nav_distance', None)\n opening_hours: Optional[str] = tags.get('opening_hours', None)\n\n lat: Optional[float] = thing.get('lat', None)\n lon: Optional[float] = thing.get('lon', None)\n amenity_type: Optional[str] = parse_thing_amenity_type(thing, tags)\n\n # use the navigation distance if it's present. but if not, set to\n # the haversine distance so that we at least get coherent results\n # for LLM.\n friendly_thing['distance'] = \"{:.3f}\".format(distance) if distance else \"unknown\"\n if nav_distance:\n friendly_thing['nav_distance'] = \"{:.3f}\".format(nav_distance) + \" km\"\n else:\n friendly_thing['nav_distance'] = f\"a bit more than {friendly_thing['distance']}km\"\n\n friendly_thing['name'] = name if name else \"unknown\"\n friendly_thing['address'] = address if address else \"unknown\"\n friendly_thing['lat'] = lat if lat else \"unknown\"\n friendly_thing['lon'] = lon if lon else \"unknown\"\n friendly_thing['amenity_type'] = amenity_type if amenity_type else \"unknown\"\n friendly_thing['opening_hours'] = opening_hours if opening_hours else \"not recorded\"\n return friendly_thing\n\ndef create_osm_link(lat, lon):\n return EXAMPLE_OSM_LINK.replace(\"\u003Clat\u003E\", str(lat)).replace(\"\u003Clon\u003E\", str(lon))\n\ndef convert_and_validate_results(\n original_location: str,\n things_nearby: List[dict],\n sort_message: str=\"closeness\",\n use_distance: bool=True\n) -\u003E Optional[str]:\n \"\"\"\n Converts the things_nearby JSON into Markdown-ish results to\n (hopefully) improve model understanding of the results. Intended\n to stop misinterpretation of GPS coordinates when creating map\n links. Also drops incomplete results. Supports Overpass and\n Nominatim results.\n \"\"\"\n entries = []\n for thing in things_nearby:\n # Convert to friendlier data, drop results without names etc.\n # No need to instruct LLM to generate map links if we do it\n # instead.\n friendly_thing = parse_and_validate_thing(thing)\n if not friendly_thing:\n continue\n\n distance = (f\" - Haversine Distance from Origin: {friendly_thing['distance']} km\\n\"\n if use_distance else \"\")\n travel_distance = (f\" - Travel Distance from Origin: {friendly_thing['nav_distance']}\\n\"\n if use_distance and 'nav_distance' in friendly_thing else \"\")\n map_link = create_osm_link(friendly_thing['lat'], friendly_thing['lon'])\n entry = (f\"## {friendly_thing['name']}\\n\"\n f\" - Latitude: {friendly_thing['lat']}\\n\"\n f\" - Longitude: {friendly_thing['lon']}\\n\"\n f\" - Address: {friendly_thing['address']}\\n\"\n f\" - Amenity Type: {friendly_thing['amenity_type']}\\n\"\n f\"{distance}\"\n f\"{travel_distance}\"\n f\" - OpenStreetMap link: {map_link}\\n\\n\"\n f\"Raw JSON data:\\n\"\n \"```json\\n\"\n f\"{str(thing)}\\n\"\n \"```\")\n\n entries.append(entry)\n\n if len(entries) == 0:\n return None\n\n result_text = \"\\n\\n\".join(entries)\n header = (\"# Search Results\\n\"\n f\"Ordered by {sort_message} to {original_location}.\")\n\n return f\"{header}\\n\\n{result_text}\"\n\nclass OsmCache:\n def __init__(self, filename=\"/tmp/osm.json\"):\n self.filename = filename\n self.data = {}\n\n # Load existing cache if it exists\n try:\n with open(self.filename, 'r') as f:\n self.data = json.load(f)\n except FileNotFoundError:\n pass\n\n def get(self, key):\n return self.data.get(key)\n\n def set(self, key, value):\n self.data[key] = value\n with open(self.filename, 'w') as f:\n json.dump(self.data, f)\n\n def get_or_set(self, key, func_to_call):\n \"\"\"\n Retrieve the value from the cache for a given key. If the key is not found,\n call `func_to_call` to generate the value and store it in the cache.\n\n :param key: The key to look up or set in the cache\n :param func_to_call: A callable function that returns the value if key is missing\n :return: The cached or generated value\n \"\"\"\n if key not in self.data:\n value = func_to_call()\n self.set(key, value)\n return self.data[key]\n\n def clear_cache(self):\n \"\"\"\n Clear all entries from the cache.\n \"\"\"\n self.data.clear()\n try:\n # Erase contents of the cache file.\n with open(self.filename, 'w'):\n pass\n except FileNotFoundError:\n pass\n\nclass OrsRouter:\n def __init__(\n self, valves, user_valves: Optional[dict], event_emitter=None,\n ):\n self.cache = OsmCache()\n self.valves = valves\n self.event_emitter = event_emitter\n self.user_valves = user_valves\n\n if self.valves.ors_api_key is not None and self.valves.ors_api_key != \"\":\n if self.valves.ors_instance is not None:\n self._client = openrouteservice.Client(\n base_url=self.valves.ors_instance,\n key=self.valves.ors_api_key\n )\n else:\n self._client = openrouteservice.Client(key=self.valves.ors_api_key)\n else:\n self._client = None\n\n\n def calculate_route(\n self, from_thing: dict, to_thing: dict\n ) -\u003E Optional[dict]:\n \"\"\"\n Calculate route between A and B. Returns the route,\n if successful, or None if the distance could not be\n calculated, or if ORS is not configured.\n \"\"\"\n if not self._client:\n return None\n\n # select profile based on distance for more accurate\n # measurements. very close haversine distances use the walking\n # profile, which should (usually?) essentially cover walking\n # and biking. further away = use car.\n if to_thing.get('distance', 9000) \u003C= 1.5:\n profile = \"foot-walking\"\n else:\n profile = \"driving-car\"\n\n coords = ((from_thing['lon'], from_thing['lat']),\n (to_thing['lon'], to_thing['lat']))\n\n # check cache first.\n cache_key = f\"ors_route_{str(coords)}\"\n cached_route = self.cache.get(cache_key)\n if cached_route:\n print(\"[OSM] Got route from cache!\")\n return cached_route\n\n resp = ors_directions(self._client, coords, profile=profile,\n preference=\"fastest\", units=\"km\")\n\n routes = resp.get('routes', [])\n if len(routes) \u003E 0:\n self.cache.set(cache_key, routes[0])\n return routes[0]\n else:\n return None\n\n def calculate_distance(\n self, from_thing: dict, to_thing: dict\n ) -\u003E Optional[float]:\n \"\"\"\n Calculate navigation distance between A and B. Returns the\n distance calculated, if successful, or None if the distance\n could not be calculated, or if ORS is not configured.\n \"\"\"\n if not self._client:\n return None\n\n route = self.calculate_route(from_thing, to_thing)\n return route.get('summary', {}).get('distance', None) if route else None\n\nclass OsmSearcher:\n def __init__(self, valves, user_valves: Optional[dict], event_emitter=None):\n self.valves = valves\n self.event_emitter = event_emitter\n self.user_valves = user_valves\n self._ors = OrsRouter(valves, user_valves, event_emitter)\n\n def create_headers(self) -\u003E Optional[dict]:\n if len(self.valves.user_agent) == 0 or len(self.valves.from_header) == 0:\n return None\n\n return {\n 'User-Agent': self.valves.user_agent,\n 'From': self.valves.from_header\n }\n\n async def event_resolving(self, done: bool=False):\n if not self.event_emitter or not self.valves.status_indicators:\n return\n\n if done:\n message = \"OpenStreetMap: resolution complete.\"\n else:\n message = \"OpenStreetMap: resolving...\"\n\n await self.event_emitter({\n \"type\": \"status\",\n \"data\": {\n \"status\": \"in_progress\",\n \"description\": message,\n \"done\": done,\n },\n })\n\n async def event_fetching(self, done: bool=False, message=\"OpenStreetMap: fetching additional info\"):\n if not self.event_emitter or not self.valves.status_indicators:\n return\n\n await self.event_emitter({\n \"type\": \"status\",\n \"data\": {\n \"status\": \"in_progress\",\n \"description\": message,\n \"done\": done,\n },\n })\n\n async def event_searching(\n self, category: str, place: str,\n status: str=\"in_progress\", done: bool=False\n ):\n if not self.event_emitter or not self.valves.status_indicators:\n return\n\n await self.event_emitter({\n \"type\": \"status\",\n \"data\": {\n \"status\": status,\n \"description\": f\"OpenStreetMap: searching for {category} near {place}\",\n \"done\": done,\n },\n })\n\n async def event_search_complete(self, category: str, place: str, num_results: int):\n if not self.event_emitter or not self.valves.status_indicators:\n return\n\n await self.event_emitter({\n \"type\": \"status\",\n \"data\": {\n \"status\": \"complete\",\n \"description\": f\"OpenStreetMap: found {num_results} '{category}' results\",\n \"done\": True,\n },\n })\n\n def create_result_document(self, thing) -\u003E Optional[dict]:\n original_thing = thing\n thing = parse_and_validate_thing(thing)\n\n if not thing:\n return None\n\n if 'address' in original_thing:\n street = get_or_none(original_thing['address'], \"road\")\n else:\n street = get_or_none(original_thing['tags'], \"addr:street\")\n\n street_name = street if street is not None else \"\"\n source_name = f\"{thing['name']} {street_name}\"\n lat, lon = thing['lat'], thing['lon']\n osm_link = create_osm_link(lat, lon)\n json_data = pretty_print_thing_json(original_thing)\n addr = f\"at {thing['address']}\" if thing['address'] != 'unknown' else 'nearby'\n\n document = (f\"\u003Cstyle\u003E{HIGHLIGHT_CSS}\u003C/style\u003E\"\n f\"\u003Cstyle\u003E{FONT_CSS}\u003C/style\u003E\"\n f\"\u003Cdiv\u003E\"\n f\"\u003Cp\u003E{thing['name']} is located {addr}.\u003C/p\u003E\"\n f\"\u003Cul\u003E\"\n f\"\u003Cli\u003E\"\n f\" \u003Cstrong\u003EOpening Hours:\u003C/strong\u003E {thing['opening_hours']}\"\n f\"\u003C/li\u003E\"\n f\"\u003C/ul\u003E\"\n f\"\u003Cp\u003ERaw JSON data:\u003C/p\u003E\"\n f\"{json_data}\"\n f\"\u003C/div\u003E\")\n\n return { \"source_name\": source_name, \"document\": document, \"osm_link\": osm_link }\n\n async def emit_result_citation(self, thing):\n if not self.event_emitter or not self.valves.status_indicators:\n return\n\n converted = self.create_result_document(thing)\n if not converted:\n return\n\n source_name = converted[\"source_name\"]\n document = converted[\"document\"]\n osm_link = converted[\"osm_link\"]\n\n await self.event_emitter({\n \"type\": \"source\",\n \"data\": {\n \"document\": [document],\n \"metadata\": [{\"source\": source_name, \"html\": True }],\n \"source\": {\"name\": source_name, \"url\": osm_link},\n }\n })\n\n async def event_error(self, exception: Exception):\n if not self.event_emitter or not self.valves.status_indicators:\n return\n\n await self.event_emitter({\n \"type\": \"status\",\n \"data\": {\n \"status\": \"error\",\n \"description\": f\"Error searching OpenStreetMap: {str(exception)}\",\n \"done\": True,\n },\n })\n\n def calculate_navigation_distance(self, start, destination) -\u003E float:\n \"\"\"Calculate real distance from A to B, instead of Haversine.\"\"\"\n return self._ors.calculate_distance(start, destination)\n\n def attempt_ors(self, origin, things_nearby) -\u003E bool:\n \"\"\"Update distances to use ORS navigable distances, if ORS enabled.\"\"\"\n used_ors = False\n cache = OsmCache()\n for thing in things_nearby:\n cache_key = f\"ors_{origin}_{thing['id']}\"\n nav_distance = cache.get(cache_key)\n\n if nav_distance:\n print(f\"[OSM] Got nav distance for {thing['id']} from cache!\")\n else:\n print(f\"[OSM] Checking ORS for {thing['id']}\")\n try:\n nav_distance = self.calculate_navigation_distance(origin, thing)\n except Exception as e:\n print(f\"[OSM] Error querying ORS: {e}\")\n print(f\"[OSM] Falling back to regular distance due to ORS error!\")\n nav_distance = thing['distance']\n\n if nav_distance:\n used_ors = True\n cache.set(cache_key, nav_distance)\n thing['nav_distance'] = round(nav_distance, 3)\n\n return used_ors\n\n def calculate_haversine(self, origin, things_nearby):\n for thing in things_nearby:\n if 'distance' not in thing:\n thing['distance'] = round(haversine_distance(origin, thing), 3)\n\n def use_detailed_interpretation_mode(self) -\u003E bool:\n # Let user valve for instruction mode override the global\n # setting.\n if self.user_valves:\n return self.user_valves.instruction_oriented_interpretation\n else:\n return self.valves.instruction_oriented_interpretation\n\n def get_result_instructions(self, tag_type_str: str) -\u003E str:\n if self.use_detailed_interpretation_mode():\n return detailed_instructions(tag_type_str)\n else:\n return simple_instructions(tag_type_str)\n\n @staticmethod\n def group_tags(tags):\n result = {}\n for tag in tags:\n key, value = tag.split('=')\n if key not in result:\n result[key] = []\n result[key].append(value)\n return result\n\n @staticmethod\n def fallback(nominatim_result):\n \"\"\"\n If we do not have Overpass Turbo results, attempt to use the\n Nominatim result instead.\n \"\"\"\n return ([nominatim_result] if 'type' in nominatim_result\n and (nominatim_result['type'] == 'amenity'\n or nominatim_result['type'] == 'shop'\n or nominatim_result['type'] == 'leisure'\n or nominatim_result['type'] == 'tourism')\n else [])\n\n\n async def nominatim_lookup_by_id(self, things, format=\"json\"):\n await self.event_fetching(done=False)\n updated_things = [] # the things with merged info.\n\n # handle last chunk, which can have nones in order due to the\n # way chunking is done.\n things = [thing for thing in things if thing is not None]\n lookups = []\n\n for thing in things:\n if thing is None:\n continue\n lookup = to_lookup(thing)\n if lookup is not None:\n lookups.append(lookup)\n\n # Nominatim likes it if we cache our data.\n cache = OsmCache()\n lookups_to_remove = []\n for lookup_id in lookups:\n from_cache = cache.get(lookup_id)\n if from_cache is not None:\n updated_things.append(from_cache)\n lookups_to_remove.append(lookup_id)\n\n # only need to look up things we do not have cached.\n lookups = [id for id in lookups if id not in lookups_to_remove]\n\n if len(lookups) == 0:\n print(\"[OSM] Got all Nominatim info from cache!\")\n await self.event_fetching(done=True)\n return updated_things\n else:\n print(f\"Looking up {len(lookups)} things from Nominatim\")\n\n\n url = urljoin(self.valves.nominatim_url, \"lookup\")\n params = {\n 'osm_ids': \",\".join(lookups),\n 'format': format\n }\n\n headers = self.create_headers()\n if not headers:\n raise ValueError(\"Headers not set\")\n\n response = requests.get(url, params=params, headers=headers)\n if response.status_code == 200:\n data = response.json()\n\n if not data:\n print(\"[OSM] No results found for lookup\")\n await self.event_fetching(done=True)\n return []\n\n addresses_by_id = {item['osm_id']: item for item in data}\n\n for thing in things:\n nominatim_result = addresses_by_id.get(thing['id'], {})\n if nominatim_result != {}:\n updated = merge_from_nominatim(thing, nominatim_result)\n if updated is not None:\n lookup = to_lookup(thing)\n cache.set(lookup, updated)\n updated_things.append(updated)\n\n await self.event_fetching(done=True)\n return updated_things\n else:\n await self.event_error(Exception(response.text))\n print(response.text)\n return []\n\n\n async def nominatim_search(self, query, format=\"json\", limit: int=1) -\u003E Optional[dict]:\n await self.event_resolving(done=False)\n cache_key = f\"nominatim_search_{query}\"\n cache = OsmCache()\n data = cache.get(cache_key)\n\n if data:\n print(f\"[OSM] Got nominatim search data for {query} from cache!\")\n await self.event_resolving(done=True)\n return data[:limit]\n\n print(f\"[OSM] Searching Nominatim for: {query}\")\n\n url = urljoin(self.valves.nominatim_url, \"search\")\n params = {\n 'q': query,\n 'format': format,\n 'addressdetails': 1,\n 'limit': limit,\n }\n\n headers = self.create_headers()\n if not headers:\n await self.event_error(\"Headers not set\")\n raise ValueError(\"Headers not set\")\n\n response = requests.get(url, params=params, headers=headers)\n if response.status_code == 200:\n await self.event_resolving(done=True)\n data = response.json()\n\n if not data:\n raise ValueError(f\"No results found for query '{query}'\")\n\n print(f\"Got result from Nominatim for: {query}\")\n cache.set(cache_key, data)\n return data[:limit]\n else:\n await self.event_error(Exception(response.text))\n print(response.text)\n return None\n\n\n async def overpass_search(\n self, place, tags, bbox, limit=5, radius=4000\n ) -\u003E (List[dict], List[dict]):\n \"\"\"\n Return a list relevant of OSM nodes and ways. Some\n post-processing is done on ways in order to add coordinates to\n them.\n \"\"\"\n print(f\"Searching Overpass Turbo around origin {place}\")\n headers = self.create_headers()\n if not headers:\n raise ValueError(\"Headers not set\")\n\n url = self.valves.overpass_turbo_url\n center = get_bounding_box_center(bbox)\n around = f\"(around:{radius},{center['lat']},{center['lon']})\"\n\n tag_groups = OsmSearcher.group_tags(tags)\n search_groups = [f'\"{tag_type}\"~\"{\"|\".join(values)}\"'\n for tag_type, values in tag_groups.items()]\n\n searches = []\n for search_group in search_groups:\n searches.append(\n f'nwr[{search_group}]{around}'\n )\n\n search = \";\\n\".join(searches)\n if len(search) \u003E 0:\n search += \";\"\n\n # \"out geom;\" is needed to get bounding box info of ways,\n # so we can calculate the coordinates.\n query = f\"\"\"\n [out:json];\n (\n {search}\n );\n out geom;\n \"\"\"\n\n print(query)\n data = { \"data\": query }\n response = requests.get(url, params=data, headers=headers)\n if response.status_code == 200:\n # nodes have have exact GPS coordinates. we also include\n # useful way entries, post-processed to remove extra data\n # and add a centered calculation of their GPS coords. any\n # way that doesn't have enough info for us to use is\n # dropped.\n results = response.json()\n results = results['elements'] if 'elements' in results else []\n nodes = []\n ways = []\n things_missing_names = []\n\n for res in results:\n if 'type' not in res or not thing_is_useful(res):\n continue\n if res['type'] == 'node':\n if thing_has_info(res):\n nodes.append(res)\n else:\n things_missing_names.append(res)\n elif res['type'] == 'way':\n processed = process_way_result(res)\n if processed is not None and thing_has_info(res):\n ways.append(processed)\n else:\n if processed is not None:\n things_missing_names.append(processed)\n\n # attempt to update ways that have no names/addresses.\n if len(things_missing_names) \u003E 0:\n print(f\"Updating {len(things_missing_names)} things with info\")\n for way_chunk in chunk_list(things_missing_names, 20):\n updated = await self.nominatim_lookup_by_id(way_chunk)\n ways = ways + updated\n\n return nodes, ways\n else:\n print(response.text)\n raise Exception(f\"Error calling Overpass API: {response.text}\")\n\n async def get_things_nearby(self, nominatim_result, place, tags, bbox, limit, radius):\n nodes, ways = await self.overpass_search(place, tags, bbox, limit, radius)\n\n # use results from overpass, but if they do not exist,\n # fall back to the nominatim result. this may or may\n # not be a good idea.\n things_nearby = (nodes + ways\n if len(nodes) \u003E 0 or len(ways) \u003E 0\n else OsmSearcher.fallback(nominatim_result))\n\n # in order to not spam ORS, we first sort by haversine\n # distance and drop number of results to the limit. then, if\n # enabled, we calculate ORS distances. then we sort again.\n origin = get_bounding_box_center(bbox)\n self.calculate_haversine(origin, things_nearby)\n\n # sort by importance + distance, drop to the liimt, then sort\n # by closeness.\n things_nearby = sort_by_rank(things_nearby)\n things_nearby = things_nearby[:limit] # drop down to requested limit\n things_nearby = sort_by_closeness(origin, things_nearby, 'distance')\n\n if self.attempt_ors(origin, things_nearby):\n things_nearby = sort_by_closeness(origin, things_nearby, 'nav_distance', 'distance')\n return things_nearby\n\n async def search_nearby(\n self, place: str, tags: List[str], limit: int=5, radius: int=4000,\n category: str=\"POIs\"\n ) -\u003E dict:\n headers = self.create_headers()\n if not headers:\n return { \"place_display_name\": place, \"results\": VALVES_NOT_SET }\n\n try:\n nominatim_result = await self.nominatim_search(place, limit=1)\n except ValueError:\n nominatim_result = []\n\n if not nominatim_result or len(nominatim_result) == 0:\n await self.event_search_complete(category, place, 0)\n return { \"place_display_name\": place, \"results\": NO_RESULTS_BAD_ADDRESS }\n\n try:\n nominatim_result = nominatim_result[0]\n\n # display friendlier searching message if possible\n if 'display_name' in nominatim_result:\n place_display_name = \",\".join(nominatim_result['display_name'].split(\",\")[:3])\n elif 'address' in nominatim_result:\n addr = parse_thing_address(nominatim_result)\n if addr is not None:\n place_display_name = \",\".join(addr.split(\",\")[:3])\n else:\n place_display_name = place\n else:\n print(f\"WARN: Could not find display name for place: {place}\")\n place_display_name = place\n\n await self.event_searching(category, place_display_name, done=False)\n\n bbox = {\n 'minlat': nominatim_result['boundingbox'][0],\n 'maxlat': nominatim_result['boundingbox'][1],\n 'minlon': nominatim_result['boundingbox'][2],\n 'maxlon': nominatim_result['boundingbox'][3]\n }\n\n print(f\"[OSM] Searching for {category} near {place_display_name}\")\n things_nearby = await self.get_things_nearby(nominatim_result, place, tags,\n bbox, limit, radius)\n\n if not things_nearby or len(things_nearby) == 0:\n await self.event_search_complete(category, place_display_name, 0)\n return { \"place_display_name\": place, \"results\": NO_RESULTS }\n\n print(f\"[OSM] Found {len(things_nearby)} {category} results near {place_display_name}\")\n\n tag_type_str = \", \".join(tags)\n\n # Only print the full result instructions if we\n # actually have something.\n search_results = convert_and_validate_results(place, things_nearby)\n if search_results:\n result_instructions = self.get_result_instructions(tag_type_str)\n else:\n result_instructions = (\"No results found at all. \"\n \"Tell the user there are no results.\")\n\n resp = (\n f\"{result_instructions}\\n\\n\"\n f\"{search_results}\"\n )\n\n # emit citations for the actual results.\n await self.event_search_complete(category, place_display_name, len(things_nearby))\n for thing in things_nearby:\n await self.emit_result_citation(thing)\n\n return { \"place_display_name\": place_display_name, \"results\": resp, \"things\": things_nearby }\n except ValueError:\n await self.event_search_complete(category, place_display_name, 0)\n return { \"place_display_name\": place_display_name, \"results\": NO_RESULTS, \"things\": [] }\n except Exception as e:\n print(e)\n await self.event_error(e)\n result = (f\"No results were found, because of an error. \"\n f\"Tell the user that there was an error finding results. \"\n f\"The error was: {e}\")\n return { \"place_display_name\": place_display_name, \"results\": result, \"things\": [] }\n\n\nasync def do_osm_search(\n valves, user_valves, place, tags,\n category=\"POIs\", event_emitter=None, limit=5, radius=4000\n):\n # handle breaking 1.0 change, in case of old Nominatim valve settings.\n if valves.nominatim_url.endswith(\"/search\"):\n message = \"Old Nominatim URL setting still in use!\"\n print(f\"[OSM] ERROR: {message}\")\n if valves.status_indicators and event_emitter is not None:\n await event_emitter({\n \"type\": \"status\",\n \"data\": {\n \"status\": \"error\",\n \"description\": f\"Error searching OpenStreetMap: {message}\",\n \"done\": True,\n },\n })\n return OLD_VALVE_SETTING.replace(\"{OLD}\", valves.nominatim_url)\n\n print(f\"[OSM] Searching for [{category}] ({tags[0]}, etc) near place: {place}\")\n searcher = OsmSearcher(valves, user_valves, event_emitter)\n search = await searcher.search_nearby(place, tags, limit=limit, radius=radius, category=category)\n return search[\"results\"]\n\nasync def do_osm_search_full(\n valves, user_valves, place, tags,\n category=\"POIs\", event_emitter=None, limit=5, radius=4000\n):\n \"\"\"Like do_osm_search, but return the full result set instead.\"\"\"\n # handle breaking 1.0 change, in case of old Nominatim valve settings.\n if valves.nominatim_url.endswith(\"/search\"):\n message = \"Old Nominatim URL setting still in use!\"\n print(f\"[OSM] ERROR: {message}\")\n if valves.status_indicators and event_emitter is not None:\n await event_emitter({\n \"type\": \"status\",\n \"data\": {\n \"status\": \"error\",\n \"description\": f\"Error searching OpenStreetMap: {message}\",\n \"done\": True,\n },\n })\n return OLD_VALVE_SETTING.replace(\"{OLD}\", valves.nominatim_url)\n\n print(f\"[OSM] Searching for [{category}] ({tags[0]}, etc) near place: {place}\")\n searcher = OsmSearcher(valves, user_valves, event_emitter)\n return await searcher.search_nearby(place, tags, limit=limit, radius=radius, category=category)\n\nclass OsmNavigator:\n def __init__(\n self, valves, user_valves: Optional[dict], event_emitter=None,\n ):\n self.valves = valves\n self.event_emitter = event_emitter\n self.user_valves = user_valves\n\n async def event_navigating(self, done: bool):\n if not self.event_emitter or not self.valves.status_indicators:\n return\n\n if done:\n message = \"OpenStreetMap: navigation complete\"\n else:\n message = \"OpenStreetMap: navigating...\"\n\n await self.event_emitter({\n \"type\": \"status\",\n \"data\": {\n \"status\": \"in_progress\",\n \"description\": message,\n \"done\": done,\n },\n })\n\n async def event_error(self, exception: Exception):\n if not self.event_emitter or not self.valves.status_indicators:\n return\n\n await self.event_emitter({\n \"type\": \"status\",\n \"data\": {\n \"status\": \"error\",\n \"description\": f\"Error navigating: {str(exception)}\",\n \"done\": True,\n },\n })\n\n async def navigate(self, start_place: str, destination_place: str):\n await self.event_navigating(done=False)\n searcher = OsmSearcher(self.valves, self.user_valves, self.event_emitter)\n router = OrsRouter(self.valves, self.user_valves, self.event_emitter)\n\n try:\n start = await searcher.nominatim_search(start_place, limit=1)\n destination = await searcher.nominatim_search(destination_place, limit=1)\n\n if not start or not destination:\n await self.event_navigating(done=True)\n return NO_RESULTS\n\n start, destination = start[0], destination[0]\n route = router.calculate_route(start, destination)\n\n if not route:\n await self.event_navigating(done=True)\n return NO_RESULTS\n\n total_distance = round(route.get('summary', {}).get('distance', ''), 2)\n travel_time = round(route.get('summary', {}).get('duration', 0) / 60.0, 2)\n travel_type = \"car\" if total_distance \u003E 1.5 else \"walking/biking\"\n\n def create_step_instruction(step):\n instruction = step['instruction']\n duration = round(step.get('duration', 0.0) / 60.0, 2)\n distance = round(step.get('distance', 0.0), 2)\n\n if duration \u003C= 0.0 or distance \u003C= 0.0:\n return f\"- {instruction}\"\n\n if duration \u003C 1.0:\n duration = f\"{round(duration * 60.0, 2)} sec\"\n else:\n duration = f\"{duration} min\"\n\n if distance \u003C 1.0:\n distance = f\"{round(distance * 1000.0, 2)}m\"\n else:\n distance = f\"{distance}km\"\n\n return f\"- {instruction} ({distance}, {duration})\"\n\n instructions = \"\\n\".join([\n create_step_instruction(step)\n for segment in route[\"segments\"]\n for step in segment[\"steps\"]]\n )\n\n markdown = (\n \"## Routing Instructions\\n\"\n f\"These are the routing instructions from \"\n f\"{start_place} to {destination_place}.\\n\\n\"\n f\" - Total Distance: {total_distance} km\\n\"\n f\" - Travel Time: {str(travel_time)} minutes\"\n \"\\n\\n\"\n \"Navigation Instructions:\\n\\n\"\n f\"{instructions}\"\n )\n\n resp = f\"{navigation_instructions(travel_type)}\\n\\n{markdown}\"\n await self.event_navigating(done=True)\n return resp\n except Exception as e:\n print(e)\n await self.event_error(e)\n return (f\"There are no results due to an error. \"\n \"Tell the user that there was an error. \"\n f\"The error was: {e}. \"\n f\"Tell the user the error message.\")\n\n\nclass Tools:\n class Valves(BaseModel):\n user_agent: str = Field(\n default=\"\", description=\"Unique user agent to identify your OSM API requests.\"\n )\n from_header: str = Field(\n default=\"\", description=\"Email address to identify your OSM requests.\"\n )\n nominatim_url: str = Field(\n default=\"https://nominatim.openstreetmap.org/\",\n description=\"URL of OSM Nominatim API for reverse geocoding (address lookup).\"\n )\n overpass_turbo_url: str = Field(\n default=\"https://overpass-api.de/api/interpreter\",\n description=\"URL of Overpass Turbo API for searching OpenStreetMap.\"\n )\n instruction_oriented_interpretation: bool = Field(\n default=True,\n description=(\"Give detailed result interpretation instructions to the model. \"\n \"Switch this off if results are inconsistent, wrong, or missing.\")\n )\n ors_api_key: Optional[str] = Field(\n default=None,\n description=(\"Provide an Open Route Service API key to calculate \"\n \"more accurate distances (leave default to disable).\")\n )\n ors_instance: Optional[str] = Field(\n default=None,\n description=\"Use a custom ORS instance (leave default to use public ORS instance).\"\n )\n status_indicators: bool = Field(\n default=True,\n description=(\"Emit status update events to the web UI.\")\n )\n pass\n\n class UserValves(BaseModel):\n instruction_oriented_interpretation: bool = Field(\n default=True,\n description=(\"Give detailed result interpretation instructions to the model. \"\n \"Switch this off if results are inconsistent, wrong, or missing.\")\n )\n pass\n\n\n def __init__(self):\n self.valves = self.Valves()\n self.user_valves = None\n\n async def find_address_for_coordinates(self, latitude: float, longitude: float, __event_emitter__) -\u003E str:\n \"\"\"\n Resolves GPS coordinates to a specific address or place.\n :param latitude: The latitude portion of the GPS coordinate.\n :param longitude: The longitude portion of the GPS coordinate.\n :return: Information about the address or place at the coordinates.\n \"\"\"\n print(f\"[OSM] Resolving [{latitude}, {longitude}] to address.\")\n return await self.find_specific_place(f\"{latitude}, {longitude}\", __event_emitter__)\n\n async def find_store_or_place_near_coordinates(\n self, store_or_business_name: str, latitude: float, longitude: float, __event_emitter__\n ) -\u003E str:\n \"\"\"\n Finds specifically named stores, businesses, or landmarks near the given\n GPS coordinates. This can be used for general or specific things,\n like 'gas station' or 'grocery store', or the name of a specific\n chain (like 'Wal-Mart' or 'Albert Heijn'). This function is intended\n for finding specific chains or businesses near the given coordinates.\n Use this if the user asks about businesses or places nearby.\n :param store_or_business_name: Name of store or business to look for.\n :param latitude: The latitude portion of the GPS coordinate.\n :param longitude: The longitude portion of the GPS coordinate.\n :return: Information about the address or places near the coordinates.\n \"\"\"\n print(f\"Searching for '{store_or_business_name}' near {latitude},{longitude}\")\n query = f\"{store_or_business_name} {latitude},{longitude}\"\n return await self.find_specific_place(query, __event_emitter__)\n\n async def find_specific_place(self, address_or_place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Looks up details on OpenStreetMap of a specific address, landmark,\n place, named building, or location. Used for when the user asks where\n a specific unique entity (like a specific museum, or church, or shopping\n center) is.\n :param address_or_place: The address or place to look up.\n :return: Information about the place, if found.\n \"\"\"\n print(f\"[OSM] Searching for info on [{address_or_place}].\")\n searcher = OsmSearcher(self.valves, self.user_valves, __event_emitter__)\n try:\n result = await searcher.nominatim_search(address_or_place, limit=5)\n if result:\n results_in_md = convert_and_validate_results(\n address_or_place, result,\n sort_message=\"importance\", use_distance=False\n )\n\n resp = f\"{specific_place_instructions()}\\n\\n{results_in_md}\"\n return resp\n else:\n return NO_RESULTS\n except Exception as e:\n print(e)\n return (f\"There are no results due to an error. \"\n \"Tell the user that there was an error. \"\n f\"The error was: {e}. \"\n f\"Tell the user the error message.\")\n\n\n async def navigate_between_places(\n self,\n start_address_or_place: str,\n destination_address_or_place: str,\n __event_emitter__\n ) -\u003E str:\n \"\"\"\n Retrieve a navigation route and associated information between two places.\n :param start_address_or_place: The address, place, or coordinates to start from.\n :param destination_address_or_place: The destination address, place, or coordinates to go.\n :return: The navigation route and associated info, if found.\n \"\"\"\n print(f\"[OSM] Navigating from [{start_address_or_place}] to [{destination_address_or_place}].\")\n navigator = OsmNavigator(self.valves, self.user_valves, __event_emitter__)\n return await navigator.navigate(start_address_or_place, destination_address_or_place)\n\n async def find_grocery_stores_near_place(\n self, place: str, __user__: dict, __event_emitter__\n ) -\u003E str:\n \"\"\"\n Finds supermarkets, grocery stores, and other food stores on\n OpenStreetMap near a given place or address. The location of the\n address or place is reverse geo-coded, then nearby results\n are fetched from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby grocery stores or supermarkets, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"shop=supermarket\", \"shop=grocery\", \"shop=convenience\", \"shop=greengrocer\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"groceries\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n\n async def find_bakeries_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds bakeries on OpenStreetMap near a given place or\n address. The location of the address or place is reverse\n geo-coded, then nearby results are fetched from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby bakeries, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"shop=bakery\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"bakeries\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_food_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds restaurants, fast food, bars, breweries, pubs, etc on\n OpenStreetMap near a given place or address. The location of the\n address or place is reverse geo-coded, then nearby results\n are fetched from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby restaurants, eateries, etc, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\n \"amenity=restaurant\",\n \"amenity=fast_food\",\n \"amenity=cafe\",\n \"amenity=pub\",\n \"amenity=bar\",\n \"amenity=eatery\",\n \"amenity=biergarten\",\n \"amenity=canteen\"\n ]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"restaurants and food\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n\n async def find_swimming_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds swimming pools, water parks, swimming areas, and other aquatic\n activities on OpenStreetMap near a given place or address. The location\n of the address or place is reverse geo-coded, then nearby results are fetched\n from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of swimming poools or places, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"leisure=swimming_pool\", \"leisure=swimming_area\",\n \"leisure=water_park\", \"tourism=theme_park\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"swimming\",\n radius=10000, place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_playgrounds_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds playgrounds and parks on OpenStreetMap near a given place, address, or coordinates.\n The location of the address or place is reverse geo-coded, then nearby results are fetched\n from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of recreational places, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"leisure=playground\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"playgrounds\",\n limit=10, place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_recreation_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds playgrounds, theme parks, parks, frisbee golf, ice skating, and other recreational\n activities on OpenStreetMap near a given place or address. The location\n of the address or place is reverse geo-coded, then nearby results are fetched\n from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of recreational places, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"leisure=horse_riding\", \"leisure=ice_rink\", \"leisure=disc_golf_course\",\n \"leisure=park\", \"leisure=amusement_arcade\", \"tourism=theme_park\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"recreational activities\",\n limit=10, radius=10000, place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_tourist_attractions_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds museums, landmarks, and other tourist attractions on OpenStreetMap near\n a given place or address. The location of the address or place is reverse geo-coded,\n then nearby results are fetched from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of tourist attractions, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"tourism=museum\", \"tourism=aquarium\", \"tourism=zoo\",\n \"tourism=attraction\", \"tourism=gallery\", \"tourism=artwork\"]\n\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"tourist attractions\",\n limit=10, radius=10000, place=place, tags=tags, event_emitter=__event_emitter__)\n\n\n\n async def find_place_of_worship_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds places of worship (churches, mosques, temples, etc) on\n OpenStreetMap near a given place or address. The location of the\n address or place is reverse geo-coded, then nearby results\n are fetched from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby places of worship, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"amenity=place_of_worship\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"places of worship\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n\n async def find_accommodation_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds accommodation (hotels, guesthouses, hostels, etc) on\n OpenStreetMap near a given place or address. The location of the\n address or place is reverse geo-coded, then nearby results\n are fetched from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby accommodation, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\n \"tourism=hotel\", \"tourism=chalet\", \"tourism=guest_house\", \"tourism=guesthouse\",\n \"tourism=motel\", \"tourism=hostel\"\n ]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"accommodation\",\n radius=10000, place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_alcohol_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds beer stores, liquor stores, and similar shops on OpenStreetMap\n near a given place or address. The location of the address or place is\n reverse geo-coded, then nearby results\n are fetched from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby alcohol shops, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"shop=alcohol\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"alcohol stores\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_drugs_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds cannabis dispensaries, coffeeshops, smartshops, and similar stores on OpenStreetMap\n near a given place or address. The location of the address or place is\n reverse geo-coded, then nearby results\n are fetched from OpenStreetMap.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby cannabis and smart shops, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"shop=coffeeshop\", \"shop=cannabis\", \"shop=headshop\", \"shop=smartshop\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"cannabis and smartshops\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_schools_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds schools (NOT universities) on OpenStreetMap near a given place or address.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby schools, if found.\n \"\"\"\n tags = [\"amenity=school\"]\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"schools\",\n limit=10, place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_universities_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds universities and colleges on OpenStreetMap near a given place or address.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby schools, if found.\n \"\"\"\n tags = [\"amenity=university\", \"amenity=college\"]\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"universities\",\n limit=10, place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_libraries_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds libraries on OpenStreetMap near a given place or address.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby libraries, if found.\n \"\"\"\n tags = [\"amenity=library\"]\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"libraries\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_public_transport_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds public transportation stops on OpenStreetMap near a given place or address.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby public transportation stops, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"highway=bus_stop\", \"amenity=bus_station\",\n \"railway=station\", \"railway=halt\", \"railway=tram_stop\",\n \"station=subway\", \"amenity=ferry_terminal\",\n \"public_transport=station\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"public transport\",\n limit=10, place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_bike_rentals_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds bike rentals on OpenStreetMap near a given place or address.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby bike rentals, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"amenity=bicycle_rental\", \"amenity=bicycle_library\", \"service:bicycle:rental=yes\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"bike rentals\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_car_rentals_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds bike rentals on OpenStreetMap near a given place or address.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby bike rentals, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"amenity=car_rental\", \"car:rental=yes\", \"rental=car\", \"car_rental=yes\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"car rentals\",\n radius=6000, place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_hardware_store_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds hardware stores, home improvement stores, and DIY stores\n near given a place or address.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby hardware/DIY stores, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"shop=doityourself\", \"shop=hardware\", \"shop=power_tools\",\n \"shop=groundskeeping\", \"shop=trade\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"hardware stores\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_electrical_store_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds electrical stores and lighting stores near a given place\n or address. These are stores that sell lighting and electrical\n equipment like wires, sockets, and so forth.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby electrical/lighting stores, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"shop=lighting\", \"shop=electrical\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"electrical stores\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_electronics_store_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds consumer electronics stores near a given place or address.\n These stores sell computers, cell phones, video games, and so on.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby electronics stores, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"shop=electronics\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves,\n category=\"consumer electronics stores\", place=place,\n tags=tags, event_emitter=__event_emitter__)\n\n async def find_doctor_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds doctors near a given place or address.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby electronics stores, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"amenity=clinic\", \"amenity=doctors\", \"healthcare=doctor\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"doctors\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_hospital_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds doctors near a given place or address.\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby electronics stores, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"healthcare=hospital\", \"amenity=hospitals\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"hospitals\",\n place=place, tags=tags, event_emitter=__event_emitter__)\n\n async def find_pharmacy_near_place(self, __user__: dict, place: str, __event_emitter__) -\u003E str:\n \"\"\"\n Finds pharmacies and health shops near a given place or address\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :return: A list of nearby electronics stores, if found.\n \"\"\"\n user_valves = __user__[\"valves\"] if \"valves\" in __user__ else None\n tags = [\"amenity=pharmacy\", \"shop=chemist\", \"shop=supplements\",\n \"shop=health_food\"]\n return await do_osm_search(valves=self.valves, user_valves=user_valves, category=\"pharmacies\",\n radius=6000, place=place, tags=tags, event_emitter=__event_emitter__)\n\n # This function exists to help catch situations where the user is\n # too generic in their query, or is looking for something the tool\n # does not yet support. By having the model pick this function, we\n # can direct it to report its capabilities and tell the user how\n # to use it. It's not perfect, but it works sometimes.\n def find_other_things_near_place(\n self,\n __user__: dict,\n place: str,\n category: str\n ) -\u003E str:\n \"\"\"\n Find shops and other places not covered by a specific\n category available in the other functions. Use this if the\n user is asking for a type of store or place that other\n functions do not support.\n\n :param place: The name of a place, an address, or GPS coordinates. City and country must be specified, if known.\n :param category: The category of place, shop, etc to look up.\n :return: A list of nearby shops.\n \"\"\"\n print(f\"[OSM] Generic catch handler called with {category}\")\n resp = (\n \"# No Results Found\\n\"\n f\"No results were found. There was an error. Finding {category} points of interest \"\n \"is not yet supported. Tell the user support will come eventually! \"\n \"Tell the user that you are only capable of finding specific \"\n \"categories of stores, amenities, and points of interest:\\n\"\n \" - Car rentals and bike rentals\\n\"\n \" - Public transport, libraries\\n\"\n \" - Education institutions (schools and universities)\\n\"\n \" - Grocery stores, supermarkets, convenience stores\\n\"\n \" - Food and restaurants\\n\"\n \" - Accommodation\\n\"\n \" - Places of Worship\\n\"\n \" - Hardware stores and home improvement centers\\n\"\n \" - Electrical and lighting stores\\n\"\n \" - Consumer electronics stores\\n\"\n \" - Healthcare (doctors, hospitals, pharmacies, and health stores)\\n\"\n \" - Various recreational and leisure activities\\n\\n\"\n \"Only mention things from the above list that you think the user \"\n \"will be interested in, given the conversation so far. Don't mention \"\n \"things not on the list. \"\n \"**IMPORTANT**: Tell the user to be specific in their \"\n \"query in their next message, so you can call the right function!\")\n return resp\n" }, "downloads": 2267, "upvotes": 0, "downvotes": 0, "updatedAt": 1734290450, "createdAt": 1722803430, "user": { "id": "c8801c6a-9835-41d0-8a71-1f9b1f475770", "username": "projectmoon", "name": "", "profileImageUrl": "https://www.gravatar.com/avatar/ca4bd75adbc67f7bea8db13bcdcdb855156594a88c376c472d03e8f4c3bb7d57?d=mp", "createdAt": 1720867374 } } ]