From 41da298b189949e24ff5271bcd671bf5f7b7a444 Mon Sep 17 00:00:00 2001 From: Sarah Hoffmann Date: Wed, 22 Mar 2023 00:07:17 +0100 Subject: [PATCH] add python implementation of reverse This adds an additional layer parameter and slightly changes the queries to do more efficient lookups for large area features. --- nominatim/api/__init__.py | 7 +- nominatim/api/core.py | 40 ++- nominatim/api/results.py | 12 +- nominatim/api/reverse.py | 509 ++++++++++++++++++++++++++++++++++++++ nominatim/api/types.py | 85 ++++++- nominatim/typing.py | 4 +- 6 files changed, 650 insertions(+), 7 deletions(-) create mode 100644 nominatim/api/reverse.py diff --git a/nominatim/api/__init__.py b/nominatim/api/__init__.py index 9494d453..cf58f27a 100644 --- a/nominatim/api/__init__.py +++ b/nominatim/api/__init__.py @@ -21,12 +21,15 @@ from .types import (PlaceID as PlaceID, OsmID as OsmID, PlaceRef as PlaceRef, Point as Point, + Bbox as Bbox, GeometryFormat as GeometryFormat, - LookupDetails as LookupDetails) + LookupDetails as LookupDetails, + DataLayer as DataLayer) from .results import (SourceTable as SourceTable, AddressLine as AddressLine, AddressLines as AddressLines, WordInfo as WordInfo, WordInfos as WordInfos, - DetailedResult as DetailedResult) + DetailedResult as DetailedResult, + ReverseResult as ReverseResult) from .localization import (Locales as Locales) diff --git a/nominatim/api/core.py b/nominatim/api/core.py index 1d2df8a8..32c9b5e5 100644 --- a/nominatim/api/core.py +++ b/nominatim/api/core.py @@ -21,8 +21,9 @@ from nominatim.config import Configuration from nominatim.api.connection import SearchConnection from nominatim.api.status import get_status, StatusResult from nominatim.api.lookup import get_place_by_id -from nominatim.api.types import PlaceRef, LookupDetails -from nominatim.api.results import DetailedResult +from nominatim.api.reverse import reverse_lookup +from nominatim.api.types import PlaceRef, LookupDetails, AnyPoint, DataLayer +from nominatim.api.results import DetailedResult, ReverseResult class NominatimAPIAsync: @@ -136,6 +137,29 @@ class NominatimAPIAsync: return await get_place_by_id(conn, place, details or LookupDetails()) + async def reverse(self, coord: AnyPoint, max_rank: Optional[int] = None, + layer: Optional[DataLayer] = None, + details: Optional[LookupDetails] = None) -> Optional[ReverseResult]: + """ Find a place by its coordinates. Also known as reverse geocoding. + + Returns the closest result that can be found or None if + no place matches the given criteria. + """ + # The following negation handles NaN correctly. Don't change. + if not abs(coord[0]) <= 180 or not abs(coord[1]) <= 90: + # There are no results to be expected outside valid coordinates. + return None + + if layer is None: + layer = DataLayer.ADDRESS | DataLayer.POI + + max_rank = max(0, min(max_rank or 30, 30)) + + async with self.begin() as conn: + return await reverse_lookup(conn, coord, max_rank, layer, + details or LookupDetails()) + + class NominatimAPI: """ API loader, synchronous version. """ @@ -172,3 +196,15 @@ class NominatimAPI: """ Get detailed information about a place in the database. """ return self._loop.run_until_complete(self._async_api.lookup(place, details)) + + + def reverse(self, coord: AnyPoint, max_rank: Optional[int] = None, + layer: Optional[DataLayer] = None, + details: Optional[LookupDetails] = None) -> Optional[ReverseResult]: + """ Find a place by its coordinates. Also known as reverse geocoding. + + Returns the closest result that can be found or None if + no place matches the given criteria. + """ + return self._loop.run_until_complete( + self._async_api.reverse(coord, max_rank, layer, details)) diff --git a/nominatim/api/results.py b/nominatim/api/results.py index a8d6588a..84d4ced9 100644 --- a/nominatim/api/results.py +++ b/nominatim/api/results.py @@ -19,7 +19,7 @@ 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 @@ -46,6 +46,8 @@ class AddressLine: names: Dict[str, str] extratags: Optional[Dict[str, str]] + local_name: Optional[str] = None + admin_level: Optional[int] fromarea: bool isaddress: bool @@ -136,6 +138,14 @@ class DetailedResult(BaseResult): 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 + + 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_')} diff --git a/nominatim/api/reverse.py b/nominatim/api/reverse.py new file mode 100644 index 00000000..053b96dd --- /dev/null +++ b/nominatim/api/reverse.py @@ -0,0 +1,509 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +# +# This file is part of Nominatim. (https://nominatim.org) +# +# Copyright (C) 2023 by the Nominatim developer community. +# For a full list of authors see the git log. +""" +Implementation of reverse geocoding. +""" +from typing import Optional + +import sqlalchemy as sa +from geoalchemy2 import WKTElement +from geoalchemy2.types import Geometry + +from nominatim.typing import SaColumn, SaSelect, SaTable, SaLabel, SaClause +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 + +def _select_from_placex(t: SaTable, wkt: Optional[str] = None) -> SaSelect: + """ Create a select statement with the columns relevant for reverse + results. + """ + if wkt is None: + distance = t.c.distance + else: + distance = t.c.geometry.ST_Distance(wkt) + + 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, + t.c.housenumber, t.c.postcode, t.c.country_code, + t.c.importance, t.c.wikipedia, + t.c.parent_place_id, t.c.rank_address, t.c.rank_search, + t.c.centroid, + distance.label('distance'), + t.c.geometry.ST_Expand(0).label('bbox')) + + +def _interpolated_housenumber(table: SaTable) -> SaLabel: + # Entries with startnumber = endnumber are legacy from version < 4.1 + 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.Integer).label('housenumber') + + +def _is_address_point(table: SaTable) -> SaClause: + return sa.and_(table.c.rank_address == 30, + sa.or_(table.c.housenumber != None, + table.c.name.has_key('housename'))) + + +class ReverseGeocoder: + """ Class implementing the logic for looking up a place from a + coordinate. + """ + + def __init__(self, conn: SearchConnection, max_rank: int, layer: DataLayer, + details: LookupDetails) -> None: + self.conn = conn + self.max_rank = max_rank + self.layer = layer + self.details = details + + + def _add_geometry_columns(self, sql: SaSelect, col: SaColumn) -> SaSelect: + if not self.details.geometry_output: + return sql + + out = [] + + if self.details.geometry_simplification > 0.0: + col = col.ST_SimplifyPreserveTopology(self.details.geometry_simplification) + + if self.details.geometry_output & GeometryFormat.GEOJSON: + out.append(col.ST_AsGeoJSON().label('geometry_geojson')) + if self.details.geometry_output & GeometryFormat.TEXT: + out.append(col.ST_AsText().label('geometry_text')) + if self.details.geometry_output & GeometryFormat.KML: + out.append(col.ST_AsKML().label('geometry_kml')) + if self.details.geometry_output & GeometryFormat.SVG: + out.append(col.ST_AsSVG().label('geometry_svg')) + + return sql.add_columns(*out) + + + def _filter_by_layer(self, table: SaTable) -> SaColumn: + if self.layer & DataLayer.MANMADE: + exclude = [] + if not (self.layer & DataLayer.RAILWAY): + exclude.append('railway') + if not (self.layer & DataLayer.NATURAL): + exclude.extend(('natural', 'water', 'waterway')) + return table.c.class_.not_in(tuple(exclude)) + + include = [] + if self.layer & DataLayer.RAILWAY: + include.append('railway') + if not (self.layer & DataLayer.NATURAL): + include.extend(('natural', 'water', 'waterway')) + return table.c.class_.in_(tuple(include)) + + + async def _find_closest_street_or_poi(self, wkt: WKTElement) -> SaRow: + """ Look up the clostest rank 26+ place in the database. + """ + t = self.conn.t.placex + + sql = _select_from_placex(t, wkt)\ + .where(t.c.geometry.ST_DWithin(wkt, distance))\ + .where(t.c.indexed_status == 0)\ + .where(t.c.linked_place_id == None)\ + .where(sa.or_(t.c.geometry.ST_GeometryType().not_in(('ST_Polygon', 'ST_MultiPolygon')), + t.c.centroid.ST_Distance(wkt) < distance))\ + .order_by('distance')\ + .limit(1) + + sql = self._add_geometry_columns(sql, t.c.geometry) + + restrict = [] + + if self.layer & DataLayer.ADDRESS: + restrict.append(sa.and_(t.c.rank_address >= 26, + t.c.rank_address <= self.max_rank)) + if self.max_rank == 30: + restrict.append(_is_address_point(t)) + if self.layer & DataLayer.POI and max_rank == 30: + restrict.append(sa.and_(t.c.rank_search == 30, + t.c.class_.not_in(('place', 'building')), + t.c.geometry.ST_GeometryType() != 'ST_LineString')) + if self.layer & (DataLayer.RAILWAY | DataLayer.MANMADE | DataLayer.NATURAL): + restrict.append(sa.and_(t.c.rank_search >= 26, + tc.rank_search <= self.max_rank, + self._filter_by_layer(t))) + + if restrict: + sql = sql.where(sa.or_(*restrict)) + + return (await self.conn.execute(sql)).one_or_none() + + + async def _find_housenumber_for_street(self, parent_place_id: int, + wkt: WKTElement) -> Optional[SaRow]: + t = conn.t.placex + + sql = _select_from_placex(t, wkt)\ + .where(t.c.geometry.ST_DWithin(wkt, 0.001))\ + .where(t.c.parent_place_id == parent_place_id)\ + .where(_is_address_point(t))\ + .where(t.c.indexed_status == 0)\ + .where(t.c.linked_place_id == None)\ + .order_by('distance')\ + .limit(1) + + sql = self._add_geometry_columns(sql, t.c.geometry) + + return (await self.conn.execute(sql)).one_or_none() + + + async def _find_interpolation_for_street(self, parent_place_id: Optional[int], + wkt: WKTElement) -> Optional[SaRow]: + t = self.conn.t.osmline + + inner = sa.select(t, + t.c.linegeo.ST_Distance(wkt).label('distance'), + t.c.linegeo.ST_LineLocatePoint(wkt).label('position'))\ + .where(t.c.linegeo.ST_DWithin(wkt, distance))\ + .order_by('distance')\ + .limit(1) + + if parent_place_id is not None: + inner = inner.where(t.c.parent_place_id == parent_place_id) + + inner = inner.subquery() + + sql = sa.select(inner.c.place_id, inner.c.osm_id, + inner.c.parent_place_id, inner.c.address, + _interpolated_housenumber(inner), + inner.c.postcode, inner.c.country_code, + inner.c.linegeo.ST_LineInterpolatePoint(inner.c.position).label('centroid'), + inner.c.distance) + + if self.details.geometry_output: + sub = sql.subquery() + sql = self._add_geometry_columns(sql, sub.c.centroid) + + return (await self.conn.execute(sql)).one_or_none() + + + async def _find_tiger_number_for_street(self, parent_place_id: int, + wkt: WKTElement) -> Optional[SaRow]: + t = self.conn.t.tiger + + inner = sa.select(t, + t.c.linegeo.ST_Distance(wkt).label('distance'), + sa.func.ST_LineLocatePoint(t.c.linegeo, wkt).label('position'))\ + .where(t.c.linegeo.ST_DWithin(wkt, 0.001))\ + .where(t.c.parent_place_id == parent_place_id)\ + .order_by('distance')\ + .limit(1)\ + .subquery() + + sql = sa.select(inner.c.place_id, + inner.c.parent_place_id, + _interpolated_housenumber(inner), + inner.c.postcode, + inner.c.linegeo.ST_LineInterpolatePoint(inner.c.position).label('centroid'), + inner.c.distance) + + if self.details.geometry_output: + sub = sql.subquery() + sql = self._add_geometry_columns(sql, sub.c.centroid) + + return (await conn.execute(sql)).one_or_none() + + + async def lookup_street_poi(self, wkt: WKTElement) -> Optional[nres.ReverseResult]: + """ Find a street or POI/address for the given WKT point. + """ + log().section('Reverse lookup on street/address level') + result = None + distance = 0.006 + parent_place_id = None + + row = await self._find_closest_street_or_poi(wkt) + log().var_dump('Result (street/building)', row) + + # If the closest result was a street, but an address was requested, + # check for a housenumber nearby which is part of the street. + if row is not None: + if self.max_rank > 27 \ + and self.layer & DataLayer.ADDRESS \ + and row.rank_address <= 27: + distance = 0.001 + parent_place_id = row.place_id + log().comment('Find housenumber for street') + addr_row = await self._find_housenumber_for_street(parent_place_id, wkt) + log().var_dump('Result (street housenumber)', addr_row) + + if addr_row is not None: + row = addr_row + distance = addr_row.distance + elif row.country_code == 'us' and parent_place_id is not None: + log().comment('Find TIGER housenumber for street') + addr_row = await self._find_tiger_number_for_street(parent_place_id, wkt) + log().var_dump('Result (street Tiger housenumber)', addr_row) + + if addr_row is not None: + result = nres.create_from_tiger_row(addr_row) + else: + distance = row.distance + + # Check for an interpolation that is either closer than our result + # or belongs to a close street found. + if self.max_rank > 27 and self.layer & DataLayer.ADDRESS: + log().comment('Find interpolation for street') + addr_row = await self._find_interpolation_for_street(parent_place_id, wkt) + log().var_dump('Result (street interpolation)', addr_row) + if addr_row is not None: + result = nres.create_from_osmline_row(addr_row) + + return result or nres.create_from_placex_row(row) + + + async def _lookup_area_address(self, wkt: WKTElement) -> Optional[SaRow]: + """ Lookup large addressable areas for the given WKT point. + """ + log().comment('Reverse lookup by larger address area features') + t = self.conn.t.placex + + # The inner SQL brings results in the right order, so that + # 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, self.max_rank))\ + .where(t.c.rank_address.between(5, 25))\ + .where(t.c.geometry.ST_GeometryType().in_(('ST_Polygon', 'ST_MultiPolygon')))\ + .where(t.c.geometry.intersects(wkt))\ + .where(t.c.name != None)\ + .where(t.c.indexed_status == 0)\ + .where(t.c.linked_place_id == None)\ + .where(t.c.type != 'postcode')\ + .order_by(sa.desc(t.c.rank_search))\ + .limit(50)\ + .subquery() + + sql = _select_from_placex(inner)\ + .where(inner.c.geometry.ST_Contains(wkt))\ + .order_by(sa.desc(inner.c.rank_search))\ + .limit(1) + + sql = self._add_geometry_columns(sql, inner.c.geometry) + + address_row = (await self.conn.execute(sql)).one_or_none() + log().var_dump('Result (area)', address_row) + + if address_row is not None and address_row.rank_search < max_rank: + log().comment('Search for better matching place nodes inside the area') + inner = sa.select(t, + t.c.geometry.ST_Distance(wkt).label('distance'))\ + .where(t.c.osm_type == 'N')\ + .where(t.c.rank_search > address_row.rank_search)\ + .where(t.c.rank_search <= max_rank)\ + .where(t.c.rank_address.between(5, 25))\ + .where(t.c.name != None)\ + .where(t.c.indexed_status == 0)\ + .where(t.c.linked_place_id == None)\ + .where(t.c.type != 'postcode')\ + .where(t.c.geometry + .ST_Buffer(sa.func.reverse_place_diameter(t.c.rank_search)) + .intersects(wkt))\ + .order_by(sa.desc(t.c.rank_search))\ + .limit(50)\ + .subquery() + + touter = conn.t.placex.alias('outer') + sql = _select_from_placex(inner)\ + .where(touter.c.place_id == address_row.place_id)\ + .where(touter.c.geometry.ST_Contains(inner.c.geometry))\ + .where(inner.c.distance < sa.func.reverse_place_diameter(inner.c.rank_search))\ + .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\ + .limit(1) + + sql = self._add_geometry_columns(sql, inner.c.geometry) + + place_address_row = (await self.conn.execute(sql)).one_or_none() + log().var_dump('Result (place node)', place_address_row) + + if place_address_row is not None: + return place_address_row + + return address_row + + + async def _lookup_area_others(self, wkt: WKTElement) -> Optional[SaRow]: + t = conn.t.placex + + inner = sa.select(t, t.c.geometry.ST_Distance(wkt).label('distance'))\ + .where(t.c.rank_address == 0)\ + .where(t.c.rank_search.between(5, self.max_rank))\ + .where(t.c.name != None)\ + .where(t.c.indexed_status == 0)\ + .where(t.c.linked_place_id == None)\ + .where(self._filter_by_layer(t))\ + .where(sa.func.reverse_buffered_extent(t.c.geometry, type_=Geometry) + .intersects(wkt))\ + .order_by(sa.desc(t.c.rank_search))\ + .limit(50) + + sql = _select_from_placex(inner)\ + .where(sa._or(inner.c.geometry.ST_GeometryType().not_in(('ST_Polygon', 'ST_MultiPolygon')), + inner.c.geometry.ST_Contains(wkt)))\ + .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\ + .limit(1) + + sql = self._add_geometry_columns(sql, inner.c.geometry) + + row = (await self.conn.execute(sql)).one_or_none() + log().var_dump('Result (non-address feature)', row) + + return row + + + async def lookup_area(self, wkt: WKTElement) -> Optional[nres.ReverseResult]: + """ Lookup large areas for the given WKT point. + """ + log().section('Reverse lookup by larger area features') + t = self.conn.t.placex + + if self.layer & DataLayer.ADDRESS: + address_row = await self._lookup_area_address(wkt) + address_distance = address_row.distance + else: + address_row = None + address_distance = 1000 + + if self.layer & (~DataLayer.ADDRESS & ~DataLayer.POI): + other_row = await self._lookup_area_others(wkt) + other_distance = other_row.distance + else: + other_row = None + other_distance = 1000 + + result = address_row if address_distance <= other_distance else other_row + + return nres.create_from_placex_row(result) + + + async def lookup_country(self, wkt: WKTElement) -> Optional[nres.ReverseResult]: + """ Lookup the country for the given WKT point. + """ + log().section('Reverse lookup by country code') + t = self.conn.t.country_grid + sql = sa.select(t.c.country_code).distinct()\ + .where(t.c.geometry.ST_Contains(wkt)) + + ccodes = tuple((r[0] for r in await self.conn.execute(sql))) + log().var_dump('Country codes', ccodes) + + if not ccodes: + return None + + if self.layer & DataLayer.ADDRESS and self.max_rank > 4: + log().comment('Search for place nodes in country') + + t = conn.t.placex + inner = sa.select(t, + t.c.geometry.ST_Distance(wkt).label('distance'))\ + .where(t.c.osm_type == 'N')\ + .where(t.c.rank_search > 4)\ + .where(t.c.rank_search <= self.max_rank)\ + .where(t.c.rank_address.between(5, 25))\ + .where(t.c.name != None)\ + .where(t.c.indexed_status == 0)\ + .where(t.c.linked_place_id == None)\ + .where(t.c.type != 'postcode')\ + .where(t.c.country_code.in_(ccodes))\ + .where(t.c.geometry + .ST_Buffer(sa.func.reverse_place_diameter(t.c.rank_search)) + .intersects(wkt))\ + .order_by(sa.desc(t.c.rank_search))\ + .limit(50)\ + .subquery() + + sql = _select_from_placex(inner)\ + .where(inner.c.distance < sa.func.reverse_place_diameter(inner.c.rank_search))\ + .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\ + .limit(1) + + sql = self._add_geometry_columns(sql, inner.c.geometry) + + address_row = (await self.conn.execute(sql)).one_or_none() + log().var_dump('Result (addressable place node)', address_row) + else: + address_row = None + + if layer & (~DataLayer.ADDRESS & ~DataLayer.POI) and self.max_rank > 4: + log().comment('Search for non-address features inside country') + + t = conn.t.placex + inner = sa.select(t, t.c.geometry.ST_Distance(wkt).label('distance'))\ + .where(t.c.rank_address == 0)\ + .where(t.c.rank_search.between(5, self.max_rank))\ + .where(t.c.name != None)\ + .where(t.c.indexed_status == 0)\ + .where(t.c.linked_place_id == None)\ + .where(self._filter_by_layer(t))\ + .where(t.c.country_code.in_(ccode))\ + .where(sa.func.reverse_buffered_extent(t.c.geometry, type_=Geometry) + .intersects(wkt))\ + .order_by(sa.desc(t.c.rank_search))\ + .limit(50)\ + .subquery() + + sql = _select_from_placex(inner)\ + .where(sa._or(inner.c.geometry.ST_GeometryType().not_in(('ST_Polygon', 'ST_MultiPolygon')), + inner.c.geometry.ST_Contains(wkt)))\ + .order_by(sa.desc(inner.c.rank_search), inner.c.distance)\ + .limit(1) + + sql = self._add_geometry_columns(sql, inner.c.geometry) + + other_row = (await self.conn.execute(sql)).one_or_none() + log().var_dump('Result (non-address feature)', other_row) + else: + other_row = None + + if layer & DataLayer.ADDRESS and address_row is None and other_row is None: + # Still nothing, then return a country with the appropriate country code. + t = conn.t.placex + sql = _select_from_placex(t, wkt)\ + .where(t.c.country_code.in_(ccodes))\ + .where(t.c.rank_address == 4)\ + .where(t.c.rank_search == 4)\ + .where(t.c.linked_place_id == None)\ + .order_by('distance') + + sql = self._add_geometry_columns(sql, inner.c.geometry) + + address_row = (await self.conn.execute(sql)).one_or_none() + + return nres.create_from_placex_row(_get_closest_row(address_row, other_row)) + + + async def lookup(self, coord: AnyPoint) -> Optional[nres.ReverseResult]: + """ 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) + + + wkt = WKTElement(f'POINT({coord[0]} {coord[1]})', srid=4326) + + result: Optional[ReverseResult] = None + + if max_rank >= 26: + result = await self.lookup_street_poi(wkt) + if result is None and max_rank > 4: + result = await self.lookup_area(wkt) + if result is None: + result = await self.lookup_country(wkt) + if result is not None: + await nres.add_result_details(self.conn, result, self.details) + + return result diff --git a/nominatim/api/types.py b/nominatim/api/types.py index 9dc3ff2e..344fd91b 100644 --- a/nominatim/api/types.py +++ b/nominatim/api/types.py @@ -7,7 +7,7 @@ """ Complex datatypes used by the Nominatim API. """ -from typing import Optional, Union, NamedTuple +from typing import Optional, Union, Tuple, NamedTuple import dataclasses import enum from struct import unpack @@ -83,6 +83,74 @@ class Point(NamedTuple): return Point(x, y) +AnyPoint = Union[Point, Tuple[float, float]] + + +class Bbox: + """ A bounding box in WSG84 projection. + + The coordinates are available as an array in the 'coord' + property in the order (minx, miny, maxx, maxy). + """ + def __init__(self, minx: float, miny: float, maxx: float, maxy: float) -> None: + self.coords = (minx, miny, maxx, maxy) + + + @property + def minlat(self) -> float: + """ Southern-most latitude, corresponding to the minimum y coordinate. + """ + return self.coords[1] + + + @property + def maxlat(self) -> float: + """ Northern-most latitude, corresponding to the maximum y coordinate. + """ + return self.coords[3] + + + @property + def minlon(self) -> float: + """ Western-most longitude, corresponding to the minimum x coordinate. + """ + return self.coords[0] + + + @property + def maxlon(self) -> float: + """ Eastern-most longitude, corresponding to the maximum x coordinate. + """ + return self.coords[2] + + + @staticmethod + def from_wkb(wkb: Optional[bytes]) -> 'Optional[Bbox]': + """ Create a Bbox from a bounding box polygon as returned by + the database. Return s None if the input value is None. + """ + if wkb is None: + return None + + if len(wkb) != 97: + raise ValueError("WKB must be a bounding box polygon") + if wkb.startswith(b'\x01\x03\x00\x00\x20\xE6\x10\x00\x00\x01\x00\x00\x00\x05\x00\x00\x00'): + x1, y1, _, _, x2, y2 = unpack('dddddd', wkb[17:65]) + else: + raise ValueError("WKB has wrong header") + + return Bbox(min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2)) + + + def from_point(pt: Point, buffer: float) -> 'Bbox': + """ Return a Bbox around the point with the buffer added to all sides. + """ + return Bbox(pt[0] - buffer, pt[1] - buffer, + pt[0] + buffer, pt[1] + buffer) + + class GeometryFormat(enum.Flag): """ Geometry output formats supported by Nominatim. """ @@ -117,3 +185,18 @@ class LookupDetails: keywords: bool = False """ Add information about the search terms used for this place. """ + geometry_simplification: float = 0.0 + """ Simplification factor for a geometry in degrees WGS. A factor of + 0.0 means the original geometry is kept. The higher the value, the + more the geometry gets simplified. + """ + + +class DataLayer(enum.Flag): + """ Layer types that can be selected for reverse and forward search. + """ + POI = enum.auto() + ADDRESS = enum.auto() + RAILWAY = enum.auto() + MANMADE = enum.auto() + NATURAL = enum.auto() diff --git a/nominatim/typing.py b/nominatim/typing.py index 07efc7ba..1946c1a6 100644 --- a/nominatim/typing.py +++ b/nominatim/typing.py @@ -53,7 +53,7 @@ else: # SQLAlchemy introduced generic types in version 2.0 making typing -# inclompatiple with older versions. Add wrappers here so we don't have +# incompatible with older versions. Add wrappers here so we don't have # to litter the code with bare-string types. if TYPE_CHECKING: @@ -66,3 +66,5 @@ SaSelect: TypeAlias = 'sa.Select[Any]' SaRow: TypeAlias = 'sa.Row[Any]' SaColumn: TypeAlias = 'sa.Column[Any]' SaLabel: TypeAlias = 'sa.Label[Any]' +SaTable: TypeAlias = 'sa.Table[Any]' +SaClause: TypeAlias = 'sa.ClauseElement[Any]' -- 2.45.2