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