safety-screen/Nominatim
1
0
Fork 0
forked from hans/Nominatim
Code Pull Requests Activity

add lookup of address interpolations

Browse Source
This commit is contained in:
Sarah Hoffmann
2023-02-02 15:01:54 +01:00
parent e1fc1566f3
commit f1ceefe9a6
4 changed files with 201 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
nominatim/api/lookup.py
View File

@@ -64,6 +64,35 @@ async def find_in_placex(conn: SearchConnection, place: ntyp.PlaceRef,
return (await conn.execute(sql)).one_or_none()
async def find_in_osmline(conn: SearchConnection, place: ntyp.PlaceRef,
details: ntyp.LookupDetails) -> Optional[SaRow]:
""" Search for the given place in the osmline table and return the
base information.
"""
t = conn.t.osmline
sql = sa.select(t.c.place_id, t.c.osm_id, t.c.parent_place_id,
t.c.indexed_date, t.c.startnumber, t.c.endnumber,
t.c.step, t.c.address, t.c.postcode, t.c.country_code,
sa.func.ST_X(sa.func.ST_Centroid(t.c.linegeo)).label('x'),
sa.func.ST_Y(sa.func.ST_Centroid(t.c.linegeo)).label('y'),
_select_column_geometry(t.c.linegeo, details.geometry_output))
if isinstance(place, ntyp.PlaceID):
sql = sql.where(t.c.place_id == place.place_id)
elif isinstance(place, ntyp.OsmID) and place.osm_type == 'W':
# There may be multiple interpolations for a single way.
# If 'class' contains a number, return the one that belongs to that number.
sql = sql.where(t.c.osm_id == place.osm_id).limit(1)
if place.osm_class and place.osm_class.isdigit():
sql = sql.order_by(sa.func.greatest(0,
sa.func.least(int(place.osm_class) - t.c.endnumber),
t.c.startnumber - int(place.osm_class)))
else:
return None
return (await conn.execute(sql)).one_or_none()
async def get_place_by_id(conn: SearchConnection, place: ntyp.PlaceRef,
details: ntyp.LookupDetails) -> Optional[nres.SearchResult]:
""" Retrieve a place with additional details from the database.
@@ -77,5 +106,11 @@ async def get_place_by_id(conn: SearchConnection, place: ntyp.PlaceRef,
await nres.add_result_details(conn, result, details)
return result
row = await find_in_osmline(conn, place, details)
if row is not None:
result = nres.create_from_osmline_row(row=row)
await nres.add_result_details(conn, result, details)
return result
# Nothing found under this ID.
return None

31
nominatim/api/results.py
View File

@@ -135,6 +135,11 @@ class SearchResult:
return '{"type": "Point","coordinates": [%f, %f]}' % self.centroid
def _filter_geometries(row: SaRow) -> Dict[str, str]:
return {k[9:]: v for k, v in row._mapping.items() # pylint: disable=W0212
if k.startswith('geometry_')}
def create_from_placex_row(row: SaRow) -> SearchResult:
""" Construct a new SearchResult and add the data from the result row
from the placex table.
@@ -157,10 +162,30 @@ def create_from_placex_row(row: SaRow) -> SearchResult:
importance=row.importance,
country_code=row.country_code,
indexed_date=getattr(row, 'indexed_date'),
centroid=Point(row.x, row.y))
centroid=Point(row.x, row.y),
geometry = _filter_geometries(row))
result.geometry = {k[9:]: v for k, v in row._mapping.items() # pylint: disable=W0212
if k.startswith('geometry_')}
return result
def create_from_osmline_row(row: SaRow) -> SearchResult:
""" Construct a new SearchResult and add the data from the result row
from the osmline table.
"""
result = SearchResult(source_table=SourceTable.OSMLINE,
place_id=row.place_id,
parent_place_id=row.parent_place_id,
osm_object=('W', row.osm_id),
category=('place', 'houses'),
address=row.address,
postcode=row.postcode,
extratags={'startnumber': str(row.startnumber),
'endnumber': str(row.endnumber),
'step': str(row.step)},
country_code=row.country_code,
indexed_date=getattr(row, 'indexed_date'),
centroid=Point(row.x, row.y),
geometry = _filter_geometries(row))
return result

16
test/python/api/conftest.py
View File

@@ -77,6 +77,22 @@ class APITester:
'isaddress': kw.get('isaddress', True)})
def add_osmline(self, **kw):
self.add_data('osmline',
{'place_id': kw.get('place_id', 10000),
'osm_id': kw.get('osm_id', 4004),
'parent_place_id': kw.get('parent_place_id'),
'indexed_date': kw.get('indexed_date',
dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
'startnumber': kw.get('startnumber', 2),
'endnumber': kw.get('endnumber', 6),
'step': kw.get('step', 2),
'address': kw.get('address'),
'postcode': kw.get('postcode'),
'country_code': kw.get('country_code'),
'linegeo': 'SRID=4326;' + kw.get('geometry', 'LINESTRING(1.1 -0.2, 1.09 -0.22)')})
async def exec_async(self, sql, *args, **kwargs):
async with self.api._async_api.begin() as conn:
return await conn.execute(sql, *args, **kwargs)

123
test/python/api/test_api_lookup.py
View File

@@ -172,7 +172,7 @@ def test_lookup_placex_with_address_details(apiobj):
]
def test_lookup_place_wth_linked_places_none_existing(apiobj):
def test_lookup_place_with_linked_places_none_existing(apiobj):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
class_='highway', type='residential', name='Street',
country_code='pl', linked_place_id=45,
@@ -253,6 +253,127 @@ def test_lookup_place_with_parented_places_existing(apiobj):
]
@pytest.mark.parametrize('idobj', (napi.PlaceID(4924), napi.OsmID('W', 9928)))
def test_lookup_in_osmline(apiobj, idobj):
import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
apiobj.add_osmline(place_id=4924, osm_id=9928,
parent_place_id=12,
startnumber=1, endnumber=4, step=1,
country_code='gb', postcode='34425',
address={'city': 'Big'},
indexed_date=import_date,
geometry='LINESTRING(23 34, 23 35)')
result = apiobj.api.lookup(idobj, napi.LookupDetails())
assert result is not None
assert result.source_table.name == 'OSMLINE'
assert result.category == ('place', 'houses')
assert result.centroid == (pytest.approx(23.0), pytest.approx(34.5))
assert result.place_id == 4924
assert result.parent_place_id == 12
assert result.linked_place_id is None
assert result.osm_object == ('W', 9928)
assert result.admin_level == 15
assert result.names is None
assert result.address == {'city': 'Big'}
assert result.extratags == {'startnumber': '1', 'endnumber': '4', 'step': '1'}
assert result.housenumber is None
assert result.postcode == '34425'
assert result.wikipedia is None
assert result.rank_search == 30
assert result.rank_address == 30
assert result.importance is None
assert result.country_code == 'gb'
assert result.indexed_date == import_date
assert result.address_rows is None
assert result.linked_rows is None
assert result.parented_rows is None
assert result.name_keywords is None
assert result.address_keywords is None
assert result.geometry == {'type': 'ST_LineString'}
def test_lookup_in_osmline_split_interpolation(apiobj):
apiobj.add_osmline(place_id=1000, osm_id=9,
startnumber=2, endnumber=4, step=1)
apiobj.add_osmline(place_id=1001, osm_id=9,
startnumber=6, endnumber=9, step=1)
apiobj.add_osmline(place_id=1002, osm_id=9,
startnumber=11, endnumber=20, step=1)
for i in range(1, 6):
result = apiobj.api.lookup(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
assert result.place_id == 1000
for i in range(7, 11):
result = apiobj.api.lookup(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
assert result.place_id == 1001
for i in range(12, 22):
result = apiobj.api.lookup(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
assert result.place_id == 1002
def test_lookup_osmline_with_address_details(apiobj):
apiobj.add_osmline(place_id=9000, osm_id=9,
startnumber=2, endnumber=4, step=1,
parent_place_id=332)
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
class_='highway', type='residential', name='Street',
country_code='pl',
rank_search=27, rank_address=26)
apiobj.add_address_placex(332, fromarea=False, isaddress=False,
distance=0.0034,
place_id=1000, osm_type='N', osm_id=3333,
class_='place', type='suburb', name='Smallplace',
country_code='pl', admin_level=13,
rank_search=24, rank_address=23)
apiobj.add_address_placex(332, fromarea=True, isaddress=True,
place_id=1001, osm_type='N', osm_id=3334,
class_='place', type='city', name='Bigplace',
country_code='pl',
rank_search=17, rank_address=16)
result = apiobj.api.lookup(napi.PlaceID(9000),
napi.LookupDetails(address_details=True))
assert result.address_rows == [
napi.AddressLine(place_id=None, osm_object=None,
category=('place', 'house_number'),
names={'ref': '2'}, extratags={},
admin_level=None, fromarea=True, isaddress=True,
rank_address=28, distance=0.0),
napi.AddressLine(place_id=332, osm_object=('W', 4),
category=('highway', 'residential'),
names={'name': 'Street'}, extratags={},
admin_level=15, fromarea=True, isaddress=True,
rank_address=26, distance=0.0),
napi.AddressLine(place_id=1000, osm_object=('N', 3333),
category=('place', 'suburb'),
names={'name': 'Smallplace'}, extratags={},
admin_level=13, fromarea=False, isaddress=True,
rank_address=23, distance=0.0034),
napi.AddressLine(place_id=1001, osm_object=('N', 3334),
category=('place', 'city'),
names={'name': 'Bigplace'}, extratags={},
admin_level=15, fromarea=True, isaddress=True,
rank_address=16, distance=0.0),
napi.AddressLine(place_id=None, osm_object=None,
category=('place', 'country_code'),
names={'ref': 'pl'}, extratags={},
admin_level=None, fromarea=True, isaddress=False,
rank_address=4, distance=0.0)
]
@pytest.mark.parametrize('gtype', (napi.GeometryFormat.KML,
napi.GeometryFormat.SVG,
napi.GeometryFormat.TEXT))