from nominatim.api.connection import SearchConnection
import nominatim.api.results as nres
from nominatim.api.logging import log
-from nominatim.api.types import AnyPoint, DataLayer, LookupDetails, GeometryFormat, Bbox
+from nominatim.api.types import AnyPoint, DataLayer, ReverseDetails, GeometryFormat, Bbox
# In SQLAlchemy expression which compare with NULL need to be expressed with
# the equal sign.
else_=table.c.linegeo.ST_LineInterpolatePoint(rounded_pos)).label('centroid')
+def _locate_interpolation(table: SaFromClause, wkt: WKTElement) -> SaLabel:
+ """ Given a position, locate the closest point on the line.
+ """
+ return sa.case((table.c.linegeo.ST_GeometryType() == 'ST_LineString',
+ sa.func.ST_LineLocatePoint(table.c.linegeo, wkt)),
+ else_=0).label('position')
+
+
def _is_address_point(table: SaFromClause) -> SaColumn:
return sa.and_(table.c.rank_address == 30,
sa.or_(table.c.housenumber != None,
coordinate.
"""
- def __init__(self, conn: SearchConnection, max_rank: int, layer: DataLayer,
- details: LookupDetails) -> None:
+ def __init__(self, conn: SearchConnection, params: ReverseDetails) -> None:
self.conn = conn
- self.max_rank = max_rank
- self.layer = layer
- self.details = details
+ self.params = params
+
+
+ @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.layer & l for l in layer)
+ return any(self.params.layers & l for l in layer)
def layer_disabled(self, *layer: DataLayer) -> bool:
""" Return true when none of the given layer types is requested.
"""
- return not any(self.layer & l for l in layer)
+ return not any(self.params.layers & l for l in layer)
def has_feature_layers(self) -> bool:
return self.layer_enabled(DataLayer.RAILWAY, DataLayer.MANMADE, DataLayer.NATURAL)
def _add_geometry_columns(self, sql: SaSelect, col: SaColumn) -> SaSelect:
- if not self.details.geometry_output:
+ if not self.has_geometries():
return sql
out = []
- if self.details.geometry_simplification > 0.0:
- col = col.ST_SimplifyPreserveTopology(self.details.geometry_simplification)
+ if self.params.geometry_simplification > 0.0:
+ col = col.ST_SimplifyPreserveTopology(self.params.geometry_simplification)
- if self.details.geometry_output & GeometryFormat.GEOJSON:
+ if self.params.geometry_output & GeometryFormat.GEOJSON:
out.append(col.ST_AsGeoJSON().label('geometry_geojson'))
- if self.details.geometry_output & GeometryFormat.TEXT:
+ if self.params.geometry_output & GeometryFormat.TEXT:
out.append(col.ST_AsText().label('geometry_text'))
- if self.details.geometry_output & GeometryFormat.KML:
+ if self.params.geometry_output & GeometryFormat.KML:
out.append(col.ST_AsKML().label('geometry_kml'))
- if self.details.geometry_output & GeometryFormat.SVG:
+ if self.params.geometry_output & GeometryFormat.SVG:
out.append(col.ST_AsSVG().label('geometry_svg'))
return sql.add_columns(*out)
sql = sa.select(t,
t.c.linegeo.ST_Distance(wkt).label('distance'),
- t.c.linegeo.ST_LineLocatePoint(wkt).label('position'))\
+ _locate_interpolation(t, wkt))\
.where(t.c.linegeo.ST_DWithin(wkt, distance))\
.where(t.c.startnumber != None)\
.order_by('distance')\
if parent_place_id is not None:
sql = sql.where(t.c.parent_place_id == parent_place_id)
- inner = sql.subquery()
+ inner = sql.subquery('ipol')
sql = sa.select(inner.c.place_id, inner.c.osm_id,
inner.c.parent_place_id, inner.c.address,
inner.c.postcode, inner.c.country_code,
inner.c.distance)
- if self.details.geometry_output:
- sub = sql.subquery()
- sql = self._add_geometry_columns(sql, sub.c.centroid)
+ if self.has_geometries():
+ sub = sql.subquery('geom')
+ sql = self._add_geometry_columns(sa.select(sub), sub.c.centroid)
return (await self.conn.execute(sql)).one_or_none()
inner = sa.select(t,
t.c.linegeo.ST_Distance(wkt).label('distance'),
- sa.func.ST_LineLocatePoint(t.c.linegeo, wkt).label('position'))\
+ _locate_interpolation(t, wkt))\
.where(t.c.linegeo.ST_DWithin(wkt, 0.001))\
.where(t.c.parent_place_id == parent_place_id)\
.order_by('distance')\
.limit(1)\
- .subquery()
+ .subquery('tiger')
sql = sa.select(inner.c.place_id,
inner.c.parent_place_id,
inner.c.postcode,
inner.c.distance)
- if self.details.geometry_output:
- sub = sql.subquery()
- sql = self._add_geometry_columns(sql, sub.c.centroid)
+ if self.has_geometries():
+ sub = sql.subquery('geom')
+ sql = self._add_geometry_columns(sa.select(sub), sub.c.centroid)
return (await self.conn.execute(sql)).one_or_none()
.where(t.c.type != 'postcode')\
.order_by(sa.desc(t.c.rank_search))\
.limit(50)\
- .subquery()
+ .subquery('area')
sql = _select_from_placex(inner)\
.where(inner.c.geometry.ST_Contains(wkt))\
.intersects(wkt))\
.order_by(sa.desc(t.c.rank_search))\
.limit(50)\
- .subquery()
+ .subquery('places')
touter = self.conn.t.placex.alias('outer')
sql = _select_from_placex(inner)\
.where(t.c.rank_address == 4)\
.where(t.c.rank_search == 4)\
.where(t.c.linked_place_id == None)\
- .order_by('distance')
+ .order_by('distance')\
+ .limit(1)
sql = self._add_geometry_columns(sql, t.c.geometry)
""" Look up a single coordinate. Returns the place information,
if a place was found near the coordinates or None otherwise.
"""
- log().function('reverse_lookup',
- coord=coord, max_rank=self.max_rank,
- layer=self.layer, details=self.details)
+ log().function('reverse_lookup', coord=coord, params=self.params)
wkt = WKTElement(f'POINT({coord[0]} {coord[1]})', srid=4326)
result.distance = row.distance
if hasattr(row, 'bbox'):
result.bbox = Bbox.from_wkb(row.bbox.data)
- await nres.add_result_details(self.conn, result, self.details)
+ await nres.add_result_details(self.conn, [result], self.params)
return result
projects / nominatim.git / blobdiff