]> git.openstreetmap.org Git - nominatim.git/blob - nominatim/api/search/db_searches.py
fix linting issues
[nominatim.git] / nominatim / api / search / db_searches.py
1 # SPDX-License-Identifier: GPL-3.0-or-later
2 #
3 # This file is part of Nominatim. (https://nominatim.org)
4 #
5 # Copyright (C) 2023 by the Nominatim developer community.
6 # For a full list of authors see the git log.
7 """
8 Implementation of the acutal database accesses for forward search.
9 """
10 from typing import List, Tuple, AsyncIterator, Dict, Any
11 import abc
12
13 import sqlalchemy as sa
14 from sqlalchemy.dialects.postgresql import ARRAY, array_agg
15
16 from nominatim.typing import SaFromClause, SaScalarSelect, SaColumn, \
17                              SaExpression, SaSelect, SaRow, SaBind
18 from nominatim.api.connection import SearchConnection
19 from nominatim.api.types import SearchDetails, DataLayer, GeometryFormat, Bbox
20 import nominatim.api.results as nres
21 from nominatim.api.search.db_search_fields import SearchData, WeightedCategories
22 from nominatim.db.sqlalchemy_types import Geometry
23
24 #pylint: disable=singleton-comparison,not-callable
25 #pylint: disable=too-many-branches,too-many-arguments,too-many-locals,too-many-statements
26
27 def _details_to_bind_params(details: SearchDetails) -> Dict[str, Any]:
28     """ Create a dictionary from search parameters that can be used
29         as bind parameter for SQL execute.
30     """
31     return {'limit': details.max_results,
32             'min_rank': details.min_rank,
33             'max_rank': details.max_rank,
34             'viewbox': details.viewbox,
35             'viewbox2': details.viewbox_x2,
36             'near': details.near,
37             'near_radius': details.near_radius,
38             'excluded': details.excluded,
39             'countries': details.countries}
40
41
42 LIMIT_PARAM: SaBind = sa.bindparam('limit')
43 MIN_RANK_PARAM: SaBind = sa.bindparam('min_rank')
44 MAX_RANK_PARAM: SaBind = sa.bindparam('max_rank')
45 VIEWBOX_PARAM: SaBind = sa.bindparam('viewbox', type_=Geometry)
46 VIEWBOX2_PARAM: SaBind = sa.bindparam('viewbox2', type_=Geometry)
47 NEAR_PARAM: SaBind = sa.bindparam('near', type_=Geometry)
48 NEAR_RADIUS_PARAM: SaBind = sa.bindparam('near_radius')
49 EXCLUDED_PARAM: SaBind = sa.bindparam('excluded')
50 COUNTRIES_PARAM: SaBind = sa.bindparam('countries')
51
52 def _select_placex(t: SaFromClause) -> SaSelect:
53     return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
54                      t.c.class_, t.c.type,
55                      t.c.address, t.c.extratags,
56                      t.c.housenumber, t.c.postcode, t.c.country_code,
57                      t.c.importance, t.c.wikipedia,
58                      t.c.parent_place_id, t.c.rank_address, t.c.rank_search,
59                      t.c.centroid,
60                      t.c.geometry.ST_Expand(0).label('bbox'))
61
62
63 def _add_geometry_columns(sql: SaSelect, col: SaColumn, details: SearchDetails) -> SaSelect:
64     if not details.geometry_output:
65         return sql
66
67     out = []
68
69     if details.geometry_simplification > 0.0:
70         col = sa.func.ST_SimplifyPreserveTopology(col, details.geometry_simplification)
71
72     if details.geometry_output & GeometryFormat.GEOJSON:
73         out.append(sa.func.ST_AsGeoJSON(col).label('geometry_geojson'))
74     if details.geometry_output & GeometryFormat.TEXT:
75         out.append(sa.func.ST_AsText(col).label('geometry_text'))
76     if details.geometry_output & GeometryFormat.KML:
77         out.append(sa.func.ST_AsKML(col).label('geometry_kml'))
78     if details.geometry_output & GeometryFormat.SVG:
79         out.append(sa.func.ST_AsSVG(col).label('geometry_svg'))
80
81     return sql.add_columns(*out)
82
83
84 def _make_interpolation_subquery(table: SaFromClause, inner: SaFromClause,
85                                  numerals: List[int], details: SearchDetails) -> SaScalarSelect:
86     all_ids = array_agg(table.c.place_id) # type: ignore[no-untyped-call]
87     sql = sa.select(all_ids).where(table.c.parent_place_id == inner.c.place_id)
88
89     if len(numerals) == 1:
90         sql = sql.where(sa.between(numerals[0], table.c.startnumber, table.c.endnumber))\
91                  .where((numerals[0] - table.c.startnumber) % table.c.step == 0)
92     else:
93         sql = sql.where(sa.or_(
94                 *(sa.and_(sa.between(n, table.c.startnumber, table.c.endnumber),
95                           (n - table.c.startnumber) % table.c.step == 0)
96                   for n in numerals)))
97
98     if details.excluded:
99         sql = sql.where(table.c.place_id.not_in(EXCLUDED_PARAM))
100
101     return sql.scalar_subquery()
102
103
104 def _filter_by_layer(table: SaFromClause, layers: DataLayer) -> SaColumn:
105     orexpr: List[SaExpression] = []
106     if layers & DataLayer.ADDRESS and layers & DataLayer.POI:
107         orexpr.append(table.c.rank_address.between(1, 30))
108     elif layers & DataLayer.ADDRESS:
109         orexpr.append(table.c.rank_address.between(1, 29))
110         orexpr.append(sa.and_(table.c.rank_address == 30,
111                               sa.or_(table.c.housenumber != None,
112                                      table.c.address.has_key('housename'))))
113     elif layers & DataLayer.POI:
114         orexpr.append(sa.and_(table.c.rank_address == 30,
115                               table.c.class_.not_in(('place', 'building'))))
116
117     if layers & DataLayer.MANMADE:
118         exclude = []
119         if not layers & DataLayer.RAILWAY:
120             exclude.append('railway')
121         if not layers & DataLayer.NATURAL:
122             exclude.extend(('natural', 'water', 'waterway'))
123         orexpr.append(sa.and_(table.c.class_.not_in(tuple(exclude)),
124                               table.c.rank_address == 0))
125     else:
126         include = []
127         if layers & DataLayer.RAILWAY:
128             include.append('railway')
129         if layers & DataLayer.NATURAL:
130             include.extend(('natural', 'water', 'waterway'))
131         orexpr.append(sa.and_(table.c.class_.in_(tuple(include)),
132                               table.c.rank_address == 0))
133
134     if len(orexpr) == 1:
135         return orexpr[0]
136
137     return sa.or_(*orexpr)
138
139
140 def _interpolated_position(table: SaFromClause, nr: SaColumn) -> SaColumn:
141     pos = sa.cast(nr - table.c.startnumber, sa.Float) / (table.c.endnumber - table.c.startnumber)
142     return sa.case(
143             (table.c.endnumber == table.c.startnumber, table.c.linegeo.ST_Centroid()),
144             else_=table.c.linegeo.ST_LineInterpolatePoint(pos)).label('centroid')
145
146
147 async def _get_placex_housenumbers(conn: SearchConnection,
148                                    place_ids: List[int],
149                                    details: SearchDetails) -> AsyncIterator[nres.SearchResult]:
150     t = conn.t.placex
151     sql = _select_placex(t).where(t.c.place_id.in_(place_ids))
152
153     sql = _add_geometry_columns(sql, t.c.geometry, details)
154
155     for row in await conn.execute(sql):
156         result = nres.create_from_placex_row(row, nres.SearchResult)
157         assert result
158         result.bbox = Bbox.from_wkb(row.bbox)
159         yield result
160
161
162 async def _get_osmline(conn: SearchConnection, place_ids: List[int],
163                        numerals: List[int],
164                        details: SearchDetails) -> AsyncIterator[nres.SearchResult]:
165     t = conn.t.osmline
166     values = sa.values(sa.Column('nr', sa.Integer()), name='housenumber')\
167                .data([(n,) for n in numerals])
168     sql = sa.select(t.c.place_id, t.c.osm_id,
169                     t.c.parent_place_id, t.c.address,
170                     values.c.nr.label('housenumber'),
171                     _interpolated_position(t, values.c.nr),
172                     t.c.postcode, t.c.country_code)\
173             .where(t.c.place_id.in_(place_ids))\
174             .join(values, values.c.nr.between(t.c.startnumber, t.c.endnumber))
175
176     if details.geometry_output:
177         sub = sql.subquery()
178         sql = _add_geometry_columns(sa.select(sub), sub.c.centroid, details)
179
180     for row in await conn.execute(sql):
181         result = nres.create_from_osmline_row(row, nres.SearchResult)
182         assert result
183         yield result
184
185
186 async def _get_tiger(conn: SearchConnection, place_ids: List[int],
187                      numerals: List[int], osm_id: int,
188                      details: SearchDetails) -> AsyncIterator[nres.SearchResult]:
189     t = conn.t.tiger
190     values = sa.values(sa.Column('nr', sa.Integer()), name='housenumber')\
191                .data([(n,) for n in numerals])
192     sql = sa.select(t.c.place_id, t.c.parent_place_id,
193                     sa.literal('W').label('osm_type'),
194                     sa.literal(osm_id).label('osm_id'),
195                     values.c.nr.label('housenumber'),
196                     _interpolated_position(t, values.c.nr),
197                     t.c.postcode)\
198             .where(t.c.place_id.in_(place_ids))\
199             .join(values, values.c.nr.between(t.c.startnumber, t.c.endnumber))
200
201     if details.geometry_output:
202         sub = sql.subquery()
203         sql = _add_geometry_columns(sa.select(sub), sub.c.centroid, details)
204
205     for row in await conn.execute(sql):
206         result = nres.create_from_tiger_row(row, nres.SearchResult)
207         assert result
208         yield result
209
210
211 class AbstractSearch(abc.ABC):
212     """ Encapuslation of a single lookup in the database.
213     """
214
215     def __init__(self, penalty: float) -> None:
216         self.penalty = penalty
217
218     @abc.abstractmethod
219     async def lookup(self, conn: SearchConnection,
220                      details: SearchDetails) -> nres.SearchResults:
221         """ Find results for the search in the database.
222         """
223
224
225 class NearSearch(AbstractSearch):
226     """ Category search of a place type near the result of another search.
227     """
228     def __init__(self, penalty: float, categories: WeightedCategories,
229                  search: AbstractSearch) -> None:
230         super().__init__(penalty)
231         self.search = search
232         self.categories = categories
233
234
235     async def lookup(self, conn: SearchConnection,
236                      details: SearchDetails) -> nres.SearchResults:
237         """ Find results for the search in the database.
238         """
239         results = nres.SearchResults()
240         base = await self.search.lookup(conn, details)
241
242         if not base:
243             return results
244
245         base.sort(key=lambda r: (r.accuracy, r.rank_search))
246         max_accuracy = base[0].accuracy + 0.5
247         base = nres.SearchResults(r for r in base if r.source_table == nres.SourceTable.PLACEX
248                                                      and r.accuracy <= max_accuracy
249                                                      and r.bbox and r.bbox.area < 20)
250
251         if base:
252             baseids = [b.place_id for b in base[:5] if b.place_id]
253
254             for category, penalty in self.categories:
255                 await self.lookup_category(results, conn, baseids, category, penalty, details)
256                 if len(results) >= details.max_results:
257                     break
258
259         return results
260
261
262     async def lookup_category(self, results: nres.SearchResults,
263                               conn: SearchConnection, ids: List[int],
264                               category: Tuple[str, str], penalty: float,
265                               details: SearchDetails) -> None:
266         """ Find places of the given category near the list of
267             place ids and add the results to 'results'.
268         """
269         table = await conn.get_class_table(*category)
270
271         t = conn.t.placex.alias('p')
272         tgeom = conn.t.placex.alias('pgeom')
273
274         sql = _select_placex(t).where(tgeom.c.place_id.in_(ids))\
275                                .where(t.c.class_ == category[0])\
276                                .where(t.c.type == category[1])
277
278         if table is None:
279             # No classtype table available, do a simplified lookup in placex.
280             sql = sql.join(tgeom, t.c.geometry.ST_DWithin(tgeom.c.centroid, 0.01))\
281                      .order_by(tgeom.c.centroid.ST_Distance(t.c.centroid))
282         else:
283             # Use classtype table. We can afford to use a larger
284             # radius for the lookup.
285             sql = sql.join(table, t.c.place_id == table.c.place_id)\
286                      .join(tgeom,
287                            sa.case((sa.and_(tgeom.c.rank_address < 9,
288                                             tgeom.c.geometry.is_area()),
289                                     tgeom.c.geometry.ST_Contains(table.c.centroid)),
290                                    else_ = tgeom.c.centroid.ST_DWithin(table.c.centroid, 0.05)))\
291                      .order_by(tgeom.c.centroid.ST_Distance(table.c.centroid))
292
293         sql = sql.where(t.c.rank_address.between(MIN_RANK_PARAM, MAX_RANK_PARAM))
294         if details.countries:
295             sql = sql.where(t.c.country_code.in_(COUNTRIES_PARAM))
296         if details.excluded:
297             sql = sql.where(t.c.place_id.not_in(EXCLUDED_PARAM))
298         if details.layers is not None:
299             sql = sql.where(_filter_by_layer(t, details.layers))
300
301         sql = sql.limit(LIMIT_PARAM)
302         for row in await conn.execute(sql, _details_to_bind_params(details)):
303             result = nres.create_from_placex_row(row, nres.SearchResult)
304             assert result
305             result.accuracy = self.penalty + penalty
306             result.bbox = Bbox.from_wkb(row.bbox)
307             results.append(result)
308
309
310
311 class PoiSearch(AbstractSearch):
312     """ Category search in a geographic area.
313     """
314     def __init__(self, sdata: SearchData) -> None:
315         super().__init__(sdata.penalty)
316         self.categories = sdata.qualifiers
317         self.countries = sdata.countries
318
319
320     async def lookup(self, conn: SearchConnection,
321                      details: SearchDetails) -> nres.SearchResults:
322         """ Find results for the search in the database.
323         """
324         bind_params = _details_to_bind_params(details)
325         t = conn.t.placex
326
327         rows: List[SaRow] = []
328
329         if details.near and details.near_radius is not None and details.near_radius < 0.2:
330             # simply search in placex table
331             sql = _select_placex(t) \
332                       .where(t.c.linked_place_id == None) \
333                       .where(t.c.geometry.ST_DWithin(NEAR_PARAM, NEAR_RADIUS_PARAM)) \
334                       .order_by(t.c.centroid.ST_Distance(NEAR_PARAM))
335
336             if self.countries:
337                 sql = sql.where(t.c.country_code.in_(self.countries.values))
338
339             if details.viewbox is not None and details.bounded_viewbox:
340                 sql = sql.where(t.c.geometry.intersects(VIEWBOX_PARAM))
341
342             classtype = self.categories.values
343             if len(classtype) == 1:
344                 sql = sql.where(t.c.class_ == classtype[0][0]) \
345                          .where(t.c.type == classtype[0][1])
346             else:
347                 sql = sql.where(sa.or_(*(sa.and_(t.c.class_ == cls, t.c.type == typ)
348                                          for cls, typ in classtype)))
349
350             sql = sql.limit(LIMIT_PARAM)
351             rows.extend(await conn.execute(sql, bind_params))
352         else:
353             # use the class type tables
354             for category in self.categories.values:
355                 table = await conn.get_class_table(*category)
356                 if table is not None:
357                     sql = _select_placex(t)\
358                                .join(table, t.c.place_id == table.c.place_id)\
359                                .where(t.c.class_ == category[0])\
360                                .where(t.c.type == category[1])
361
362                     if details.viewbox is not None and details.bounded_viewbox:
363                         sql = sql.where(table.c.centroid.intersects(VIEWBOX_PARAM))
364
365                     if details.near and details.near_radius is not None:
366                         sql = sql.order_by(table.c.centroid.ST_Distance(NEAR_PARAM))\
367                                  .where(table.c.centroid.ST_DWithin(NEAR_PARAM,
368                                                                     NEAR_RADIUS_PARAM))
369
370                     if self.countries:
371                         sql = sql.where(t.c.country_code.in_(self.countries.values))
372
373                     sql = sql.limit(LIMIT_PARAM)
374                     rows.extend(await conn.execute(sql, bind_params))
375
376         results = nres.SearchResults()
377         for row in rows:
378             result = nres.create_from_placex_row(row, nres.SearchResult)
379             assert result
380             result.accuracy = self.penalty + self.categories.get_penalty((row.class_, row.type))
381             result.bbox = Bbox.from_wkb(row.bbox)
382             results.append(result)
383
384         return results
385
386
387 class CountrySearch(AbstractSearch):
388     """ Search for a country name or country code.
389     """
390     def __init__(self, sdata: SearchData) -> None:
391         super().__init__(sdata.penalty)
392         self.countries = sdata.countries
393
394
395     async def lookup(self, conn: SearchConnection,
396                      details: SearchDetails) -> nres.SearchResults:
397         """ Find results for the search in the database.
398         """
399         t = conn.t.placex
400
401         sql = _select_placex(t)\
402                 .where(t.c.country_code.in_(self.countries.values))\
403                 .where(t.c.rank_address == 4)
404
405         sql = _add_geometry_columns(sql, t.c.geometry, details)
406
407         if details.excluded:
408             sql = sql.where(t.c.place_id.not_in(EXCLUDED_PARAM))
409
410         if details.viewbox is not None and details.bounded_viewbox:
411             sql = sql.where(t.c.geometry.intersects(VIEWBOX_PARAM))
412
413         if details.near is not None and details.near_radius is not None:
414             sql = sql.where(t.c.geometry.ST_DWithin(NEAR_PARAM, NEAR_RADIUS_PARAM))
415
416         results = nres.SearchResults()
417         for row in await conn.execute(sql, _details_to_bind_params(details)):
418             result = nres.create_from_placex_row(row, nres.SearchResult)
419             assert result
420             result.accuracy = self.penalty + self.countries.get_penalty(row.country_code, 5.0)
421             results.append(result)
422
423         return results or await self.lookup_in_country_table(conn, details)
424
425
426     async def lookup_in_country_table(self, conn: SearchConnection,
427                                       details: SearchDetails) -> nres.SearchResults:
428         """ Look up the country in the fallback country tables.
429         """
430         # Avoid the fallback search when this is a more search. Country results
431         # usually are in the first batch of results and it is not possible
432         # to exclude these fallbacks.
433         if details.excluded:
434             return nres.SearchResults()
435
436         t = conn.t.country_name
437         tgrid = conn.t.country_grid
438
439         sql = sa.select(tgrid.c.country_code,
440                         tgrid.c.geometry.ST_Centroid().ST_Collect().ST_Centroid()
441                               .label('centroid'))\
442                 .where(tgrid.c.country_code.in_(self.countries.values))\
443                 .group_by(tgrid.c.country_code)
444
445         if details.viewbox is not None and details.bounded_viewbox:
446             sql = sql.where(tgrid.c.geometry.intersects(VIEWBOX_PARAM))
447         if details.near is not None and details.near_radius is not None:
448             sql = sql.where(tgrid.c.geometry.ST_DWithin(NEAR_PARAM, NEAR_RADIUS_PARAM))
449
450         sub = sql.subquery('grid')
451
452         sql = sa.select(t.c.country_code,
453                         (t.c.name
454                          + sa.func.coalesce(t.c.derived_name,
455                                             sa.cast('', type_=conn.t.types.Composite))
456                         ).label('name'),
457                         sub.c.centroid)\
458                 .join(sub, t.c.country_code == sub.c.country_code)
459
460         results = nres.SearchResults()
461         for row in await conn.execute(sql, _details_to_bind_params(details)):
462             result = nres.create_from_country_row(row, nres.SearchResult)
463             assert result
464             result.accuracy = self.penalty + self.countries.get_penalty(row.country_code, 5.0)
465             results.append(result)
466
467         return results
468
469
470
471 class PostcodeSearch(AbstractSearch):
472     """ Search for a postcode.
473     """
474     def __init__(self, extra_penalty: float, sdata: SearchData) -> None:
475         super().__init__(sdata.penalty + extra_penalty)
476         self.countries = sdata.countries
477         self.postcodes = sdata.postcodes
478         self.lookups = sdata.lookups
479         self.rankings = sdata.rankings
480
481
482     async def lookup(self, conn: SearchConnection,
483                      details: SearchDetails) -> nres.SearchResults:
484         """ Find results for the search in the database.
485         """
486         t = conn.t.postcode
487
488         sql = sa.select(t.c.place_id, t.c.parent_place_id,
489                         t.c.rank_search, t.c.rank_address,
490                         t.c.postcode, t.c.country_code,
491                         t.c.geometry.label('centroid'))\
492                 .where(t.c.postcode.in_(self.postcodes.values))
493
494         sql = _add_geometry_columns(sql, t.c.geometry, details)
495
496         penalty: SaExpression = sa.literal(self.penalty)
497
498         if details.viewbox is not None:
499             if details.bounded_viewbox:
500                 sql = sql.where(t.c.geometry.intersects(VIEWBOX_PARAM))
501             else:
502                 penalty += sa.case((t.c.geometry.intersects(VIEWBOX_PARAM), 0.0),
503                                    (t.c.geometry.intersects(VIEWBOX2_PARAM), 1.0),
504                                    else_=2.0)
505
506         if details.near is not None:
507             if details.near_radius is not None:
508                 sql = sql.where(t.c.geometry.ST_DWithin(NEAR_PARAM, NEAR_RADIUS_PARAM))
509             sql = sql.order_by(t.c.geometry.ST_Distance(NEAR_PARAM))
510
511         if self.countries:
512             sql = sql.where(t.c.country_code.in_(self.countries.values))
513
514         if details.excluded:
515             sql = sql.where(t.c.place_id.not_in(EXCLUDED_PARAM))
516
517         if self.lookups:
518             assert len(self.lookups) == 1
519             assert self.lookups[0].lookup_type == 'restrict'
520             tsearch = conn.t.search_name
521             sql = sql.where(tsearch.c.place_id == t.c.parent_place_id)\
522                      .where(sa.func.array_cat(tsearch.c.name_vector,
523                                               tsearch.c.nameaddress_vector,
524                                               type_=ARRAY(sa.Integer))
525                                     .contains(self.lookups[0].tokens))
526
527         for ranking in self.rankings:
528             penalty += ranking.sql_penalty(conn.t.search_name)
529         penalty += sa.case(*((t.c.postcode == v, p) for v, p in self.postcodes),
530                        else_=1.0)
531
532
533         sql = sql.add_columns(penalty.label('accuracy'))
534         sql = sql.order_by('accuracy').limit(LIMIT_PARAM)
535
536         results = nres.SearchResults()
537         for row in await conn.execute(sql, _details_to_bind_params(details)):
538             result = nres.create_from_postcode_row(row, nres.SearchResult)
539             assert result
540             result.accuracy = row.accuracy
541             results.append(result)
542
543         return results
544
545
546
547 class PlaceSearch(AbstractSearch):
548     """ Generic search for an address or named place.
549     """
550     def __init__(self, extra_penalty: float, sdata: SearchData, expected_count: int) -> None:
551         super().__init__(sdata.penalty + extra_penalty)
552         self.countries = sdata.countries
553         self.postcodes = sdata.postcodes
554         self.housenumbers = sdata.housenumbers
555         self.qualifiers = sdata.qualifiers
556         self.lookups = sdata.lookups
557         self.rankings = sdata.rankings
558         self.expected_count = expected_count
559
560
561     async def lookup(self, conn: SearchConnection,
562                      details: SearchDetails) -> nres.SearchResults:
563         """ Find results for the search in the database.
564         """
565         t = conn.t.placex.alias('p')
566         tsearch = conn.t.search_name.alias('s')
567
568         sql = sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
569                         t.c.class_, t.c.type,
570                         t.c.address, t.c.extratags,
571                         t.c.housenumber, t.c.postcode, t.c.country_code,
572                         t.c.wikipedia,
573                         t.c.parent_place_id, t.c.rank_address, t.c.rank_search,
574                         t.c.centroid,
575                         t.c.geometry.ST_Expand(0).label('bbox'))\
576                 .where(t.c.place_id == tsearch.c.place_id)
577
578
579         sql = _add_geometry_columns(sql, t.c.geometry, details)
580
581         penalty: SaExpression = sa.literal(self.penalty)
582         for ranking in self.rankings:
583             penalty += ranking.sql_penalty(tsearch)
584
585         for lookup in self.lookups:
586             sql = sql.where(lookup.sql_condition(tsearch))
587
588         if self.countries:
589             sql = sql.where(tsearch.c.country_code.in_(self.countries.values))
590
591         if self.postcodes:
592             # if a postcode is given, don't search for state or country level objects
593             sql = sql.where(tsearch.c.address_rank > 9)
594             tpc = conn.t.postcode
595             if self.expected_count > 1000:
596                 # Many results expected. Restrict by postcode.
597                 sql = sql.where(sa.select(tpc.c.postcode)
598                                   .where(tpc.c.postcode.in_(self.postcodes.values))
599                                   .where(tsearch.c.centroid.ST_DWithin(tpc.c.geometry, 0.12))
600                                   .exists())
601
602             # Less results, only have a preference for close postcodes
603             pc_near = sa.select(sa.func.min(tpc.c.geometry.ST_Distance(tsearch.c.centroid)))\
604                       .where(tpc.c.postcode.in_(self.postcodes.values))\
605                       .scalar_subquery()
606             penalty += sa.case((t.c.postcode.in_(self.postcodes.values), 0.0),
607                                else_=sa.func.coalesce(pc_near, 2.0))
608
609         if details.viewbox is not None:
610             if details.bounded_viewbox:
611                 sql = sql.where(tsearch.c.centroid.intersects(VIEWBOX_PARAM))
612             else:
613                 penalty += sa.case((t.c.geometry.intersects(VIEWBOX_PARAM), 0.0),
614                                    (t.c.geometry.intersects(VIEWBOX2_PARAM), 1.0),
615                                    else_=2.0)
616
617         if details.near is not None:
618             if details.near_radius is not None:
619                 sql = sql.where(tsearch.c.centroid.ST_DWithin(NEAR_PARAM, NEAR_RADIUS_PARAM))
620             sql = sql.add_columns(-tsearch.c.centroid.ST_Distance(NEAR_PARAM)
621                                       .label('importance'))
622             sql = sql.order_by(sa.desc(sa.text('importance')))
623         else:
624             sql = sql.order_by(penalty - sa.case((tsearch.c.importance > 0, tsearch.c.importance),
625                                   else_=0.75001-(sa.cast(tsearch.c.search_rank, sa.Float())/40)))
626             sql = sql.add_columns(t.c.importance)
627
628
629         sql = sql.add_columns(penalty.label('accuracy'))\
630                  .order_by(sa.text('accuracy'))
631
632         if self.housenumbers:
633             hnr_regexp = f"\\m({'|'.join(self.housenumbers.values)})\\M"
634             sql = sql.where(tsearch.c.address_rank.between(16, 30))\
635                      .where(sa.or_(tsearch.c.address_rank < 30,
636                                   t.c.housenumber.op('~*')(hnr_regexp)))
637
638             # Cross check for housenumbers, need to do that on a rather large
639             # set. Worst case there are 40.000 main streets in OSM.
640             inner = sql.limit(10000).subquery()
641
642             # Housenumbers from placex
643             thnr = conn.t.placex.alias('hnr')
644             pid_list = array_agg(thnr.c.place_id) # type: ignore[no-untyped-call]
645             place_sql = sa.select(pid_list)\
646                           .where(thnr.c.parent_place_id == inner.c.place_id)\
647                           .where(thnr.c.housenumber.op('~*')(hnr_regexp))\
648                           .where(thnr.c.linked_place_id == None)\
649                           .where(thnr.c.indexed_status == 0)
650
651             if details.excluded:
652                 place_sql = place_sql.where(thnr.c.place_id.not_in(EXCLUDED_PARAM))
653             if self.qualifiers:
654                 place_sql = place_sql.where(self.qualifiers.sql_restrict(thnr))
655
656             numerals = [int(n) for n in self.housenumbers.values if n.isdigit()]
657             interpol_sql: SaExpression
658             tiger_sql: SaExpression
659             if numerals and \
660                (not self.qualifiers or ('place', 'house') in self.qualifiers.values):
661                 # Housenumbers from interpolations
662                 interpol_sql = _make_interpolation_subquery(conn.t.osmline, inner,
663                                                             numerals, details)
664                 # Housenumbers from Tiger
665                 tiger_sql = sa.case((inner.c.country_code == 'us',
666                                      _make_interpolation_subquery(conn.t.tiger, inner,
667                                                                   numerals, details)
668                                     ), else_=None)
669             else:
670                 interpol_sql = sa.literal_column('NULL')
671                 tiger_sql = sa.literal_column('NULL')
672
673             unsort = sa.select(inner, place_sql.scalar_subquery().label('placex_hnr'),
674                                interpol_sql.label('interpol_hnr'),
675                                tiger_sql.label('tiger_hnr')).subquery('unsort')
676             sql = sa.select(unsort)\
677                     .order_by(sa.case((unsort.c.placex_hnr != None, 1),
678                                       (unsort.c.interpol_hnr != None, 2),
679                                       (unsort.c.tiger_hnr != None, 3),
680                                       else_=4),
681                               unsort.c.accuracy)
682         else:
683             sql = sql.where(t.c.linked_place_id == None)\
684                      .where(t.c.indexed_status == 0)
685             if self.qualifiers:
686                 sql = sql.where(self.qualifiers.sql_restrict(t))
687             if details.excluded:
688                 sql = sql.where(tsearch.c.place_id.not_in(EXCLUDED_PARAM))
689             if details.min_rank > 0:
690                 sql = sql.where(sa.or_(tsearch.c.address_rank >= MIN_RANK_PARAM,
691                                        tsearch.c.search_rank >= MIN_RANK_PARAM))
692             if details.max_rank < 30:
693                 sql = sql.where(sa.or_(tsearch.c.address_rank <= MAX_RANK_PARAM,
694                                        tsearch.c.search_rank <= MAX_RANK_PARAM))
695             if details.layers is not None:
696                 sql = sql.where(_filter_by_layer(t, details.layers))
697
698         sql = sql.limit(LIMIT_PARAM)
699
700         results = nres.SearchResults()
701         for row in await conn.execute(sql, _details_to_bind_params(details)):
702             result = nres.create_from_placex_row(row, nres.SearchResult)
703             assert result
704             result.bbox = Bbox.from_wkb(row.bbox)
705             result.accuracy = row.accuracy
706             if not details.excluded or not result.place_id in details.excluded:
707                 results.append(result)
708
709             if self.housenumbers and row.rank_address < 30:
710                 if row.placex_hnr:
711                     subs = _get_placex_housenumbers(conn, row.placex_hnr, details)
712                 elif row.interpol_hnr:
713                     subs = _get_osmline(conn, row.interpol_hnr, numerals, details)
714                 elif row.tiger_hnr:
715                     subs = _get_tiger(conn, row.tiger_hnr, numerals, row.osm_id, details)
716                 else:
717                     subs = None
718
719                 if subs is not None:
720                     async for sub in subs:
721                         assert sub.housenumber
722                         sub.accuracy = result.accuracy
723                         if not any(nr in self.housenumbers.values
724                                    for nr in sub.housenumber.split(';')):
725                             sub.accuracy += 0.6
726                         results.append(sub)
727
728                 result.accuracy += 1.0 # penalty for missing housenumber
729
730         return results