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
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
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
@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
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
+ 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'.
""" 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:
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
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:
""" % 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
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.
"""
- result.parented_rows = []
+ result.parented_rows = AddressLines()
if result.source_table != SourceTable.PLACEX:
return
projects / nominatim.git / blobdiff