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 table of Tiger addresses and return the
  99         base information.
 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 get_place_by_id(conn: SearchConnection, place: ntyp.PlaceRef,
 118                           details: ntyp.LookupDetails) -> Optional[nres.SearchResult]:
 119     """ Retrieve a place with additional details from the database.
 120     """
 121     if details.geometry_output and details.geometry_output != ntyp.GeometryFormat.GEOJSON:
 122         raise ValueError("lookup only supports geojosn polygon output.")
 123
 124     row = await find_in_placex(conn, place, details)
 125     if row is not None:
 126         result = nres.create_from_placex_row(row)
 127         await nres.add_result_details(conn, result, details)
 128         return result
 129
 130     row = await find_in_osmline(conn, place, details)
 131     if row is not None:
 132         result = nres.create_from_osmline_row(row)
 133         await nres.add_result_details(conn, result, details)
 134         return result
 135
 136     row = await find_in_tiger(conn, place, details)
 137     if row is not None:
 138         result = nres.create_from_tiger_row(row)
 139         await nres.add_result_details(conn, result, details)
 140         return result
 141
 142     # Nothing found under this ID.
 143     return None