"""
Implementation of reverse geocoding.
"""
-from typing import Optional, List
+from typing import Optional, List, Callable, Type, Tuple
import sqlalchemy as sa
from geoalchemy2 import WKTElement
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, LookupDetails, GeometryFormat
+from nominatim.api.types import AnyPoint, DataLayer, LookupDetails, GeometryFormat, Bbox
# In SQLAlchemy expression which compare with NULL need to be expressed with
# the equal sign.
# pylint: disable=singleton-comparison
+RowFunc = Callable[[Optional[SaRow], Type[nres.ReverseResult]], Optional[nres.ReverseResult]]
+
def _select_from_placex(t: SaFromClause, wkt: Optional[str] = None) -> SaSelect:
""" Create a select statement with the columns relevant for reverse
results.
"""
if wkt is None:
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')
+
return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
t.c.class_, t.c.type,
t.c.housenumber, t.c.postcode, t.c.country_code,
t.c.importance, t.c.wikipedia,
t.c.parent_place_id, t.c.rank_address, t.c.rank_search,
- t.c.centroid,
+ centroid,
distance.label('distance'),
t.c.geometry.ST_Expand(0).label('bbox'))
sa.Integer).label('housenumber')
+def _interpolated_position(table: SaFromClause) -> SaLabel:
+ fac = sa.cast(table.c.step, sa.Float) / (table.c.endnumber - table.c.startnumber)
+ rounded_pos = sa.func.round(table.c.position / fac) * fac
+ return sa.case(
+ (table.c.endnumber == table.c.startnumber, table.c.linegeo.ST_Centroid()),
+ else_=table.c.linegeo.ST_LineInterpolatePoint(rounded_pos)).label('centroid')
+
+
def _is_address_point(table: SaFromClause) -> SaColumn:
return sa.and_(table.c.rank_address == 30,
sa.or_(table.c.housenumber != None,
self.layer = layer
self.details = details
+ def layer_enabled(self, *layer: DataLayer) -> bool:
+ """ Return true when any of the given layer types are requested.
+ """
+ return any(self.layer & l for l in layer)
+
+
+ def layer_disabled(self, *layer: DataLayer) -> bool:
+ """ Return true when none of the given layer types is requested.
+ """
+ return not any(self.layer & l for l in layer)
+
+
+ def has_feature_layers(self) -> bool:
+ """ Return true if any layer other than ADDRESS or POI is requested.
+ """
+ return self.layer_enabled(DataLayer.RAILWAY, DataLayer.MANMADE, DataLayer.NATURAL)
def _add_geometry_columns(self, sql: SaSelect, col: SaColumn) -> SaSelect:
if not self.details.geometry_output:
def _filter_by_layer(self, table: SaFromClause) -> SaColumn:
- if self.layer & DataLayer.MANMADE:
+ if self.layer_enabled(DataLayer.MANMADE):
exclude = []
- if not self.layer & DataLayer.RAILWAY:
+ if self.layer_disabled(DataLayer.RAILWAY):
exclude.append('railway')
- if not self.layer & DataLayer.NATURAL:
+ if self.layer_disabled(DataLayer.NATURAL):
exclude.extend(('natural', 'water', 'waterway'))
return table.c.class_.not_in(tuple(exclude))
include = []
- if self.layer & DataLayer.RAILWAY:
+ if self.layer_enabled(DataLayer.RAILWAY):
include.append('railway')
- if self.layer & DataLayer.NATURAL:
+ if self.layer_enabled(DataLayer.NATURAL):
include.extend(('natural', 'water', 'waterway'))
return table.c.class_.in_(tuple(include))
restrict: List[SaColumn] = []
- if self.layer & DataLayer.ADDRESS:
+ if self.layer_enabled(DataLayer.ADDRESS):
restrict.append(sa.and_(t.c.rank_address >= 26,
t.c.rank_address <= min(29, self.max_rank)))
if self.max_rank == 30:
restrict.append(_is_address_point(t))
- if self.layer & DataLayer.POI and self.max_rank == 30:
+ 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'))
- if self.layer & (DataLayer.RAILWAY | DataLayer.MANMADE | DataLayer.NATURAL):
+ if self.has_feature_layers():
restrict.append(sa.and_(t.c.rank_search.between(26, self.max_rank),
t.c.rank_address == 0,
self._filter_by_layer(t)))
sql = sa.select(inner.c.place_id, inner.c.osm_id,
inner.c.parent_place_id, inner.c.address,
_interpolated_housenumber(inner),
+ _interpolated_position(inner),
inner.c.postcode, inner.c.country_code,
- inner.c.linegeo.ST_LineInterpolatePoint(inner.c.position).label('centroid'),
inner.c.distance)
if self.details.geometry_output:
async def _find_tiger_number_for_street(self, parent_place_id: int,
+ parent_type: str, parent_id: int,
wkt: WKTElement) -> Optional[SaRow]:
t = self.conn.t.tiger
sql = sa.select(inner.c.place_id,
inner.c.parent_place_id,
+ sa.literal(parent_type).label('osm_type'),
+ sa.literal(parent_id).label('osm_id'),
_interpolated_housenumber(inner),
+ _interpolated_position(inner),
inner.c.postcode,
- inner.c.linegeo.ST_LineInterpolatePoint(inner.c.position).label('centroid'),
inner.c.distance)
if self.details.geometry_output:
return (await self.conn.execute(sql)).one_or_none()
- async def lookup_street_poi(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
+ async def lookup_street_poi(self,
+ wkt: WKTElement) -> Tuple[Optional[SaRow], RowFunc]:
""" Find a street or POI/address for the given WKT point.
"""
log().section('Reverse lookup on street/address level')
- result = None
distance = 0.006
parent_place_id = None
row = await self._find_closest_street_or_poi(wkt, distance)
+ row_func: RowFunc = nres.create_from_placex_row
log().var_dump('Result (street/building)', row)
# If the closest result was a street, but an address was requested,
# check for a housenumber nearby which is part of the street.
if row is not None:
if self.max_rank > 27 \
- and self.layer & DataLayer.ADDRESS \
+ and self.layer_enabled(DataLayer.ADDRESS) \
and row.rank_address <= 27:
distance = 0.001
parent_place_id = row.place_id
if addr_row is not None:
row = addr_row
+ row_func = nres.create_from_placex_row
distance = addr_row.distance
elif row.country_code == 'us' and parent_place_id is not None:
log().comment('Find TIGER housenumber for street')
- addr_row = await self._find_tiger_number_for_street(parent_place_id, wkt)
+ addr_row = await self._find_tiger_number_for_street(parent_place_id,
+ row.osm_type,
+ row.osm_id,
+ wkt)
log().var_dump('Result (street Tiger housenumber)', addr_row)
if addr_row is not None:
- result = nres.create_from_tiger_row(addr_row, nres.ReverseResult)
+ row = addr_row
+ row_func = nres.create_from_tiger_row
else:
distance = row.distance
# Check for an interpolation that is either closer than our result
# or belongs to a close street found.
- if self.max_rank > 27 and self.layer & DataLayer.ADDRESS:
+ 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)
log().var_dump('Result (street interpolation)', addr_row)
if addr_row is not None:
- result = nres.create_from_osmline_row(addr_row, nres.ReverseResult)
+ row = addr_row
+ row_func = nres.create_from_osmline_row
- return result or nres.create_from_placex_row(row, nres.ReverseResult)
+ return row, row_func
async def _lookup_area_address(self, wkt: WKTElement) -> Optional[SaRow]:
return row
- async def lookup_area(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
+ async def lookup_area(self, wkt: WKTElement) -> Optional[SaRow]:
""" Lookup large areas for the given WKT point.
"""
log().section('Reverse lookup by larger area features')
- if self.layer & DataLayer.ADDRESS:
+ if self.layer_enabled(DataLayer.ADDRESS):
address_row = await self._lookup_area_address(wkt)
else:
address_row = None
- if self.layer & (~DataLayer.ADDRESS & ~DataLayer.POI):
+ if self.has_feature_layers():
other_row = await self._lookup_area_others(wkt)
else:
other_row = None
- return nres.create_from_placex_row(_get_closest(address_row, other_row), nres.ReverseResult)
+ return _get_closest(address_row, other_row)
- async def lookup_country(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
+ async def lookup_country(self, wkt: WKTElement) -> Optional[SaRow]:
""" Lookup the country for the given WKT point.
"""
log().section('Reverse lookup by country code')
address_row = (await self.conn.execute(sql)).one_or_none()
- return nres.create_from_placex_row(address_row, nres.ReverseResult)
+ return address_row
async def lookup(self, coord: AnyPoint) -> Optional[nres.ReverseResult]:
wkt = WKTElement(f'POINT({coord[0]} {coord[1]})', srid=4326)
- result: Optional[nres.ReverseResult] = None
+ row: Optional[SaRow] = None
+ row_func: RowFunc = nres.create_from_placex_row
if self.max_rank >= 26:
- result = await self.lookup_street_poi(wkt)
- if result is None and self.max_rank > 4:
- result = await self.lookup_area(wkt)
- if result is None and self.layer & DataLayer.ADDRESS:
- result = await self.lookup_country(wkt)
+ row, tmp_row_func = await self.lookup_street_poi(wkt)
+ 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)
+ if row is None and self.layer_enabled(DataLayer.ADDRESS):
+ row = await self.lookup_country(wkt)
+
+ 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)
await nres.add_result_details(self.conn, result, self.details)
return result
projects / nominatim.git / blobdiff