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