AddressLines as AddressLines,
WordInfo as WordInfo,
WordInfos as WordInfos,
- SearchResult as SearchResult)
+ DetailedResult as DetailedResult)
from .localization import (Locales as Locales)
from nominatim.api.status import get_status, StatusResult
from nominatim.api.lookup import get_place_by_id
from nominatim.api.types import PlaceRef, LookupDetails
-from nominatim.api.results import SearchResult
+from nominatim.api.results import DetailedResult
class NominatimAPIAsync:
async def lookup(self, place: PlaceRef,
- details: LookupDetails) -> Optional[SearchResult]:
+ details: LookupDetails) -> Optional[DetailedResult]:
""" Get detailed information about a place in the database.
Returns None if there is no entry under the given ID.
def lookup(self, place: PlaceRef,
- details: LookupDetails) -> Optional[SearchResult]:
+ details: LookupDetails) -> Optional[DetailedResult]:
""" Get detailed information about a place in the database.
"""
return self._loop.run_until_complete(self._async_api.lookup(place, details))
Implementation of place lookup by ID.
"""
from typing import Optional
+import datetime as dt
import sqlalchemy as sa
async def get_place_by_id(conn: SearchConnection, place: ntyp.PlaceRef,
- details: ntyp.LookupDetails) -> Optional[nres.SearchResult]:
+ details: ntyp.LookupDetails) -> Optional[nres.DetailedResult]:
""" Retrieve a place with additional details from the database.
"""
log().function('get_place_by_id', place=place, details=details)
raise ValueError("lookup only supports geojosn polygon output.")
row = await find_in_placex(conn, place, details)
+ log().var_dump('Result (placex)', row)
if row is not None:
- result = nres.create_from_placex_row(row)
- log().var_dump('Result', result)
- 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)
- log().var_dump('Result', result)
- await nres.add_result_details(conn, result, details)
- return result
-
- row = await find_in_postcode(conn, place, details)
- if row is not None:
- result = nres.create_from_postcode_row(row)
- log().var_dump('Result', result)
- await nres.add_result_details(conn, result, details)
- return result
-
- row = await find_in_tiger(conn, place, details)
- if row is not None:
- result = nres.create_from_tiger_row(row)
- log().var_dump('Result', result)
- await nres.add_result_details(conn, result, details)
- return result
-
- # Nothing found under this ID.
- return None
+ result = nres.create_from_placex_row(row, nres.DetailedResult)
+ else:
+ row = await find_in_osmline(conn, place, details)
+ log().var_dump('Result (osmline)', row)
+ if row is not None:
+ result = nres.create_from_osmline_row(row, nres.DetailedResult)
+ else:
+ row = await find_in_postcode(conn, place, details)
+ log().var_dump('Result (postcode)', row)
+ if row is not None:
+ result = nres.create_from_postcode_row(row, nres.DetailedResult)
+ else:
+ row = await find_in_tiger(conn, place, details)
+ log().var_dump('Result (tiger)', row)
+ if row is not None:
+ result = nres.create_from_tiger_row(row, nres.DetailedResult)
+ else:
+ return None
+
+ # add missing details
+ assert result is not None
+ result.parent_place_id = row.parent_place_id
+ result.linked_place_id = getattr(row, 'linked_place_id', None)
+ indexed_date = getattr(row, 'indexed_date', None)
+ if indexed_date is not None:
+ result.indexed_date = indexed_date.replace(tzinfo=dt.timezone.utc)
+
+ await nres.add_result_details(conn, result, details)
+
+ return result
internal use only. That's why they are implemented as free-standing functions
instead of member functions.
"""
-from typing import Optional, Tuple, Dict, Sequence
+from typing import Optional, Tuple, Dict, Sequence, TypeVar, Type
import enum
import dataclasses
import datetime as dt
@dataclasses.dataclass
-class SearchResult:
- """ Data class collecting all available information about a search result.
+class BaseResult:
+ """ Data class collecting information common to all
+ types of search results.
"""
source_table: SourceTable
category: Tuple[str, str]
centroid: Point
place_id : Optional[int] = None
- parent_place_id: Optional[int] = None
- linked_place_id: Optional[int] = None
osm_object: Optional[Tuple[str, int]] = None
admin_level: int = 15
country_code: Optional[str] = None
- indexed_date: Optional[dt.datetime] = None
-
address_rows: Optional[AddressLines] = None
linked_rows: Optional[AddressLines] = None
parented_rows: Optional[AddressLines] = None
geometry: Dict[str, str] = dataclasses.field(default_factory=dict)
- def __post_init__(self) -> None:
- if self.indexed_date is not None and self.indexed_date.tzinfo is None:
- self.indexed_date = self.indexed_date.replace(tzinfo=dt.timezone.utc)
-
@property
def lat(self) -> float:
""" Get the latitude (or y) of the center point of the place.
"""
return self.importance or (0.7500001 - (self.rank_search/40.0))
+BaseResultT = TypeVar('BaseResultT', bound=BaseResult)
+
+@dataclasses.dataclass
+class DetailedResult(BaseResult):
+ """ A search result with more internal information from the database
+ added.
+ """
+ parent_place_id: Optional[int] = None
+ linked_place_id: Optional[int] = None
+ indexed_date: Optional[dt.datetime] = None
+
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.
+def create_from_placex_row(row: Optional[SaRow],
+ class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
+ """ Construct a new result and add the data from the result row
+ from the placex table. 'class_type' defines the type of result
+ to return. Returns None if the row is None.
"""
- return SearchResult(source_table=SourceTable.PLACEX,
- place_id=row.place_id,
- parent_place_id=row.parent_place_id,
- linked_place_id=row.linked_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,
- housenumber=row.housenumber,
- postcode=row.postcode,
- wikipedia=row.wikipedia,
- rank_address=row.rank_address,
- rank_search=row.rank_search,
- importance=row.importance,
- country_code=row.country_code,
- indexed_date=getattr(row, 'indexed_date'),
- centroid=Point.from_wkb(row.centroid.data),
- geometry=_filter_geometries(row))
-
-
-def create_from_osmline_row(row: SaRow) -> SearchResult:
- """ Construct a new SearchResult and add the data from the result row
- from the osmline table.
+ if row is None:
+ return None
+
+ return class_type(source_table=SourceTable.PLACEX,
+ 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,
+ housenumber=row.housenumber,
+ postcode=row.postcode,
+ wikipedia=row.wikipedia,
+ rank_address=row.rank_address,
+ rank_search=row.rank_search,
+ importance=row.importance,
+ country_code=row.country_code,
+ centroid=Point.from_wkb(row.centroid.data),
+ geometry=_filter_geometries(row))
+
+
+def create_from_osmline_row(row: Optional[SaRow],
+ class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
+ """ Construct a new result and add the data from the result row
+ from the address interpolation table osmline. 'class_type' defines
+ the type of result to return. Returns None if the row is None.
+
+ If the row contains a housenumber, then the housenumber is filled out.
+ Otherwise the result contains the interpolation information in extratags.
"""
- return 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.from_wkb(row.centroid.data),
- geometry=_filter_geometries(row))
-
-
-def create_from_tiger_row(row: SaRow) -> SearchResult:
- """ Construct a new SearchResult and add the data from the result row
- from the Tiger table.
+ if row is None:
+ return None
+
+ hnr = getattr(row, 'housenumber', None)
+
+ res = class_type(source_table=SourceTable.OSMLINE,
+ place_id=row.place_id,
+ osm_object=('W', row.osm_id),
+ category=('place', 'houses' if hnr is None else 'house'),
+ address=row.address,
+ postcode=row.postcode,
+ country_code=row.country_code,
+ centroid=Point.from_wkb(row.centroid.data),
+ geometry=_filter_geometries(row))
+
+ if hnr is None:
+ res.extratags = {'startnumber': str(row.startnumber),
+ 'endnumber': str(row.endnumber),
+ 'step': str(row.step)}
+ else:
+ res.housenumber = str(hnr)
+
+ return res
+
+
+def create_from_tiger_row(row: Optional[SaRow],
+ class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
+ """ Construct a new result and add the data from the result row
+ from the Tiger data interpolation table. 'class_type' defines
+ the type of result to return. Returns None if the row is None.
+
+ If the row contains a housenumber, then the housenumber is filled out.
+ Otherwise the result contains the interpolation information in extratags.
"""
- return SearchResult(source_table=SourceTable.TIGER,
- place_id=row.place_id,
- parent_place_id=row.parent_place_id,
- category=('place', 'houses'),
- postcode=row.postcode,
- extratags={'startnumber': str(row.startnumber),
- 'endnumber': str(row.endnumber),
- 'step': str(row.step)},
- country_code='us',
- centroid=Point.from_wkb(row.centroid.data),
- geometry=_filter_geometries(row))
-
-
-def create_from_postcode_row(row: SaRow) -> SearchResult:
- """ Construct a new SearchResult and add the data from the result row
- from the postcode centroid table.
+ if row is None:
+ return None
+
+ hnr = getattr(row, 'housenumber', None)
+
+ res = class_type(source_table=SourceTable.TIGER,
+ place_id=row.place_id,
+ category=('place', 'houses' if hnr is None else 'house'),
+ postcode=row.postcode,
+ country_code='us',
+ centroid=Point.from_wkb(row.centroid.data),
+ geometry=_filter_geometries(row))
+
+ if hnr is None:
+ res.extratags = {'startnumber': str(row.startnumber),
+ 'endnumber': str(row.endnumber),
+ 'step': str(row.step)}
+ else:
+ res.housenumber = str(hnr)
+
+ return res
+
+
+def create_from_postcode_row(row: Optional[SaRow],
+ class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
+ """ Construct a new result and add the data from the result row
+ from the postcode table. 'class_type' defines
+ the type of result to return. Returns None if the row is None.
"""
- return SearchResult(source_table=SourceTable.POSTCODE,
- place_id=row.place_id,
- parent_place_id=row.parent_place_id,
- category=('place', 'postcode'),
- names={'ref': row.postcode},
- rank_search=row.rank_search,
- rank_address=row.rank_address,
- country_code=row.country_code,
- centroid=Point.from_wkb(row.centroid.data),
- indexed_date=row.indexed_date,
- geometry=_filter_geometries(row))
-
-
-async def add_result_details(conn: SearchConnection, result: SearchResult,
+ if row is None:
+ return None
+
+ return class_type(source_table=SourceTable.POSTCODE,
+ place_id=row.place_id,
+ category=('place', 'postcode'),
+ names={'ref': row.postcode},
+ rank_search=row.rank_search,
+ rank_address=row.rank_address,
+ country_code=row.country_code,
+ centroid=Point.from_wkb(row.centroid.data),
+ geometry=_filter_geometries(row))
+
+
+async def add_result_details(conn: SearchConnection, result: BaseResult,
details: LookupDetails) -> None:
""" Retrieve more details from the database according to the
parameters specified in 'details'.
distance=row.distance)
-async def complete_address_details(conn: SearchConnection, result: SearchResult) -> None:
+async def complete_address_details(conn: SearchConnection, result: BaseResult) -> None:
""" Retrieve information about places that make up the address of the result.
"""
housenumber = -1
for row in await conn.execute(sql):
result.address_rows.append(_result_row_to_address_row(row))
+
# pylint: disable=consider-using-f-string
def _placex_select_address_row(conn: SearchConnection,
centroid: Point) -> SaSelect:
""" % centroid).label('distance'))
-async def complete_linked_places(conn: SearchConnection, result: SearchResult) -> None:
+async def complete_linked_places(conn: SearchConnection, result: BaseResult) -> None:
""" Retrieve information about places that link to the result.
"""
result.linked_rows = []
result.linked_rows.append(_result_row_to_address_row(row))
-async def complete_keywords(conn: SearchConnection, result: SearchResult) -> None:
+async def complete_keywords(conn: SearchConnection, result: BaseResult) -> None:
""" Retrieve information about the search terms used for this place.
"""
t = conn.t.search_name
result.address_keywords.append(WordInfo(*row))
-async def complete_parented_places(conn: SearchConnection, result: SearchResult) -> None:
+async def complete_parented_places(conn: SearchConnection, result: BaseResult) -> None:
""" Retrieve information about places that the result provides the
address for.
"""
writer.end_object().next()
-@dispatch.format_func(napi.SearchResult, 'details-json')
-def _format_search_json(result: napi.SearchResult, options: Mapping[str, Any]) -> str:
+@dispatch.format_func(napi.DetailedResult, 'json')
+def _format_search_json(result: napi.DetailedResult, options: Mapping[str, Any]) -> str:
locales = options.get('locales', napi.Locales())
geom = result.geometry.get('geojson')
centroid = result.centroid.to_geojson()
raise params.error('No place with that OSM ID found.', status=404)
output = formatting.format_result(
- result,
- 'details-json',
+ result, 'json',
{'locales': locales,
'group_hierarchy': params.get_bool('group_hierarchy', False),
'icon_base_url': params.config().MAPICON_URL})
if result:
output = api_output.format_result(
result,
- 'details-json',
+ 'json',
{'locales': locales,
'group_hierarchy': args.group_hierarchy})
# reformat the result, so it is pretty-printed
assert result == '{"status":0,"message":"OK","data_updated":"2010-02-07T20:20:03+00:00","software_version":"%s","database_version":"5.6"}' % (NOMINATIM_VERSION, )
-# SearchResult
+# DetailedResult
def test_search_details_minimal():
- search = napi.SearchResult(napi.SourceTable.PLACEX,
- ('place', 'thing'),
- napi.Point(1.0, 2.0))
+ search = napi.DetailedResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0))
- result = api_impl.format_result(search, 'details-json', {})
+ result = api_impl.format_result(search, 'json', {})
assert json.loads(result) == \
{'category': 'place',
def test_search_details_full():
- import_date = dt.datetime(2010, 2, 7, 20, 20, 3, 0)
- search = napi.SearchResult(
+ import_date = dt.datetime(2010, 2, 7, 20, 20, 3, 0, tzinfo=dt.timezone.utc)
+ search = napi.DetailedResult(
source_table=napi.SourceTable.PLACEX,
category=('amenity', 'bank'),
centroid=napi.Point(56.947, -87.44),
indexed_date = import_date
)
- result = api_impl.format_result(search, 'details-json', {})
+ result = api_impl.format_result(search, 'json', {})
assert json.loads(result) == \
{'place_id': 37563,
('ST_Polygon', True),
('ST_MultiPolygon', True)])
def test_search_details_no_geometry(gtype, isarea):
- search = napi.SearchResult(napi.SourceTable.PLACEX,
+ search = napi.DetailedResult(napi.SourceTable.PLACEX,
('place', 'thing'),
napi.Point(1.0, 2.0),
geometry={'type': gtype})
- result = api_impl.format_result(search, 'details-json', {})
+ result = api_impl.format_result(search, 'json', {})
js = json.loads(result)
assert js['geometry'] == {'type': 'Point', 'coordinates': [1.0, 2.0]}
def test_search_details_with_geometry():
- search = napi.SearchResult(napi.SourceTable.PLACEX,
- ('place', 'thing'),
- napi.Point(1.0, 2.0),
- geometry={'geojson': '{"type":"Point","coordinates":[56.947,-87.44]}'})
+ search = napi.DetailedResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0),
+ geometry={'geojson': '{"type":"Point","coordinates":[56.947,-87.44]}'})
- result = api_impl.format_result(search, 'details-json', {})
+ result = api_impl.format_result(search, 'json', {})
js = json.loads(result)
assert js['geometry'] == {'type': 'Point', 'coordinates': [56.947, -87.44]}
def test_search_details_with_address_minimal():
- search = napi.SearchResult(napi.SourceTable.PLACEX,
- ('place', 'thing'),
- napi.Point(1.0, 2.0),
- address_rows=[
+ search = napi.DetailedResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0),
+ address_rows=[
napi.AddressLine(place_id=None,
osm_object=None,
category=('bnd', 'note'),
isaddress=False,
rank_address=10,
distance=0.0)
- ])
+ ])
- result = api_impl.format_result(search, 'details-json', {})
+ result = api_impl.format_result(search, 'json', {})
js = json.loads(result)
assert js['address'] == [{'localname': '',
def test_search_details_with_address_full():
- search = napi.SearchResult(napi.SourceTable.PLACEX,
- ('place', 'thing'),
- napi.Point(1.0, 2.0),
- address_rows=[
+ search = napi.DetailedResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0),
+ address_rows=[
napi.AddressLine(place_id=3498,
osm_object=('R', 442),
category=('bnd', 'note'),
isaddress=True,
rank_address=10,
distance=0.034)
- ])
+ ])
- result = api_impl.format_result(search, 'details-json', {})
+ result = api_impl.format_result(search, 'json', {})
js = json.loads(result)
assert js['address'] == [{'localname': 'Trespass',
--- /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 result datatype helper functions.
+"""
+import struct
+
+import pytest
+import pytest_asyncio
+import sqlalchemy as sa
+
+
+from nominatim.api import SourceTable, DetailedResult, Point
+import nominatim.api.results as nresults
+
+class FakeCentroid:
+ def __init__(self, x, y):
+ self.data = struct.pack("=biidd", 1, 0x20000001, 4326,
+ x, y)
+
+class FakeRow:
+ def __init__(self, **kwargs):
+ for k, v in kwargs.items():
+ setattr(self, k, v)
+ self._mapping = kwargs
+
+
+def test_minimal_detailed_result():
+ res = DetailedResult(SourceTable.PLACEX,
+ ('amenity', 'post_box'),
+ Point(23.1, 0.5))
+
+ assert res.lon == 23.1
+ assert res.lat == 0.5
+ assert res.calculated_importance() == pytest.approx(0.0000001)
+
+def test_detailed_result_custom_importance():
+ res = DetailedResult(SourceTable.PLACEX,
+ ('amenity', 'post_box'),
+ Point(23.1, 0.5),
+ importance=0.4563)
+
+ assert res.calculated_importance() == 0.4563
+
+
+@pytest.mark.parametrize('func', (nresults.create_from_placex_row,
+ nresults.create_from_osmline_row,
+ nresults.create_from_tiger_row,
+ nresults.create_from_postcode_row))
+def test_create_row_none(func):
+ assert func(None, DetailedResult) is None
+
+
+@pytest.mark.parametrize('func', (nresults.create_from_osmline_row,
+ nresults.create_from_tiger_row))
+def test_create_row_with_housenumber(func):
+ row = FakeRow(place_id = 2345, osm_id = 111, housenumber = 4,
+ address = None, postcode = '99900', country_code = 'xd',
+ centroid = FakeCentroid(0, 0))
+
+ res = func(row, DetailedResult)
+
+ assert res.housenumber == '4'
+ assert res.extratags is None
+ assert res.category == ('place', 'house')
+
+
+@pytest.mark.parametrize('func', (nresults.create_from_osmline_row,
+ nresults.create_from_tiger_row))
+def test_create_row_without_housenumber(func):
+ row = FakeRow(place_id=2345, osm_id=111,
+ startnumber=1, endnumber=11, step=2,
+ address=None, postcode='99900', country_code='xd',
+ centroid=FakeCentroid(0, 0))
+
+ res = func(row, DetailedResult)
+
+ assert res.housenumber is None
+ assert res.extratags == {'startnumber': '1', 'endnumber': '11', 'step': '2'}
+ assert res.category == ('place', 'houses')
@pytest.fixture(autouse=True)
def setup_status_mock(self, monkeypatch):
- result = napi.SearchResult(napi.SourceTable.PLACEX, ('place', 'thing'),
- napi.Point(1.0, -3.0))
+ result = napi.DetailedResult(napi.SourceTable.PLACEX, ('place', 'thing'),
+ napi.Point(1.0, -3.0))
monkeypatch.setattr(napi.NominatimAPI, 'lookup',
lambda *args: result)
projects / nominatim.git / commitdiff