open-webui-filters/osm.py

"""
title: OpenStreetMap Tool
author: projectmoon
author_url: https://git.agnos.is/projectmoon/open-webui-filters
version: 2.2.0
license: AGPL-3.0+
required_open_webui_version: 0.4.3
requirements: openrouteservice, pygments
"""
import itertools
import json
import math
import requests

from pygments import highlight
from pygments.lexers import JsonLexer
from pygments.formatters import HtmlFormatter

import openrouteservice
from openrouteservice.directions import directions as ors_directions

from urllib.parse import urljoin
from operator import itemgetter
from typing import List, Optional
from pydantic import BaseModel, Field

# Yoinked from the OpenWebUI CSS
FONTS = ",".join([
    "-apple-system", "BlinkMacSystemFont", "Inter",
    "ui-sans-serif", "system-ui", "Segoe UI",
    "Roboto", "Ubuntu", "Cantarell", "Noto Sans",
    "sans-serif", "Helvetica Neue", "Arial",
    "\"Apple Color Emoji\"", "\"Segoe UI Emoji\"",
    "Segoe UI Symbol", "\"Noto Color Emoji\""
])

FONT_CSS = f"""
html {{ font-family: {FONTS}; }}

@media (prefers-color-scheme: dark) {{
  html {{
    --tw-text-opacity: 1;
    color: rgb(227 227 227 / var(--tw-text-opacity));
  }}
}}
"""

HIGHLIGHT_CSS = HtmlFormatter().get_style_defs('.highlight')

NOMINATIM_LOOKUP_TYPES = {
    "node": "N",
    "route": "R",
    "way": "W"
}

OLD_VALVE_SETTING = """ Tell the user that you cannot search
OpenStreetMap until the configuration is fixed. The Nominatim URL
valve setting needs to be updated. There has been a breaking change in
1.0 of the OpenStreetMap tool. The valve setting is currently set to:
`{OLD}`.

It shoule be set to the root URL of the Nominatim endpoint, for
example:

`https://nominatim.openstreetmap.org/`

Inform the user they need to fix this configuration setting.
""".replace("\n", " ").strip()

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_RESULTS_BAD_ADDRESS = ("No results found. Tell the user you found no results because "
                          "OpenStreetMap could not resolve the address. "
                          "Print the exact address or location you searched for. "
                          "Suggest to the user that they refine their "
                          "question, for example removing the apartment number, sub-unit, "
                          "etc. Example: If `123 Main Street, Apt 4` returned no results, "
                          "suggest that the user searc for `123 Main Street` instead. "
                          "Use the address the user searched for in your example.")

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.")

# Give examples of OSM links to help prevent wonky generated links
# with correct GPS coords but incorrect URLs.
EXAMPLE_OSM_LINK = "https://www.openstreetmap.org/#map=19/<lat>/<lon>"
OSM_LINK_INSTRUCTIONS = (
    "Make friendly human-readable OpenStreetMap links when possible, "
    "by using the latitude and longitude of the amenities: "
    f"{EXAMPLE_OSM_LINK}\n\n"
)

def chunk_list(input_list, chunk_size):
    it = iter(input_list)
    return list(
        itertools.zip_longest(*[iter(it)] * chunk_size, fillvalue=None)
    )

def to_lookup(thing) -> Optional[str]:
    lookup_type = NOMINATIM_LOOKUP_TYPES.get(thing['type'])
    if lookup_type is not None:
        return f"{lookup_type}{thing['id']}"

def specific_place_instructions() -> str:
    return (
        "# Result Instructions\n"
        "These are search results ordered by relevance for the "
        "address, place, landmark, or location the user is asking "
        "about. **IMPORTANT!:** Tell the user all relevant information, "
        "including address, contact information, and the OpenStreetMap link. "
        "Make the map link into a nice human-readable markdown link."
    )

def navigation_instructions(travel_type) -> str:
    return (
        "# Result Instructions\n"
        "This is the navigation route that the user has requested. "
        f"These instructions are for travel by {travel_type}. "
        "Tell the user the total distance, "
        "and estimated travel time. "
        "If the user **specifically asked for it**, also tell "
        "them the route itself. When telling the route, you must tell "
        f"the user that it's a **{travel_type}** route."
    )

def detailed_instructions(tag_type_str: str) -> str:
    """
    Produce detailed instructions for models good at following
    detailed instructions.
    """
    return (
        "# Detailed Search Result Instructions\n"
        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 closer results "
        "first. Closer results are higher in the list. When telling the "
        "user the distance, use the TRAVEL DISTANCE. Do not say one "
        "distance is farther away than another. Just say what the "
        "distances are. "
        f"{OSM_LINK_INSTRUCTIONS}"
        "Give map links friendly, contextual labels. Don't just print "
        f"the naked link:\n"
        f' - Example: You can view it on [OpenStreetMap]({EXAMPLE_OSM_LINK})\n'
        f' - Example: Here it is on [OpenStreetMap]({EXAMPLE_OSM_LINK})\n'
        f' - Example: You can find it on [OpenStreetMap]({EXAMPLE_OSM_LINK})\n'
        "\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"
        "Remember that the CLOSEST result is first, and you should use "
        "that result first.\n\n"
        "The results (if present) are below, in Markdown format.\n\n"
        "**ALWAYS SAY THE CLOSEST RESULT FIRST!**"
    )

def simple_instructions(tag_type_str: str) -> str:
    """
    Produce simpler markdown-oriented instructions for models that do
    better with that.
    """
    return (
        "# OpenStreetMap Result Instructions\n"
        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. "
        "For each result, report the following information: \n"
        " - Name\n"
        " - Address\n"
        " - OpenStreetMap Link (make it a human readable link like 'View on OpenStreetMap')\n"
        " - Contact information (address, phone, website, email, etc)\n\n"
        "Tell the user about ALL the results, and give the CLOSEST result "
        "first. The results are ordered by closeness as the crow flies. "
        "When telling the user about distances, use the TRAVEL DISTANCE only. "
        "Only use relevant results. If there are no relevant results, "
        "say so. Do not make up answers or hallucinate. "
        "Make sure that your results are in the actual location the user is talking about, "
        "and not a place of the same name in a different country."
        "The search results are below."
    )

def merge_from_nominatim(thing, nominatim_result) -> Optional[dict]:
    """Merge information into object missing all or some of it."""
    if thing is None:
        return None

    if 'address' not in nominatim_result:
        return None

    nominatim_address = nominatim_result['address']

    # prioritize actual name, road name, then display name. display
    # name is often the full address, which is a bit much.
    nominatim_name = nominatim_result.get('name')
    nominatim_road = nominatim_address.get('road')
    nominatim_display_name = nominatim_result.get('display_name')
    thing_name = thing.get('name')

    if nominatim_name and not thing_name:
        thing['name'] = nominatim_name.strip()
    elif nominatim_road and not thing_name:
        thing['name'] = nominatim_road.strip()
    elif nominatim_display_name and not thing_name:
        thing['name'] = nominatim_display_name.strip()

    tags = thing.get('tags', {})

    for key in nominatim_address:
        obj_key = f"addr:{key}"
        if obj_key not in tags:
            tags[obj_key] = nominatim_address[key]

    thing['tags'] = tags
    return thing

def pretty_print_thing_json(thing):
    """Converts an OSM thing to nice JSON HTML."""
    formatted_json_str = json.dumps(thing, indent=2)
    lexer = JsonLexer()
    formatter = HtmlFormatter(style='colorful')
    return highlight(formatted_json_str, lexer, formatter)


def thing_is_useful(thing):
    """
    Determine if an OSM way entry is useful to us. This means it
    has something more than just its main classification tag, and
    (usually) has at least a name. Some exceptions are made for ways
    that do not have names.
    """
    tags = thing.get('tags', {})
    has_tags = len(tags) > 1
    has_useful_tags = (
        'leisure' in tags or
        'shop' in tags or
        'amenity' in tags or
        'car:rental' in tags or
        'rental' in tags or
        'car_rental' in tags or
        'service:bicycle:rental' in tags or
        'tourism' in tags
    )

    # there can be a lot of artwork in city centers. drop ones that
    # aren't as notable. we define notable by the thing having wiki
    # entries, or by being tagged as historical.
    if tags.get('tourism', '') == 'artwork':
        notable = (
            'wikipedia' in tags or
            'wikimedia_commons' in tags
        )
    else:
        notable = True

    return has_tags and has_useful_tags and notable

def thing_has_info(thing):
    has_name = any('name' in tag for tag in thing['tags'])
    return thing_is_useful(thing) and has_name
    # is_exception = way['tags'].get('leisure', None) is not None
    # return has_tags and (has_name or is_exception)

def process_way_result(way) -> Optional[dict]:
    """
    Post-process an OSM Way dict to remove the geometry and node
    info, and calculate a single GPS coordinate from its bounding
    box.
    """
    if 'nodes' in way:
        del way['nodes']

    if 'geometry' in way:
        del way['geometry']

    if 'bounds' in way:
        way_center = get_bounding_box_center(way['bounds'])
        way['lat'] = way_center['lat']
        way['lon'] = way_center['lon']
        del way['bounds']
        return way

    return None

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, 'minlat')
    min_lon = convert(bbox, 'minlon')
    max_lat = convert(bbox, 'maxlat')
    max_lon = convert(bbox, 'maxlon')

    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, *keys: str):
    """
    Sorts a list of { lat, lon }-like dicts by closeness to an origin point.
    The origin is a dict with keys of { lat, lon }.
    """
    return sorted(points, key=itemgetter(*keys))

def sort_by_rank(things):
    """
    Calculate a rank for a list of things. More important ones are
    pushed towards the top. Currently only for tourism tags.
    """
    def rank_thing(thing: dict) -> int:
        tags = thing.get('tags', {})
        if not 'tourism' in tags:
            return 0

        rank = len([name for name in tags.keys()
                         if name.startswith("name")])
        rank += 5 if 'historic' in tags else 0
        rank += 5 if 'wikipedia' in tags else 0
        rank += 1 if 'wikimedia_commons' in tags else 0
        rank += 5 if tags.get('tourism', '') == 'museum' else 0
        rank += 5 if tags.get('tourism', '') == 'aquarium' else 0
        rank += 5 if tags.get('tourism', '') == 'zoo' else 0
        return rank

    return sorted(things, reverse=True, key=lambda t: (rank_thing(t), -t['distance']))

def get_or_none(tags: dict, *keys: str) -> Optional[str]:
    """
    Try to extract a value from a dict by trying keys in order, or
    return None if none of the keys were found.
    """
    for key in keys:
        if key in tags:
            return tags[key]

    return None

def all_are_none(*args) -> bool:
    for arg in args:
        if arg is not None:
            return False

    return True

def friendly_shop_name(shop_type: str) -> str:
    """
    Make certain shop types more friendly for LLM interpretation.
    """
    if shop_type == "doityourself":
        return "hardware"
    else:
        return shop_type

def parse_thing_address(thing: dict) -> Optional[str]:
    """
    Parse address from either an Overpass result or Nominatim
    result.
    """
    if 'address' in thing:
        # nominatim result
        return parse_address_from_address_obj(thing['address'])
    else:
        return parse_address_from_tags(thing['tags'])

def parse_address_from_address_obj(address) -> Optional[str]:
    """Parse address from Nominatim address object."""
    house_number = get_or_none(address, "house_number")
    street = get_or_none(address, "road")
    city = get_or_none(address, "city")
    state = get_or_none(address, "state")
    postal_code = get_or_none(address, "postcode")

    # if all are none, that means we don't know the address at all.
    if all_are_none(house_number, street, city, state, postal_code):
        return None

    # Handle missing values to create complete-ish addresses, even if
    # we have missing data. We will get either a partly complete
    # address, or None if all the values are missing.
    line1 = filter(None, [street, house_number])
    line2 = filter(None, [city, state, postal_code])
    line1 = " ".join(line1).strip()
    line2 = " ".join(line2).strip()
    full_address = filter(None, [line1, line2])
    full_address = ", ".join(full_address).strip()
    return full_address if len(full_address) > 0 else None

def parse_address_from_tags(tags: dict) -> Optional[str]:
    """Parse address from Overpass tags object."""
    house_number = get_or_none(tags, "addr:housenumber", "addr:house_number")
    street = get_or_none(tags, "addr:street")
    city = get_or_none(tags, "addr:city")
    state = get_or_none(tags, "addr:state", "addr:province")
    postal_code = get_or_none(
        tags,
        "addr:postcode", "addr:post_code", "addr:postal_code",
        "addr:zipcode", "addr:zip_code"
    )

    # if all are none, that means we don't know the address at all.
    if all_are_none(house_number, street, city, state, postal_code):
        return None

    # Handle missing values to create complete-ish addresses, even if
    # we have missing data. We will get either a partly complete
    # address, or None if all the values are missing.
    line1 = filter(None, [street, house_number])
    line2 = filter(None, [city, state, postal_code])
    line1 = " ".join(line1).strip()
    line2 = " ".join(line2).strip()
    full_address = filter(None, [line1, line2])
    full_address = ", ".join(full_address).strip()
    return full_address if len(full_address) > 0 else None

def parse_thing_amenity_type(thing: dict, tags: dict) -> Optional[dict]:
    """
    Extract amenity type or other identifying category from
    Nominatim or Overpass result object.
    """
    if 'amenity' in tags:
        return tags['amenity']

    if thing.get('class') == 'amenity' or thing.get('class') == 'shop':
        return thing.get('type')

    # fall back to tag categories, like shop=*
    if 'shop' in tags:
        return friendly_shop_name(tags['shop'])
    if 'leisure' in tags:
        return friendly_shop_name(tags['leisure'])

    return None

def parse_and_validate_thing(thing: dict) -> Optional[dict]:
    """
    Parse an OSM result (node or post-processed way) and make it
    more friendly to work with. Helps remove ambiguity of the LLM
    interpreting the raw JSON data. If there is not enough data,
    discard the result.
    """
    tags: dict = thing['tags'] if 'tags' in thing else {}

    # Currently we define "enough data" as at least having lat, lon,
    # and a name. nameless things are allowed if they are in a certain
    # class of POIs (leisure).
    has_name = 'name' in tags or 'name' in thing
    is_leisure = 'leisure' in tags or 'leisure' in thing
    if 'lat' not in thing or 'lon' not in thing:
        return None

    if not has_name and not is_leisure:
        return None

    friendly_thing = {}
    name: str = (tags['name'] if 'name' in tags
                 else thing['name'] if 'name' in thing
                 else str(thing['id']) if 'id' in thing
                 else str(thing['osm_id']) if 'osm_id' in thing
                 else "unknown")

    address: str = parse_thing_address(thing)
    distance: Optional[float] = thing.get('distance', None)
    nav_distance: Optional[float] = thing.get('nav_distance', None)
    opening_hours: Optional[str] = tags.get('opening_hours', None)

    lat: Optional[float] = thing.get('lat', None)
    lon: Optional[float] = thing.get('lon', None)
    amenity_type: Optional[str] = parse_thing_amenity_type(thing, tags)

    # use the navigation distance if it's present. but if not, set to
    # the haversine distance so that we at least get coherent results
    # for LLM.
    friendly_thing['distance'] = "{:.3f}".format(distance) if distance else "unknown"
    if nav_distance:
        friendly_thing['nav_distance'] = "{:.3f}".format(nav_distance) + " km"
    else:
        friendly_thing['nav_distance'] = f"a bit more than {friendly_thing['distance']}km"

    friendly_thing['name'] = name if name else "unknown"
    friendly_thing['address'] = address if address else "unknown"
    friendly_thing['lat'] = lat if lat else "unknown"
    friendly_thing['lon'] = lon if lon else "unknown"
    friendly_thing['amenity_type'] = amenity_type if amenity_type else "unknown"
    friendly_thing['opening_hours'] = opening_hours if opening_hours else "not recorded"
    return friendly_thing

def create_osm_link(lat, lon):
    return EXAMPLE_OSM_LINK.replace("<lat>", str(lat)).replace("<lon>", str(lon))

def convert_and_validate_results(
    original_location: str,
    things_nearby: List[dict],
    sort_message: str="closeness",
    use_distance: bool=True
) -> Optional[str]:
    """
    Converts the things_nearby JSON into Markdown-ish results to
    (hopefully) improve model understanding of the results. Intended
    to stop misinterpretation of GPS coordinates when creating map
    links. Also drops incomplete results. Supports Overpass and
    Nominatim results.
    """
    entries = []
    for thing in things_nearby:
        # Convert to friendlier data, drop results without names etc.
        # No need to instruct LLM to generate map links if we do it
        # instead.
        friendly_thing = parse_and_validate_thing(thing)
        if not friendly_thing:
            continue

        distance = (f" - Haversine Distance from Origin: {friendly_thing['distance']} km\n"
                    if use_distance else "")
        travel_distance = (f" - Travel Distance from Origin: {friendly_thing['nav_distance']}\n"
                        if use_distance and 'nav_distance' in friendly_thing else "")
        map_link = create_osm_link(friendly_thing['lat'], friendly_thing['lon'])
        entry = (f"## {friendly_thing['name']}\n"
                 f" - Latitude: {friendly_thing['lat']}\n"
                 f" - Longitude: {friendly_thing['lon']}\n"
                 f" - Address: {friendly_thing['address']}\n"
                 f" - Amenity Type: {friendly_thing['amenity_type']}\n"
                 f"{distance}"
                 f"{travel_distance}"
                 f" - OpenStreetMap link: {map_link}\n\n"
                 f"Raw JSON data:\n"
                 "```json\n"
                 f"{str(thing)}\n"
                 "```")

        entries.append(entry)

    if len(entries) == 0:
        return None

    result_text = "\n\n".join(entries)
    header = ("# Search Results\n"
              f"Ordered by {sort_message} to {original_location}.")

    return f"{header}\n\n{result_text}"

class OsmCache:
    def __init__(self, filename="/tmp/osm.json"):
        self.filename = filename
        self.data = {}

        # Load existing cache if it exists
        try:
            with open(self.filename, 'r') as f:
                self.data = json.load(f)
        except FileNotFoundError:
            pass

    def get(self, key):
        return self.data.get(key)

    def set(self, key, value):
        self.data[key] = value
        with open(self.filename, 'w') as f:
            json.dump(self.data, f)

    def get_or_set(self, key, func_to_call):
        """
        Retrieve the value from the cache for a given key. If the key is not found,
        call `func_to_call` to generate the value and store it in the cache.

        :param key: The key to look up or set in the cache
        :param func_to_call: A callable function that returns the value if key is missing
        :return: The cached or generated value
        """
        if key not in self.data:
            value = func_to_call()
            self.set(key, value)
        return self.data[key]

    def clear_cache(self):
        """
        Clear all entries from the cache.
        """
        self.data.clear()
        try:
            # Erase contents of the cache file.
            with open(self.filename, 'w'):
                pass
        except FileNotFoundError:
            pass

class OrsRouter:
    def __init__(
            self, valves, user_valves: Optional[dict], event_emitter=None,
    ):
        self.cache = OsmCache()
        self.valves = valves
        self.event_emitter = event_emitter
        self.user_valves = user_valves

        if self.valves.ors_api_key is not None and self.valves.ors_api_key != "":
            if self.valves.ors_instance is not None:
                self._client = openrouteservice.Client(
                    base_url=self.valves.ors_instance,
                    key=self.valves.ors_api_key
                )
            else:
                self._client = openrouteservice.Client(key=self.valves.ors_api_key)
        else:
            self._client = None


    def calculate_route(
            self, from_thing: dict, to_thing: dict
    ) -> Optional[dict]:
        """
        Calculate route between A and B. Returns the route,
        if successful, or None if the distance could not be
        calculated, or if ORS is not configured.
        """
        if not self._client:
            return None

        # select profile based on distance for more accurate
        # measurements. very close haversine distances use the walking
        # profile, which should (usually?) essentially cover walking
        # and biking. further away = use car.
        if to_thing.get('distance', 9000) <= 1.5:
            profile = "foot-walking"
        else:
            profile = "driving-car"

        coords = ((from_thing['lon'], from_thing['lat']),
                  (to_thing['lon'], to_thing['lat']))

        # check cache first.
        cache_key = f"ors_route_{str(coords)}"
        cached_route = self.cache.get(cache_key)
        if cached_route:
            print("[OSM] Got route from cache!")
            return cached_route

        resp = ors_directions(self._client, coords, profile=profile,
                              preference="fastest", units="km")

        routes = resp.get('routes', [])
        if len(routes) > 0:
            self.cache.set(cache_key, routes[0])
            return routes[0]
        else:
            return None

    def calculate_distance(
            self, from_thing: dict, to_thing: dict
    ) -> Optional[float]:
        """
        Calculate navigation distance between A and B. Returns the
        distance calculated, if successful, or None if the distance
        could not be calculated, or if ORS is not configured.
        """
        if not self._client:
            return None

        route = self.calculate_route(from_thing, to_thing)
        return route.get('summary', {}).get('distance', None) if route else None

class OsmSearcher:
    def __init__(self, valves, user_valves: Optional[dict], event_emitter=None):
        self.valves = valves
        self.event_emitter = event_emitter
        self.user_valves = user_valves
        self._ors = OrsRouter(valves, user_valves, event_emitter)

    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
        }

    async def event_resolving(self, done: bool=False):
        if not self.event_emitter or not self.valves.status_indicators:
            return

        if done:
            message = "OpenStreetMap: resolution complete."
        else:
            message = "OpenStreetMap: resolving..."

        await self.event_emitter({
            "type": "status",
            "data": {
                "status": "in_progress",
                "description": message,
                "done": done,
            },
        })

    async def event_fetching(self, done: bool=False, message="OpenStreetMap: fetching additional info"):
        if not self.event_emitter or not self.valves.status_indicators:
            return

        await self.event_emitter({
            "type": "status",
            "data": {
                "status": "in_progress",
                "description": message,
                "done": done,
            },
        })

    async def event_searching(
            self, category: str, place: str,
            status: str="in_progress", done: bool=False
    ):
        if not self.event_emitter or not self.valves.status_indicators:
            return

        await self.event_emitter({
            "type": "status",
            "data": {
                "status": status,
                "description": f"OpenStreetMap: searching for {category} near {place}",
                "done": done,
            },
        })

    async def event_search_complete(self, category: str, place: str, num_results: int):
        if not self.event_emitter or not self.valves.status_indicators:
            return

        await self.event_emitter({
            "type": "status",
            "data": {
                "status": "complete",
                "description": f"OpenStreetMap: found {num_results} '{category}' results",
                "done": True,
            },
        })

    def create_result_document(self, thing) -> Optional[dict]:
        original_thing = thing
        thing = parse_and_validate_thing(thing)

        if not thing:
            return None

        if 'address' in original_thing:
            street = get_or_none(original_thing['address'], "road")
        else:
            street = get_or_none(original_thing['tags'], "addr:street")

        street_name = street if street is not None else ""
        source_name = f"{thing['name']} {street_name}"
        lat, lon = thing['lat'], thing['lon']
        osm_link = create_osm_link(lat, lon)
        json_data = pretty_print_thing_json(original_thing)
        addr = f"at {thing['address']}" if thing['address'] != 'unknown' else 'nearby'

        document = (f"<style>{HIGHLIGHT_CSS}</style>"
                    f"<style>{FONT_CSS}</style>"
                    f"<div>"
                    f"<p>{thing['name']} is located {addr}.</p>"
                    f"<ul>"
                    f"<li>"
                    f"  <strong>Opening Hours:</strong> {thing['opening_hours']}"
                    f"</li>"
                    f"</ul>"
                    f"<p>Raw JSON data:</p>"
                    f"{json_data}"
                    f"</div>")

        return { "source_name": source_name, "document": document, "osm_link": osm_link }

    async def emit_result_citation(self, thing):
        if not self.event_emitter or not self.valves.status_indicators:
            return

        converted = self.create_result_document(thing)
        if not converted:
            return

        source_name = converted["source_name"]
        document = converted["document"]
        osm_link = converted["osm_link"]

        await self.event_emitter({
            "type": "source",
            "data": {
                "document": [document],
                "metadata": [{"source": source_name, "html": True }],
                "source": {"name": source_name, "url": osm_link},
            }
        })

    async def event_error(self, exception: Exception):
        if not self.event_emitter or not self.valves.status_indicators:
            return

        await self.event_emitter({
            "type": "status",
            "data": {
                "status": "error",
                "description": f"Error searching OpenStreetMap: {str(exception)}",
                "done": True,
            },
        })

    def calculate_navigation_distance(self, start, destination) -> float:
        """Calculate real distance from A to B, instead of Haversine."""
        return self._ors.calculate_distance(start, destination)

    def attempt_ors(self, origin, things_nearby) -> bool:
        """Update distances to use ORS navigable distances, if ORS enabled."""
        used_ors = False
        cache = OsmCache()
        for thing in things_nearby:
            cache_key = f"ors_{origin}_{thing['id']}"
            nav_distance = cache.get(cache_key)

            if nav_distance:
                print(f"[OSM] Got nav distance for {thing['id']} from cache!")
            else:
                print(f"[OSM] Checking ORS for {thing['id']}")
                try:
                    nav_distance = self.calculate_navigation_distance(origin, thing)
                except Exception as e:
                    print(f"[OSM] Error querying ORS: {e}")
                    print(f"[OSM] Falling back to regular distance due to ORS error!")
                    nav_distance = thing['distance']

            if nav_distance:
                used_ors = True
                cache.set(cache_key, nav_distance)
                thing['nav_distance'] = nav_distance