import sqlalchemy as sa
-from nominatim_core.typing import SaColumn, SaSelect, SaFromClause, SaLabel, SaRow,\
- SaBind, SaLambdaSelect
-from nominatim_core.db.sqlalchemy_types import Geometry
+from .typing import SaColumn, SaSelect, SaFromClause, SaLabel, SaRow, \
+ SaBind, SaLambdaSelect
+from .sql.sqlalchemy_types import Geometry
from .connection import SearchConnection
from . import results as nres
from .logging import log
from .types import AnyPoint, DataLayer, ReverseDetails, GeometryFormat, Bbox
-# In SQLAlchemy expression which compare with NULL need to be expressed with
-# the equal sign.
-# pylint: disable=singleton-comparison
RowFunc = Callable[[Optional[SaRow], Type[nres.ReverseResult]], Optional[nres.ReverseResult]]
WKT_PARAM: SaBind = sa.bindparam('wkt', type_=Geometry)
MAX_RANK_PARAM: SaBind = sa.bindparam('max_rank')
+
def no_index(expr: SaColumn) -> SaColumn:
""" Wrap the given expression, so that the query planner will
refrain from using the expression for index lookup.
"""
- return sa.func.coalesce(sa.null(), expr) # pylint: disable=not-callable
+ return sa.func.coalesce(sa.null(), expr)
def _select_from_placex(t: SaFromClause, use_wkt: bool = True) -> SaSelect:
centroid = sa.case((t.c.geometry.is_line_like(), t.c.geometry.ST_ClosestPoint(WKT_PARAM)),
else_=t.c.centroid).label('centroid')
-
return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
t.c.class_, t.c.type,
t.c.address, t.c.extratags,
def _interpolated_housenumber(table: SaFromClause) -> SaLabel:
return sa.cast(table.c.startnumber
- + sa.func.round(((table.c.endnumber - table.c.startnumber) * table.c.position)
- / table.c.step) * table.c.step,
+ + sa.func.round(((table.c.endnumber - table.c.startnumber) * table.c.position)
+ / table.c.step) * table.c.step,
sa.Integer).label('housenumber')
fac = sa.cast(table.c.step, sa.Float) / (table.c.endnumber - table.c.startnumber)
rounded_pos = sa.func.round(table.c.position / fac) * fac
return sa.case(
- (table.c.endnumber == table.c.startnumber, table.c.linegeo.ST_Centroid()),
- else_=table.c.linegeo.ST_LineInterpolatePoint(rounded_pos)).label('centroid')
+ (table.c.endnumber == table.c.startnumber, table.c.linegeo.ST_Centroid()),
+ else_=table.c.linegeo.ST_LineInterpolatePoint(rounded_pos)).label('centroid')
def _locate_interpolation(table: SaFromClause) -> SaLabel:
self.bind_params: Dict[str, Any] = {'max_rank': params.max_rank}
-
@property
def max_rank(self) -> int:
""" Return the maximum configured rank.
"""
return self.params.max_rank
-
def has_geometries(self) -> bool:
""" Check if any geometries are requested.
"""
return bool(self.params.geometry_output)
-
def layer_enabled(self, *layer: DataLayer) -> bool:
""" Return true when any of the given layer types are requested.
"""
- return any(self.params.layers & l for l in layer)
-
+ return any(self.params.layers & ly for ly in layer)
def layer_disabled(self, *layer: DataLayer) -> bool:
""" Return true when none of the given layer types is requested.
"""
- return not any(self.params.layers & l for l in layer)
-
+ return not any(self.params.layers & ly for ly in layer)
def has_feature_layers(self) -> bool:
""" Return true if any layer other than ADDRESS or POI is requested.
"""
return self.layer_enabled(DataLayer.RAILWAY, DataLayer.MANMADE, DataLayer.NATURAL)
-
def _add_geometry_columns(self, sql: SaLambdaSelect, col: SaColumn) -> SaSelect:
out = []
return sql.add_columns(*out)
-
def _filter_by_layer(self, table: SaFromClause) -> SaColumn:
if self.layer_enabled(DataLayer.MANMADE):
exclude = []
include.extend(('natural', 'water', 'waterway'))
return table.c.class_.in_(tuple(include))
-
async def _find_closest_street_or_poi(self, distance: float) -> Optional[SaRow]:
""" Look up the closest rank 26+ place in the database, which
is closer than the given distance.
# when used with prepared statements
diststr = sa.text(f"{distance}")
- sql: SaLambdaSelect = sa.lambda_stmt(lambda: _select_from_placex(t)
- .where(t.c.geometry.within_distance(WKT_PARAM, diststr))
- .where(t.c.indexed_status == 0)
- .where(t.c.linked_place_id == None)
- .where(sa.or_(sa.not_(t.c.geometry.is_area()),
- t.c.centroid.ST_Distance(WKT_PARAM) < diststr))
- .order_by('distance')
- .limit(2))
+ sql: SaLambdaSelect = sa.lambda_stmt(
+ lambda: _select_from_placex(t)
+ .where(t.c.geometry.within_distance(WKT_PARAM, diststr))
+ .where(t.c.indexed_status == 0)
+ .where(t.c.linked_place_id == None)
+ .where(sa.or_(sa.not_(t.c.geometry.is_area()),
+ t.c.centroid.ST_Distance(WKT_PARAM) < diststr))
+ .order_by('distance')
+ .limit(2))
if self.has_geometries():
sql = self._add_geometry_columns(sql, t.c.geometry)
return prev_row
-
async def _find_housenumber_for_street(self, parent_place_id: int) -> Optional[SaRow]:
t = self.conn.t.placex
return (await self.conn.execute(sql, self.bind_params)).one_or_none()
-
async def _find_interpolation_for_street(self, parent_place_id: Optional[int],
distance: float) -> Optional[SaRow]:
t = self.conn.t.osmline
inner = sql.subquery('ipol')
sql = sa.select(inner.c.place_id, inner.c.osm_id,
- inner.c.parent_place_id, inner.c.address,
- _interpolated_housenumber(inner),
- _interpolated_position(inner),
- inner.c.postcode, inner.c.country_code,
- inner.c.distance)
+ inner.c.parent_place_id, inner.c.address,
+ _interpolated_housenumber(inner),
+ _interpolated_position(inner),
+ inner.c.postcode, inner.c.country_code,
+ inner.c.distance)
if self.has_geometries():
sub = sql.subquery('geom')
return (await self.conn.execute(sql, self.bind_params)).one_or_none()
-
async def _find_tiger_number_for_street(self, parent_place_id: int) -> Optional[SaRow]:
t = self.conn.t.tiger
return (await self.conn.execute(sql, self.bind_params)).one_or_none()
-
async def lookup_street_poi(self) -> Tuple[Optional[SaRow], RowFunc]:
""" Find a street or POI/address for the given WKT point.
"""
return row, row_func
-
async def _lookup_area_address(self) -> Optional[SaRow]:
""" Lookup large addressable areas for the given WKT point.
"""
# later only a minimum of results needs to be checked with ST_Contains.
inner = sa.select(t, sa.literal(0.0).label('distance'))\
.where(t.c.rank_search.between(5, MAX_RANK_PARAM))\
+ .where(t.c.rank_address != 5)\
+ .where(t.c.rank_address != 11)\
.where(t.c.geometry.intersects(WKT_PARAM))\
.where(sa.func.PlacexGeometryReverseLookuppolygon())\
.order_by(sa.desc(t.c.rank_search))\
.subquery('area')
return _select_from_placex(inner, False)\
- .where(inner.c.geometry.ST_Contains(WKT_PARAM))\
- .order_by(sa.desc(inner.c.rank_search))\
- .limit(1)
+ .where(inner.c.geometry.ST_Contains(WKT_PARAM))\
+ .order_by(sa.desc(inner.c.rank_search))\
+ .limit(1)
sql: SaLambdaSelect = sa.lambda_stmt(_base_query)
if self.has_geometries():
def _place_inside_area_query() -> SaSelect:
inner = \
- sa.select(t,
- t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\
- .where(t.c.rank_search > address_rank)\
- .where(t.c.rank_search <= MAX_RANK_PARAM)\
- .where(t.c.indexed_status == 0)\
- .where(sa.func.IntersectsReverseDistance(t, WKT_PARAM))\
- .order_by(sa.desc(t.c.rank_search))\
- .limit(50)\
- .subquery('places')
+ sa.select(t, t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\
+ .where(t.c.rank_search > address_rank)\
+ .where(t.c.rank_search <= MAX_RANK_PARAM)\
+ .where(t.c.indexed_status == 0)\
+ .where(sa.func.IntersectsReverseDistance(t, WKT_PARAM))\
+ .order_by(sa.desc(t.c.rank_search))\
+ .limit(50)\
+ .subquery('places')
touter = t.alias('outer')
return _select_from_placex(inner, False)\
return address_row
-
async def _lookup_area_others(self) -> Optional[SaRow]:
t = self.conn.t.placex
.subquery()
sql = _select_from_placex(inner, False)\
- .where(sa.or_(sa.not_(inner.c.geometry.is_area()),
- inner.c.geometry.ST_Contains(WKT_PARAM)))\
- .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
- .limit(1)
+ .where(sa.or_(sa.not_(inner.c.geometry.is_area()),
+ inner.c.geometry.ST_Contains(WKT_PARAM)))\
+ .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\
+ .limit(1)
if self.has_geometries():
sql = self._add_geometry_columns(sql, inner.c.geometry)
return row
-
async def lookup_area(self) -> Optional[SaRow]:
""" Lookup large areas for the current search.
"""
return _get_closest(address_row, other_row)
-
async def lookup_country_codes(self) -> List[str]:
""" Lookup the country for the current search.
"""
log().var_dump('Country codes', ccodes)
return ccodes
-
- async def lookup_country(self, ccodes: List[str]) -> Optional[SaRow]:
+ async def lookup_country(self, ccodes: List[str]) -> Tuple[Optional[SaRow], RowFunc]:
""" Lookup the country for the current search.
"""
+ row_func = nres.create_from_placex_row
if not ccodes:
ccodes = await self.lookup_country_codes()
if not ccodes:
- return None
+ return None, row_func
t = self.conn.t.placex
if self.max_rank > 4:
log().comment('Search for place nodes in country')
def _base_query() -> SaSelect:
- inner = \
- sa.select(t,
- t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\
- .where(t.c.rank_search > 4)\
- .where(t.c.rank_search <= MAX_RANK_PARAM)\
- .where(t.c.indexed_status == 0)\
- .where(t.c.country_code.in_(ccodes))\
- .where(sa.func.IntersectsReverseDistance(t, WKT_PARAM))\
- .order_by(sa.desc(t.c.rank_search))\
- .limit(50)\
- .subquery('area')
+ inner = sa.select(t, t.c.geometry.ST_Distance(WKT_PARAM).label('distance'))\
+ .where(t.c.rank_search > 4)\
+ .where(t.c.rank_search <= MAX_RANK_PARAM)\
+ .where(t.c.indexed_status == 0)\
+ .where(t.c.country_code.in_(ccodes))\
+ .where(sa.func.IntersectsReverseDistance(t, WKT_PARAM))\
+ .order_by(sa.desc(t.c.rank_search))\
+ .limit(50)\
+ .subquery('area')
return _select_from_placex(inner, False)\
.where(sa.func.IsBelowReverseDistance(inner.c.distance, inner.c.rank_search))\
address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
- return address_row
+ if address_row is None:
+ # finally fall back to country table
+ t = self.conn.t.country_name
+ tgrid = self.conn.t.country_grid
+
+ sql = sa.select(tgrid.c.country_code,
+ tgrid.c.geometry.ST_Centroid().ST_Collect().ST_Centroid()
+ .label('centroid'),
+ tgrid.c.geometry.ST_Collect().ST_Expand(0).label('bbox'))\
+ .where(tgrid.c.country_code.in_(ccodes))\
+ .group_by(tgrid.c.country_code)
+
+ sub = sql.subquery('grid')
+ sql = sa.select(t.c.country_code,
+ t.c.name.merge(t.c.derived_name).label('name'),
+ sub.c.centroid, sub.c.bbox)\
+ .join(sub, t.c.country_code == sub.c.country_code)\
+ .order_by(t.c.country_code)\
+ .limit(1)
+ sql = self._add_geometry_columns(sql, sub.c.centroid)
+
+ address_row = (await self.conn.execute(sql, self.bind_params)).one_or_none()
+ row_func = nres.create_from_country_row
+
+ return address_row, row_func
async def lookup(self, coord: AnyPoint) -> Optional[nres.ReverseResult]:
""" Look up a single coordinate. Returns the place information,
"""
log().function('reverse_lookup', coord=coord, params=self.params)
-
self.bind_params['wkt'] = f'POINT({coord[0]} {coord[1]})'
row: Optional[SaRow] = None
if self.max_rank > 4:
row = await self.lookup_area()
if row is None and self.layer_enabled(DataLayer.ADDRESS):
- row = await self.lookup_country(ccodes)
+ row, row_func = await self.lookup_country(ccodes)
result = row_func(row, nres.ReverseResult)
if result is not None:
assert row is not None
- result.distance = row.distance
+ result.distance = getattr(row, 'distance', 0)
if hasattr(row, 'bbox'):
result.bbox = Bbox.from_wkb(row.bbox)
await nres.add_result_details(self.conn, [result], self.params)
projects / nominatim.git / blobdiff