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
  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
  26 # This file defines complex result data classes.
  27 # pylint: disable=too-many-instance-attributes
  28
  29 class SourceTable(enum.Enum):
  30     """ Enumeration of kinds of results.
  31     """
  32     PLACEX = 1
  33     OSMLINE = 2
  34     TIGER = 3
  35     POSTCODE = 4
  36     COUNTRY = 5
  37
  38
  39 @dataclasses.dataclass
  40 class AddressLine:
  41     """ Detailed information about a related place.
  42     """
  43     place_id: Optional[int]
  44     osm_object: Optional[Tuple[str, int]]
  45     category: Tuple[str, str]
  46     names: Dict[str, str]
  47     extratags: Optional[Dict[str, str]]
  48
  49     admin_level: Optional[int]
  50     fromarea: bool
  51     isaddress: bool
  52     rank_address: int
  53     distance: float
  54
  55
  56 AddressLines = Sequence[AddressLine]
  57
  58
  59 @dataclasses.dataclass
  60 class WordInfo:
  61     """ Detailed information about a search term.
  62     """
  63     word_id: int
  64     word_token: str
  65     word: Optional[str] = None
  66
  67
  68 WordInfos = Sequence[WordInfo]
  69
  70
  71 @dataclasses.dataclass
  72 class BaseResult:
  73     """ Data class collecting information common to all
  74         types of search results.
  75     """
  76     source_table: SourceTable
  77     category: Tuple[str, str]
  78     centroid: Point
  79
  80     place_id : Optional[int] = None
  81     osm_object: Optional[Tuple[str, int]] = None
  82
  83     names: Optional[Dict[str, str]] = None
  84     address: Optional[Dict[str, str]] = None
  85     extratags: Optional[Dict[str, str]] = None
  86
  87     housenumber: Optional[str] = None
  88     postcode: Optional[str] = None
  89     wikipedia: Optional[str] = None
  90
  91     rank_address: int = 30
  92     rank_search: int = 30
  93     importance: Optional[float] = None
  94
  95     country_code: Optional[str] = None
  96
  97     address_rows: Optional[AddressLines] = None
  98     linked_rows: Optional[AddressLines] = None
  99     parented_rows: Optional[AddressLines] = None
 100     name_keywords: Optional[WordInfos] = None
 101     address_keywords: Optional[WordInfos] = None
 102
 103     geometry: Dict[str, str] = dataclasses.field(default_factory=dict)
 104
 105     @property
 106     def lat(self) -> float:
 107         """ Get the latitude (or y) of the center point of the place.
 108         """
 109         return self.centroid[1]
 110
 111
 112     @property
 113     def lon(self) -> float:
 114         """ Get the longitude (or x) of the center point of the place.
 115         """
 116         return self.centroid[0]
 117
 118
 119     def calculated_importance(self) -> float:
 120         """ Get a valid importance value. This is either the stored importance
 121             of the value or an artificial value computed from the place's
 122             search rank.
 123         """
 124         return self.importance or (0.7500001 - (self.rank_search/40.0))
 125
 126 BaseResultT = TypeVar('BaseResultT', bound=BaseResult)
 127
 128 @dataclasses.dataclass
 129 class DetailedResult(BaseResult):
 130     """ A search result with more internal information from the database
 131         added.
 132     """
 133     parent_place_id: Optional[int] = None
 134     linked_place_id: Optional[int] = None
 135     admin_level: int = 15
 136     indexed_date: Optional[dt.datetime] = None
 137
 138
 139 @dataclasses.dataclass
 140 class ReverseResult(BaseResult):
 141     """ A search result for reverse geocoding.
 142     """
 143     distance: Optional[float] = None
 144     bbox: Optional[Bbox] = None
 145
 146
 147 def _filter_geometries(row: SaRow) -> Dict[str, str]:
 148     return {k[9:]: v for k, v in row._mapping.items() # pylint: disable=W0212
 149             if k.startswith('geometry_')}
 150
 151
 152 def create_from_placex_row(row: Optional[SaRow],
 153                            class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
 154     """ Construct a new result and add the data from the result row
 155         from the placex table. 'class_type' defines the type of result
 156         to return. Returns None if the row is None.
 157     """
 158     if row is None:
 159         return None
 160
 161     return class_type(source_table=SourceTable.PLACEX,
 162                       place_id=row.place_id,
 163                       osm_object=(row.osm_type, row.osm_id),
 164                       category=(row.class_, row.type),
 165                       names=row.name,
 166                       address=row.address,
 167                       extratags=row.extratags,
 168                       housenumber=row.housenumber,
 169                       postcode=row.postcode,
 170                       wikipedia=row.wikipedia,
 171                       rank_address=row.rank_address,
 172                       rank_search=row.rank_search,
 173                       importance=row.importance,
 174                       country_code=row.country_code,
 175                       centroid=Point.from_wkb(row.centroid.data),
 176                       geometry=_filter_geometries(row))
 177
 178
 179 def create_from_osmline_row(row: Optional[SaRow],
 180                             class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
 181     """ Construct a new result and add the data from the result row
 182         from the address interpolation table osmline. 'class_type' defines
 183         the type of result to return. Returns None if the row is None.
 184
 185         If the row contains a housenumber, then the housenumber is filled out.
 186         Otherwise the result contains the interpolation information in extratags.
 187     """
 188     if row is None:
 189         return None
 190
 191     hnr = getattr(row, 'housenumber', None)
 192
 193     res = class_type(source_table=SourceTable.OSMLINE,
 194                      place_id=row.place_id,
 195                      osm_object=('W', row.osm_id),
 196                      category=('place', 'houses' if hnr is None else 'house'),
 197                      address=row.address,
 198                      postcode=row.postcode,
 199                      country_code=row.country_code,
 200                      centroid=Point.from_wkb(row.centroid.data),
 201                      geometry=_filter_geometries(row))
 202
 203     if hnr is None:
 204         res.extratags = {'startnumber': str(row.startnumber),
 205                          'endnumber': str(row.endnumber),
 206                          'step': str(row.step)}
 207     else:
 208         res.housenumber = str(hnr)
 209
 210     return res
 211
 212
 213 def create_from_tiger_row(row: Optional[SaRow],
 214                           class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
 215     """ Construct a new result and add the data from the result row
 216         from the Tiger data interpolation table. 'class_type' defines
 217         the type of result to return. Returns None if the row is None.
 218
 219         If the row contains a housenumber, then the housenumber is filled out.
 220         Otherwise the result contains the interpolation information in extratags.
 221     """
 222     if row is None:
 223         return None
 224
 225     hnr = getattr(row, 'housenumber', None)
 226
 227     res = class_type(source_table=SourceTable.TIGER,
 228                      place_id=row.place_id,
 229                      category=('place', 'houses' if hnr is None else 'house'),
 230                      postcode=row.postcode,
 231                      country_code='us',
 232                      centroid=Point.from_wkb(row.centroid.data),
 233                      geometry=_filter_geometries(row))
 234
 235     if hnr is None:
 236         res.extratags = {'startnumber': str(row.startnumber),
 237                          'endnumber': str(row.endnumber),
 238                          'step': str(row.step)}
 239     else:
 240         res.housenumber = str(hnr)
 241
 242     return res
 243
 244
 245 def create_from_postcode_row(row: Optional[SaRow],
 246                           class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
 247     """ Construct a new result and add the data from the result row
 248         from the postcode table. 'class_type' defines
 249         the type of result to return. Returns None if the row is None.
 250     """
 251     if row is None:
 252         return None
 253
 254     return class_type(source_table=SourceTable.POSTCODE,
 255                       place_id=row.place_id,
 256                       category=('place', 'postcode'),
 257                       names={'ref': row.postcode},
 258                       rank_search=row.rank_search,
 259                       rank_address=row.rank_address,
 260                       country_code=row.country_code,
 261                       centroid=Point.from_wkb(row.centroid.data),
 262                       geometry=_filter_geometries(row))
 263
 264
 265 async def add_result_details(conn: SearchConnection, result: BaseResult,
 266                              details: LookupDetails) -> None:
 267     """ Retrieve more details from the database according to the
 268         parameters specified in 'details'.
 269     """
 270     log().section('Query details for result')
 271     if details.address_details:
 272         log().comment('Query address details')
 273         await complete_address_details(conn, result)
 274     if details.linked_places:
 275         log().comment('Query linked places')
 276         await complete_linked_places(conn, result)
 277     if details.parented_places:
 278         log().comment('Query parent places')
 279         await complete_parented_places(conn, result)
 280     if details.keywords:
 281         log().comment('Query keywords')
 282         await complete_keywords(conn, result)
 283
 284
 285 def _result_row_to_address_row(row: SaRow) -> AddressLine:
 286     """ Create a new AddressLine from the results of a datbase query.
 287     """
 288     extratags: Dict[str, str] = getattr(row, 'extratags', {})
 289     if 'place_type' in row:
 290         extratags['place_type'] = row.place_type
 291
 292     names = row.name
 293     if getattr(row, 'housenumber', None) is not None:
 294         if names is None:
 295             names = {}
 296         names['housenumber'] = row.housenumber
 297
 298     return AddressLine(place_id=row.place_id,
 299                        osm_object=None if row.osm_type is None else (row.osm_type, row.osm_id),
 300                        category=(getattr(row, 'class'), row.type),
 301                        names=names,
 302                        extratags=extratags,
 303                        admin_level=row.admin_level,
 304                        fromarea=row.fromarea,
 305                        isaddress=getattr(row, 'isaddress', True),
 306                        rank_address=row.rank_address,
 307                        distance=row.distance)
 308
 309
 310 async def complete_address_details(conn: SearchConnection, result: BaseResult) -> None:
 311     """ Retrieve information about places that make up the address of the result.
 312     """
 313     housenumber = -1
 314     if result.source_table in (SourceTable.TIGER, SourceTable.OSMLINE):
 315         if result.housenumber is not None:
 316             housenumber = int(result.housenumber)
 317         elif result.extratags is not None and 'startnumber' in result.extratags:
 318             # details requests do not come with a specific house number
 319             housenumber = int(result.extratags['startnumber'])
 320
 321     sfn = sa.func.get_addressdata(result.place_id, housenumber)\
 322             .table_valued( # type: ignore[no-untyped-call]
 323                 sa.column('place_id', type_=sa.Integer),
 324                 'osm_type',
 325                 sa.column('osm_id', type_=sa.BigInteger),
 326                 sa.column('name', type_=conn.t.types.Composite),
 327                 'class', 'type', 'place_type',
 328                 sa.column('admin_level', type_=sa.Integer),
 329                 sa.column('fromarea', type_=sa.Boolean),
 330                 sa.column('isaddress', type_=sa.Boolean),
 331                 sa.column('rank_address', type_=sa.SmallInteger),
 332                 sa.column('distance', type_=sa.Float))
 333     sql = sa.select(sfn).order_by(sa.column('rank_address').desc(),
 334                                   sa.column('isaddress').desc())
 335
 336     result.address_rows = []
 337     for row in await conn.execute(sql):
 338         result.address_rows.append(_result_row_to_address_row(row))
 339
 340
 341 # pylint: disable=consider-using-f-string
 342 def _placex_select_address_row(conn: SearchConnection,
 343                                centroid: Point) -> SaSelect:
 344     t = conn.t.placex
 345     return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
 346                      t.c.class_.label('class'), t.c.type,
 347                      t.c.admin_level, t.c.housenumber,
 348                      sa.literal_column("""ST_GeometryType(geometry) in
 349                                         ('ST_Polygon','ST_MultiPolygon')""").label('fromarea'),
 350                      t.c.rank_address,
 351                      sa.literal_column(
 352                          """ST_DistanceSpheroid(geometry, 'SRID=4326;POINT(%f %f)'::geometry,
 353                               'SPHEROID["WGS 84",6378137,298.257223563, AUTHORITY["EPSG","7030"]]')
 354                          """ % centroid).label('distance'))
 355
 356
 357 async def complete_linked_places(conn: SearchConnection, result: BaseResult) -> None:
 358     """ Retrieve information about places that link to the result.
 359     """
 360     result.linked_rows = []
 361     if result.source_table != SourceTable.PLACEX:
 362         return
 363
 364     sql = _placex_select_address_row(conn, result.centroid)\
 365             .where(conn.t.placex.c.linked_place_id == result.place_id)
 366
 367     for row in await conn.execute(sql):
 368         result.linked_rows.append(_result_row_to_address_row(row))
 369
 370
 371 async def complete_keywords(conn: SearchConnection, result: BaseResult) -> None:
 372     """ Retrieve information about the search terms used for this place.
 373     """
 374     t = conn.t.search_name
 375     sql = sa.select(t.c.name_vector, t.c.nameaddress_vector)\
 376             .where(t.c.place_id == result.place_id)
 377
 378     result.name_keywords = []
 379     result.address_keywords = []
 380     for name_tokens, address_tokens in await conn.execute(sql):
 381         t = conn.t.word
 382         sel = sa.select(t.c.word_id, t.c.word_token, t.c.word)
 383
 384         for row in await conn.execute(sel.where(t.c.word_id == sa.any_(name_tokens))):
 385             result.name_keywords.append(WordInfo(*row))
 386
 387         for row in await conn.execute(sel.where(t.c.word_id == sa.any_(address_tokens))):
 388             result.address_keywords.append(WordInfo(*row))
 389
 390
 391 async def complete_parented_places(conn: SearchConnection, result: BaseResult) -> None:
 392     """ Retrieve information about places that the result provides the
 393         address for.
 394     """
 395     result.parented_rows = []
 396     if result.source_table != SourceTable.PLACEX:
 397         return
 398
 399     sql = _placex_select_address_row(conn, result.centroid)\
 400             .where(conn.t.placex.c.parent_place_id == result.place_id)\
 401             .where(conn.t.placex.c.rank_search == 30)
 402
 403     for row in await conn.execute(sql):
 404         result.parented_rows.append(_result_row_to_address_row(row))