nominatim/api/results.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 Dataclasses for search results and helper functions to fill them.
   9
  10 Data classes are part of the public API while the functions are for
  11 internal use only. That's why they are implemented as free-standing functions
  12 instead of member functions.
  13 """
  14 from typing import Optional, Tuple, Dict, Sequence, TypeVar, Type, List, Any
  15 import enum
  16 import dataclasses
  17 import datetime as dt
  18
  19 import sqlalchemy as sa
  20
  21 from nominatim.typing import SaSelect, SaRow
  22 from nominatim.api.types import Point, Bbox, LookupDetails
  23 from nominatim.api.connection import SearchConnection
  24 from nominatim.api.logging import log
  25 from nominatim.api.localization import Locales
  26
  27 # This file defines complex result data classes.
  28 # pylint: disable=too-many-instance-attributes
  29
  30 def _mingle_name_tags(names: Optional[Dict[str, str]]) -> Optional[Dict[str, str]]:
  31     """ Mix-in names from linked places, so that they show up
  32         as standard names where necessary.
  33     """
  34     if not names:
  35         return None
  36
  37     out = {}
  38     for k, v in names.items():
  39         if k.startswith('_place_'):
  40             outkey = k[7:]
  41             out[k if outkey in names else outkey] = v
  42         else:
  43             out[k] = v
  44
  45     return out
  46
  47
  48 class SourceTable(enum.Enum):
  49     """ Enumeration of kinds of results.
  50     """
  51     PLACEX = 1
  52     OSMLINE = 2
  53     TIGER = 3
  54     POSTCODE = 4
  55     COUNTRY = 5
  56
  57
  58 @dataclasses.dataclass
  59 class AddressLine:
  60     """ Detailed information about a related place.
  61     """
  62     place_id: Optional[int]
  63     osm_object: Optional[Tuple[str, int]]
  64     category: Tuple[str, str]
  65     names: Dict[str, str]
  66     extratags: Optional[Dict[str, str]]
  67
  68     admin_level: Optional[int]
  69     fromarea: bool
  70     isaddress: bool
  71     rank_address: int
  72     distance: float
  73
  74     local_name: Optional[str] = None
  75
  76
  77 class AddressLines(List[AddressLine]):
  78     """ Sequence of address lines order in descending order by their rank.
  79     """
  80
  81     def localize(self, locales: Locales) -> List[str]:
  82         """ Set the local name of address parts according to the chosen
  83             locale. Return the list of local names without duplications.
  84
  85             Only address parts that are marked as isaddress are localized
  86             and returned.
  87         """
  88         label_parts: List[str] = []
  89
  90         for line in self:
  91             if line.isaddress and line.names:
  92                 line.local_name = locales.display_name(line.names)
  93                 if not label_parts or label_parts[-1] != line.local_name:
  94                     label_parts.append(line.local_name)
  95
  96         return label_parts
  97
  98
  99
 100 @dataclasses.dataclass
 101 class WordInfo:
 102     """ Detailed information about a search term.
 103     """
 104     word_id: int
 105     word_token: str
 106     word: Optional[str] = None
 107
 108
 109 WordInfos = Sequence[WordInfo]
 110
 111
 112 @dataclasses.dataclass
 113 class BaseResult:
 114     """ Data class collecting information common to all
 115         types of search results.
 116     """
 117     source_table: SourceTable
 118     category: Tuple[str, str]
 119     centroid: Point
 120
 121     place_id : Optional[int] = None
 122     osm_object: Optional[Tuple[str, int]] = None
 123
 124     locale_name: Optional[str] = None
 125     display_name: Optional[str] = None
 126
 127     names: Optional[Dict[str, str]] = None
 128     address: Optional[Dict[str, str]] = None
 129     extratags: Optional[Dict[str, str]] = None
 130
 131     housenumber: Optional[str] = None
 132     postcode: Optional[str] = None
 133     wikipedia: Optional[str] = None
 134
 135     rank_address: int = 30
 136     rank_search: int = 30
 137     importance: Optional[float] = None
 138
 139     country_code: Optional[str] = None
 140
 141     address_rows: Optional[AddressLines] = None
 142     linked_rows: Optional[AddressLines] = None
 143     parented_rows: Optional[AddressLines] = None
 144     name_keywords: Optional[WordInfos] = None
 145     address_keywords: Optional[WordInfos] = None
 146
 147     geometry: Dict[str, str] = dataclasses.field(default_factory=dict)
 148
 149     @property
 150     def lat(self) -> float:
 151         """ Get the latitude (or y) of the center point of the place.
 152         """
 153         return self.centroid[1]
 154
 155
 156     @property
 157     def lon(self) -> float:
 158         """ Get the longitude (or x) of the center point of the place.
 159         """
 160         return self.centroid[0]
 161
 162
 163     def calculated_importance(self) -> float:
 164         """ Get a valid importance value. This is either the stored importance
 165             of the value or an artificial value computed from the place's
 166             search rank.
 167         """
 168         return self.importance or (0.7500001 - (self.rank_search/40.0))
 169
 170
 171     def localize(self, locales: Locales) -> None:
 172         """ Fill the locale_name and the display_name field for the
 173             place and, if available, its address information.
 174         """
 175         self.locale_name = locales.display_name(self.names)
 176         if self.address_rows:
 177             self.display_name = ', '.join(self.address_rows.localize(locales))
 178         else:
 179             self.display_name = self.locale_name
 180
 181
 182
 183 BaseResultT = TypeVar('BaseResultT', bound=BaseResult)
 184
 185 @dataclasses.dataclass
 186 class DetailedResult(BaseResult):
 187     """ A search result with more internal information from the database
 188         added.
 189     """
 190     parent_place_id: Optional[int] = None
 191     linked_place_id: Optional[int] = None
 192     admin_level: int = 15
 193     indexed_date: Optional[dt.datetime] = None
 194
 195
 196 @dataclasses.dataclass
 197 class ReverseResult(BaseResult):
 198     """ A search result for reverse geocoding.
 199     """
 200     distance: Optional[float] = None
 201     bbox: Optional[Bbox] = None
 202
 203
 204 class ReverseResults(List[ReverseResult]):
 205     """ Sequence of reverse lookup results ordered by distance.
 206         May be empty when no result was found.
 207     """
 208
 209
 210 @dataclasses.dataclass
 211 class SearchResult(BaseResult):
 212     """ A search result for forward geocoding.
 213     """
 214     bbox: Optional[Bbox] = None
 215     accuracy: float = 0.0
 216
 217
 218     @property
 219     def ranking(self) -> float:
 220         """ Return the ranking, a combined measure of accuracy and importance.
 221         """
 222         return (self.accuracy if self.accuracy is not None else 1) \
 223                - self.calculated_importance()
 224
 225
 226 class SearchResults(List[SearchResult]):
 227     """ Sequence of forward lookup results ordered by relevance.
 228         May be empty when no result was found.
 229     """
 230
 231     def localize(self, locales: Locales) -> None:
 232         """ Apply the given locales to all results.
 233         """
 234         for result in self:
 235             result.localize(locales)
 236
 237
 238 def _filter_geometries(row: SaRow) -> Dict[str, str]:
 239     return {k[9:]: v for k, v in row._mapping.items() # pylint: disable=W0212
 240             if k.startswith('geometry_')}
 241
 242
 243 def create_from_placex_row(row: Optional[SaRow],
 244                            class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
 245     """ Construct a new result and add the data from the result row
 246         from the placex table. 'class_type' defines the type of result
 247         to return. Returns None if the row is None.
 248     """
 249     if row is None:
 250         return None
 251
 252     return class_type(source_table=SourceTable.PLACEX,
 253                       place_id=row.place_id,
 254                       osm_object=(row.osm_type, row.osm_id),
 255                       category=(row.class_, row.type),
 256                       names=_mingle_name_tags(row.name),
 257                       address=row.address,
 258                       extratags=row.extratags,
 259                       housenumber=row.housenumber,
 260                       postcode=row.postcode,
 261                       wikipedia=row.wikipedia,
 262                       rank_address=row.rank_address,
 263                       rank_search=row.rank_search,
 264                       importance=row.importance,
 265                       country_code=row.country_code,
 266                       centroid=Point.from_wkb(row.centroid),
 267                       geometry=_filter_geometries(row))
 268
 269
 270 def create_from_osmline_row(row: Optional[SaRow],
 271                             class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
 272     """ Construct a new result and add the data from the result row
 273         from the address interpolation table osmline. 'class_type' defines
 274         the type of result to return. Returns None if the row is None.
 275
 276         If the row contains a housenumber, then the housenumber is filled out.
 277         Otherwise the result contains the interpolation information in extratags.
 278     """
 279     if row is None:
 280         return None
 281
 282     hnr = getattr(row, 'housenumber', None)
 283
 284     res = class_type(source_table=SourceTable.OSMLINE,
 285                      place_id=row.place_id,
 286                      osm_object=('W', row.osm_id),
 287                      category=('place', 'houses' if hnr is None else 'house'),
 288                      address=row.address,
 289                      postcode=row.postcode,
 290                      country_code=row.country_code,
 291                      centroid=Point.from_wkb(row.centroid),
 292                      geometry=_filter_geometries(row))
 293
 294     if hnr is None:
 295         res.extratags = {'startnumber': str(row.startnumber),
 296                          'endnumber': str(row.endnumber),
 297                          'step': str(row.step)}
 298     else:
 299         res.housenumber = str(hnr)
 300
 301     return res
 302
 303
 304 def create_from_tiger_row(row: Optional[SaRow],
 305                           class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
 306     """ Construct a new result and add the data from the result row
 307         from the Tiger data interpolation table. 'class_type' defines
 308         the type of result to return. Returns None if the row is None.
 309
 310         If the row contains a housenumber, then the housenumber is filled out.
 311         Otherwise the result contains the interpolation information in extratags.
 312     """
 313     if row is None:
 314         return None
 315
 316     hnr = getattr(row, 'housenumber', None)
 317
 318     res = class_type(source_table=SourceTable.TIGER,
 319                      place_id=row.place_id,
 320                      osm_object=(row.osm_type, row.osm_id),
 321                      category=('place', 'houses' if hnr is None else 'house'),
 322                      postcode=row.postcode,
 323                      country_code='us',
 324                      centroid=Point.from_wkb(row.centroid),
 325                      geometry=_filter_geometries(row))
 326
 327     if hnr is None:
 328         res.extratags = {'startnumber': str(row.startnumber),
 329                          'endnumber': str(row.endnumber),
 330                          'step': str(row.step)}
 331     else:
 332         res.housenumber = str(hnr)
 333
 334     return res
 335
 336
 337 def create_from_postcode_row(row: Optional[SaRow],
 338                           class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
 339     """ Construct a new result and add the data from the result row
 340         from the postcode table. 'class_type' defines
 341         the type of result to return. Returns None if the row is None.
 342     """
 343     if row is None:
 344         return None
 345
 346     return class_type(source_table=SourceTable.POSTCODE,
 347                       place_id=row.place_id,
 348                       category=('place', 'postcode'),
 349                       names={'ref': row.postcode},
 350                       rank_search=row.rank_search,
 351                       rank_address=row.rank_address,
 352                       country_code=row.country_code,
 353                       centroid=Point.from_wkb(row.centroid),
 354                       geometry=_filter_geometries(row))
 355
 356
 357 def create_from_country_row(row: Optional[SaRow],
 358                         class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
 359     """ Construct a new result and add the data from the result row
 360         from the fallback country tables. 'class_type' defines
 361         the type of result to return. Returns None if the row is None.
 362     """
 363     if row is None:
 364         return None
 365
 366     return class_type(source_table=SourceTable.COUNTRY,
 367                       category=('place', 'country'),
 368                       centroid=Point.from_wkb(row.centroid),
 369                       names=row.name,
 370                       rank_address=4, rank_search=4,
 371                       country_code=row.country_code)
 372
 373
 374 async def add_result_details(conn: SearchConnection, results: List[BaseResultT],
 375                              details: LookupDetails) -> None:
 376     """ Retrieve more details from the database according to the
 377         parameters specified in 'details'.
 378     """
 379     if results:
 380         log().section('Query details for result')
 381         if details.address_details:
 382             log().comment('Query address details')
 383             await complete_address_details(conn, results)
 384         if details.linked_places:
 385             log().comment('Query linked places')
 386             for result in results:
 387                 await complete_linked_places(conn, result)
 388         if details.parented_places:
 389             log().comment('Query parent places')
 390             for result in results:
 391                 await complete_parented_places(conn, result)
 392         if details.keywords:
 393             log().comment('Query keywords')
 394             for result in results:
 395                 await complete_keywords(conn, result)
 396
 397
 398 def _result_row_to_address_row(row: SaRow) -> AddressLine:
 399     """ Create a new AddressLine from the results of a datbase query.
 400     """
 401     extratags: Dict[str, str] = getattr(row, 'extratags', {})
 402     if hasattr(row, 'place_type') and row.place_type:
 403         extratags['place'] = row.place_type
 404
 405     names = _mingle_name_tags(row.name) or {}
 406     if getattr(row, 'housenumber', None) is not None:
 407         names['housenumber'] = row.housenumber
 408
 409     return AddressLine(place_id=row.place_id,
 410                        osm_object=None if row.osm_type is None else (row.osm_type, row.osm_id),
 411                        category=(getattr(row, 'class'), row.type),
 412                        names=names,
 413                        extratags=extratags,
 414                        admin_level=row.admin_level,
 415                        fromarea=row.fromarea,
 416                        isaddress=getattr(row, 'isaddress', True),
 417                        rank_address=row.rank_address,
 418                        distance=row.distance)
 419
 420
 421 async def complete_address_details(conn: SearchConnection, results: List[BaseResultT]) -> None:
 422     """ Retrieve information about places that make up the address of the result.
 423     """
 424     def get_hnr(result: BaseResult) -> Tuple[int, int]:
 425         housenumber = -1
 426         if result.source_table in (SourceTable.TIGER, SourceTable.OSMLINE):
 427             if result.housenumber is not None:
 428                 housenumber = int(result.housenumber)
 429             elif result.extratags is not None and 'startnumber' in result.extratags:
 430                 # details requests do not come with a specific house number
 431                 housenumber = int(result.extratags['startnumber'])
 432         assert result.place_id
 433         return result.place_id, housenumber
 434
 435     data: List[Tuple[Any, ...]] = [get_hnr(r) for r in results if r.place_id]
 436
 437     if not data:
 438         return
 439
 440     values = sa.values(sa.column('place_id', type_=sa.Integer),
 441                        sa.column('housenumber', type_=sa.Integer),
 442                        name='places',
 443                        literal_binds=True).data(data)
 444
 445     sfn = sa.func.get_addressdata(values.c.place_id, values.c.housenumber)\
 446                 .table_valued( # type: ignore[no-untyped-call]
 447                     sa.column('place_id', type_=sa.Integer),
 448                     'osm_type',
 449                     sa.column('osm_id', type_=sa.BigInteger),
 450                     sa.column('name', type_=conn.t.types.Composite),
 451                     'class', 'type', 'place_type',
 452                     sa.column('admin_level', type_=sa.Integer),
 453                     sa.column('fromarea', type_=sa.Boolean),
 454                     sa.column('isaddress', type_=sa.Boolean),
 455                     sa.column('rank_address', type_=sa.SmallInteger),
 456                     sa.column('distance', type_=sa.Float),
 457                     joins_implicitly=True)
 458
 459     sql = sa.select(values.c.place_id.label('result_place_id'), sfn)\
 460             .order_by(values.c.place_id,
 461                       sa.column('rank_address').desc(),
 462                       sa.column('isaddress').desc())
 463
 464     current_result = None
 465     for row in await conn.execute(sql):
 466         if current_result is None or row.result_place_id != current_result.place_id:
 467             for result in results:
 468                 if result.place_id == row.result_place_id:
 469                     current_result = result
 470                     break
 471             else:
 472                 assert False
 473             current_result.address_rows = AddressLines()
 474         current_result.address_rows.append(_result_row_to_address_row(row))
 475
 476
 477 # pylint: disable=consider-using-f-string
 478 def _placex_select_address_row(conn: SearchConnection,
 479                                centroid: Point) -> SaSelect:
 480     t = conn.t.placex
 481     return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
 482                      t.c.class_.label('class'), t.c.type,
 483                      t.c.admin_level, t.c.housenumber,
 484                      sa.literal_column("""ST_GeometryType(geometry) in
 485                                         ('ST_Polygon','ST_MultiPolygon')""").label('fromarea'),
 486                      t.c.rank_address,
 487                      sa.literal_column(
 488                          """ST_DistanceSpheroid(geometry, 'SRID=4326;POINT(%f %f)'::geometry,
 489                               'SPHEROID["WGS 84",6378137,298.257223563, AUTHORITY["EPSG","7030"]]')
 490                          """ % centroid).label('distance'))
 491
 492
 493 async def complete_linked_places(conn: SearchConnection, result: BaseResult) -> None:
 494     """ Retrieve information about places that link to the result.
 495     """
 496     result.linked_rows = AddressLines()
 497     if result.source_table != SourceTable.PLACEX:
 498         return
 499
 500     sql = _placex_select_address_row(conn, result.centroid)\
 501             .where(conn.t.placex.c.linked_place_id == result.place_id)
 502
 503     for row in await conn.execute(sql):
 504         result.linked_rows.append(_result_row_to_address_row(row))
 505
 506
 507 async def complete_keywords(conn: SearchConnection, result: BaseResult) -> None:
 508     """ Retrieve information about the search terms used for this place.
 509
 510         Requires that the query analyzer was initialised to get access to
 511         the word table.
 512     """
 513     t = conn.t.search_name
 514     sql = sa.select(t.c.name_vector, t.c.nameaddress_vector)\
 515             .where(t.c.place_id == result.place_id)
 516
 517     result.name_keywords = []
 518     result.address_keywords = []
 519
 520     t = conn.t.meta.tables['word']
 521     sel = sa.select(t.c.word_id, t.c.word_token, t.c.word)
 522
 523     for name_tokens, address_tokens in await conn.execute(sql):
 524         for row in await conn.execute(sel.where(t.c.word_id == sa.any_(name_tokens))):
 525             result.name_keywords.append(WordInfo(*row))
 526
 527         for row in await conn.execute(sel.where(t.c.word_id == sa.any_(address_tokens))):
 528             result.address_keywords.append(WordInfo(*row))
 529
 530
 531 async def complete_parented_places(conn: SearchConnection, result: BaseResult) -> None:
 532     """ Retrieve information about places that the result provides the
 533         address for.
 534     """
 535     result.parented_rows = AddressLines()
 536     if result.source_table != SourceTable.PLACEX:
 537         return
 538
 539     sql = _placex_select_address_row(conn, result.centroid)\
 540             .where(conn.t.placex.c.parent_place_id == result.place_id)\
 541             .where(conn.t.placex.c.rank_search == 30)
 542
 543     for row in await conn.execute(sql):
 544         result.parented_rows.append(_result_row_to_address_row(row))