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