Merge pull request #4101 from Itz-Agasta/artificial-postcodes

Add stable postcode refs for details lookup and search exclusion

Author: lonvia

docs/api/Details.md

@@ -14,7 +14,7 @@ versions of Nominatim. Do not rely on the output in scripts or applications.
 
 
 
-The details API supports the following two request formats:
+The details API supports the following three request formats:
 
 ``` xml
 https://nominatim.openstreetmap.org/details?osmtype=[N|W|R]&osmid=<value>&class=<value>
@@ -39,6 +39,16 @@ for a place is different between Nominatim installation (servers) and
 changes when data gets reimported. Therefore it cannot be used as
 a permanent id and shouldn't be used in bug reports.
 
+``` xml
+https://nominatim.openstreetmap.org/details?postcode=<country_code>:<postcode_id>
+```
+
+Artificial postcodes do not always have an OSM object reference. For these,
+Nominatim provides a stable postcode reference that combines the country code
+and postcode. Spaces in the postcode id are replaced with underscores. For
+example, `us:94110` refers to postcode `94110` in the United States and
+`gb:EH4_7EA` refers to postcode `EH4 7EA` in Great Britain.
+
 !!! danger "Deprecation warning"
     The API can also be used with the URL
     `https://nominatim.openstreetmap.org/details.php`. This is now deprecated

docs/api/Search.md

@@ -231,7 +231,7 @@ to the address layer (see above).
 
 | Parameter | Value | Default |
 |-----------| ----- | ------- |
-| exclude_place_ids | comma-separated list of ids (OSM IDs where possible, otherwise place_ids) |
+| exclude_place_ids | comma-separated list of ids (OSM IDs where possible, otherwise place_ids or stable postcode refs) |
 
 If you do not want certain OSM objects to appear in the search
 result, give a comma separated list of the ids you want to skip.
@@ -240,8 +240,9 @@ Each entry may be one of:
 
 * a Nominatim internal `place_id` (for example `125279639`)
 * an OSM object reference in the form `<osm_type><osm_id>` where `<osm_type>` is one of `N` (node), `W` (way) or `R` (relation), for example `N107775`
+* a stable postcode reference in the form `P<country_code>:<postcode_id>`, for example `Pus:94110` or `Pgb:EH4_7EA`
 
-Usage of OSM IDs is recommended because they are server independent. `place_id`s are stil required for results without an OSM object reference (for example, postcodes and countries). When a street is excluded via its OSM ID, then interpolations and TIGER data derived from that street are excluded as well.
+Usage of OSM IDs is recommended because they are server independent. Stable postcode refs should be used for artificial postcode results without an OSM object reference. In postcode refs, spaces are replaced with underscores. `place_id`s are still required for results without either kind of stable reference (for example, countries). When a street is excluded via its OSM ID, then interpolations and TIGER data derived from that street are excluded as well.
 
 This can be used to retrieve additional search results. For example, if a
 previous query only returned a few results, then including those here would

src/nominatim_api/__init__.py

@@ -20,7 +20,9 @@
 from .status import (StatusResult as StatusResult)
 from .types import (PlaceID as PlaceID,
                     OsmID as OsmID,
+                    PostcodeRef as PostcodeRef,
                     PlaceRef as PlaceRef,
+                    parse_postcode_param as parse_postcode_param,
                     Point as Point,
                     Bbox as Bbox,
                     GeometryFormat as GeometryFormat,

src/nominatim_api/lookup.py

@@ -90,6 +90,10 @@ def enumerate_free_osm_ids(self) -> Iterable[Tuple[int, ntyp.OsmID]]:
         return ((i, p.pid) for i, p in enumerate(self.lookups)
                 if p.result is None and isinstance(p.pid, ntyp.OsmID))
 
+    def enumerate_free_postcode_refs(self) -> Iterable[Tuple[int, ntyp.PostcodeRef]]:
+        return ((i, p.pid) for i, p in enumerate(self.lookups)
+                if p.result is None and isinstance(p.pid, ntyp.PostcodeRef))
+
 
 class DetailedCollector:
     """ Result collector for detailed lookup.
@@ -139,6 +143,11 @@ def enumerate_free_osm_ids(self) -> Iterable[Tuple[int, ntyp.OsmID]]:
             return [(0, self.place)]
         return []
 
+    def enumerate_free_postcode_refs(self) -> Iterable[Tuple[int, ntyp.PostcodeRef]]:
+        if self.result is None and isinstance(self.place, ntyp.PostcodeRef):
+            return [(0, self.place)]
+        return []
+
 
 Collector = Union[LookupCollector, DetailedCollector]
 
@@ -301,6 +310,26 @@ async def find_in_postcode(conn: SearchConnection, collector: Collector) -> bool
                                              nres.create_from_postcode_row):
             return True
 
+    postcode_refs = [{'i': i, 'cc': p.country_code, 'pc': p.postcode}
+                     for i, p in collector.enumerate_free_postcode_refs()]
+
+    if postcode_refs:
+        ref_tab = sa.func.JsonArrayEach(sa.type_coerce(postcode_refs, sa.JSON))\
+                    .table_valued(sa.column('value', type_=sa.JSON))
+        t = conn.t.postcode
+        sql = sa.select(ref_tab.c.value['i'].as_integer().label('_idx'),
+                        t.c.osm_id, t.c.place_id, t.c.parent_place_id,
+                        t.c.rank_search,
+                        t.c.indexed_date, t.c.postcode, t.c.country_code,
+                        t.c.centroid)\
+                .where(t.c.osm_id.is_(None))\
+                .where(t.c.country_code == ref_tab.c.value['cc'].as_string())\
+                .where(t.c.postcode == ref_tab.c.value['pc'].as_string())
+
+        if await collector.add_rows_from_sql(conn, sql, t.c.geometry,
+                                             nres.create_from_postcode_row):
+            return True
+
     osm_ids = [{'i': i, 'oi': p.osm_id}
                for i, p in collector.enumerate_free_osm_ids() if p.osm_type == 'R']
 

src/nominatim_api/search/geocoder.py

@@ -15,7 +15,7 @@
 import sqlalchemy as sa
 
 from ..connection import SearchConnection
-from ..types import PlaceRef, SearchDetails, PlaceID, OsmID
+from ..types import PlaceRef, SearchDetails, PlaceID, OsmID, PostcodeRef
 from ..results import SearchResult, SearchResults, add_result_details
 from ..timeout import Timeout
 from ..logging import log
@@ -53,6 +53,19 @@ async def _resolve_excluded_osm_ids(self) -> None:
 
         place_ids: List[PlaceRef] = [e for e in excluded if isinstance(e, PlaceID)]
         osm_ids = [e for e in excluded if isinstance(e, OsmID)]
+        postcode_refs = [e for e in excluded if isinstance(e, PostcodeRef)]
+
+        if postcode_refs:
+            p = self.conn.t.postcode
+            conditions = [
+                sa.and_(p.c.osm_id.is_(None),
+                        p.c.country_code == ref.country_code,
+                        p.c.postcode == ref.postcode)
+                for ref in postcode_refs
+            ]
+            sql = sa.select(p.c.place_id).where(sa.or_(*conditions))
+            place_ids.extend(PlaceID(row.place_id)
+                             for row in await self.conn.execute(sql))
 
         if osm_ids:
             t = self.conn.t.placex

src/nominatim_api/types.py

@@ -14,6 +14,7 @@
 import datetime as dt
 import enum
 import math
+import re
 from struct import unpack
 from binascii import unhexlify
 
@@ -22,6 +23,10 @@
 from .errors import UsageError
 
 
+POSTCODE_REF_RE = re.compile(r'^P(?P<cc>[A-Za-z]{2}):(?P<postcode>.+)$')
+POSTCODE_PARAM_RE = re.compile(r'^(?P<cc>[A-Za-z]{2}):(?P<postcode>.+)$')
+
+
 @dataclasses.dataclass
 class PlaceID:
     """ Reference a place by Nominatim's internal ID.
@@ -81,7 +86,64 @@ def class_as_housenumber(self) -> Optional[int]:
         return None
 
 
-PlaceRef = Union[PlaceID, OsmID]
+@dataclasses.dataclass
+class PostcodeRef:
+    """ Reference an artificial postcode by country code and postcode.
+    """
+    country_code: str
+    postcode: str
+
+    def __post_init__(self) -> None:
+        if len(self.country_code) != 2 or not self.country_code.isalpha():
+            raise ValueError('Country code must be two letters.')
+        if not self.postcode:
+            raise ValueError('Postcode must not be empty.')
+        self.country_code = self.country_code.lower()
+        self.postcode = self.postcode.replace('_', ' ')
+
+    def __str__(self) -> str:
+        return f"P{self.country_code}:{self.place_id_part()}"
+
+    def place_id_part(self) -> str:
+        """Return a URL-safe identifier part for APIs.
+        """
+        return self.postcode.replace(' ', '_')
+
+
+PlaceRef = Union[PlaceID, OsmID, PostcodeRef]
+
+
+def parse_place_ref(ref: Any) -> PlaceRef:
+    """ Parse a stable place reference string.
+    """
+    if isinstance(ref, (PlaceID, OsmID, PostcodeRef)):
+        return ref
+
+    if not isinstance(ref, str):
+        raise UsageError("Parameter 'place_ref' must be a string.")
+
+    if match := POSTCODE_REF_RE.fullmatch(ref):
+        return PostcodeRef(match.group('cc'), match.group('postcode'))
+
+    raise UsageError(f"Invalid place_ref: {ref}")
+
+
+def parse_postcode_param(ref: Any) -> PostcodeRef:
+    """Parse the /details postcode parameter.
+    """
+    if isinstance(ref, PostcodeRef):
+        return ref
+
+    if not isinstance(ref, str):
+        raise UsageError("Parameter 'postcode' must be a string.")
+
+    if match := POSTCODE_REF_RE.fullmatch(ref):
+        return PostcodeRef(match.group('cc'), match.group('postcode'))
+
+    if match := POSTCODE_PARAM_RE.fullmatch(ref):
+        return PostcodeRef(match.group('cc'), match.group('postcode'))
+
+    raise UsageError(f"Invalid postcode ID: {ref}")
 
 
 class Point(NamedTuple):
@@ -427,7 +489,7 @@ def format_excluded(ids: Any) -> List[PlaceRef]:
     for i in plist:
         if not i:
             continue
-        if isinstance(i, (PlaceID, OsmID)):
+        if isinstance(i, (PlaceID, OsmID, PostcodeRef)):
             result.append(i)
         elif isinstance(i, int):
             if i > 0:
@@ -439,6 +501,8 @@ def format_excluded(ids: Any) -> List[PlaceRef]:
             elif len(i) > 1 and i[0].upper() in ('N', 'W', 'R') and i[1:].isdigit():
                 if int(i[1:]) > 0:
                     result.append(OsmID(i[0].upper(), int(i[1:])))
+            elif match := POSTCODE_REF_RE.fullmatch(i):
+                result.append(PostcodeRef(match.group('cc'), match.group('postcode')))
             else:
                 raise UsageError(f"Invalid exclude ID: {i}")
         else:

src/nominatim_api/v1/format_json.py

@@ -12,7 +12,7 @@
 from ..utils.json_writer import JsonWriter
 from ..results import AddressLines, ReverseResults, SearchResults
 from . import classtypes as cl
-from .helpers import _add_admin_level
+from .helpers import _add_admin_level, result_to_exclude_id
 from ..types import EntranceDetails
 
 
@@ -108,6 +108,10 @@ def format_base_json(results: Union[ReverseResults, SearchResults],
 
         _write_osm_id(out, result.osm_object)
 
+        if (postcode_id := result_to_exclude_id(result)) is not None \
+           and postcode_id.startswith('P'):
+            out.keyval('postcode_id', postcode_id)
+
         # lat and lon must be string values
         out.keyval('lat', f"{result.centroid.lat:0.7f}")\
            .keyval('lon', f"{result.centroid.lon:0.7f}")\
@@ -192,6 +196,10 @@ def format_base_geojson(results: Union[ReverseResults, SearchResults],
 
         _write_osm_id(out, result.osm_object)
 
+        if (postcode_id := result_to_exclude_id(result)) is not None \
+           and postcode_id.startswith('P'):
+            out.keyval('postcode_id', postcode_id)
+
         out.keyval('place_rank', result.rank_search)\
            .keyval('category', result.category[0])\
            .keyval('type', result.category[1])\
@@ -262,6 +270,10 @@ def format_base_geocodejson(results: Union[ReverseResults, SearchResults],
 
         _write_osm_id(out, result.osm_object)
 
+        if (postcode_id := result_to_exclude_id(result)) is not None \
+           and postcode_id.startswith('P'):
+            out.keyval('postcode_id', postcode_id)
+
         out.keyval('osm_key', result.category[0])\
            .keyval('osm_value', result.category[1])\
            .keyval('type', GEOCODEJSON_RANKS[max(3, min(28, result.rank_address))])\

src/nominatim_api/v1/helpers.py

@@ -13,7 +13,7 @@
 import re
 
 from ..results import SearchResults, SourceTable, BaseResult
-from ..types import SearchDetails, GeometryFormat
+from ..types import SearchDetails, GeometryFormat, PostcodeRef
 
 
 def _add_admin_level(result: BaseResult) -> Optional[Dict[str, str]]:
@@ -107,6 +107,22 @@ def extend_query_parts(queryparts: Dict[str, Any], details: Dict[str, Any],
         queryparts['featureType'] = feature_type
 
 
+def result_to_exclude_id(result: BaseResult) -> Optional[str]:
+    """Return the best stable follow-up identifier for a search result.
+    """
+    if result.osm_object:
+        return f"{result.osm_object[0]}{result.osm_object[1]}"
+
+    if result.category == ('place', 'postcode') and result.country_code \
+       and result.names and 'ref' in result.names:
+        return str(PostcodeRef(result.country_code, result.names['ref']))
+
+    if result.place_id:
+        return str(result.place_id)
+
+    return None
+
+
 def deduplicate_results(results: SearchResults, max_results: int) -> SearchResults:
     """ Remove results that look like duplicates.
 

src/nominatim_api/v1/server_glue.py

@@ -20,7 +20,7 @@
 from .. import logging as loglib
 from ..core import NominatimAPIAsync
 from .format import RawDataList
-from ..types import DataLayer, GeometryFormat, PlaceRef, PlaceID, OsmID, Point
+from ..types import DataLayer, GeometryFormat, PlaceRef, PlaceID, OsmID, Point, parse_postcode_param
 from ..status import StatusResult
 from ..results import DetailedResult, ReverseResults, SearchResult, SearchResults
 from ..localization import Locales
@@ -155,11 +155,15 @@ async def details_endpoint(api: NominatimAPIAsync, params: ASGIAdaptor) -> Any:
     place: PlaceRef
     if place_id:
         place = PlaceID(place_id)
-    else:
-        osmtype = params.get('osmtype')
-        if osmtype is None:
-            params.raise_error("Missing ID parameter 'place_id' or 'osmtype'.")
+    elif (postcode := params.get('postcode')) is not None:
+        try:
+            place = parse_postcode_param(postcode)
+        except UsageError as err:
+            params.raise_error(str(err))
+    elif (osmtype := params.get('osmtype')) is not None:
         place = OsmID(osmtype, params.get_int('osmid'), params.get('class'))
+    else:
+        params.raise_error("Missing ID parameter 'place_id', 'postcode' or 'osmtype'.")
 
     debug = setup_debugging(params)
 
@@ -179,7 +183,7 @@ async def details_endpoint(api: NominatimAPIAsync, params: ASGIAdaptor) -> Any:
         return build_response(params, loglib.get_and_disable())
 
     if result is None:
-        params.raise_error('No place with that OSM ID found.', status=404)
+        params.raise_error('No place with that ID found.', status=404)
 
     locales = Locales.from_accept_languages(get_accepted_languages(params),
                                             params.config().OUTPUT_NAMES)
@@ -374,11 +378,9 @@ async def search_endpoint(api: NominatimAPIAsync, params: ASGIAdaptor) -> Any:
                                    params.get('featureType', ''),
                                    params.get_bool('namedetails', False),
                                    params.get_bool('extratags', False),
-                                   (f"{r.osm_object[0]}{r.osm_object[1]}"
-                                    if r.osm_object
-                                    else str(r.place_id)
-                                    for r in results
-                                    if r.osm_object or r.place_id))
+                                   (rid for rid in (helpers.result_to_exclude_id(r)
+                                                    for r in results)
+                                    if rid is not None))
         queryparts['format'] = fmt
 
         moreurl = params.base_uri() + '/search?' + urlencode(queryparts)

test/python/api/test_api_details.py

@@ -534,6 +534,29 @@ def test_lookup_in_postcode(apiobj, frontend):
     assert result.geometry == {'type': 'ST_Polygon'}
 
 
+def test_lookup_in_postcode_by_stable_ref(apiobj, frontend):
+    import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
+    apiobj.add_postcode(place_id=554,
+                        parent_place_id=152,
+                        postcode='94110',
+                        country_code='us',
+                        rank_search=20,
+                        indexed_date=import_date,
+                        centroid='POINT(-122.4170874 37.7536873)',
+                        geometry='POLYGON((-122.47 37.70, -122.47 37.80, '
+                                 '-122.36 37.75, -122.47 37.70))')
+
+    api = frontend(apiobj, options={'details'})
+    result = api.details(napi.PostcodeRef('us', '94110'))
+
+    assert result is not None
+    assert result.source_table.name == 'POSTCODE'
+    assert result.place_id == 554
+    assert result.osm_object is None
+    assert result.names == {'ref': '94110'}
+    assert result.country_code == 'us'
+
+
 @pytest.mark.parametrize('lookup', [napi.PlaceID(9000),
                                     napi.OsmID('R', 12)])
 def test_lookup_postcode_with_address_details(apiobj, frontend, lookup):