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