add python implementation of reverse

This adds an additional layer parameter and slightly changes the
queries to do more efficient lookups for large area features.

nominatim/api/__init__.py

@@ -21,12 +21,15 @@ from .types import (PlaceID as PlaceID,
                     OsmID as OsmID,
                     PlaceRef as PlaceRef,
                     Point as Point,
+                    Bbox as Bbox,
                     GeometryFormat as GeometryFormat,
-                    LookupDetails as LookupDetails)
+                    LookupDetails as LookupDetails,
+                    DataLayer as DataLayer)
 from .results import (SourceTable as SourceTable,
                       AddressLine as AddressLine,
                       AddressLines as AddressLines,
                       WordInfo as WordInfo,
                       WordInfos as WordInfos,
-                      DetailedResult as DetailedResult)
+                      DetailedResult as DetailedResult,
+                      ReverseResult as ReverseResult)
 from .localization import (Locales as Locales)

nominatim/api/core.py

@@ -21,8 +21,9 @@ from nominatim.config import Configuration
 from nominatim.api.connection import SearchConnection
 from nominatim.api.status import get_status, StatusResult
 from nominatim.api.lookup import get_place_by_id
-from nominatim.api.types import PlaceRef, LookupDetails
-from nominatim.api.results import DetailedResult
+from nominatim.api.reverse import reverse_lookup
+from nominatim.api.types import PlaceRef, LookupDetails, AnyPoint, DataLayer
+from nominatim.api.results import DetailedResult, ReverseResult
 
 
 class NominatimAPIAsync:
@@ -136,6 +137,29 @@ class NominatimAPIAsync:
             return await get_place_by_id(conn, place, details or LookupDetails())
 
 
+    async def reverse(self, coord: AnyPoint, max_rank: Optional[int] = None,
+                      layer: Optional[DataLayer] = None,
+                      details: Optional[LookupDetails] = None) -> Optional[ReverseResult]:
+        """ Find a place by its coordinates. Also known as reverse geocoding.
+
+            Returns the closest result that can be found or None if
+            no place matches the given criteria.
+        """
+        # The following negation handles NaN correctly. Don't change.
+        if not abs(coord[0]) <= 180 or not abs(coord[1]) <= 90:
+            # There are no results to be expected outside valid coordinates.
+            return None
+
+        if layer is None:
+            layer = DataLayer.ADDRESS | DataLayer.POI
+
+        max_rank = max(0, min(max_rank or 30, 30))
+
+        async with self.begin() as conn:
+            return await reverse_lookup(conn, coord, max_rank, layer,
+                                        details or LookupDetails())
+
+
 class NominatimAPI:
     """ API loader, synchronous version.
     """
@@ -172,3 +196,15 @@ class NominatimAPI:
         """ Get detailed information about a place in the database.
         """
         return self._loop.run_until_complete(self._async_api.lookup(place, details))
+
+
+    def reverse(self, coord: AnyPoint, max_rank: Optional[int] = None,
+                layer: Optional[DataLayer] = None,
+                details: Optional[LookupDetails] = None) -> Optional[ReverseResult]:
+        """ Find a place by its coordinates. Also known as reverse geocoding.
+
+            Returns the closest result that can be found or None if
+            no place matches the given criteria.
+        """
+        return self._loop.run_until_complete(
+                   self._async_api.reverse(coord, max_rank, layer, details))

nominatim/api/results.py

@@ -19,7 +19,7 @@ import datetime as dt
 import sqlalchemy as sa
 
 from nominatim.typing import SaSelect, SaRow
-from nominatim.api.types import Point, LookupDetails
+from nominatim.api.types import Point, Bbox, LookupDetails
 from nominatim.api.connection import SearchConnection
 from nominatim.api.logging import log
 
@@ -46,6 +46,8 @@ class AddressLine:
     names: Dict[str, str]
     extratags: Optional[Dict[str, str]]
 
+    local_name: Optional[str] = None
+
     admin_level: Optional[int]
     fromarea: bool
     isaddress: bool
@@ -136,6 +138,14 @@ class DetailedResult(BaseResult):
     indexed_date: Optional[dt.datetime] = None
 
 
+@dataclasses.dataclass
+class ReverseResult(BaseResult):
+    """ A search result for reverse geocoding.
+    """
+    distance: Optional[float] = None
+    bbox: Optional[Bbox] = None
+
+
 def _filter_geometries(row: SaRow) -> Dict[str, str]:
     return {k[9:]: v for k, v in row._mapping.items() # pylint: disable=W0212
             if k.startswith('geometry_')}

nominatim/api/reverse.py

@@ -0,0 +1,509 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2023 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Implementation of reverse geocoding.
+"""
+from typing import Optional
+
+import sqlalchemy as sa
+from geoalchemy2 import WKTElement
+from geoalchemy2.types import Geometry
+
+from nominatim.typing import SaColumn, SaSelect, SaTable, SaLabel, SaClause
+from nominatim.api.connection import SearchConnection
+import nominatim.api.results as nres
+from nominatim.api.logging import log
+from nominatim.api.types import AnyPoint, DataLayer, LookupDetails, GeometryFormat
+
+def _select_from_placex(t: SaTable, wkt: Optional[str] = None) -> SaSelect:
+    """ Create a select statement with the columns relevant for reverse
+        results.
+    """
+    if wkt is None:
+        distance = t.c.distance
+    else:
+        distance = t.c.geometry.ST_Distance(wkt)
+
+    return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
+                     t.c.class_, t.c.type,
+                     t.c.address, t.c.extratags,
+                     t.c.housenumber, t.c.postcode, t.c.country_code,
+                     t.c.importance, t.c.wikipedia,
+                     t.c.parent_place_id, t.c.rank_address, t.c.rank_search,
+                     t.c.centroid,
+                     distance.label('distance'),
+                     t.c.geometry.ST_Expand(0).label('bbox'))
+
+
+def _interpolated_housenumber(table: SaTable) -> SaLabel:
+    # Entries with startnumber = endnumber are legacy from version < 4.1
+    return sa.cast(table.c.startnumber
+                    + sa.func.round(((table.c.endnumber - table.c.startnumber) * table.c.position)
+                                    / table.c.step) * table.c.step,
+                   sa.Integer).label('housenumber')
+
+
+def _is_address_point(table: SaTable) -> SaClause:
+    return sa.and_(table.c.rank_address == 30,
+                   sa.or_(table.c.housenumber != None,
+                          table.c.name.has_key('housename')))
+
+
+class ReverseGeocoder:
+    """ Class implementing the logic for looking up a place from a
+        coordinate.
+    """
+
+    def __init__(self, conn: SearchConnection, max_rank: int, layer: DataLayer,
+                 details: LookupDetails) -> None:
+        self.conn = conn
+        self.max_rank = max_rank
+        self.layer = layer
+        self.details = details
+
+
+    def _add_geometry_columns(self, sql: SaSelect, col: SaColumn) -> SaSelect:
+        if not self.details.geometry_output:
+            return sql
+
+        out = []
+
+        if self.details.geometry_simplification > 0.0:
+            col = col.ST_SimplifyPreserveTopology(self.details.geometry_simplification)
+
+        if self.details.geometry_output & GeometryFormat.GEOJSON:
+            out.append(col.ST_AsGeoJSON().label('geometry_geojson'))
+        if self.details.geometry_output & GeometryFormat.TEXT:
+            out.append(col.ST_AsText().label('geometry_text'))
+        if self.details.geometry_output & GeometryFormat.KML:
+            out.append(col.ST_AsKML().label('geometry_kml'))
+        if self.details.geometry_output & GeometryFormat.SVG:
+            out.append(col.ST_AsSVG().label('geometry_svg'))
+
+        return sql.add_columns(*out)
+
+
+    def _filter_by_layer(self, table: SaTable) -> SaColumn:
+        if self.layer & DataLayer.MANMADE:
+            exclude = []
+            if not (self.layer & DataLayer.RAILWAY):
+                exclude.append('railway')
+            if not (self.layer & DataLayer.NATURAL):
+                exclude.extend(('natural', 'water', 'waterway'))
+            return table.c.class_.not_in(tuple(exclude))
+
+        include = []
+        if self.layer & DataLayer.RAILWAY:
+            include.append('railway')
+        if not (self.layer & DataLayer.NATURAL):
+            include.extend(('natural', 'water', 'waterway'))
+        return table.c.class_.in_(tuple(include))
+
+
+    async def _find_closest_street_or_poi(self, wkt: WKTElement) -> SaRow:
+        """ Look up the clostest rank 26+ place in the database.
+        """
+        t = self.conn.t.placex
+
+        sql = _select_from_placex(t, wkt)\
+                .where(t.c.geometry.ST_DWithin(wkt, distance))\
+                .where(t.c.indexed_status == 0)\
+                .where(t.c.linked_place_id == None)\
+                .where(sa.or_(t.c.geometry.ST_GeometryType().not_in(('ST_Polygon', 'ST_MultiPolygon')),
+                              t.c.centroid.ST_Distance(wkt) < distance))\
+                .order_by('distance')\
+                .limit(1)
+
+        sql = self._add_geometry_columns(sql, t.c.geometry)
+
+        restrict = []
+
+        if self.layer & DataLayer.ADDRESS:
+            restrict.append(sa.and_(t.c.rank_address >= 26,
+                                    t.c.rank_address <= self.max_rank))
+            if self.max_rank == 30:
+                restrict.append(_is_address_point(t))
+        if self.layer & DataLayer.POI and max_rank == 30:
+            restrict.append(sa.and_(t.c.rank_search == 30,
+                                    t.c.class_.not_in(('place', 'building')),
+                                    t.c.geometry.ST_GeometryType() != 'ST_LineString'))
+        if self.layer & (DataLayer.RAILWAY | DataLayer.MANMADE | DataLayer.NATURAL):
+            restrict.append(sa.and_(t.c.rank_search >= 26,
+                                    tc.rank_search <= self.max_rank,
+                                    self._filter_by_layer(t)))
+
+        if restrict:
+            sql = sql.where(sa.or_(*restrict))
+
+        return (await self.conn.execute(sql)).one_or_none()
+
+
+    async def _find_housenumber_for_street(self, parent_place_id: int,
+                                           wkt: WKTElement) -> Optional[SaRow]:
+        t = conn.t.placex
+
+        sql = _select_from_placex(t, wkt)\
+                .where(t.c.geometry.ST_DWithin(wkt, 0.001))\
+                .where(t.c.parent_place_id == parent_place_id)\
+                .where(_is_address_point(t))\
+                .where(t.c.indexed_status == 0)\
+                .where(t.c.linked_place_id == None)\
+                .order_by('distance')\
+                .limit(1)
+
+        sql = self._add_geometry_columns(sql, t.c.geometry)
+
+        return (await self.conn.execute(sql)).one_or_none()
+
+
+    async def _find_interpolation_for_street(self, parent_place_id: Optional[int],
+                                             wkt: WKTElement) -> Optional[SaRow]:
+        t = self.conn.t.osmline
+
+        inner = sa.select(t,
+                          t.c.linegeo.ST_Distance(wkt).label('distance'),
+                          t.c.linegeo.ST_LineLocatePoint(wkt).label('position'))\
+                  .where(t.c.linegeo.ST_DWithin(wkt, distance))\
+                  .order_by('distance')\
+                  .limit(1)
+
+        if parent_place_id is not None:
+            inner = inner.where(t.c.parent_place_id == parent_place_id)
+
+        inner = inner.subquery()
+
+        sql = sa.select(inner.c.place_id, inner.c.osm_id,
+                        inner.c.parent_place_id, inner.c.address,
+                        _interpolated_housenumber(inner),
+                        inner.c.postcode, inner.c.country_code,
+                        inner.c.linegeo.ST_LineInterpolatePoint(inner.c.position).label('centroid'),
+                        inner.c.distance)
+
+        if self.details.geometry_output:
+            sub = sql.subquery()
+            sql = self._add_geometry_columns(sql, sub.c.centroid)
+
+        return (await self.conn.execute(sql)).one_or_none()
+
+
+    async def _find_tiger_number_for_street(self, parent_place_id: int,
+                                            wkt: WKTElement) -> Optional[SaRow]:
+        t = self.conn.t.tiger
+
+        inner = sa.select(t,
+                          t.c.linegeo.ST_Distance(wkt).label('distance'),
+                          sa.func.ST_LineLocatePoint(t.c.linegeo, wkt).label('position'))\
+                  .where(t.c.linegeo.ST_DWithin(wkt, 0.001))\
+                  .where(t.c.parent_place_id == parent_place_id)\
+                  .order_by('distance')\
+                  .limit(1)\
+                  .subquery()
+
+        sql = sa.select(inner.c.place_id,
+                        inner.c.parent_place_id,
+                        _interpolated_housenumber(inner),
+                        inner.c.postcode,
+                        inner.c.linegeo.ST_LineInterpolatePoint(inner.c.position).label('centroid'),
+                        inner.c.distance)
+
+        if self.details.geometry_output:
+            sub = sql.subquery()
+            sql = self._add_geometry_columns(sql, sub.c.centroid)
+
+        return (await conn.execute(sql)).one_or_none()
+
+
+    async def lookup_street_poi(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
+        """ Find a street or POI/address for the given WKT point.
+        """
+        log().section('Reverse lookup on street/address level')
+        result = None
+        distance = 0.006
+        parent_place_id = None
+
+        row = await self._find_closest_street_or_poi(wkt)
+        log().var_dump('Result (street/building)', row)
+
+        # If the closest result was a street, but an address was requested,
+        # check for a housenumber nearby which is part of the street.
+        if row is not None:
+            if self.max_rank > 27 \
+               and self.layer & DataLayer.ADDRESS \
+               and row.rank_address <= 27:
+                distance = 0.001
+                parent_place_id = row.place_id
+                log().comment('Find housenumber for street')
+                addr_row = await self._find_housenumber_for_street(parent_place_id, wkt)
+                log().var_dump('Result (street housenumber)', addr_row)
+
+                if addr_row is not None:
+                    row = addr_row
+                    distance = addr_row.distance
+                elif row.country_code == 'us' and parent_place_id is not None:
+                    log().comment('Find TIGER housenumber for street')
+                    addr_row = await self._find_tiger_number_for_street(parent_place_id, wkt)
+                    log().var_dump('Result (street Tiger housenumber)', addr_row)
+
+                    if addr_row is not None:
+                        result = nres.create_from_tiger_row(addr_row)
+            else:
+                distance = row.distance
+
+        # Check for an interpolation that is either closer than our result
+        # or belongs to a close street found.
+        if self.max_rank > 27 and self.layer & DataLayer.ADDRESS:
+            log().comment('Find interpolation for street')
+            addr_row = await self._find_interpolation_for_street(parent_place_id, wkt)
+            log().var_dump('Result (street interpolation)', addr_row)
+            if addr_row is not None:
+                result = nres.create_from_osmline_row(addr_row)
+
+        return result or nres.create_from_placex_row(row)
+
+
+    async def _lookup_area_address(self, wkt: WKTElement) -> Optional[SaRow]:
+        """ Lookup large addressable areas for the given WKT point.
+        """
+        log().comment('Reverse lookup by larger address area features')
+        t = self.conn.t.placex
+
+        # The inner SQL brings results in the right order, so that
+        # later only a minimum of results needs to be checked with ST_Contains.
+        inner = sa.select(t, sa.literal(0.0).label('distance'))\
+                  .where(t.c.rank_search.between(5, self.max_rank))\
+                  .where(t.c.rank_address.between(5, 25))\
+                  .where(t.c.geometry.ST_GeometryType().in_(('ST_Polygon', 'ST_MultiPolygon')))\
+                  .where(t.c.geometry.intersects(wkt))\
+                  .where(t.c.name != None)\
+                  .where(t.c.indexed_status == 0)\
+                  .where(t.c.linked_place_id == None)\
+                  .where(t.c.type != 'postcode')\
+                  .order_by(sa.desc(t.c.rank_search))\
+                  .limit(50)\
+                  .subquery()
+
+        sql = _select_from_placex(inner)\
+                  .where(inner.c.geometry.ST_Contains(wkt))\
+                  .order_by(sa.desc(inner.c.rank_search))\
+                  .limit(1)
+
+        sql = self._add_geometry_columns(sql, inner.c.geometry)
+
+        address_row = (await self.conn.execute(sql)).one_or_none()
+        log().var_dump('Result (area)', address_row)
+
+        if address_row is not None and address_row.rank_search < max_rank:
+            log().comment('Search for better matching place nodes inside the area')
+            inner = sa.select(t,
+                              t.c.geometry.ST_Distance(wkt).label('distance'))\
+                      .where(t.c.osm_type == 'N')\
+                      .where(t.c.rank_search > address_row.rank_search)\
+                      .where(t.c.rank_search <= max_rank)\
+                      .where(t.c.rank_address.between(5, 25))\
+                      .where(t.c.name != None)\
+                      .where(t.c.indexed_status == 0)\
+                      .where(t.c.linked_place_id == None)\
+                      .where(t.c.type != 'postcode')\
+                      .where(t.c.geometry
+                                .ST_Buffer(sa.func.reverse_place_diameter(t.c.rank_search))
+                                .intersects(wkt))\
+                      .order_by(sa.desc(t.c.rank_search))\
+                      .limit(50)\
+                      .subquery()
+
+            touter = conn.t.placex.alias('outer')
+            sql = _select_from_placex(inner)\
+                  .where(touter.c.place_id == address_row.place_id)\
+                  .where(touter.c.geometry.ST_Contains(inner.c.geometry))\
+                  .where(inner.c.distance < sa.func.reverse_place_diameter(inner.c.rank_search))\
+                  .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
+                  .limit(1)
+
+            sql = self._add_geometry_columns(sql, inner.c.geometry)
+
+            place_address_row = (await self.conn.execute(sql)).one_or_none()
+            log().var_dump('Result (place node)', place_address_row)
+
+            if place_address_row is not None:
+                return place_address_row
+
+        return address_row
+
+
+    async def _lookup_area_others(self, wkt: WKTElement) -> Optional[SaRow]:
+        t = conn.t.placex
+
+        inner = sa.select(t, t.c.geometry.ST_Distance(wkt).label('distance'))\
+                  .where(t.c.rank_address == 0)\
+                  .where(t.c.rank_search.between(5, self.max_rank))\
+                  .where(t.c.name != None)\
+                  .where(t.c.indexed_status == 0)\
+                  .where(t.c.linked_place_id == None)\
+                  .where(self._filter_by_layer(t))\
+                  .where(sa.func.reverse_buffered_extent(t.c.geometry, type_=Geometry)
+                                .intersects(wkt))\
+                  .order_by(sa.desc(t.c.rank_search))\
+                  .limit(50)
+
+        sql = _select_from_placex(inner)\
+                  .where(sa._or(inner.c.geometry.ST_GeometryType().not_in(('ST_Polygon', 'ST_MultiPolygon')),
+                                inner.c.geometry.ST_Contains(wkt)))\
+                  .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
+                  .limit(1)
+
+        sql = self._add_geometry_columns(sql, inner.c.geometry)
+
+        row = (await self.conn.execute(sql)).one_or_none()
+        log().var_dump('Result (non-address feature)', row)
+
+        return row
+
+
+    async def lookup_area(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
+        """ Lookup large areas for the given WKT point.
+        """
+        log().section('Reverse lookup by larger area features')
+        t = self.conn.t.placex
+
+        if self.layer & DataLayer.ADDRESS:
+            address_row = await self._lookup_area_address(wkt)
+            address_distance = address_row.distance
+        else:
+            address_row = None
+            address_distance = 1000
+
+        if self.layer & (~DataLayer.ADDRESS & ~DataLayer.POI):
+            other_row = await self._lookup_area_others(wkt)
+            other_distance = other_row.distance
+        else:
+            other_row = None
+            other_distance = 1000
+
+        result = address_row if address_distance <= other_distance else other_row
+
+        return nres.create_from_placex_row(result)
+
+
+    async def lookup_country(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
+        """ Lookup the country for the given WKT point.
+        """
+        log().section('Reverse lookup by country code')
+        t = self.conn.t.country_grid
+        sql = sa.select(t.c.country_code).distinct()\
+                .where(t.c.geometry.ST_Contains(wkt))
+
+        ccodes = tuple((r[0] for r in await self.conn.execute(sql)))
+        log().var_dump('Country codes', ccodes)
+
+        if not ccodes:
+            return None
+
+        if self.layer & DataLayer.ADDRESS and self.max_rank > 4:
+            log().comment('Search for place nodes in country')
+
+            t = conn.t.placex
+            inner = sa.select(t,
+                              t.c.geometry.ST_Distance(wkt).label('distance'))\
+                      .where(t.c.osm_type == 'N')\
+                      .where(t.c.rank_search > 4)\
+                      .where(t.c.rank_search <= self.max_rank)\
+                      .where(t.c.rank_address.between(5, 25))\
+                      .where(t.c.name != None)\
+                      .where(t.c.indexed_status == 0)\
+                      .where(t.c.linked_place_id == None)\
+                      .where(t.c.type != 'postcode')\
+                      .where(t.c.country_code.in_(ccodes))\
+                      .where(t.c.geometry
+                                .ST_Buffer(sa.func.reverse_place_diameter(t.c.rank_search))
+                                .intersects(wkt))\
+                      .order_by(sa.desc(t.c.rank_search))\
+                      .limit(50)\
+                      .subquery()
+
+            sql = _select_from_placex(inner)\
+                  .where(inner.c.distance < sa.func.reverse_place_diameter(inner.c.rank_search))\
+                  .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
+                  .limit(1)
+
+            sql = self._add_geometry_columns(sql, inner.c.geometry)
+
+            address_row = (await self.conn.execute(sql)).one_or_none()
+            log().var_dump('Result (addressable place node)', address_row)
+        else:
+            address_row = None
+
+        if layer & (~DataLayer.ADDRESS & ~DataLayer.POI) and self.max_rank > 4:
+            log().comment('Search for non-address features inside country')
+
+            t = conn.t.placex
+            inner = sa.select(t, t.c.geometry.ST_Distance(wkt).label('distance'))\
+                      .where(t.c.rank_address == 0)\
+                      .where(t.c.rank_search.between(5, self.max_rank))\
+                      .where(t.c.name != None)\
+                      .where(t.c.indexed_status == 0)\
+                      .where(t.c.linked_place_id == None)\
+                      .where(self._filter_by_layer(t))\
+                      .where(t.c.country_code.in_(ccode))\
+                      .where(sa.func.reverse_buffered_extent(t.c.geometry, type_=Geometry)
+                                    .intersects(wkt))\
+                      .order_by(sa.desc(t.c.rank_search))\
+                      .limit(50)\
+                      .subquery()
+
+            sql = _select_from_placex(inner)\
+                      .where(sa._or(inner.c.geometry.ST_GeometryType().not_in(('ST_Polygon', 'ST_MultiPolygon')),
+                                    inner.c.geometry.ST_Contains(wkt)))\
+                      .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
+                      .limit(1)
+
+            sql = self._add_geometry_columns(sql, inner.c.geometry)
+
+            other_row = (await self.conn.execute(sql)).one_or_none()
+            log().var_dump('Result (non-address feature)', other_row)
+        else:
+            other_row = None
+
+        if layer & DataLayer.ADDRESS and address_row is None and other_row is None:
+            # Still nothing, then return a country with the appropriate country code.
+            t = conn.t.placex
+            sql = _select_from_placex(t, wkt)\
+                      .where(t.c.country_code.in_(ccodes))\
+                      .where(t.c.rank_address == 4)\
+                      .where(t.c.rank_search == 4)\
+                      .where(t.c.linked_place_id == None)\
+                      .order_by('distance')
+
+            sql = self._add_geometry_columns(sql, inner.c.geometry)
+
+            address_row = (await self.conn.execute(sql)).one_or_none()
+
+        return nres.create_from_placex_row(_get_closest_row(address_row, other_row))
+
+
+    async def lookup(self, coord: AnyPoint) -> Optional[nres.ReverseResult]:
+        """ Look up a single coordinate. Returns the place information,
+            if a place was found near the coordinates or None otherwise.
+        """
+        log().function('reverse_lookup',
+                       coord=coord, max_rank=self.max_rank,
+                       layer=self.layer, details=self.details)
+
+
+        wkt = WKTElement(f'POINT({coord[0]} {coord[1]})', srid=4326)
+
+        result: Optional[ReverseResult] = None
+
+        if max_rank >= 26:
+            result = await self.lookup_street_poi(wkt)
+        if result is None and max_rank > 4:
+            result = await self.lookup_area(wkt)
+        if result is None:
+            result = await self.lookup_country(wkt)
+        if result is not None:
+            await nres.add_result_details(self.conn, result, self.details)
+
+        return result

nominatim/api/types.py

@@ -7,7 +7,7 @@
 """
 Complex datatypes used by the Nominatim API.
 """
-from typing import Optional, Union, NamedTuple
+from typing import Optional, Union, Tuple, NamedTuple
 import dataclasses
 import enum
 from struct import unpack
@@ -83,6 +83,74 @@ class Point(NamedTuple):
         return Point(x, y)
 
 
+AnyPoint = Union[Point, Tuple[float, float]]
+
+
+class Bbox:
+    """ A bounding box in WSG84 projection.
+
+        The coordinates are available as an array in the 'coord'
+        property in the order (minx, miny, maxx, maxy).
+    """
+    def __init__(self, minx: float, miny: float, maxx: float, maxy: float) -> None:
+        self.coords = (minx, miny, maxx, maxy)
+
+
+    @property
+    def minlat(self) -> float:
+        """ Southern-most latitude, corresponding to the minimum y coordinate.
+        """
+        return self.coords[1]
+
+
+    @property
+    def maxlat(self) -> float:
+        """ Northern-most latitude, corresponding to the maximum y coordinate.
+        """
+        return self.coords[3]
+
+
+    @property
+    def minlon(self) -> float:
+        """ Western-most longitude, corresponding to the minimum x coordinate.
+        """
+        return self.coords[0]
+
+
+    @property
+    def maxlon(self) -> float:
+        """ Eastern-most longitude, corresponding to the maximum x coordinate.
+        """
+        return self.coords[2]
+
+
+    @staticmethod
+    def from_wkb(wkb: Optional[bytes]) -> 'Optional[Bbox]':
+        """ Create a Bbox from a bounding box polygon as returned by
+            the database. Return s None if the input value is None.
+        """
+        if wkb is None:
+            return None
+
+        if len(wkb) != 97:
+            raise ValueError("WKB must be a bounding box polygon")
+        if wkb.startswith(b'\x01\x03\x00\x00\x20\xE6\x10\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00'):
+            x1, y1, _, _, x2, y2 = unpack('<dddddd', wkb[17:65])
+        elif wkb.startswith(b'\x00\x20\x00\x00\x03\x00\x00\x10\xe6\x00\x00\x00\x01\x00\x00\x00\x05'):
+            x1, y1, _, _, x2, y2 = unpack('>dddddd', wkb[17:65])
+        else:
+            raise ValueError("WKB has wrong header")
+
+        return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
+
+
+    def from_point(pt: Point, buffer: float) -> 'Bbox':
+        """ Return a Bbox around the point with the buffer added to all sides.
+        """
+        return Bbox(pt[0] - buffer, pt[1] - buffer,
+                    pt[0] + buffer, pt[1] + buffer)
+
+
 class GeometryFormat(enum.Flag):
     """ Geometry output formats supported by Nominatim.
     """
@@ -117,3 +185,18 @@ class LookupDetails:
     keywords: bool = False
     """ Add information about the search terms used for this place.
     """
+    geometry_simplification: float = 0.0
+    """ Simplification factor for a geometry in degrees WGS. A factor of
+        0.0 means the original geometry is kept. The higher the value, the
+        more the geometry gets simplified.
+    """
+
+
+class DataLayer(enum.Flag):
+    """ Layer types that can be selected for reverse and forward search.
+    """
+    POI = enum.auto()
+    ADDRESS = enum.auto()
+    RAILWAY = enum.auto()
+    MANMADE = enum.auto()
+    NATURAL = enum.auto()

nominatim/typing.py

@@ -53,7 +53,7 @@ else:
 
 
 # SQLAlchemy introduced generic types in version 2.0 making typing
-# inclompatiple with older versions. Add wrappers here so we don't have
+# incompatible with older versions. Add wrappers here so we don't have
 # to litter the code with bare-string types.
 
 if TYPE_CHECKING:
@@ -66,3 +66,5 @@ SaSelect: TypeAlias = 'sa.Select[Any]'
 SaRow: TypeAlias = 'sa.Row[Any]'
 SaColumn: TypeAlias = 'sa.Column[Any]'
 SaLabel: TypeAlias = 'sa.Label[Any]'
+SaTable: TypeAlias = 'sa.Table[Any]'
+SaClause: TypeAlias = 'sa.ClauseElement[Any]'