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