import sqlalchemy as sa
-from nominatim_core.typing import SaSelect, SaRow
-from nominatim_core.db.sqlalchemy_types import Geometry
+from .typing import SaSelect, SaRow
+from .sql.sqlalchemy_types import Geometry
from .types import Point, Bbox, LookupDetails
from .connection import SearchConnection
from .logging import log
from .localization import Locales
# This file defines complex result data classes.
-# pylint: disable=too-many-instance-attributes
+
def _mingle_name_tags(names: Optional[Dict[str, str]]) -> Optional[Dict[str, str]]:
""" Mix-in names from linked places, so that they show up
local_name: Optional[str] = None
""" Place holder for localization of this address part. See
- [Localization](#localization) below.
+ [Localization](Result-Handling.md#localization) below.
"""
return label_parts
-
@dataclasses.dataclass
class WordInfo:
""" Each entry in the list of search terms contains the
category: Tuple[str, str]
centroid: Point
- place_id : Optional[int] = None
+ place_id: Optional[int] = None
osm_object: Optional[Tuple[str, int]] = None
parent_place_id: Optional[int] = None
linked_place_id: Optional[int] = None
"""
return self.centroid[1]
-
@property
def lon(self) -> float:
""" Get the longitude (or x) of the center point of the place.
"""
return self.centroid[0]
-
def calculated_importance(self) -> float:
""" Get a valid importance value. This is either the stored importance
of the value or an artificial value computed from the place's
"""
return self.importance or (0.40001 - (self.rank_search/75.0))
-
def localize(self, locales: Locales) -> None:
""" Fill the locale_name and the display_name field for the
place and, if available, its address information.
self.display_name = self.locale_name
-
BaseResultT = TypeVar('BaseResultT', bound=BaseResult)
+
@dataclasses.dataclass
class DetailedResult(BaseResult):
""" A search result with more internal information from the database
bbox: Optional[Bbox] = None
accuracy: float = 0.0
-
@property
def ranking(self) -> float:
""" Return the ranking, a combined measure of accuracy and importance.
"""
return (self.accuracy if self.accuracy is not None else 1) \
- - self.calculated_importance()
+ - self.calculated_importance()
class SearchResults(List[SearchResult]):
def _filter_geometries(row: SaRow) -> Dict[str, str]:
- return {k[9:]: v for k, v in row._mapping.items() # pylint: disable=W0212
+ return {k[9:]: v for k, v in row._mapping.items()
if k.startswith('geometry_')}
place_id=row.place_id,
osm_object=(row.osm_type, row.osm_id),
category=(row.class_, row.type),
- parent_place_id = row.parent_place_id,
- linked_place_id = getattr(row, 'linked_place_id', None),
- admin_level = getattr(row, 'admin_level', 15),
+ parent_place_id=row.parent_place_id,
+ linked_place_id=getattr(row, 'linked_place_id', None),
+ admin_level=getattr(row, 'admin_level', 15),
names=_mingle_name_tags(row.name),
address=row.address,
extratags=row.extratags,
res = class_type(source_table=SourceTable.OSMLINE,
place_id=row.place_id,
- parent_place_id = row.parent_place_id,
+ parent_place_id=row.parent_place_id,
osm_object=('W', row.osm_id),
category=('place', 'houses' if hnr is None else 'house'),
address=row.address,
res = class_type(source_table=SourceTable.TIGER,
place_id=row.place_id,
- parent_place_id = row.parent_place_id,
+ parent_place_id=row.parent_place_id,
osm_object=(osm_type or row.osm_type, osm_id or row.osm_id),
category=('place', 'houses' if hnr is None else 'house'),
postcode=row.postcode,
def create_from_postcode_row(row: Optional[SaRow],
- class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
+ 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 class_type(source_table=SourceTable.POSTCODE,
place_id=row.place_id,
- parent_place_id = row.parent_place_id,
+ parent_place_id=row.parent_place_id,
category=('place', 'postcode'),
names={'ref': row.postcode},
rank_search=row.rank_search,
def create_from_country_row(row: Optional[SaRow],
- class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
+ class_type: Type[BaseResultT]) -> Optional[BaseResultT]:
""" Construct a new result and add the data from the result row
from the fallback country tables. 'class_type' defines
the type of result to return. Returns None if the row is None.
distance=0.0))
result.address_rows.append(AddressLine(
category=('place', 'country_code'),
- names={'ref': result.country_code}, extratags = {},
+ names={'ref': result.country_code}, extratags={},
fromarea=True, isaddress=False, rank_address=4,
distance=0.0))
for result in results:
_setup_address_details(result)
- ### Lookup entries from place_address line
+ # Lookup entries from place_address line
lookup_ids = [{'pid': r.place_id,
'lid': _get_address_lookup_id(r),
'names': list(r.address.values()) if r.address else [],
- 'c': ('SRID=4326;' + r.centroid.to_wkt()) if r.centroid else '' }
+ 'c': ('SRID=4326;' + r.centroid.to_wkt()) if r.centroid else ''}
for r in results if r.place_id]
if not lookup_ids:
.order_by(taddr.c.distance.desc())\
.order_by(t.c.rank_search.desc())
-
current_result = None
current_rank_address = -1
for row in await conn.execute(sql):
for result in results:
await _finalize_entry(conn, result)
-
- ### Finally add the record for the parent entry where necessary.
+ # Finally add the record for the parent entry where necessary.
parent_lookup_ids = list(filter(lambda e: e['pid'] != e['lid'], lookup_ids))
if parent_lookup_ids:
t.c.class_, t.c.type, t.c.extratags,
t.c.admin_level,
t.c.rank_address)\
- .where(t.c.place_id == ltab.c.value['lid'].as_integer())
+ .where(t.c.place_id == ltab.c.value['lid'].as_integer())
for row in await conn.execute(sql):
current_result = next((r for r in results if r.place_id == row.src_place_id), None)
fromarea=True, isaddress=True,
rank_address=row.rank_address, distance=0.0))
- ### Now sort everything
+ # Now sort everything
def mk_sort_key(place_id: Optional[int]) -> Callable[[AddressLine], Tuple[bool, int, bool]]:
return lambda a: (a.place_id != place_id, -a.rank_address, a.isaddress)
return
sql = _placex_select_address_row(conn, result.centroid)\
- .where(conn.t.placex.c.linked_place_id == result.place_id)
+ .where(conn.t.placex.c.linked_place_id == result.place_id)
for row in await conn.execute(sql):
result.linked_rows.append(_result_row_to_address_row(row))
return
sql = _placex_select_address_row(conn, result.centroid)\
- .where(conn.t.placex.c.parent_place_id == result.place_id)\
- .where(conn.t.placex.c.rank_search == 30)
+ .where(conn.t.placex.c.parent_place_id == result.place_id)\
+ .where(conn.t.placex.c.rank_search == 30)
for row in await conn.execute(sql):
result.parented_rows.append(_result_row_to_address_row(row))
projects / nominatim.git / blobdiff