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