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                     sa.func.ST_X(t.c.centroid).label('x'),
  50                     sa.func.ST_Y(t.c.centroid).label('y'),
  51                     _select_column_geometry(t.c.geometry, details.geometry_output))
  52
  53     if isinstance(place, ntyp.PlaceID):
  54         sql = sql.where(t.c.place_id == place.place_id)
  55     elif isinstance(place, ntyp.OsmID):
  56         sql = sql.where(t.c.osm_type == place.osm_type)\
  57                  .where(t.c.osm_id == place.osm_id)
  58         if place.osm_class:
  59             sql = sql.where(t.c.class_ == place.osm_class)
  60         else:
  61             sql = sql.order_by(t.c.class_)
  62         sql = sql.limit(1)
  63     else:
  64         return None
  65
  66     return (await conn.execute(sql)).one_or_none()
  67
  68
  69 async def find_in_osmline(conn: SearchConnection, place: ntyp.PlaceRef,
  70                           details: ntyp.LookupDetails) -> Optional[SaRow]:
  71     """ Search for the given place in the osmline table and return the
  72         base information.
  73     """
  74     log().section("Find in interpolation table")
  75     t = conn.t.osmline
  76     sql = sa.select(t.c.place_id, t.c.osm_id, t.c.parent_place_id,
  77                     t.c.indexed_date, t.c.startnumber, t.c.endnumber,
  78                     t.c.step, t.c.address, t.c.postcode, t.c.country_code,
  79                     sa.func.ST_X(sa.func.ST_Centroid(t.c.linegeo)).label('x'),
  80                     sa.func.ST_Y(sa.func.ST_Centroid(t.c.linegeo)).label('y'),
  81                     _select_column_geometry(t.c.linegeo, details.geometry_output))
  82
  83     if isinstance(place, ntyp.PlaceID):
  84         sql = sql.where(t.c.place_id == place.place_id)
  85     elif isinstance(place, ntyp.OsmID) and place.osm_type == 'W':
  86         # There may be multiple interpolations for a single way.
  87         # If 'class' contains a number, return the one that belongs to that number.
  88         sql = sql.where(t.c.osm_id == place.osm_id).limit(1)
  89         if place.osm_class and place.osm_class.isdigit():
  90             sql = sql.order_by(sa.func.greatest(0,
  91                                     sa.func.least(int(place.osm_class) - t.c.endnumber),
  92                                            t.c.startnumber - int(place.osm_class)))
  93     else:
  94         return None
  95
  96     return (await conn.execute(sql)).one_or_none()
  97
  98
  99 async def find_in_tiger(conn: SearchConnection, place: ntyp.PlaceRef,
 100                         details: ntyp.LookupDetails) -> Optional[SaRow]:
 101     """ Search for the given place in the table of Tiger addresses and return
 102         the base information. Only lookup by place ID is supported.
 103     """
 104     log().section("Find in TIGER table")
 105     t = conn.t.tiger
 106     sql = sa.select(t.c.place_id, t.c.parent_place_id,
 107                     t.c.startnumber, t.c.endnumber, t.c.step,
 108                     t.c.postcode,
 109                     sa.func.ST_X(sa.func.ST_Centroid(t.c.linegeo)).label('x'),
 110                     sa.func.ST_Y(sa.func.ST_Centroid(t.c.linegeo)).label('y'),
 111                     _select_column_geometry(t.c.linegeo, details.geometry_output))
 112
 113     if isinstance(place, ntyp.PlaceID):
 114         sql = sql.where(t.c.place_id == place.place_id)
 115     else:
 116         return None
 117
 118     return (await conn.execute(sql)).one_or_none()
 119
 120
 121 async def find_in_postcode(conn: SearchConnection, place: ntyp.PlaceRef,
 122                            details: ntyp.LookupDetails) -> Optional[SaRow]:
 123     """ Search for the given place in the postcode table and return the
 124         base information. Only lookup by place ID is supported.
 125     """
 126     log().section("Find in postcode table")
 127     t = conn.t.postcode
 128     sql = sa.select(t.c.place_id, t.c.parent_place_id,
 129                     t.c.rank_search, t.c.rank_address,
 130                     t.c.indexed_date, t.c.postcode, t.c.country_code,
 131                     sa.func.ST_X(t.c.geometry).label('x'),
 132                     sa.func.ST_Y(t.c.geometry).label('y'),
 133                     _select_column_geometry(t.c.geometry, details.geometry_output))
 134
 135     if isinstance(place, ntyp.PlaceID):
 136         sql = sql.where(t.c.place_id == place.place_id)
 137     else:
 138         return None
 139
 140     return (await conn.execute(sql)).one_or_none()
 141
 142
 143 async def get_place_by_id(conn: SearchConnection, place: ntyp.PlaceRef,
 144                           details: ntyp.LookupDetails) -> Optional[nres.SearchResult]:
 145     """ Retrieve a place with additional details from the database.
 146     """
 147     log().function('get_place_by_id', place=place, details=details)
 148
 149     if details.geometry_output and details.geometry_output != ntyp.GeometryFormat.GEOJSON:
 150         raise ValueError("lookup only supports geojosn polygon output.")
 151
 152     row = await find_in_placex(conn, place, details)
 153     if row is not None:
 154         result = nres.create_from_placex_row(row)
 155         log().var_dump('Result', result)
 156         await nres.add_result_details(conn, result, details)
 157         return result
 158
 159     row = await find_in_osmline(conn, place, details)
 160     if row is not None:
 161         result = nres.create_from_osmline_row(row)
 162         log().var_dump('Result', result)
 163         await nres.add_result_details(conn, result, details)
 164         return result
 165
 166     row = await find_in_postcode(conn, place, details)
 167     if row is not None:
 168         result = nres.create_from_postcode_row(row)
 169         log().var_dump('Result', result)
 170         await nres.add_result_details(conn, result, details)
 171         return result
 172
 173     row = await find_in_tiger(conn, place, details)
 174     if row is not None:
 175         result = nres.create_from_tiger_row(row)
 176         log().var_dump('Result', result)
 177         await nres.add_result_details(conn, result, details)
 178         return result
 179
 180     # Nothing found under this ID.
 181     return None