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