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     result = 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     return result
 169
 170
 171 def create_from_osmline_row(row: SaRow) -> SearchResult:
 172     """ Construct a new SearchResult and add the data from the result row
 173         from the osmline table.
 174     """
 175     result = SearchResult(source_table=SourceTable.OSMLINE,
 176                           place_id=row.place_id,
 177                           parent_place_id=row.parent_place_id,
 178                           osm_object=('W', row.osm_id),
 179                           category=('place', 'houses'),
 180                           address=row.address,
 181                           postcode=row.postcode,
 182                           extratags={'startnumber': str(row.startnumber),
 183                                      'endnumber': str(row.endnumber),
 184                                      'step': str(row.step)},
 185                           country_code=row.country_code,
 186                           indexed_date=getattr(row, 'indexed_date'),
 187                           centroid=Point(row.x, row.y),
 188                           geometry = _filter_geometries(row))
 189
 190     return result
 191
 192
 193 async def add_result_details(conn: SearchConnection, result: SearchResult,
 194                              details: LookupDetails) -> None:
 195     """ Retrieve more details from the database according to the
 196         parameters specified in 'details'.
 197     """
 198     if details.address_details:
 199         await complete_address_details(conn, result)
 200     if details.linked_places:
 201         await complete_linked_places(conn, result)
 202     if details.parented_places:
 203         await complete_parented_places(conn, result)
 204     if details.keywords:
 205         await complete_keywords(conn, result)
 206
 207
 208 def _result_row_to_address_row(row: SaRow) -> AddressLine:
 209     """ Create a new AddressLine from the results of a datbase query.
 210     """
 211     extratags: Dict[str, str] = getattr(row, 'extratags', {})
 212     if 'place_type' in row:
 213         extratags['place_type'] = row.place_type
 214
 215     names = row.name
 216     if getattr(row, 'housenumber', None) is not None:
 217         if names is None:
 218             names = {}
 219         names['housenumber'] = row.housenumber
 220
 221     return AddressLine(place_id=row.place_id,
 222                        osm_object=None if row.osm_type is None else (row.osm_type, row.osm_id),
 223                        category=(getattr(row, 'class'), row.type),
 224                        names=names,
 225                        extratags=extratags,
 226                        admin_level=row.admin_level,
 227                        fromarea=row.fromarea,
 228                        isaddress=getattr(row, 'isaddress', True),
 229                        rank_address=row.rank_address,
 230                        distance=row.distance)
 231
 232
 233 async def complete_address_details(conn: SearchConnection, result: SearchResult) -> None:
 234     """ Retrieve information about places that make up the address of the result.
 235     """
 236     housenumber = -1
 237     if result.source_table in (SourceTable.TIGER, SourceTable.OSMLINE):
 238         if result.housenumber is not None:
 239             housenumber = int(result.housenumber)
 240         elif result.extratags is not None and 'startnumber' in result.extratags:
 241             # details requests do not come with a specific house number
 242             housenumber = int(result.extratags['startnumber'])
 243
 244     sfn = sa.func.get_addressdata(result.place_id, housenumber)\
 245             .table_valued( # type: ignore[no-untyped-call]
 246                 sa.column('place_id', type_=sa.Integer),
 247                 'osm_type',
 248                 sa.column('osm_id', type_=sa.BigInteger),
 249                 sa.column('name', type_=conn.t.types.Composite),
 250                 'class', 'type', 'place_type',
 251                 sa.column('admin_level', type_=sa.Integer),
 252                 sa.column('fromarea', type_=sa.Boolean),
 253                 sa.column('isaddress', type_=sa.Boolean),
 254                 sa.column('rank_address', type_=sa.SmallInteger),
 255                 sa.column('distance', type_=sa.Float))
 256     sql = sa.select(sfn).order_by(sa.column('rank_address').desc(),
 257                                   sa.column('isaddress').desc())
 258
 259     result.address_rows = []
 260     for row in await conn.execute(sql):
 261         result.address_rows.append(_result_row_to_address_row(row))
 262
 263 # pylint: disable=consider-using-f-string
 264 def _placex_select_address_row(conn: SearchConnection,
 265                                centroid: Point) -> SaSelect:
 266     t = conn.t.placex
 267     return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
 268                      t.c.class_.label('class'), t.c.type,
 269                      t.c.admin_level, t.c.housenumber,
 270                      sa.literal_column("""ST_GeometryType(geometry) in
 271                                         ('ST_Polygon','ST_MultiPolygon')""").label('fromarea'),
 272                      t.c.rank_address,
 273                      sa.literal_column(
 274                          """ST_DistanceSpheroid(geometry, 'SRID=4326;POINT(%f %f)'::geometry,
 275                               'SPHEROID["WGS 84",6378137,298.257223563, AUTHORITY["EPSG","7030"]]')
 276                          """ % centroid).label('distance'))
 277
 278
 279 async def complete_linked_places(conn: SearchConnection, result: SearchResult) -> None:
 280     """ Retrieve information about places that link to the result.
 281     """
 282     result.linked_rows = []
 283     if result.source_table != SourceTable.PLACEX:
 284         return
 285
 286     sql = _placex_select_address_row(conn, result.centroid)\
 287             .where(conn.t.placex.c.linked_place_id == result.place_id)
 288
 289     for row in await conn.execute(sql):
 290         result.linked_rows.append(_result_row_to_address_row(row))
 291
 292
 293 async def complete_keywords(conn: SearchConnection, result: SearchResult) -> None:
 294     """ Retrieve information about the search terms used for this place.
 295     """
 296     t = conn.t.search_name
 297     sql = sa.select(t.c.name_vector, t.c.nameaddress_vector)\
 298             .where(t.c.place_id == result.place_id)
 299
 300     result.name_keywords = []
 301     result.address_keywords = []
 302     for name_tokens, address_tokens in await conn.execute(sql):
 303         t = conn.t.word
 304         sel = sa.select(t.c.word_id, t.c.word_token, t.c.word)
 305
 306         for row in await conn.execute(sel.where(t.c.word_id == sa.any_(name_tokens))):
 307             result.name_keywords.append(WordInfo(*row))
 308
 309         for row in await conn.execute(sel.where(t.c.word_id == sa.any_(address_tokens))):
 310             result.address_keywords.append(WordInfo(*row))
 311
 312
 313 async def complete_parented_places(conn: SearchConnection, result: SearchResult) -> None:
 314     """ Retrieve information about places that the result provides the
 315         address for.
 316     """
 317     result.parented_rows = []
 318     if result.source_table != SourceTable.PLACEX:
 319         return
 320
 321     sql = _placex_select_address_row(conn, result.centroid)\
 322             .where(conn.t.placex.c.parent_place_id == result.place_id)\
 323             .where(conn.t.placex.c.rank_search == 30)
 324
 325     for row in await conn.execute(sql):
 326         result.parented_rows.append(_result_row_to_address_row(row))