internal use only. That's why they are implemented as free-standing functions
instead of member functions.
"""
-from typing import Optional, Tuple, Dict, Sequence, Any
+from typing import Optional, Tuple, Dict, Sequence
import enum
import dataclasses
import datetime as dt
from nominatim.typing import SaSelect, SaRow
from nominatim.api.types import Point, LookupDetails
from nominatim.api.connection import SearchConnection
+from nominatim.api.logging import log
# This file defines complex result data classes.
# pylint: disable=too-many-instance-attributes
names: Dict[str, str]
extratags: Optional[Dict[str, str]]
- admin_level: int
+ admin_level: Optional[int]
fromarea: bool
isaddress: bool
rank_address: int
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:
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.
"""
- result = 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(row.x, row.y))
-
- result.geometry = {k[9:]: v for k, v in row._mapping.items() # pylint: disable=W0212
- if k.startswith('geometry_')}
-
- return result
+ 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(row.x, row.y),
+ 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.
+ """
+ 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(row.x, row.y),
+ 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.
+ """
+ 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(row.x, row.y),
+ 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.
+ """
+ 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(row.x, row.y),
+ indexed_date=row.indexed_date,
+ geometry=_filter_geometries(row))
async def add_result_details(conn: SearchConnection, result: SearchResult,
""" Retrieve more details from the database according to the
parameters specified in 'details'.
"""
+ log().section('Query details for result')
if details.address_details:
+ log().comment('Query address details')
await complete_address_details(conn, result)
if details.linked_places:
+ log().comment('Query linked places')
await complete_linked_places(conn, result)
if details.parented_places:
+ log().comment('Query parent places')
await complete_parented_places(conn, result)
if details.keywords:
+ log().comment('Query keywords')
await complete_keywords(conn, result)
if 'place_type' in row:
extratags['place_type'] = row.place_type
+ names = row.name
+ if getattr(row, 'housenumber', None) is not None:
+ if names is None:
+ names = {}
+ names['housenumber'] = row.housenumber
+
return AddressLine(place_id=row.place_id,
- osm_object=(row.osm_type, row.osm_id),
+ osm_object=None if row.osm_type is None else (row.osm_type, row.osm_id),
category=(getattr(row, 'class'), row.type),
- names=row.name,
+ names=names,
extratags=extratags,
admin_level=row.admin_level,
fromarea=row.fromarea,
t = conn.t.placex
return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
t.c.class_.label('class'), t.c.type,
- t.c.admin_level,
+ t.c.admin_level, t.c.housenumber,
sa.literal_column("""ST_GeometryType(geometry) in
('ST_Polygon','ST_MultiPolygon')""").label('fromarea'),
t.c.rank_address,
projects / nominatim.git / blobdiff