async def _find_housenumber_for_street(self, parent_place_id: int) -> Optional[SaRow]:
t = self.conn.t.placex
- sql: SaLambdaSelect = sa.lambda_stmt(lambda: _select_from_placex(t)
- .where(t.c.geometry.within_distance(WKT_PARAM, 0.001))
- .where(t.c.parent_place_id == parent_place_id)
- .where(sa.func.IsAddressPoint(t))
- .where(t.c.indexed_status == 0)
- .where(t.c.linked_place_id == None)
- .order_by('distance')
- .limit(1))
-
+ def _base_query() -> SaSelect:
+ return _select_from_placex(t)\
+ .where(t.c.geometry.within_distance(WKT_PARAM, 0.001))\
+ .where(t.c.parent_place_id == parent_place_id)\
+ .where(sa.func.IsAddressPoint(t))\
+ .where(t.c.indexed_status == 0)\
+ .where(t.c.linked_place_id == None)\
+ .order_by('distance')\
+ .limit(1)
+
+ sql: SaLambdaSelect
if self.has_geometries():
- sql = self._add_geometry_columns(sql, t.c.geometry)
+ sql = self._add_geometry_columns(_base_query(), t.c.geometry)
+ else:
+ sql = sa.lambda_stmt(_base_query)
return (await self.conn.execute(sql, self.bind_params)).one_or_none()
distance: float) -> Optional[SaRow]:
t = self.conn.t.osmline
- sql: Any = sa.lambda_stmt(lambda:
- sa.select(t,
- t.c.linegeo.ST_Distance(WKT_PARAM).label('distance'),
- _locate_interpolation(t))
- .where(t.c.linegeo.within_distance(WKT_PARAM, distance))
- .where(t.c.startnumber != None)
- .order_by('distance')
- .limit(1))
+ sql = sa.select(t,
+ t.c.linegeo.ST_Distance(WKT_PARAM).label('distance'),
+ _locate_interpolation(t))\
+ .where(t.c.linegeo.within_distance(WKT_PARAM, distance))\
+ .where(t.c.startnumber != None)\
+ .order_by('distance')\
+ .limit(1)
if parent_place_id is not None:
- sql += lambda s: s.where(t.c.parent_place_id == parent_place_id)
+ sql = sql.where(t.c.parent_place_id == parent_place_id)
- def _wrap_query(base_sql: SaLambdaSelect) -> SaSelect:
- inner = base_sql.subquery('ipol')
+ inner = sql.subquery('ipol')
- return sa.select(inner.c.place_id, inner.c.osm_id,
+ 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.distance)
- sql += _wrap_query
-
if self.has_geometries():
sub = sql.subquery('geom')
sql = self._add_geometry_columns(sa.select(sub), sub.c.centroid)
inner.c.postcode,
inner.c.distance)
- sql: SaLambdaSelect = sa.lambda_stmt(_base_query)
-
+ sql: SaLambdaSelect
if self.has_geometries():
- sub = sql.subquery('geom')
+ sub = _base_query().subquery('geom')
sql = self._add_geometry_columns(sa.select(sub), sub.c.centroid)
+ else:
+ sql = sa.lambda_stmt(_base_query)
return (await self.conn.execute(sql, self.bind_params)).one_or_none()
.order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
.limit(1)
- sql = sa.lambda_stmt(_place_inside_area_query)
if self.has_geometries():
- sql = self._add_geometry_columns(sql, sa.literal_column('places.geometry'))
+ sql = self._add_geometry_columns(_place_inside_area_query(),
+ sa.literal_column('places.geometry'))
+ else:
+ sql = sa.lambda_stmt(_place_inside_area_query)
place_address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
log().var_dump('Result (place node)', place_address_row)
.order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
.limit(1)
- sql: SaLambdaSelect = sa.lambda_stmt(_base_query)
+ sql: SaLambdaSelect
if self.has_geometries():
- sql = self._add_geometry_columns(sql, sa.literal_column('area.geometry'))
+ sql = self._add_geometry_columns(_base_query(),
+ sa.literal_column('area.geometry'))
+ else:
+ sql = sa.lambda_stmt(_base_query)
address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
log().var_dump('Result (addressable place node)', address_row)
if address_row is None:
# Still nothing, then return a country with the appropriate country code.
- sql = sa.lambda_stmt(lambda: _select_from_placex(t)\
- .where(t.c.country_code.in_(ccodes))\
- .where(t.c.rank_address == 4)\
- .where(t.c.rank_search == 4)\
- .where(t.c.linked_place_id == None)\
- .order_by('distance')\
- .limit(1))
+ def _country_base_query() -> SaSelect:
+ return _select_from_placex(t)\
+ .where(t.c.country_code.in_(ccodes))\
+ .where(t.c.rank_address == 4)\
+ .where(t.c.rank_search == 4)\
+ .where(t.c.linked_place_id == None)\
+ .order_by('distance')\
+ .limit(1)
if self.has_geometries():
- sql = self._add_geometry_columns(sql, t.c.geometry)
+ sql = self._add_geometry_columns(_country_base_query(), t.c.geometry)
+ else:
+ sql = sa.lambda_stmt(_country_base_query)
address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
layers=napi.DataLayer.POI) is None
-def test_reverse_housenumber_on_street(apiobj, frontend):
+@pytest.mark.parametrize('with_geom', [True, False])
+def test_reverse_housenumber_on_street(apiobj, frontend, with_geom):
apiobj.add_placex(place_id=990, class_='highway', type='service',
rank_search=27, rank_address=27,
name = {'name': 'My Street'},
rank_search=30, rank_address=30,
housenumber='23',
centroid=(10.0, 10.00001))
+ apiobj.add_placex(place_id=1990, class_='highway', type='service',
+ rank_search=27, rank_address=27,
+ name = {'name': 'Other Street'},
+ centroid=(10.0, 1.0),
+ geometry='LINESTRING(9.995 1, 10.005 1)')
+ apiobj.add_placex(place_id=1991, class_='place', type='house',
+ parent_place_id=1990,
+ rank_search=30, rank_address=30,
+ housenumber='23',
+ centroid=(10.0, 1.00001))
+
+ params = {'geometry_output': napi.GeometryFormat.TEXT} if with_geom else {}
api = frontend(apiobj, options=API_OPTIONS)
- assert api.reverse((10.0, 10.0), max_rank=30).place_id == 991
+ assert api.reverse((10.0, 10.0), max_rank=30, **params).place_id == 991
assert api.reverse((10.0, 10.0), max_rank=27).place_id == 990
assert api.reverse((10.0, 10.00001), max_rank=30).place_id == 991
+ assert api.reverse((10.0, 1.0), **params).place_id == 1991
-def test_reverse_housenumber_interpolation(apiobj, frontend):
+@pytest.mark.parametrize('with_geom', [True, False])
+def test_reverse_housenumber_interpolation(apiobj, frontend, with_geom):
apiobj.add_placex(place_id=990, class_='highway', type='service',
rank_search=27, rank_address=27,
name = {'name': 'My Street'},
startnumber=1, endnumber=3, step=1,
centroid=(10.0, 10.00001),
geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
+ apiobj.add_placex(place_id=1990, class_='highway', type='service',
+ rank_search=27, rank_address=27,
+ name = {'name': 'Other Street'},
+ centroid=(10.0, 20.0),
+ geometry='LINESTRING(9.995 20, 10.005 20)')
+ apiobj.add_osmline(place_id=1992,
+ parent_place_id=1990,
+ startnumber=1, endnumber=3, step=1,
+ centroid=(10.0, 20.00001),
+ geometry='LINESTRING(9.995 20.00001, 10.005 20.00001)')
+
+ params = {'geometry_output': napi.GeometryFormat.TEXT} if with_geom else {}
api = frontend(apiobj, options=API_OPTIONS)
- assert api.reverse((10.0, 10.0)).place_id == 992
+ assert api.reverse((10.0, 10.0), **params).place_id == 992
+ assert api.reverse((10.0, 20.0), **params).place_id == 1992
def test_reverse_housenumber_point_interpolation(apiobj, frontend):
@pytest.mark.parametrize('rank', [4, 30])
-def test_reverse_country_lookup_country_only(apiobj, frontend, rank):
+@pytest.mark.parametrize('with_geom', [True, False])
+def test_reverse_country_lookup_country_only(apiobj, frontend, rank, with_geom):
apiobj.add_country('xx', 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))')
+ apiobj.add_country('yy', 'POLYGON((10 0, 10 1, 11 1, 11 0, 10 0))')
apiobj.add_placex(place_id=225, class_='place', type='country',
name={'name': 'My Country'},
rank_address=4,
country_code='xx',
centroid=(0.7, 0.7))
+ params = {'max_rank': rank}
+ if with_geom:
+ params['geometry_output'] = napi.GeometryFormat.TEXT
+
api = frontend(apiobj, options=API_OPTIONS)
- assert api.reverse((0.5, 0.5), max_rank=rank).place_id == 225
+ assert api.reverse((0.5, 0.5), **params).place_id == 225
+ assert api.reverse((10.5, 0.5), **params) is None
-def test_reverse_country_lookup_place_node_inside(apiobj, frontend):
+@pytest.mark.parametrize('with_geom', [True, False])
+def test_reverse_country_lookup_place_node_inside(apiobj, frontend, with_geom):
apiobj.add_country('xx', 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))')
+ apiobj.add_country('yy', 'POLYGON((10 0, 10 1, 11 1, 11 0, 10 0))')
apiobj.add_placex(place_id=225, class_='place', type='state',
osm_type='N',
name={'name': 'My State'},
rank_search=6,
country_code='xx',
centroid=(0.5, 0.505))
+ apiobj.add_placex(place_id=425, class_='place', type='state',
+ osm_type='N',
+ name={'name': 'Other State'},
+ rank_address=6,
+ rank_search=6,
+ country_code='yy',
+ centroid=(10.5, 0.505))
+
+ params = {'geometry_output': napi.GeometryFormat.KML} if with_geom else {}
api = frontend(apiobj, options=API_OPTIONS)
- assert api.reverse((0.5, 0.5)).place_id == 225
+ assert api.reverse((0.5, 0.5), **params).place_id == 225
+ assert api.reverse((10.5, 0.5), **params).place_id == 425
@pytest.mark.parametrize('gtype', list(napi.GeometryFormat))
startnumber=1, endnumber=3, step=1,
centroid=(10.0, 10.00001),
geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
+ apiobj.add_placex(place_id=1000, class_='highway', type='service',
+ rank_search=27, rank_address=27,
+ name = {'name': 'My Street'},
+ centroid=(11.0, 11.0),
+ country_code='us',
+ geometry='LINESTRING(10.995 11, 11.005 11)')
+ apiobj.add_tiger(place_id=1001,
+ parent_place_id=1000,
+ startnumber=1, endnumber=3, step=1,
+ centroid=(11.0, 11.00001),
+ geometry='LINESTRING(10.995 11.00001, 11.005 11.00001)')
api = frontend(apiobj, options=API_OPTIONS)
- output = api.reverse((10.0, 10.0),
- geometry_output=napi.GeometryFormat.GEOJSON).geometry['geojson']
- assert json.loads(output) == {'coordinates': [10, 10.00001], 'type': 'Point'}
+ params = {'geometry_output': napi.GeometryFormat.GEOJSON}
+
+ output = api.reverse((10.0, 10.0), **params)
+ assert json.loads(output.geometry['geojson']) == {'coordinates': [10, 10.00001], 'type': 'Point'}
+
+ output = api.reverse((11.0, 11.0), **params)
+ assert json.loads(output.geometry['geojson']) == {'coordinates': [11, 11.00001], 'type': 'Point'}
projects / nominatim.git / commitdiff