X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/ee0c5e24bb2bd89288a2fa7a9dada2c7ee8f44a9..1f0e1bec0ee66df7641e53c6b55e84d7eb88cec2:/nominatim/api/results.py diff --git a/nominatim/api/results.py b/nominatim/api/results.py index 10f03393..98b13380 100644 --- a/nominatim/api/results.py +++ b/nominatim/api/results.py @@ -11,7 +11,7 @@ Data classes are part of the public API while the functions are for 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, List import enum import dataclasses import datetime as dt @@ -19,9 +19,10 @@ import datetime as dt import sqlalchemy as sa from nominatim.typing import SaSelect, SaRow -from nominatim.api.types import Point, LookupDetails +from nominatim.api.types import Point, Bbox, LookupDetails from nominatim.api.connection import SearchConnection from nominatim.api.logging import log +from nominatim.api.localization import Locales # This file defines complex result data classes. # pylint: disable=too-many-instance-attributes @@ -52,8 +53,30 @@ class AddressLine: rank_address: int distance: float + local_name: Optional[str] = None + + +class AddressLines(List[AddressLine]): + """ Sequence of address lines order in descending order by their rank. + """ + + def localize(self, locales: Locales) -> List[str]: + """ Set the local name of address parts according to the chosen + locale. Return the list of local names without duplications. + + Only address parts that are marked as isaddress are localized + and returned. + """ + label_parts: List[str] = [] + + for line in self: + if line.isaddress and line.names: + line.local_name = locales.display_name(line.names) + if not label_parts or label_parts[-1] != line.local_name: + label_parts.append(line.local_name) + + return label_parts -AddressLines = Sequence[AddressLine] @dataclasses.dataclass @@ -69,18 +92,16 @@ WordInfos = Sequence[WordInfo] @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 names: Optional[Dict[str, str]] = None address: Optional[Dict[str, str]] = None @@ -96,8 +117,6 @@ class SearchResult: 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 @@ -106,10 +125,6 @@ class SearchResult: 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. @@ -131,93 +146,166 @@ class SearchResult: """ 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 + admin_level: int = 15 + indexed_date: Optional[dt.datetime] = None + + +@dataclasses.dataclass +class ReverseResult(BaseResult): + """ A search result for reverse geocoding. + """ + distance: Optional[float] = None + bbox: Optional[Bbox] = None + + +class ReverseResults(List[ReverseResult]): + """ Sequence of reverse lookup results ordered by distance. + May be empty when no result was found. + """ + + +@dataclasses.dataclass +class SearchResult(BaseResult): + """ A search result for forward geocoding. + """ + bbox: Optional[Bbox] = None + + +class SearchResults(List[SearchResult]): + """ Sequence of forward lookup results ordered by relevance. + May be empty when no result was found. + """ + 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), + 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, + osm_object=(row.osm_type, row.osm_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'. @@ -241,8 +329,8 @@ def _result_row_to_address_row(row: SaRow) -> AddressLine: """ Create a new AddressLine from the results of a datbase query. """ extratags: Dict[str, str] = getattr(row, 'extratags', {}) - if 'place_type' in row: - extratags['place_type'] = row.place_type + if hasattr(row, 'place_type') and row.place_type: + extratags['place'] = row.place_type names = row.name if getattr(row, 'housenumber', None) is not None: @@ -262,7 +350,7 @@ def _result_row_to_address_row(row: SaRow) -> AddressLine: 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 @@ -288,10 +376,11 @@ async def complete_address_details(conn: SearchConnection, result: SearchResult) sql = sa.select(sfn).order_by(sa.column('rank_address').desc(), sa.column('isaddress').desc()) - result.address_rows = [] + result.address_rows = AddressLines() 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: @@ -308,10 +397,10 @@ def _placex_select_address_row(conn: SearchConnection, """ % 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 = AddressLines() if result.source_table != SourceTable.PLACEX: return @@ -322,7 +411,7 @@ async def complete_linked_places(conn: SearchConnection, result: SearchResult) - 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 @@ -342,11 +431,11 @@ async def complete_keywords(conn: SearchConnection, result: SearchResult) -> Non 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. """ - result.parented_rows = [] + result.parented_rows = AddressLines() if result.source_table != SourceTable.PLACEX: return