nominatim/api/lookup.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 Implementation of place lookup by ID.
   9 """
  10 from typing import Optional
  11
  12 import sqlalchemy as sa
  13
  14 from nominatim.typing import SaColumn, SaLabel, SaRow
  15 from nominatim.api.connection import SearchConnection
  16 import nominatim.api.types as ntyp
  17 import nominatim.api.results as nres
  18 from nominatim.api.logging import log
  19
  20 def _select_column_geometry(column: SaColumn,
  21                             geometry_output: ntyp.GeometryFormat) -> SaLabel:
  22     """ Create the appropriate column expression for selecting a
  23         geometry for the details response.
  24     """
  25     if geometry_output & ntyp.GeometryFormat.GEOJSON:
  26         return sa.literal_column(f"""
  27                   ST_AsGeoJSON(CASE WHEN ST_NPoints({column.name}) > 5000
  28                                THEN ST_SimplifyPreserveTopology({column.name}, 0.0001)
  29                                ELSE {column.name} END)
  30                   """).label('geometry_geojson')
  31
  32     return sa.func.ST_GeometryType(column).label('geometry_type')
  33
  34
  35 async def find_in_placex(conn: SearchConnection, place: ntyp.PlaceRef,
  36                          details: ntyp.LookupDetails) -> Optional[SaRow]:
  37     """ Search for the given place in the placex table and return the
  38         base information.
  39     """
  40     log().section("Find in placex table")
  41     t = conn.t.placex
  42     sql = sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
  43                     t.c.class_, t.c.type, t.c.admin_level,
  44                     t.c.address, t.c.extratags,
  45                     t.c.housenumber, t.c.postcode, t.c.country_code,
  46                     t.c.importance, t.c.wikipedia, t.c.indexed_date,
  47                     t.c.parent_place_id, t.c.rank_address, t.c.rank_search,
  48                     t.c.linked_place_id,
  49                     t.c.centroid,
  50                     _select_column_geometry(t.c.geometry, details.geometry_output))
  51
  52     if isinstance(place, ntyp.PlaceID):
  53         sql = sql.where(t.c.place_id == place.place_id)
  54     elif isinstance(place, ntyp.OsmID):
  55         sql = sql.where(t.c.osm_type == place.osm_type)\
  56                  .where(t.c.osm_id == place.osm_id)
  57         if place.osm_class:
  58             sql = sql.where(t.c.class_ == place.osm_class)
  59         else:
  60             sql = sql.order_by(t.c.class_)
  61         sql = sql.limit(1)
  62     else:
  63         return None
  64
  65     return (await conn.execute(sql)).one_or_none()
  66
  67
  68 async def find_in_osmline(conn: SearchConnection, place: ntyp.PlaceRef,
  69                           details: ntyp.LookupDetails) -> Optional[SaRow]:
  70     """ Search for the given place in the osmline table and return the
  71         base information.
  72     """
  73     log().section("Find in interpolation table")
  74     t = conn.t.osmline
  75     sql = sa.select(t.c.place_id, t.c.osm_id, t.c.parent_place_id,
  76                     t.c.indexed_date, t.c.startnumber, t.c.endnumber,
  77                     t.c.step, t.c.address, t.c.postcode, t.c.country_code,
  78                     t.c.linegeo.ST_Centroid().label('centroid'),
  79                     _select_column_geometry(t.c.linegeo, details.geometry_output))
  80
  81     if isinstance(place, ntyp.PlaceID):
  82         sql = sql.where(t.c.place_id == place.place_id)
  83     elif isinstance(place, ntyp.OsmID) and place.osm_type == 'W':
  84         # There may be multiple interpolations for a single way.
  85         # If 'class' contains a number, return the one that belongs to that number.
  86         sql = sql.where(t.c.osm_id == place.osm_id).limit(1)
  87         if place.osm_class and place.osm_class.isdigit():
  88             sql = sql.order_by(sa.func.greatest(0,
  89                                     sa.func.least(int(place.osm_class) - t.c.endnumber),
  90                                            t.c.startnumber - int(place.osm_class)))
  91     else:
  92         return None
  93
  94     return (await conn.execute(sql)).one_or_none()
  95
  96
  97 async def find_in_tiger(conn: SearchConnection, place: ntyp.PlaceRef,
  98                         details: ntyp.LookupDetails) -> Optional[SaRow]:
  99     """ Search for the given place in the table of Tiger addresses and return
 100         the base information. Only lookup by place ID is supported.
 101     """
 102     log().section("Find in TIGER table")
 103     t = conn.t.tiger
 104     sql = sa.select(t.c.place_id, t.c.parent_place_id,
 105                     t.c.startnumber, t.c.endnumber, t.c.step,
 106                     t.c.postcode,
 107                     t.c.linegeo.ST_Centroid().label('centroid'),
 108                     _select_column_geometry(t.c.linegeo, details.geometry_output))
 109
 110     if isinstance(place, ntyp.PlaceID):
 111         sql = sql.where(t.c.place_id == place.place_id)
 112     else:
 113         return None
 114
 115     return (await conn.execute(sql)).one_or_none()
 116
 117
 118 async def find_in_postcode(conn: SearchConnection, place: ntyp.PlaceRef,
 119                            details: ntyp.LookupDetails) -> Optional[SaRow]:
 120     """ Search for the given place in the postcode table and return the
 121         base information. Only lookup by place ID is supported.
 122     """
 123     log().section("Find in postcode table")
 124     t = conn.t.postcode
 125     sql = sa.select(t.c.place_id, t.c.parent_place_id,
 126                     t.c.rank_search, t.c.rank_address,
 127                     t.c.indexed_date, t.c.postcode, t.c.country_code,
 128                     t.c.geometry.label('centroid'),
 129                     _select_column_geometry(t.c.geometry, details.geometry_output))
 130
 131     if isinstance(place, ntyp.PlaceID):
 132         sql = sql.where(t.c.place_id == place.place_id)
 133     else:
 134         return None
 135
 136     return (await conn.execute(sql)).one_or_none()
 137
 138
 139 async def get_place_by_id(conn: SearchConnection, place: ntyp.PlaceRef,
 140                           details: ntyp.LookupDetails) -> Optional[nres.SearchResult]:
 141     """ Retrieve a place with additional details from the database.
 142     """
 143     log().function('get_place_by_id', place=place, details=details)
 144
 145     if details.geometry_output and details.geometry_output != ntyp.GeometryFormat.GEOJSON:
 146         raise ValueError("lookup only supports geojosn polygon output.")
 147
 148     row = await find_in_placex(conn, place, details)
 149     if row is not None:
 150         result = nres.create_from_placex_row(row)
 151         log().var_dump('Result', result)
 152         await nres.add_result_details(conn, result, details)
 153         return result
 154
 155     row = await find_in_osmline(conn, place, details)
 156     if row is not None:
 157         result = nres.create_from_osmline_row(row)
 158         log().var_dump('Result', result)
 159         await nres.add_result_details(conn, result, details)
 160         return result
 161
 162     row = await find_in_postcode(conn, place, details)
 163     if row is not None:
 164         result = nres.create_from_postcode_row(row)
 165         log().var_dump('Result', result)
 166         await nres.add_result_details(conn, result, details)
 167         return result
 168
 169     row = await find_in_tiger(conn, place, details)
 170     if row is not None:
 171         result = nres.create_from_tiger_row(row)
 172         log().var_dump('Result', result)
 173         await nres.add_result_details(conn, result, details)
 174         return result
 175
 176     # Nothing found under this ID.
 177     return None