from typing import Optional, List, Callable, Type, Tuple
import sqlalchemy as sa
-from geoalchemy2 import WKTElement
from nominatim.typing import SaColumn, SaSelect, SaFromClause, SaLabel, SaRow
from nominatim.api.connection import SearchConnection
import nominatim.api.results as nres
from nominatim.api.logging import log
from nominatim.api.types import AnyPoint, DataLayer, ReverseDetails, GeometryFormat, Bbox
+from nominatim.db.sqlalchemy_types import Geometry
# In SQLAlchemy expression which compare with NULL need to be expressed with
# the equal sign.
RowFunc = Callable[[Optional[SaRow], Type[nres.ReverseResult]], Optional[nres.ReverseResult]]
-def _select_from_placex(t: SaFromClause, wkt: Optional[str] = None) -> SaSelect:
+WKT_PARAM = sa.bindparam('wkt', type_=Geometry)
+MAX_RANK_PARAM = sa.bindparam('max_rank')
+
+def _select_from_placex(t: SaFromClause, use_wkt: bool = True) -> SaSelect:
""" Create a select statement with the columns relevant for reverse
results.
"""
- if wkt is None:
+ if not use_wkt:
distance = t.c.distance
centroid = t.c.centroid
else:
- distance = t.c.geometry.ST_Distance(wkt)
- centroid = sa.case(
- (t.c.geometry.ST_GeometryType().in_(('ST_LineString',
- 'ST_MultiLineString')),
- t.c.geometry.ST_ClosestPoint(wkt)),
- else_=t.c.centroid).label('centroid')
+ distance = t.c.geometry.ST_Distance(WKT_PARAM)
+ centroid = sa.case((t.c.geometry.is_line_like(), t.c.geometry.ST_ClosestPoint(WKT_PARAM)),
+ else_=t.c.centroid).label('centroid')
return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
else_=table.c.linegeo.ST_LineInterpolatePoint(rounded_pos)).label('centroid')
-def _locate_interpolation(table: SaFromClause, wkt: WKTElement) -> SaLabel:
+def _locate_interpolation(table: SaFromClause) -> SaLabel:
""" Given a position, locate the closest point on the line.
"""
- return sa.case((table.c.linegeo.ST_GeometryType() == 'ST_LineString',
- sa.func.ST_LineLocatePoint(table.c.linegeo, wkt)),
+ return sa.case((table.c.linegeo.is_line_like(),
+ table.c.linegeo.ST_LineLocatePoint(WKT_PARAM)),
else_=0).label('position')
sa.or_(table.c.housenumber != None,
table.c.name.has_key('housename')))
+
def _get_closest(*rows: Optional[SaRow]) -> Optional[SaRow]:
return min(rows, key=lambda row: 1000 if row is None else row.distance)
+
class ReverseGeocoder:
""" Class implementing the logic for looking up a place from a
coordinate.
self.conn = conn
self.params = params
+ self.bind_params = {'max_rank': params.max_rank}
+
@property
def max_rank(self) -> int:
"""
return self.layer_enabled(DataLayer.RAILWAY, DataLayer.MANMADE, DataLayer.NATURAL)
+
def _add_geometry_columns(self, sql: SaSelect, col: SaColumn) -> SaSelect:
if not self.has_geometries():
return sql
out = []
if self.params.geometry_simplification > 0.0:
- col = col.ST_SimplifyPreserveTopology(self.params.geometry_simplification)
+ col = sa.func.ST_SimplifyPreserveTopology(col, self.params.geometry_simplification)
if self.params.geometry_output & GeometryFormat.GEOJSON:
- out.append(col.ST_AsGeoJSON().label('geometry_geojson'))
+ out.append(sa.func.ST_AsGeoJSON(col).label('geometry_geojson'))
if self.params.geometry_output & GeometryFormat.TEXT:
- out.append(col.ST_AsText().label('geometry_text'))
+ out.append(sa.func.ST_AsText(col).label('geometry_text'))
if self.params.geometry_output & GeometryFormat.KML:
- out.append(col.ST_AsKML().label('geometry_kml'))
+ out.append(sa.func.ST_AsKML(col).label('geometry_kml'))
if self.params.geometry_output & GeometryFormat.SVG:
- out.append(col.ST_AsSVG().label('geometry_svg'))
+ out.append(sa.func.ST_AsSVG(col).label('geometry_svg'))
return sql.add_columns(*out)
return table.c.class_.in_(tuple(include))
- async def _find_closest_street_or_poi(self, wkt: WKTElement,
- distance: float) -> Optional[SaRow]:
+ async def _find_closest_street_or_poi(self, distance: float) -> Optional[SaRow]:
""" Look up the closest rank 26+ place in the database, which
is closer than the given distance.
"""
t = self.conn.t.placex
- sql = _select_from_placex(t, wkt)\
- .where(t.c.geometry.ST_DWithin(wkt, distance))\
+ sql = _select_from_placex(t)\
+ .where(t.c.geometry.ST_DWithin(WKT_PARAM, distance))\
.where(t.c.indexed_status == 0)\
.where(t.c.linked_place_id == None)\
- .where(sa.or_(t.c.geometry.ST_GeometryType()
- .not_in(('ST_Polygon', 'ST_MultiPolygon')),
- t.c.centroid.ST_Distance(wkt) < distance))\
+ .where(sa.or_(sa.not_(t.c.geometry.is_area()),
+ t.c.centroid.ST_Distance(WKT_PARAM) < distance))\
.order_by('distance')\
.limit(1)
if self.layer_enabled(DataLayer.POI) and self.max_rank == 30:
restrict.append(sa.and_(t.c.rank_search == 30,
t.c.class_.not_in(('place', 'building')),
- t.c.geometry.ST_GeometryType() != 'ST_LineString'))
+ sa.not_(t.c.geometry.is_line_like())))
if self.has_feature_layers():
- restrict.append(sa.and_(t.c.rank_search.between(26, self.max_rank),
+ restrict.append(sa.and_(t.c.rank_search.between(26, MAX_RANK_PARAM),
t.c.rank_address == 0,
self._filter_by_layer(t)))
if not restrict:
return None
- return (await self.conn.execute(sql.where(sa.or_(*restrict)))).one_or_none()
+ sql = sql.where(sa.or_(*restrict))
+
+ return (await self.conn.execute(sql, self.bind_params)).one_or_none()
- async def _find_housenumber_for_street(self, parent_place_id: int,
- wkt: WKTElement) -> Optional[SaRow]:
+ async def _find_housenumber_for_street(self, parent_place_id: int) -> Optional[SaRow]:
t = self.conn.t.placex
- sql = _select_from_placex(t, wkt)\
- .where(t.c.geometry.ST_DWithin(wkt, 0.001))\
+ sql = _select_from_placex(t)\
+ .where(t.c.geometry.ST_DWithin(WKT_PARAM, 0.001))\
.where(t.c.parent_place_id == parent_place_id)\
.where(_is_address_point(t))\
.where(t.c.indexed_status == 0)\
sql = self._add_geometry_columns(sql, t.c.geometry)
- return (await self.conn.execute(sql)).one_or_none()
+ return (await self.conn.execute(sql, self.bind_params)).one_or_none()
async def _find_interpolation_for_street(self, parent_place_id: Optional[int],
- wkt: WKTElement,
distance: float) -> Optional[SaRow]:
t = self.conn.t.osmline
sql = sa.select(t,
- t.c.linegeo.ST_Distance(wkt).label('distance'),
- _locate_interpolation(t, wkt))\
- .where(t.c.linegeo.ST_DWithin(wkt, distance))\
+ t.c.linegeo.ST_Distance(WKT_PARAM).label('distance'),
+ _locate_interpolation(t))\
+ .where(t.c.linegeo.ST_DWithin(WKT_PARAM, distance))\
.where(t.c.startnumber != None)\
.order_by('distance')\
.limit(1)
sub = sql.subquery('geom')
sql = self._add_geometry_columns(sa.select(sub), sub.c.centroid)
- return (await self.conn.execute(sql)).one_or_none()
+ return (await self.conn.execute(sql, self.bind_params)).one_or_none()
async def _find_tiger_number_for_street(self, parent_place_id: int,
- parent_type: str, parent_id: int,
- wkt: WKTElement) -> Optional[SaRow]:
+ parent_type: str,
+ parent_id: int) -> Optional[SaRow]:
t = self.conn.t.tiger
inner = sa.select(t,
- t.c.linegeo.ST_Distance(wkt).label('distance'),
- _locate_interpolation(t, wkt))\
- .where(t.c.linegeo.ST_DWithin(wkt, 0.001))\
+ t.c.linegeo.ST_Distance(WKT_PARAM).label('distance'),
+ _locate_interpolation(t))\
+ .where(t.c.linegeo.ST_DWithin(WKT_PARAM, 0.001))\
.where(t.c.parent_place_id == parent_place_id)\
.order_by('distance')\
.limit(1)\
sub = sql.subquery('geom')
sql = self._add_geometry_columns(sa.select(sub), sub.c.centroid)
- return (await self.conn.execute(sql)).one_or_none()
+ return (await self.conn.execute(sql, self.bind_params)).one_or_none()
- async def lookup_street_poi(self,
- wkt: WKTElement) -> Tuple[Optional[SaRow], RowFunc]:
+ async def lookup_street_poi(self) -> Tuple[Optional[SaRow], RowFunc]:
""" Find a street or POI/address for the given WKT point.
"""
log().section('Reverse lookup on street/address level')
distance = 0.006
parent_place_id = None
- row = await self._find_closest_street_or_poi(wkt, distance)
+ row = await self._find_closest_street_or_poi(distance)
row_func: RowFunc = nres.create_from_placex_row
log().var_dump('Result (street/building)', row)
distance = 0.001
parent_place_id = row.place_id
log().comment('Find housenumber for street')
- addr_row = await self._find_housenumber_for_street(parent_place_id, wkt)
+ addr_row = await self._find_housenumber_for_street(parent_place_id)
log().var_dump('Result (street housenumber)', addr_row)
if addr_row is not None:
log().comment('Find TIGER housenumber for street')
addr_row = await self._find_tiger_number_for_street(parent_place_id,
row.osm_type,
- row.osm_id,
- wkt)
+ row.osm_id)
log().var_dump('Result (street Tiger housenumber)', addr_row)
if addr_row is not None:
if self.max_rank > 27 and self.layer_enabled(DataLayer.ADDRESS):
log().comment('Find interpolation for street')
addr_row = await self._find_interpolation_for_street(parent_place_id,
- wkt, distance)
+ distance)
log().var_dump('Result (street interpolation)', addr_row)
if addr_row is not None:
row = addr_row
return row, row_func
- async def _lookup_area_address(self, wkt: WKTElement) -> Optional[SaRow]:
+ async def _lookup_area_address(self) -> Optional[SaRow]:
""" Lookup large addressable areas for the given WKT point.
"""
log().comment('Reverse lookup by larger address area features')
# The inner SQL brings results in the right order, so that
# later only a minimum of results needs to be checked with ST_Contains.
inner = sa.select(t, sa.literal(0.0).label('distance'))\
- .where(t.c.rank_search.between(5, self.max_rank))\
+ .where(t.c.rank_search.between(5, MAX_RANK_PARAM))\
.where(t.c.rank_address.between(5, 25))\
- .where(t.c.geometry.ST_GeometryType().in_(('ST_Polygon', 'ST_MultiPolygon')))\
- .where(t.c.geometry.intersects(wkt))\
+ .where(t.c.geometry.is_area())\
+ .where(t.c.geometry.intersects(WKT_PARAM))\
.where(t.c.name != None)\
.where(t.c.indexed_status == 0)\
.where(t.c.linked_place_id == None)\
.limit(50)\
.subquery('area')
- sql = _select_from_placex(inner)\
- .where(inner.c.geometry.ST_Contains(wkt))\
+ sql = _select_from_placex(inner, False)\
+ .where(inner.c.geometry.ST_Contains(WKT_PARAM))\
.order_by(sa.desc(inner.c.rank_search))\
.limit(1)
sql = self._add_geometry_columns(sql, inner.c.geometry)
- address_row = (await self.conn.execute(sql)).one_or_none()
+ address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
log().var_dump('Result (area)', address_row)
if address_row is not None and address_row.rank_search < self.max_rank:
log().comment('Search for better matching place nodes inside the area')
inner = sa.select(t,
- t.c.geometry.ST_Distance(wkt).label('distance'))\
+ t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\
.where(t.c.osm_type == 'N')\
.where(t.c.rank_search > address_row.rank_search)\
- .where(t.c.rank_search <= self.max_rank)\
+ .where(t.c.rank_search <= MAX_RANK_PARAM)\
.where(t.c.rank_address.between(5, 25))\
.where(t.c.name != None)\
.where(t.c.indexed_status == 0)\
.where(t.c.type != 'postcode')\
.where(t.c.geometry
.ST_Buffer(sa.func.reverse_place_diameter(t.c.rank_search))
- .intersects(wkt))\
+ .intersects(WKT_PARAM))\
.order_by(sa.desc(t.c.rank_search))\
.limit(50)\
.subquery('places')
touter = self.conn.t.placex.alias('outer')
- sql = _select_from_placex(inner)\
+ sql = _select_from_placex(inner, False)\
.join(touter, touter.c.geometry.ST_Contains(inner.c.geometry))\
.where(touter.c.place_id == address_row.place_id)\
.where(inner.c.distance < sa.func.reverse_place_diameter(inner.c.rank_search))\
sql = self._add_geometry_columns(sql, inner.c.geometry)
- place_address_row = (await self.conn.execute(sql)).one_or_none()
+ place_address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
log().var_dump('Result (place node)', place_address_row)
if place_address_row is not None:
return address_row
- async def _lookup_area_others(self, wkt: WKTElement) -> Optional[SaRow]:
+ async def _lookup_area_others(self) -> Optional[SaRow]:
t = self.conn.t.placex
- inner = sa.select(t, t.c.geometry.ST_Distance(wkt).label('distance'))\
+ inner = sa.select(t, t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\
.where(t.c.rank_address == 0)\
- .where(t.c.rank_search.between(5, self.max_rank))\
+ .where(t.c.rank_search.between(5, MAX_RANK_PARAM))\
.where(t.c.name != None)\
.where(t.c.indexed_status == 0)\
.where(t.c.linked_place_id == None)\
.where(self._filter_by_layer(t))\
.where(t.c.geometry
.ST_Buffer(sa.func.reverse_place_diameter(t.c.rank_search))
- .intersects(wkt))\
+ .intersects(WKT_PARAM))\
.order_by(sa.desc(t.c.rank_search))\
.limit(50)\
.subquery()
- sql = _select_from_placex(inner)\
- .where(sa.or_(inner.c.geometry.ST_GeometryType()
- .not_in(('ST_Polygon', 'ST_MultiPolygon')),
- inner.c.geometry.ST_Contains(wkt)))\
+ sql = _select_from_placex(inner, False)\
+ .where(sa.or_(sa.not_(inner.c.geometry.is_area()),
+ inner.c.geometry.ST_Contains(WKT_PARAM)))\
.order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
.limit(1)
sql = self._add_geometry_columns(sql, inner.c.geometry)
- row = (await self.conn.execute(sql)).one_or_none()
+ row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
log().var_dump('Result (non-address feature)', row)
return row
- async def lookup_area(self, wkt: WKTElement) -> Optional[SaRow]:
- """ Lookup large areas for the given WKT point.
+ async def lookup_area(self) -> Optional[SaRow]:
+ """ Lookup large areas for the current search.
"""
log().section('Reverse lookup by larger area features')
if self.layer_enabled(DataLayer.ADDRESS):
- address_row = await self._lookup_area_address(wkt)
+ address_row = await self._lookup_area_address()
else:
address_row = None
if self.has_feature_layers():
- other_row = await self._lookup_area_others(wkt)
+ other_row = await self._lookup_area_others()
else:
other_row = None
return _get_closest(address_row, other_row)
- async def lookup_country(self, wkt: WKTElement) -> Optional[SaRow]:
- """ Lookup the country for the given WKT point.
+ async def lookup_country(self) -> Optional[SaRow]:
+ """ Lookup the country for the current search.
"""
log().section('Reverse lookup by country code')
t = self.conn.t.country_grid
sql = sa.select(t.c.country_code).distinct()\
- .where(t.c.geometry.ST_Contains(wkt))
+ .where(t.c.geometry.ST_Contains(WKT_PARAM))
- ccodes = tuple((r[0] for r in await self.conn.execute(sql)))
+ ccodes = tuple((r[0] for r in await self.conn.execute(sql, self.bind_params)))
log().var_dump('Country codes', ccodes)
if not ccodes:
log().comment('Search for place nodes in country')
inner = sa.select(t,
- t.c.geometry.ST_Distance(wkt).label('distance'))\
+ t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\
.where(t.c.osm_type == 'N')\
.where(t.c.rank_search > 4)\
- .where(t.c.rank_search <= self.max_rank)\
+ .where(t.c.rank_search <= MAX_RANK_PARAM)\
.where(t.c.rank_address.between(5, 25))\
.where(t.c.name != None)\
.where(t.c.indexed_status == 0)\
.where(t.c.country_code.in_(ccodes))\
.where(t.c.geometry
.ST_Buffer(sa.func.reverse_place_diameter(t.c.rank_search))
- .intersects(wkt))\
+ .intersects(WKT_PARAM))\
.order_by(sa.desc(t.c.rank_search))\
.limit(50)\
.subquery()
- sql = _select_from_placex(inner)\
+ sql = _select_from_placex(inner, False)\
.where(inner.c.distance < sa.func.reverse_place_diameter(inner.c.rank_search))\
.order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
.limit(1)
sql = self._add_geometry_columns(sql, inner.c.geometry)
- address_row = (await self.conn.execute(sql)).one_or_none()
+ address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
log().var_dump('Result (addressable place node)', address_row)
else:
address_row = None
if address_row is None:
# Still nothing, then return a country with the appropriate country code.
- sql = _select_from_placex(t, wkt)\
+ sql = _select_from_placex(t)\
.where(t.c.country_code.in_(ccodes))\
.where(t.c.rank_address == 4)\
.where(t.c.rank_search == 4)\
sql = self._add_geometry_columns(sql, t.c.geometry)
- address_row = (await self.conn.execute(sql)).one_or_none()
+ address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
return address_row
log().function('reverse_lookup', coord=coord, params=self.params)
- wkt = WKTElement(f'POINT({coord[0]} {coord[1]})', srid=4326)
+ self.bind_params['wkt'] = f'SRID=4326;POINT({coord[0]} {coord[1]})'
row: Optional[SaRow] = None
row_func: RowFunc = nres.create_from_placex_row
if self.max_rank >= 26:
- row, tmp_row_func = await self.lookup_street_poi(wkt)
+ row, tmp_row_func = await self.lookup_street_poi()
if row is not None:
row_func = tmp_row_func
if row is None and self.max_rank > 4:
- row = await self.lookup_area(wkt)
+ row = await self.lookup_area()
if row is None and self.layer_enabled(DataLayer.ADDRESS):
- row = await self.lookup_country(wkt)
+ row = await self.lookup_country()
result = row_func(row, nres.ReverseResult)
if result is not None:
assert row is not None
result.distance = row.distance
if hasattr(row, 'bbox'):
- result.bbox = Bbox.from_wkb(row.bbox.data)
+ result.bbox = Bbox.from_wkb(row.bbox)
await nres.add_result_details(self.conn, [result], self.params)
return result
projects / nominatim.git / blobdiff