Merge remote-tracking branch 'upstream/master'

[nominatim.git] / nominatim / api / reverse.py

diff --git a/nominatim/api/reverse.py b/nominatim/api/reverse.py
index 053b96dd96f3a1c3db683f547eef76bde5ffc16c..e16742cfa34170f6a8afbd0da75b6c945d7635d7 100644 (file)
--- a/nominatim/api/reverse.py
+++ b/nominatim/api/reverse.py
@@ -7,26 +7,47 @@
 """
 Implementation of reverse geocoding.
 """
 """
 Implementation of reverse geocoding.
 """

-from typing import Optional
+from typing import Optional, List, Callable, Type, Tuple, Dict, Any, cast, Union
+import functools

 
 import sqlalchemy as sa
 
 import sqlalchemy as sa

-from geoalchemy2 import WKTElement
-from geoalchemy2.types import Geometry

 
 

-from nominatim.typing import SaColumn, SaSelect, SaTable, SaLabel, SaClause
+from nominatim.typing import SaColumn, SaSelect, SaFromClause, SaLabel, SaRow,\
+                             SaBind, SaLambdaSelect

 from nominatim.api.connection import SearchConnection
 import nominatim.api.results as nres
 from nominatim.api.logging import log
 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
+from nominatim.api.types import AnyPoint, DataLayer, ReverseDetails, GeometryFormat, Bbox
+from nominatim.db.sqlalchemy_types import Geometry

 
 

-def _select_from_placex(t: SaTable, wkt: Optional[str] = None) -> SaSelect:
+# In SQLAlchemy expression which compare with NULL need to be expressed with
+# the equal sign.
+# pylint: disable=singleton-comparison
+
+RowFunc = Callable[[Optional[SaRow], Type[nres.ReverseResult]], Optional[nres.ReverseResult]]
+
+WKT_PARAM: SaBind = sa.bindparam('wkt', type_=Geometry)
+MAX_RANK_PARAM: SaBind = sa.bindparam('max_rank')
+
+def no_index(expr: SaColumn) -> SaColumn:
+    """ Wrap the given expression, so that the query planner will
+        refrain from using the expression for index lookup.
+    """
+    return sa.func.coalesce(sa.null(), expr) # pylint: disable=not-callable
+
+
+def _select_from_placex(t: SaFromClause, use_wkt: bool = True) -> SaSelect:

     """ Create a select statement with the columns relevant for reverse
         results.
     """
     """ Create a select statement with the columns relevant for reverse
         results.
     """

-    if wkt is None:
+    if not use_wkt:

         distance = t.c.distance
         distance = t.c.distance

+        centroid = t.c.centroid

     else:
     else:

-        distance = t.c.geometry.ST_Distance(wkt)
+        distance = t.c.geometry.ST_Distance(WKT_PARAM)
+        centroid = sa.case((t.c.geometry.is_line_like(), t.c.geometry.ST_ClosestPoint(WKT_PARAM)),
+                           else_=t.c.centroid).label('centroid')
+

 
     return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
                      t.c.class_, t.c.type,
 
     return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
                      t.c.class_, t.c.type,


@@ -34,23 +55,37 @@ def _select_from_placex(t: SaTable, wkt: Optional[str] = None) -> SaSelect:
                      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.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,
+                     centroid,
+                     t.c.linked_place_id, t.c.admin_level,

                      distance.label('distance'),
                      t.c.geometry.ST_Expand(0).label('bbox'))
 
 
                      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
+def _interpolated_housenumber(table: SaFromClause) -> SaLabel:

     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')
 
 
     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')))
+def _interpolated_position(table: SaFromClause) -> SaLabel:
+    fac = sa.cast(table.c.step, sa.Float) / (table.c.endnumber - table.c.startnumber)
+    rounded_pos = sa.func.round(table.c.position / fac) * fac
+    return sa.case(
+             (table.c.endnumber == table.c.startnumber, table.c.linegeo.ST_Centroid()),
+              else_=table.c.linegeo.ST_LineInterpolatePoint(rounded_pos)).label('centroid')
+
+
+def _locate_interpolation(table: SaFromClause) -> SaLabel:
+    """ Given a position, locate the closest point on the line.
+    """
+    return sa.case((table.c.linegeo.is_line_like(),
+                    table.c.linegeo.ST_LineLocatePoint(WKT_PARAM)),
+                   else_=0).label('position')
+
+
+def _get_closest(*rows: Optional[SaRow]) -> Optional[SaRow]:
+    return min(rows, key=lambda row: 1000 if row is None else row.distance)

 
 
 class ReverseGeocoder:
 
 
 class ReverseGeocoder:


@@ -58,274 +93,328 @@ class ReverseGeocoder:
         coordinate.
     """
 
         coordinate.
     """
 

-    def __init__(self, conn: SearchConnection, max_rank: int, layer: DataLayer,
-                 details: LookupDetails) -> None:
+    def __init__(self, conn: SearchConnection, params: ReverseDetails,
+                 restrict_to_country_areas: bool = False) -> None:

         self.conn = conn
         self.conn = conn

-        self.max_rank = max_rank
-        self.layer = layer
-        self.details = details
+        self.params = params
+        self.restrict_to_country_areas = restrict_to_country_areas
+
+        self.bind_params: Dict[str, Any] = {'max_rank': params.max_rank}
+
+
+    @property
+    def max_rank(self) -> int:
+        """ Return the maximum configured rank.
+        """
+        return self.params.max_rank
+
+
+    def has_geometries(self) -> bool:
+        """ Check if any geometries are requested.
+        """
+        return bool(self.params.geometry_output)
+
+
+    def layer_enabled(self, *layer: DataLayer) -> bool:
+        """ Return true when any of the given layer types are requested.
+        """
+        return any(self.params.layers & l for l in layer)
+
+
+    def layer_disabled(self, *layer: DataLayer) -> bool:
+        """ Return true when none of the given layer types is requested.
+        """
+        return not any(self.params.layers & l for l in layer)
+

 
 

+    def has_feature_layers(self) -> bool:
+        """ Return true if any layer other than ADDRESS or POI is requested.
+        """
+        return self.layer_enabled(DataLayer.RAILWAY, DataLayer.MANMADE, DataLayer.NATURAL)

 
 

-    def _add_geometry_columns(self, sql: SaSelect, col: SaColumn) -> SaSelect:
-        if not self.details.geometry_output:
-            return sql

 
 

+    def _add_geometry_columns(self, sql: SaLambdaSelect, col: SaColumn) -> SaSelect:

         out = []
 
         out = []
 

-        if self.details.geometry_simplification > 0.0:
-            col = col.ST_SimplifyPreserveTopology(self.details.geometry_simplification)
+        if self.params.geometry_simplification > 0.0:
+            col = sa.func.ST_SimplifyPreserveTopology(col, self.params.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'))
+        if self.params.geometry_output & GeometryFormat.GEOJSON:
+            out.append(sa.func.ST_AsGeoJSON(col, 7).label('geometry_geojson'))
+        if self.params.geometry_output & GeometryFormat.TEXT:
+            out.append(sa.func.ST_AsText(col).label('geometry_text'))
+        if self.params.geometry_output & GeometryFormat.KML:
+            out.append(sa.func.ST_AsKML(col, 7).label('geometry_kml'))
+        if self.params.geometry_output & GeometryFormat.SVG:
+            out.append(sa.func.ST_AsSVG(col, 0, 7).label('geometry_svg'))

 
         return sql.add_columns(*out)
 
 
 
         return sql.add_columns(*out)
 
 

-    def _filter_by_layer(self, table: SaTable) -> SaColumn:
-        if self.layer & DataLayer.MANMADE:
+    def _filter_by_layer(self, table: SaFromClause) -> SaColumn:
+        if self.layer_enabled(DataLayer.MANMADE):

             exclude = []
             exclude = []

-            if not (self.layer & DataLayer.RAILWAY):
+            if self.layer_disabled(DataLayer.RAILWAY):

                 exclude.append('railway')
                 exclude.append('railway')

-            if not (self.layer & DataLayer.NATURAL):
+            if self.layer_disabled(DataLayer.NATURAL):

                 exclude.extend(('natural', 'water', 'waterway'))
             return table.c.class_.not_in(tuple(exclude))
 
         include = []
                 exclude.extend(('natural', 'water', 'waterway'))
             return table.c.class_.not_in(tuple(exclude))
 
         include = []

-        if self.layer & DataLayer.RAILWAY:
+        if self.layer_enabled(DataLayer.RAILWAY):

             include.append('railway')
             include.append('railway')

-        if not (self.layer & DataLayer.NATURAL):
+        if self.layer_enabled(DataLayer.NATURAL):

             include.extend(('natural', 'water', 'waterway'))
         return table.c.class_.in_(tuple(include))
 
 
             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.
+    async def _find_closest_street_or_poi(self, distance: float) -> Optional[SaRow]:
+        """ Look up the closest rank 26+ place in the database, which
+            is closer than the given distance.

         """
         t = self.conn.t.placex
 
         """
         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)
+        # PostgreSQL must not get the distance as a parameter because
+        # there is a danger it won't be able to properly estimate index use
+        # when used with prepared statements
+        diststr = sa.text(f"{distance}")
+
+        sql: SaLambdaSelect = sa.lambda_stmt(lambda: _select_from_placex(t)
+                .where(t.c.geometry.within_distance(WKT_PARAM, diststr))
+                .where(t.c.indexed_status == 0)
+                .where(t.c.linked_place_id == None)
+                .where(sa.or_(sa.not_(t.c.geometry.is_area()),
+                              t.c.centroid.ST_Distance(WKT_PARAM) < diststr))
+                .order_by('distance')
+                .limit(1))

 
 

-        sql = self._add_geometry_columns(sql, t.c.geometry)
+        if self.has_geometries():
+            sql = self._add_geometry_columns(sql, t.c.geometry)

 
 

-        restrict = []
+        restrict: List[Union[SaColumn, Callable[[], SaColumn]]] = []

 
 

-        if self.layer & DataLayer.ADDRESS:
-            restrict.append(sa.and_(t.c.rank_address >= 26,
-                                    t.c.rank_address <= self.max_rank))
+        if self.layer_enabled(DataLayer.ADDRESS):
+            max_rank = min(29, self.max_rank)
+            restrict.append(lambda: no_index(t.c.rank_address).between(26, max_rank))

             if self.max_rank == 30:
             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,
+                restrict.append(lambda: sa.func.IsAddressPoint(t))
+        if self.layer_enabled(DataLayer.POI) and self.max_rank == 30:
+            restrict.append(lambda: sa.and_(no_index(t.c.rank_search) == 30,
+                                            t.c.class_.not_in(('place', 'building')),
+                                            sa.not_(t.c.geometry.is_line_like())))
+        if self.has_feature_layers():
+            restrict.append(sa.and_(no_index(t.c.rank_search).between(26, MAX_RANK_PARAM),
+                                    no_index(t.c.rank_address) == 0,

                                     self._filter_by_layer(t)))
 
                                     self._filter_by_layer(t)))
 

-        if restrict:
-            sql = sql.where(sa.or_(*restrict))
+        if not restrict:
+            return None

 
 

-        return (await self.conn.execute(sql)).one_or_none()
+        sql = sql.where(sa.or_(*restrict))

 
 

+        return (await self.conn.execute(sql, self.bind_params)).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))\
+    async def _find_housenumber_for_street(self, parent_place_id: int) -> Optional[SaRow]:
+        t = self.conn.t.placex
+
+        def _base_query() -> SaSelect:
+            return _select_from_placex(t)\
+                .where(t.c.geometry.within_distance(WKT_PARAM, 0.001))\

                 .where(t.c.parent_place_id == parent_place_id)\
                 .where(t.c.parent_place_id == parent_place_id)\

-                .where(_is_address_point(t))\
+                .where(sa.func.IsAddressPoint(t))\

                 .where(t.c.indexed_status == 0)\
                 .where(t.c.linked_place_id == None)\
                 .order_by('distance')\
                 .limit(1)
 
                 .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)
+        sql: SaLambdaSelect
+        if self.has_geometries():
+            sql = self._add_geometry_columns(_base_query(), t.c.geometry)
+        else:
+            sql = sa.lambda_stmt(_base_query)

 
 

-        return (await self.conn.execute(sql)).one_or_none()
+        return (await self.conn.execute(sql, self.bind_params)).one_or_none()

 
 
     async def _find_interpolation_for_street(self, parent_place_id: Optional[int],
 
 
     async def _find_interpolation_for_street(self, parent_place_id: Optional[int],

-                                             wkt: WKTElement) -> Optional[SaRow]:
+                                             distance: float) -> Optional[SaRow]:

         t = self.conn.t.osmline
 
         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)
+        sql = sa.select(t,
+                        t.c.linegeo.ST_Distance(WKT_PARAM).label('distance'),
+                        _locate_interpolation(t))\
+                .where(t.c.linegeo.within_distance(WKT_PARAM, distance))\
+                .where(t.c.startnumber != None)\
+                .order_by('distance')\
+                .limit(1)

 
         if parent_place_id is not None:
 
         if parent_place_id is not None:

-            inner = inner.where(t.c.parent_place_id == parent_place_id)
+            sql = sql.where(t.c.parent_place_id == parent_place_id)

 
 

-        inner = inner.subquery()
+        inner = sql.subquery('ipol')

 
         sql = sa.select(inner.c.place_id, inner.c.osm_id,
 
         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)
+                             inner.c.parent_place_id, inner.c.address,
+                             _interpolated_housenumber(inner),
+                             _interpolated_position(inner),
+                             inner.c.postcode, inner.c.country_code,
+                             inner.c.distance)

 
 

-        if self.details.geometry_output:
-            sub = sql.subquery()
-            sql = self._add_geometry_columns(sql, sub.c.centroid)
+        if self.has_geometries():
+            sub = sql.subquery('geom')
+            sql = self._add_geometry_columns(sa.select(sub), sub.c.centroid)

 
 

-        return (await self.conn.execute(sql)).one_or_none()
+        return (await self.conn.execute(sql, self.bind_params)).one_or_none()

 
 
 
 

-    async def _find_tiger_number_for_street(self, parent_place_id: int,
-                                            wkt: WKTElement) -> Optional[SaRow]:
+    async def _find_tiger_number_for_street(self, parent_place_id: int) -> Optional[SaRow]:

         t = self.conn.t.tiger
 
         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)
+        def _base_query() -> SaSelect:
+            inner = sa.select(t,
+                              t.c.linegeo.ST_Distance(WKT_PARAM).label('distance'),
+                              _locate_interpolation(t))\
+                      .where(t.c.linegeo.within_distance(WKT_PARAM, 0.001))\
+                      .where(t.c.parent_place_id == parent_place_id)\
+                      .order_by('distance')\
+                      .limit(1)\
+                      .subquery('tiger')
+
+            return sa.select(inner.c.place_id,
+                             inner.c.parent_place_id,
+                             _interpolated_housenumber(inner),
+                             _interpolated_position(inner),
+                             inner.c.postcode,
+                             inner.c.distance)
+
+        sql: SaLambdaSelect
+        if self.has_geometries():
+            sub = _base_query().subquery('geom')
+            sql = self._add_geometry_columns(sa.select(sub), sub.c.centroid)
+        else:
+            sql = sa.lambda_stmt(_base_query)

 
 

-        return (await conn.execute(sql)).one_or_none()
+        return (await self.conn.execute(sql, self.bind_params)).one_or_none()

 
 
 
 

-    async def lookup_street_poi(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
+    async def lookup_street_poi(self) -> Tuple[Optional[SaRow], RowFunc]:

         """ Find a street or POI/address for the given WKT point.
         """
         log().section('Reverse lookup on street/address level')
         """ 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
 
         distance = 0.006
         parent_place_id = None
 

-        row = await self._find_closest_street_or_poi(wkt)
+        row = await self._find_closest_street_or_poi(distance)
+        row_func: RowFunc = nres.create_from_placex_row

         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 \
         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 self.layer_enabled(DataLayer.ADDRESS) \

                and row.rank_address <= 27:
                 distance = 0.001
                 parent_place_id = row.place_id
                 log().comment('Find housenumber for street')
                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)
+                addr_row = await self._find_housenumber_for_street(parent_place_id)

                 log().var_dump('Result (street housenumber)', addr_row)
 
                 if addr_row is not None:
                     row = addr_row
                 log().var_dump('Result (street housenumber)', addr_row)
 
                 if addr_row is not None:
                     row = addr_row

+                    row_func = nres.create_from_placex_row

                     distance = addr_row.distance
                 elif row.country_code == 'us' and parent_place_id is not None:
                     log().comment('Find TIGER housenumber for street')
                     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)
+                    addr_row = await self._find_tiger_number_for_street(parent_place_id)

                     log().var_dump('Result (street Tiger housenumber)', addr_row)
 
                     if addr_row is not None:
                     log().var_dump('Result (street Tiger housenumber)', addr_row)
 
                     if addr_row is not None:

-                        result = nres.create_from_tiger_row(addr_row)
+                        row_func = cast(RowFunc,
+                                        functools.partial(nres.create_from_tiger_row,
+                                                          osm_type=row.osm_type,
+                                                          osm_id=row.osm_id))
+                        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.
             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:
+        if self.max_rank > 27 and self.layer_enabled(DataLayer.ADDRESS):

             log().comment('Find interpolation for street')
             log().comment('Find interpolation for street')

-            addr_row = await self._find_interpolation_for_street(parent_place_id, wkt)
+            addr_row = await self._find_interpolation_for_street(parent_place_id,
+                                                                 distance)

             log().var_dump('Result (street interpolation)', addr_row)
             if addr_row is not None:
             log().var_dump('Result (street interpolation)', addr_row)
             if addr_row is not None:

-                result = nres.create_from_osmline_row(addr_row)
+                row = addr_row
+                row_func = nres.create_from_osmline_row

 
 

-        return result or nres.create_from_placex_row(row)
+        return row, row_func

 
 
 
 

-    async def _lookup_area_address(self, wkt: WKTElement) -> Optional[SaRow]:
+    async def _lookup_area_address(self) -> 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
 
         """ 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()
+        def _base_query() -> SaSelect:
+            # 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, MAX_RANK_PARAM))\
+                      .where(t.c.geometry.intersects(WKT_PARAM))\
+                      .where(sa.func.PlacexGeometryReverseLookuppolygon())\
+                      .order_by(sa.desc(t.c.rank_search))\
+                      .limit(50)\
+                      .subquery('area')

 
 

-        sql = _select_from_placex(inner)\
-                  .where(inner.c.geometry.ST_Contains(wkt))\
-                  .order_by(sa.desc(inner.c.rank_search))\
-                  .limit(1)
+            return _select_from_placex(inner, False)\
+                      .where(inner.c.geometry.ST_Contains(WKT_PARAM))\
+                      .order_by(sa.desc(inner.c.rank_search))\
+                      .limit(1)

 
 

-        sql = self._add_geometry_columns(sql, inner.c.geometry)
+        sql: SaLambdaSelect = sa.lambda_stmt(_base_query)
+        if self.has_geometries():
+            sql = self._add_geometry_columns(sql, sa.literal_column('area.geometry'))

 
 

-        address_row = (await self.conn.execute(sql)).one_or_none()
+        address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()

         log().var_dump('Result (area)', address_row)
 
         log().var_dump('Result (area)', address_row)
 

-        if address_row is not None and address_row.rank_search < max_rank:
+        if address_row is not None and address_row.rank_search < self.max_rank:

             log().comment('Search for better matching place nodes inside the area')
             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)\
+
+            address_rank = address_row.rank_search
+            address_id = address_row.place_id
+
+            def _place_inside_area_query() -> SaSelect:
+                inner = \
+                    sa.select(t,
+                              t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\
+                      .where(t.c.rank_search > address_rank)\
+                      .where(t.c.rank_search <= MAX_RANK_PARAM)\

                       .where(t.c.indexed_status == 0)\
                       .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))\
+                      .where(sa.func.IntersectsReverseDistance(t, WKT_PARAM))\

                       .order_by(sa.desc(t.c.rank_search))\
                       .limit(50)\
                       .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)
+                      .subquery('places')
+
+                touter = t.alias('outer')
+                return _select_from_placex(inner, False)\
+                    .join(touter, touter.c.geometry.ST_Contains(inner.c.geometry))\
+                    .where(touter.c.place_id == address_id)\
+                    .where(sa.func.IsBelowReverseDistance(inner.c.distance, inner.c.rank_search))\
+                    .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
+                    .limit(1)
+
+            if self.has_geometries():
+                sql = self._add_geometry_columns(_place_inside_area_query(),
+                                                 sa.literal_column('places.geometry'))
+            else:
+                sql = sa.lambda_stmt(_place_inside_area_query)

 
 

-            place_address_row = (await self.conn.execute(sql)).one_or_none()
+            place_address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()

             log().var_dump('Result (place node)', place_address_row)
 
             if place_address_row is not None:
             log().var_dump('Result (place node)', place_address_row)
 
             if place_address_row is not None:


@@ -334,176 +423,168 @@ class ReverseGeocoder:
         return address_row
 
 
         return address_row
 
 

-    async def _lookup_area_others(self, wkt: WKTElement) -> Optional[SaRow]:
-        t = conn.t.placex
+    async def _lookup_area_others(self) -> Optional[SaRow]:
+        t = self.conn.t.placex

 
 

-        inner = sa.select(t, t.c.geometry.ST_Distance(wkt).label('distance'))\
+        inner = sa.select(t, t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\

                   .where(t.c.rank_address == 0)\
                   .where(t.c.rank_address == 0)\

-                  .where(t.c.rank_search.between(5, self.max_rank))\
+                  .where(t.c.rank_search.between(5, MAX_RANK_PARAM))\

                   .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.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))\
+                  .where(t.c.geometry.intersects(sa.func.ST_Expand(WKT_PARAM, 0.007)))\

                   .order_by(sa.desc(t.c.rank_search))\
                   .order_by(sa.desc(t.c.rank_search))\

-                  .limit(50)
+                  .order_by('distance')\
+                  .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)))\
+        sql = _select_from_placex(inner, False)\
+                  .where(sa.or_(sa.not_(inner.c.geometry.is_area()),
+                                inner.c.geometry.ST_Contains(WKT_PARAM)))\

                   .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
                   .limit(1)
 
                   .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
                   .limit(1)
 

-        sql = self._add_geometry_columns(sql, inner.c.geometry)
+        if self.has_geometries():
+            sql = self._add_geometry_columns(sql, inner.c.geometry)

 
 

-        row = (await self.conn.execute(sql)).one_or_none()
+        row = (await self.conn.execute(sql, self.bind_params)).one_or_none()

         log().var_dump('Result (non-address feature)', row)
 
         return row
 
 
         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.
+    async def lookup_area(self) -> Optional[SaRow]:
+        """ Lookup large areas for the current search.

         """
         log().section('Reverse lookup by larger area features')
         """
         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
+        if self.layer_enabled(DataLayer.ADDRESS):
+            address_row = await self._lookup_area_address()

         else:
             address_row = None
         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
+        if self.has_feature_layers():
+            other_row = await self._lookup_area_others()

         else:
             other_row = None
         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)
+        return _get_closest(address_row, other_row)

 
 
 
 

-    async def lookup_country(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
-        """ Lookup the country for the given WKT point.
+    async def lookup_country_codes(self) -> List[str]:
+        """ Lookup the country for the current search.

         """
         log().section('Reverse lookup by country code')
         t = self.conn.t.country_grid
         sql = sa.select(t.c.country_code).distinct()\
         """
         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))
+                .where(t.c.geometry.ST_Contains(WKT_PARAM))

 
 

-        ccodes = tuple((r[0] for r in await self.conn.execute(sql)))
+        ccodes = [cast(str, r[0]) for r in await self.conn.execute(sql, self.bind_params)]

         log().var_dump('Country codes', ccodes)
         log().var_dump('Country codes', ccodes)

+        return ccodes
+
+
+    async def lookup_country(self, ccodes: List[str]) -> Optional[SaRow]:
+        """ Lookup the country for the current search.
+        """
+        if not ccodes:
+            ccodes = await self.lookup_country_codes()

 
         if not ccodes:
             return None
 
 
         if not ccodes:
             return None
 

-        if self.layer & DataLayer.ADDRESS and self.max_rank > 4:
+        t = self.conn.t.placex
+        if self.max_rank > 4:

             log().comment('Search for place nodes in country')
 
             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')\
+            def _base_query() -> SaSelect:
+                inner = \
+                    sa.select(t,
+                              t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\

                       .where(t.c.rank_search > 4)\
                       .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.rank_search <= MAX_RANK_PARAM)\

                       .where(t.c.indexed_status == 0)\
                       .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.country_code.in_(ccodes))\

-                      .where(t.c.geometry
-                                .ST_Buffer(sa.func.reverse_place_diameter(t.c.rank_search))
-                                .intersects(wkt))\
+                      .where(sa.func.IntersectsReverseDistance(t, WKT_PARAM))\

                       .order_by(sa.desc(t.c.rank_search))\
                       .limit(50)\
                       .order_by(sa.desc(t.c.rank_search))\
                       .limit(50)\

-                      .subquery()
+                      .subquery('area')

 
 

-            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)
+                return _select_from_placex(inner, False)\
+                    .where(sa.func.IsBelowReverseDistance(inner.c.distance, 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)
+            sql: SaLambdaSelect
+            if self.has_geometries():
+                sql = self._add_geometry_columns(_base_query(),
+                                                 sa.literal_column('area.geometry'))
+            else:
+                sql = sa.lambda_stmt(_base_query)

 
 

-            address_row = (await self.conn.execute(sql)).one_or_none()
+            address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()

             log().var_dump('Result (addressable place node)', address_row)
         else:
             address_row = 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:
+        if address_row is None:

             # Still nothing, then return a country with the appropriate country code.
             # 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)
+            def _country_base_query() -> SaSelect:
+                return _select_from_placex(t)\
+                         .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')\
+                         .limit(1)
+
+            if self.has_geometries():
+                sql = self._add_geometry_columns(_country_base_query(), t.c.geometry)
+            else:
+                sql = sa.lambda_stmt(_country_base_query)

 
 

-            address_row = (await self.conn.execute(sql)).one_or_none()
+            address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()

 
 

-        return nres.create_from_placex_row(_get_closest_row(address_row, other_row))
+        return address_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.
         """
 
 
     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)
+        log().function('reverse_lookup', coord=coord, params=self.params)
+

 
 

+        self.bind_params['wkt'] = f'POINT({coord[0]} {coord[1]})'

 
 

-        wkt = WKTElement(f'POINT({coord[0]} {coord[1]})', srid=4326)
+        row: Optional[SaRow] = None
+        row_func: RowFunc = nres.create_from_placex_row
+
+        if self.max_rank >= 26:
+            row, tmp_row_func = await self.lookup_street_poi()
+            if row is not None:
+                row_func = tmp_row_func
+
+        if row is None:
+            if self.restrict_to_country_areas:
+                ccodes = await self.lookup_country_codes()
+                if not ccodes:
+                    return None
+            else:
+                ccodes = []

 
 

-        result: Optional[ReverseResult] = None
+            if self.max_rank > 4:
+                row = await self.lookup_area()
+            if row is None and self.layer_enabled(DataLayer.ADDRESS):
+                row = await self.lookup_country(ccodes)

 
 

-        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)
+        result = row_func(row, nres.ReverseResult)

         if result is not None:
         if result is not None:

-            await nres.add_result_details(self.conn, result, self.details)
+            assert row is not None
+            result.distance = row.distance
+            if hasattr(row, 'bbox'):
+                result.bbox = Bbox.from_wkb(row.bbox)
+            await nres.add_result_details(self.conn, [result], self.params)

 
         return result
 
         return result