# 'too-many-ancestors' is triggered already by deriving from UserDict
# 'not-context-manager' disabled because it causes false positives once
# typed Python is enabled. See also https://github.com/PyCQA/pylint/issues/5273
-disable=too-few-public-methods,duplicate-code,too-many-ancestors,bad-option-value,no-self-use,not-context-manager,use-dict-literal
+disable=too-few-public-methods,duplicate-code,too-many-ancestors,bad-option-value,no-self-use,not-context-manager,use-dict-literal,chained-comparison
-good-names=i,x,y,m,t,fd,db,cc
+good-names=i,x,y,m,t,fd,db,cc,x1,x2,y1,y2,pt
from nominatim.api.connection import SearchConnection
from nominatim.api.status import get_status, StatusResult
from nominatim.api.lookup import get_place_by_id
-from nominatim.api.reverse import reverse_lookup
+from nominatim.api.reverse import ReverseGeocoder
from nominatim.api.types import PlaceRef, LookupDetails, AnyPoint, DataLayer
from nominatim.api.results import DetailedResult, ReverseResult
max_rank = max(0, min(max_rank or 30, 30))
async with self.begin() as conn:
- return await reverse_lookup(conn, coord, max_rank, layer,
- details or LookupDetails())
+ geocoder = ReverseGeocoder(conn, max_rank, layer,
+ details or LookupDetails())
+ return await geocoder.lookup(coord)
class NominatimAPI:
""" Print the SQL for the given statement.
"""
+ def format_sql(self, conn: AsyncConnection, statement: 'sa.Executable') -> str:
+ """ Return the comiled version of the statement.
+ """
+ try:
+ return str(cast('sa.ClauseElement', statement)
+ .compile(conn.sync_engine, compile_kwargs={"literal_binds": True}))
+ except sa.exc.CompileError:
+ pass
+
+ return str(cast('sa.ClauseElement', statement).compile(conn.sync_engine))
+
class HTMLLogger(BaseLogger):
""" Logger that formats messages in HTML.
def sql(self, conn: AsyncConnection, statement: 'sa.Executable') -> None:
- sqlstr = str(cast('sa.ClauseElement', statement)
- .compile(conn.sync_engine, compile_kwargs={"literal_binds": True}))
+ sqlstr = self.format_sql(conn, statement)
if CODE_HIGHLIGHT:
sqlstr = highlight(sqlstr, PostgresLexer(),
HtmlFormatter(nowrap=True, lineseparator='<br />'))
def sql(self, conn: AsyncConnection, statement: 'sa.Executable') -> None:
- sqlstr = str(cast('sa.ClauseElement', statement)
- .compile(conn.sync_engine, compile_kwargs={"literal_binds": True}))
- sqlstr = '\n| '.join(textwrap.wrap(sqlstr, width=78))
+ sqlstr = '\n| '.join(textwrap.wrap(self.format_sql(conn, statement), width=78))
self._write(f"| {sqlstr}\n\n")
assert result is not None
result.parent_place_id = row.parent_place_id
result.linked_place_id = getattr(row, 'linked_place_id', None)
+ result.admin_level = getattr(row, 'admin_level', 15)
indexed_date = getattr(row, 'indexed_date', None)
if indexed_date is not None:
result.indexed_date = indexed_date.replace(tzinfo=dt.timezone.utc)
names: Dict[str, str]
extratags: Optional[Dict[str, str]]
- local_name: Optional[str] = None
-
admin_level: Optional[int]
fromarea: bool
isaddress: bool
place_id : Optional[int] = None
osm_object: Optional[Tuple[str, int]] = None
- admin_level: int = 15
names: Optional[Dict[str, str]] = None
address: Optional[Dict[str, str]] = None
"""
parent_place_id: Optional[int] = None
linked_place_id: Optional[int] = None
+ admin_level: int = 15
indexed_date: Optional[dt.datetime] = None
place_id=row.place_id,
osm_object=(row.osm_type, row.osm_id),
category=(row.class_, row.type),
- admin_level=row.admin_level,
names=row.name,
address=row.address,
extratags=row.extratags,
"""
Implementation of reverse geocoding.
"""
-from typing import Optional
+from typing import Optional, List
import sqlalchemy as sa
from geoalchemy2 import WKTElement
-from geoalchemy2.types import Geometry
-from nominatim.typing import SaColumn, SaSelect, SaTable, SaLabel, SaClause
+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, LookupDetails, GeometryFormat
-def _select_from_placex(t: SaTable, wkt: Optional[str] = None) -> SaSelect:
+# In SQLAlchemy expression which compare with NULL need to be expressed with
+# the equal sign.
+# pylint: disable=singleton-comparison
+
+def _select_from_placex(t: SaFromClause, wkt: Optional[str] = None) -> SaSelect:
""" Create a select statement with the columns relevant for reverse
results.
"""
t.c.geometry.ST_Expand(0).label('bbox'))
-def _interpolated_housenumber(table: SaTable) -> SaLabel:
- # Entries with startnumber = endnumber are legacy from version < 4.1
+def _interpolated_housenumber(table: SaFromClause) -> SaLabel:
return sa.cast(table.c.startnumber
+ sa.func.round(((table.c.endnumber - table.c.startnumber) * table.c.position)
/ table.c.step) * table.c.step,
sa.Integer).label('housenumber')
-def _is_address_point(table: SaTable) -> SaClause:
+def _is_address_point(table: SaFromClause) -> SaColumn:
return sa.and_(table.c.rank_address == 30,
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
return sql.add_columns(*out)
- def _filter_by_layer(self, table: SaTable) -> SaColumn:
+ def _filter_by_layer(self, table: SaFromClause) -> SaColumn:
if self.layer & DataLayer.MANMADE:
exclude = []
- if not (self.layer & DataLayer.RAILWAY):
+ if not self.layer & DataLayer.RAILWAY:
exclude.append('railway')
- if not (self.layer & DataLayer.NATURAL):
+ if not self.layer & DataLayer.NATURAL:
exclude.extend(('natural', 'water', 'waterway'))
return table.c.class_.not_in(tuple(exclude))
include = []
if self.layer & DataLayer.RAILWAY:
include.append('railway')
- if not (self.layer & DataLayer.NATURAL):
+ if self.layer & DataLayer.NATURAL:
include.extend(('natural', 'water', 'waterway'))
return table.c.class_.in_(tuple(include))
- async def _find_closest_street_or_poi(self, wkt: WKTElement) -> SaRow:
- """ Look up the clostest rank 26+ place in the database.
+ async def _find_closest_street_or_poi(self, wkt: WKTElement,
+ 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
.where(t.c.geometry.ST_DWithin(wkt, 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')),
+ .where(sa.or_(t.c.geometry.ST_GeometryType()
+ .not_in(('ST_Polygon', 'ST_MultiPolygon')),
t.c.centroid.ST_Distance(wkt) < distance))\
.order_by('distance')\
.limit(1)
sql = self._add_geometry_columns(sql, t.c.geometry)
- restrict = []
+ restrict: List[SaColumn] = []
if self.layer & DataLayer.ADDRESS:
restrict.append(sa.and_(t.c.rank_address >= 26,
- t.c.rank_address <= self.max_rank))
+ 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 max_rank == 30:
+ if self.layer & 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):
- restrict.append(sa.and_(t.c.rank_search >= 26,
- tc.rank_search <= self.max_rank,
+ restrict.append(sa.and_(t.c.rank_search.between(26, self.max_rank),
+ t.c.rank_address == 0,
self._filter_by_layer(t)))
- if restrict:
- sql = sql.where(sa.or_(*restrict))
+ if not restrict:
+ return None
- return (await self.conn.execute(sql)).one_or_none()
+ return (await self.conn.execute(sql.where(sa.or_(*restrict)))).one_or_none()
async def _find_housenumber_for_street(self, parent_place_id: int,
wkt: WKTElement) -> Optional[SaRow]:
- t = conn.t.placex
+ t = self.conn.t.placex
sql = _select_from_placex(t, wkt)\
.where(t.c.geometry.ST_DWithin(wkt, 0.001))\
async def _find_interpolation_for_street(self, parent_place_id: Optional[int],
- wkt: WKTElement) -> Optional[SaRow]:
+ wkt: WKTElement,
+ distance: float) -> Optional[SaRow]:
t = self.conn.t.osmline
- inner = sa.select(t,
- t.c.linegeo.ST_Distance(wkt).label('distance'),
- t.c.linegeo.ST_LineLocatePoint(wkt).label('position'))\
- .where(t.c.linegeo.ST_DWithin(wkt, distance))\
- .order_by('distance')\
- .limit(1)
+ sql = sa.select(t,
+ t.c.linegeo.ST_Distance(wkt).label('distance'),
+ t.c.linegeo.ST_LineLocatePoint(wkt).label('position'))\
+ .where(t.c.linegeo.ST_DWithin(wkt, distance))\
+ .order_by('distance')\
+ .limit(1)
if parent_place_id is not None:
- inner = inner.where(t.c.parent_place_id == parent_place_id)
+ sql = sql.where(t.c.parent_place_id == parent_place_id)
- inner = inner.subquery()
+ inner = sql.subquery()
sql = sa.select(inner.c.place_id, inner.c.osm_id,
inner.c.parent_place_id, inner.c.address,
sub = sql.subquery()
sql = self._add_geometry_columns(sql, sub.c.centroid)
- return (await conn.execute(sql)).one_or_none()
+ return (await self.conn.execute(sql)).one_or_none()
async def lookup_street_poi(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
distance = 0.006
parent_place_id = None
- row = await self._find_closest_street_or_poi(wkt)
+ row = await self._find_closest_street_or_poi(wkt, distance)
log().var_dump('Result (street/building)', row)
# If the closest result was a street, but an address was requested,
log().var_dump('Result (street Tiger housenumber)', addr_row)
if addr_row is not None:
- result = nres.create_from_tiger_row(addr_row)
+ result = nres.create_from_tiger_row(addr_row, nres.ReverseResult)
else:
distance = row.distance
# or belongs to a close street found.
if self.max_rank > 27 and self.layer & DataLayer.ADDRESS:
log().comment('Find interpolation for street')
- addr_row = await self._find_interpolation_for_street(parent_place_id, wkt)
+ 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)
+ result = nres.create_from_osmline_row(addr_row, nres.ReverseResult)
- return result or nres.create_from_placex_row(row)
+ return result or nres.create_from_placex_row(row, nres.ReverseResult)
async def _lookup_area_address(self, wkt: WKTElement) -> Optional[SaRow]:
address_row = (await self.conn.execute(sql)).one_or_none()
log().var_dump('Result (area)', address_row)
- if address_row is not None and address_row.rank_search < max_rank:
+ 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'))\
.where(t.c.osm_type == 'N')\
.where(t.c.rank_search > address_row.rank_search)\
- .where(t.c.rank_search <= max_rank)\
+ .where(t.c.rank_search <= self.max_rank)\
.where(t.c.rank_address.between(5, 25))\
.where(t.c.name != None)\
.where(t.c.indexed_status == 0)\
.limit(50)\
.subquery()
- touter = conn.t.placex.alias('outer')
+ touter = self.conn.t.placex.alias('outer')
sql = _select_from_placex(inner)\
+ .join(touter, touter.c.geometry.ST_Contains(inner.c.geometry))\
.where(touter.c.place_id == address_row.place_id)\
- .where(touter.c.geometry.ST_Contains(inner.c.geometry))\
.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)
async def _lookup_area_others(self, wkt: WKTElement) -> Optional[SaRow]:
- t = conn.t.placex
+ t = self.conn.t.placex
inner = sa.select(t, t.c.geometry.ST_Distance(wkt).label('distance'))\
.where(t.c.rank_address == 0)\
.where(t.c.indexed_status == 0)\
.where(t.c.linked_place_id == None)\
.where(self._filter_by_layer(t))\
- .where(sa.func.reverse_buffered_extent(t.c.geometry, type_=Geometry)
+ .where(t.c.geometry
+ .ST_Buffer(sa.func.reverse_place_diameter(t.c.rank_search))
.intersects(wkt))\
.order_by(sa.desc(t.c.rank_search))\
- .limit(50)
+ .limit(50)\
+ .subquery()
sql = _select_from_placex(inner)\
- .where(sa._or(inner.c.geometry.ST_GeometryType().not_in(('ST_Polygon', 'ST_MultiPolygon')),
+ .where(sa.or_(inner.c.geometry.ST_GeometryType()
+ .not_in(('ST_Polygon', 'ST_MultiPolygon')),
inner.c.geometry.ST_Contains(wkt)))\
.order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
.limit(1)
""" Lookup large areas for the given WKT point.
"""
log().section('Reverse lookup by larger area features')
- t = self.conn.t.placex
if self.layer & DataLayer.ADDRESS:
address_row = await self._lookup_area_address(wkt)
- address_distance = address_row.distance
else:
address_row = None
- address_distance = 1000
if self.layer & (~DataLayer.ADDRESS & ~DataLayer.POI):
other_row = await self._lookup_area_others(wkt)
- other_distance = other_row.distance
else:
other_row = None
- other_distance = 1000
-
- result = address_row if address_distance <= other_distance else other_row
- return nres.create_from_placex_row(result)
+ return nres.create_from_placex_row(_get_closest(address_row, other_row), nres.ReverseResult)
async def lookup_country(self, wkt: WKTElement) -> Optional[nres.ReverseResult]:
if not ccodes:
return None
- if self.layer & DataLayer.ADDRESS and self.max_rank > 4:
+ t = self.conn.t.placex
+ if self.max_rank > 4:
log().comment('Search for place nodes in country')
- t = conn.t.placex
inner = sa.select(t,
t.c.geometry.ST_Distance(wkt).label('distance'))\
.where(t.c.osm_type == 'N')\
else:
address_row = None
- if layer & (~DataLayer.ADDRESS & ~DataLayer.POI) and self.max_rank > 4:
- log().comment('Search for non-address features inside country')
-
- t = conn.t.placex
- inner = sa.select(t, t.c.geometry.ST_Distance(wkt).label('distance'))\
- .where(t.c.rank_address == 0)\
- .where(t.c.rank_search.between(5, self.max_rank))\
- .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.country_code.in_(ccode))\
- .where(sa.func.reverse_buffered_extent(t.c.geometry, type_=Geometry)
- .intersects(wkt))\
- .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)))\
- .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
- .limit(1)
-
- sql = self._add_geometry_columns(sql, inner.c.geometry)
-
- other_row = (await self.conn.execute(sql)).one_or_none()
- log().var_dump('Result (non-address feature)', other_row)
- else:
- other_row = None
-
- if layer & DataLayer.ADDRESS and address_row is None and other_row is None:
+ if address_row is None:
# Still nothing, then return a country with the appropriate country code.
- t = conn.t.placex
sql = _select_from_placex(t, wkt)\
.where(t.c.country_code.in_(ccodes))\
.where(t.c.rank_address == 4)\
.where(t.c.linked_place_id == None)\
.order_by('distance')
- sql = self._add_geometry_columns(sql, inner.c.geometry)
+ sql = self._add_geometry_columns(sql, t.c.geometry)
address_row = (await self.conn.execute(sql)).one_or_none()
- return nres.create_from_placex_row(_get_closest_row(address_row, other_row))
+ return nres.create_from_placex_row(address_row, nres.ReverseResult)
async def lookup(self, coord: AnyPoint) -> Optional[nres.ReverseResult]:
wkt = WKTElement(f'POINT({coord[0]} {coord[1]})', srid=4326)
- result: Optional[ReverseResult] = None
+ result: Optional[nres.ReverseResult] = None
- if max_rank >= 26:
+ if self.max_rank >= 26:
result = await self.lookup_street_poi(wkt)
- if result is None and max_rank > 4:
+ if result is None and self.max_rank > 4:
result = await self.lookup_area(wkt)
- if result is None:
+ if result is None and self.layer & DataLayer.ADDRESS:
result = await self.lookup_country(wkt)
if result is not None:
await nres.add_result_details(self.conn, result, self.details)
AnyPoint = Union[Point, Tuple[float, float]]
+WKB_BBOX_HEADER_LE = b'\x01\x03\x00\x00\x20\xE6\x10\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00'
+WKB_BBOX_HEADER_BE = b'\x00\x20\x00\x00\x03\x00\x00\x10\xe6\x00\x00\x00\x01\x00\x00\x00\x05'
class Bbox:
""" A bounding box in WSG84 projection.
if len(wkb) != 97:
raise ValueError("WKB must be a bounding box polygon")
- if wkb.startswith(b'\x01\x03\x00\x00\x20\xE6\x10\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00'):
+ if wkb.startswith(WKB_BBOX_HEADER_LE):
x1, y1, _, _, x2, y2 = unpack('<dddddd', wkb[17:65])
- elif wkb.startswith(b'\x00\x20\x00\x00\x03\x00\x00\x10\xe6\x00\x00\x00\x01\x00\x00\x00\x05'):
+ elif wkb.startswith(WKB_BBOX_HEADER_BE):
x1, y1, _, _, x2, y2 = unpack('>dddddd', wkb[17:65])
else:
raise ValueError("WKB has wrong header")
return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2))
+ @staticmethod
def from_point(pt: Point, buffer: float) -> 'Bbox':
""" Return a Bbox around the point with the buffer added to all sides.
"""
SaSelect: TypeAlias = 'sa.Select[Any]'
SaRow: TypeAlias = 'sa.Row[Any]'
-SaColumn: TypeAlias = 'sa.Column[Any]'
+SaColumn: TypeAlias = 'sa.ColumnElement[Any]'
SaLabel: TypeAlias = 'sa.Label[Any]'
-SaTable: TypeAlias = 'sa.Table[Any]'
-SaClause: TypeAlias = 'sa.ClauseElement[Any]'
+SaFromClause: TypeAlias = 'sa.FromClause'
+SaSelectable: TypeAlias = 'sa.Selectable'
if isinstance(name, str):
name = {'name': name}
+ centroid = kw.get('centroid', (23.0, 34.0))
+ geometry = kw.get('geometry', 'POINT(%f %f)' % centroid)
+
self.add_data('placex',
{'place_id': kw.get('place_id', 1000),
'osm_type': kw.get('osm_type', 'W'),
'rank_search': kw.get('rank_search', 30),
'rank_address': kw.get('rank_address', 30),
'importance': kw.get('importance'),
- 'centroid': 'SRID=4326;POINT(%f %f)' % kw.get('centroid', (23.0, 34.0)),
+ 'centroid': 'SRID=4326;POINT(%f %f)' % centroid,
+ 'indexed_status': kw.get('indexed_status', 0),
'indexed_date': kw.get('indexed_date',
dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
- 'geometry': 'SRID=4326;' + kw.get('geometry', 'POINT(23 34)')})
+ 'geometry': 'SRID=4326;' + geometry})
def add_address_placex(self, object_id, **kw):
'geometry': 'SRID=4326;' + kw.get('geometry', 'POINT(23 34)')})
+ def add_country(self, country_code, geometry):
+ self.add_data('country_grid',
+ {'country_code': country_code,
+ 'area': 0.1,
+ 'geometry': 'SRID=4326;' + geometry})
+
+
async def exec_async(self, sql, *args, **kwargs):
async with self.api._async_api.begin() as conn:
return await conn.execute(sql, *args, **kwargs)
testapi = APITester()
testapi.async_to_sync(testapi.create_tables())
- SQLPreprocessor(temp_db_conn, testapi.api.config)\
- .run_sql_file(temp_db_conn, 'functions/address_lookup.sql')
+ proc = SQLPreprocessor(temp_db_conn, testapi.api.config)
+ proc.run_sql_file(temp_db_conn, 'functions/address_lookup.sql')
+ proc.run_sql_file(temp_db_conn, 'functions/ranking.sql')
loglib.set_log_output('text')
yield testapi
--- /dev/null
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2023 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Tests for reverse API call.
+
+These tests make sure that all Python code is correct and executable.
+Functional tests can be found in the BDD test suite.
+"""
+import json
+
+import pytest
+
+import nominatim.api as napi
+
+def test_reverse_rank_30(apiobj):
+ apiobj.add_placex(place_id=223, class_='place', type='house',
+ housenumber='1',
+ centroid=(1.3, 0.7),
+ geometry='POINT(1.3 0.7)')
+
+ result = apiobj.api.reverse((1.3, 0.7))
+
+ assert result is not None
+ assert result.place_id == 223
+
+
+@pytest.mark.parametrize('country', ['de', 'us'])
+def test_reverse_street(apiobj, country):
+ apiobj.add_placex(place_id=990, class_='highway', type='service',
+ rank_search=27, rank_address=27,
+ name = {'name': 'My Street'},
+ centroid=(10.0, 10.0),
+ country_code=country,
+ geometry='LINESTRING(9.995 10, 10.005 10)')
+
+ assert apiobj.api.reverse((9.995, 10)).place_id == 990
+
+
+def test_reverse_ignore_unindexed(apiobj):
+ apiobj.add_placex(place_id=223, class_='place', type='house',
+ housenumber='1',
+ indexed_status=2,
+ centroid=(1.3, 0.7),
+ geometry='POINT(1.3 0.7)')
+
+ result = apiobj.api.reverse((1.3, 0.7))
+
+ assert result is None
+
+
+@pytest.mark.parametrize('y,layer,place_id', [(0.7, napi.DataLayer.ADDRESS, 223),
+ (0.70001, napi.DataLayer.POI, 224),
+ (0.7, napi.DataLayer.ADDRESS | napi.DataLayer.POI, 224),
+ (0.70001, napi.DataLayer.ADDRESS | napi.DataLayer.POI, 223),
+ (0.7, napi.DataLayer.MANMADE, 225),
+ (0.7, napi.DataLayer.RAILWAY, 226),
+ (0.7, napi.DataLayer.NATURAL, 227),
+ (0.70003, napi.DataLayer.MANMADE | napi.DataLayer.RAILWAY, 225),
+ (0.70003, napi.DataLayer.MANMADE | napi.DataLayer.NATURAL, 225)])
+def test_reverse_rank_30_layers(apiobj, y, layer, place_id):
+ apiobj.add_placex(place_id=223, class_='place', type='house',
+ housenumber='1',
+ rank_address=30,
+ rank_search=30,
+ centroid=(1.3, 0.70001))
+ apiobj.add_placex(place_id=224, class_='amenity', type='toilet',
+ rank_address=30,
+ rank_search=30,
+ centroid=(1.3, 0.7))
+ apiobj.add_placex(place_id=225, class_='man_made', type='tower',
+ rank_address=0,
+ rank_search=30,
+ centroid=(1.3, 0.70003))
+ apiobj.add_placex(place_id=226, class_='railway', type='station',
+ rank_address=0,
+ rank_search=30,
+ centroid=(1.3, 0.70004))
+ apiobj.add_placex(place_id=227, class_='natural', type='cave',
+ rank_address=0,
+ rank_search=30,
+ centroid=(1.3, 0.70005))
+
+ assert apiobj.api.reverse((1.3, y), layer=layer).place_id == place_id
+
+
+def test_reverse_poi_layer_with_no_pois(apiobj):
+ apiobj.add_placex(place_id=223, class_='place', type='house',
+ housenumber='1',
+ rank_address=30,
+ rank_search=30,
+ centroid=(1.3, 0.70001))
+
+ assert apiobj.api.reverse((1.3, 0.70001), max_rank=29,
+ layer=napi.DataLayer.POI) is None
+
+
+def test_reverse_housenumber_on_street(apiobj):
+ apiobj.add_placex(place_id=990, class_='highway', type='service',
+ rank_search=27, rank_address=27,
+ name = {'name': 'My Street'},
+ centroid=(10.0, 10.0),
+ geometry='LINESTRING(9.995 10, 10.005 10)')
+ apiobj.add_placex(place_id=991, class_='place', type='house',
+ parent_place_id=990,
+ rank_search=30, rank_address=30,
+ housenumber='23',
+ centroid=(10.0, 10.00001))
+
+ assert apiobj.api.reverse((10.0, 10.0), max_rank=30).place_id == 991
+ assert apiobj.api.reverse((10.0, 10.0), max_rank=27).place_id == 990
+ assert apiobj.api.reverse((10.0, 10.00001), max_rank=30).place_id == 991
+
+
+def test_reverse_housenumber_interpolation(apiobj):
+ apiobj.add_placex(place_id=990, class_='highway', type='service',
+ rank_search=27, rank_address=27,
+ name = {'name': 'My Street'},
+ centroid=(10.0, 10.0),
+ geometry='LINESTRING(9.995 10, 10.005 10)')
+ apiobj.add_placex(place_id=991, class_='place', type='house',
+ parent_place_id=990,
+ rank_search=30, rank_address=30,
+ housenumber='23',
+ centroid=(10.0, 10.00002))
+ apiobj.add_osmline(place_id=992,
+ parent_place_id=990,
+ startnumber=1, endnumber=3, step=1,
+ centroid=(10.0, 10.00001),
+ geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
+
+ assert apiobj.api.reverse((10.0, 10.0)).place_id == 992
+
+
+def test_reverse_tiger_number(apiobj):
+ apiobj.add_placex(place_id=990, class_='highway', type='service',
+ rank_search=27, rank_address=27,
+ name = {'name': 'My Street'},
+ centroid=(10.0, 10.0),
+ country_code='us',
+ geometry='LINESTRING(9.995 10, 10.005 10)')
+ apiobj.add_tiger(place_id=992,
+ parent_place_id=990,
+ startnumber=1, endnumber=3, step=1,
+ centroid=(10.0, 10.00001),
+ geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
+
+ assert apiobj.api.reverse((10.0, 10.0)).place_id == 992
+ assert apiobj.api.reverse((10.0, 10.00001)).place_id == 992
+
+
+def test_reverse_low_zoom_address(apiobj):
+ apiobj.add_placex(place_id=1001, class_='place', type='house',
+ housenumber='1',
+ rank_address=30,
+ rank_search=30,
+ centroid=(59.3, 80.70001))
+ apiobj.add_placex(place_id=1002, class_='place', type='town',
+ name={'name': 'Town'},
+ rank_address=16,
+ rank_search=16,
+ centroid=(59.3, 80.70001),
+ geometry="""POLYGON((59.3 80.70001, 59.3001 80.70001,
+ 59.3001 80.70101, 59.3 80.70101, 59.3 80.70001))""")
+
+ assert apiobj.api.reverse((59.30005, 80.7005)).place_id == 1001
+ assert apiobj.api.reverse((59.30005, 80.7005), max_rank=18).place_id == 1002
+
+
+def test_reverse_place_node_in_area(apiobj):
+ apiobj.add_placex(place_id=1002, class_='place', type='town',
+ name={'name': 'Town Area'},
+ rank_address=16,
+ rank_search=16,
+ centroid=(59.3, 80.70001),
+ geometry="""POLYGON((59.3 80.70001, 59.3001 80.70001,
+ 59.3001 80.70101, 59.3 80.70101, 59.3 80.70001))""")
+ apiobj.add_placex(place_id=1003, class_='place', type='suburb',
+ name={'name': 'Suburb Point'},
+ osm_type='N',
+ rank_address=18,
+ rank_search=18,
+ centroid=(59.30004, 80.70055))
+
+ assert apiobj.api.reverse((59.30004, 80.70055)).place_id == 1003
+
+
+@pytest.mark.parametrize('layer,place_id', [(napi.DataLayer.MANMADE, 225),
+ (napi.DataLayer.RAILWAY, 226),
+ (napi.DataLayer.NATURAL, 227),
+ (napi.DataLayer.MANMADE | napi.DataLayer.RAILWAY, 225),
+ (napi.DataLayer.MANMADE | napi.DataLayer.NATURAL, 225)])
+def test_reverse_larger_area_layers(apiobj, layer, place_id):
+ apiobj.add_placex(place_id=225, class_='man_made', type='dam',
+ name={'name': 'Dam'},
+ rank_address=0,
+ rank_search=25,
+ centroid=(1.3, 0.70003))
+ apiobj.add_placex(place_id=226, class_='railway', type='yard',
+ name={'name': 'Dam'},
+ rank_address=0,
+ rank_search=20,
+ centroid=(1.3, 0.70004))
+ apiobj.add_placex(place_id=227, class_='natural', type='spring',
+ name={'name': 'Dam'},
+ rank_address=0,
+ rank_search=16,
+ centroid=(1.3, 0.70005))
+
+ assert apiobj.api.reverse((1.3, 0.7), layer=layer).place_id == place_id
+
+
+def test_reverse_country_lookup_no_objects(apiobj):
+ apiobj.add_country('xx', 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))')
+
+ assert apiobj.api.reverse((0.5, 0.5)) is None
+
+
+@pytest.mark.parametrize('rank', [4, 30])
+def test_reverse_country_lookup_country_only(apiobj, rank):
+ apiobj.add_country('xx', 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))')
+ apiobj.add_placex(place_id=225, class_='place', type='country',
+ name={'name': 'My Country'},
+ rank_address=4,
+ rank_search=4,
+ country_code='xx',
+ centroid=(0.7, 0.7))
+
+ assert apiobj.api.reverse((0.5, 0.5), max_rank=rank).place_id == 225
+
+
+def test_reverse_country_lookup_place_node_inside(apiobj):
+ apiobj.add_country('xx', 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))')
+ apiobj.add_placex(place_id=225, class_='place', type='state',
+ osm_type='N',
+ name={'name': 'My State'},
+ rank_address=6,
+ rank_search=6,
+ country_code='xx',
+ centroid=(0.5, 0.505))
+
+ assert apiobj.api.reverse((0.5, 0.5)).place_id == 225
+
+
+@pytest.mark.parametrize('gtype', list(napi.GeometryFormat))
+def test_reverse_geometry_output_placex(apiobj, gtype):
+ apiobj.add_country('xx', 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))')
+ apiobj.add_placex(place_id=1001, class_='place', type='house',
+ housenumber='1',
+ rank_address=30,
+ rank_search=30,
+ centroid=(59.3, 80.70001))
+ apiobj.add_placex(place_id=1003, class_='place', type='suburb',
+ name={'name': 'Suburb Point'},
+ osm_type='N',
+ rank_address=18,
+ rank_search=18,
+ country_code='xx',
+ centroid=(0.5, 0.5))
+
+ details = napi.LookupDetails(geometry_output=gtype)
+
+ assert apiobj.api.reverse((59.3, 80.70001), details=details).place_id == 1001
+ assert apiobj.api.reverse((0.5, 0.5), details=details).place_id == 1003
+
+
+def test_reverse_simplified_geometry(apiobj):
+ apiobj.add_placex(place_id=1001, class_='place', type='house',
+ housenumber='1',
+ rank_address=30,
+ rank_search=30,
+ centroid=(59.3, 80.70001))
+
+ details = napi.LookupDetails(geometry_output=napi.GeometryFormat.GEOJSON,
+ geometry_simplification=0.1)
+ assert apiobj.api.reverse((59.3, 80.70001), details=details).place_id == 1001
+
+
+def test_reverse_interpolation_geometry(apiobj):
+ apiobj.add_osmline(place_id=992,
+ parent_place_id=990,
+ startnumber=1, endnumber=3, step=1,
+ centroid=(10.0, 10.00001),
+ geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
+
+ details = napi.LookupDetails(geometry_output=napi.GeometryFormat.TEXT)
+ assert apiobj.api.reverse((10.0, 10.0), details=details)\
+ .geometry['text'] == 'POINT(10 10.00001)'
+
+
+def test_reverse_tiger_geometry(apiobj):
+ apiobj.add_placex(place_id=990, class_='highway', type='service',
+ rank_search=27, rank_address=27,
+ name = {'name': 'My Street'},
+ centroid=(10.0, 10.0),
+ country_code='us',
+ geometry='LINESTRING(9.995 10, 10.005 10)')
+ apiobj.add_tiger(place_id=992,
+ parent_place_id=990,
+ startnumber=1, endnumber=3, step=1,
+ centroid=(10.0, 10.00001),
+ geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
+
+ details = napi.LookupDetails(geometry_output=napi.GeometryFormat.GEOJSON)
+ output = apiobj.api.reverse((10.0, 10.0), details=details).geometry['geojson']
+
+ assert json.loads(output) == {'coordinates': [10, 10.00001], 'type': 'Point'}
+
projects / nominatim.git / commitdiff