add output formatters for ReverseResults

These formatters are written in a way that they can be reused for
search results later.

nominatim/api/__init__.py

@@ -31,5 +31,6 @@ from .results import (SourceTable as SourceTable,
                       WordInfo as WordInfo,
                       WordInfos as WordInfos,
                       DetailedResult as DetailedResult,
-                      ReverseResult as ReverseResult)
+                      ReverseResult as ReverseResult,
+                      ReverseResults as ReverseResults)
 from .localization import (Locales as Locales)

nominatim/api/results.py

@@ -11,7 +11,7 @@ Data classes are part of the public API while the functions are for
 internal use only. That's why they are implemented as free-standing functions
 instead of member functions.
 """
-from typing import Optional, Tuple, Dict, Sequence, TypeVar, Type
+from typing import Optional, Tuple, Dict, Sequence, TypeVar, Type, List
 import enum
 import dataclasses
 import datetime as dt
@@ -22,6 +22,7 @@ from nominatim.typing import SaSelect, SaRow
 from nominatim.api.types import Point, Bbox, LookupDetails
 from nominatim.api.connection import SearchConnection
 from nominatim.api.logging import log
+from nominatim.api.localization import Locales
 
 # This file defines complex result data classes.
 # pylint: disable=too-many-instance-attributes
@@ -52,8 +53,30 @@ class AddressLine:
     rank_address: int
     distance: float
 
+    local_name: Optional[str] = None
+
+
+class AddressLines(List[AddressLine]):
+    """ Sequence of address lines order in descending order by their rank.
+    """
+
+    def localize(self, locales: Locales) -> List[str]:
+        """ Set the local name of address parts according to the chosen
+            locale. Return the list of local names without duplications.
+
+            Only address parts that are marked as isaddress are localized
+            and returned.
+        """
+        label_parts: List[str] = []
+
+        for line in self:
+            if line.isaddress and line.names:
+                line.local_name = locales.display_name(line.names)
+                if not label_parts or label_parts[-1] != line.local_name:
+                    label_parts.append(line.local_name)
+
+        return label_parts
 
-AddressLines = Sequence[AddressLine]
 
 
 @dataclasses.dataclass
@@ -144,6 +167,12 @@ class ReverseResult(BaseResult):
     bbox: Optional[Bbox] = None
 
 
+class ReverseResults(List[ReverseResult]):
+    """ Sequence of reverse lookup results ordered by distance.
+        May be empty when no result was found.
+    """
+
+
 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_')}
@@ -333,7 +362,7 @@ async def complete_address_details(conn: SearchConnection, result: BaseResult) -
     sql = sa.select(sfn).order_by(sa.column('rank_address').desc(),
                                   sa.column('isaddress').desc())
 
-    result.address_rows = []
+    result.address_rows = AddressLines()
     for row in await conn.execute(sql):
         result.address_rows.append(_result_row_to_address_row(row))
 
@@ -357,7 +386,7 @@ def _placex_select_address_row(conn: SearchConnection,
 async def complete_linked_places(conn: SearchConnection, result: BaseResult) -> None:
     """ Retrieve information about places that link to the result.
     """
-    result.linked_rows = []
+    result.linked_rows = AddressLines()
     if result.source_table != SourceTable.PLACEX:
         return
 
@@ -392,7 +421,7 @@ async def complete_parented_places(conn: SearchConnection, result: BaseResult) -
     """ Retrieve information about places that the result provides the
         address for.
     """
-    result.parented_rows = []
+    result.parented_rows = AddressLines()
     if result.source_table != SourceTable.PLACEX:
         return
 

nominatim/api/v1/classtypes.py

@@ -10,6 +10,52 @@ Hard-coded information about tag catagories.
 These tables have been copied verbatim from the old PHP code. For future
 version a more flexible formatting is required.
 """
+from typing import Tuple, Optional, Mapping
+
+def get_label_tag(category: Tuple[str, str], extratags: Optional[Mapping[str, str]],
+                  rank: int, country: Optional[str]) -> str:
+    """ Create a label tag for the given place that can be used as an XML name.
+    """
+    if rank < 26 and extratags and 'place'in extratags:
+        label = extratags['place']
+    elif category == ('boundary', 'administrative'):
+        label = ADMIN_LABELS.get((country or '', int(rank/2)))\
+                or ADMIN_LABELS.get(('', int(rank/2)))\
+                or 'Administrative'
+    elif category[1] == 'postal_code':
+        label = 'postcode'
+    elif rank < 26:
+        label = category[1] if category[1] != 'yes' else category[0]
+    elif rank < 28:
+        label = 'road'
+    elif category[0] == 'place'\
+         and category[1] in ('house_number', 'house_name', 'country_code'):
+        label = category[1]
+    else:
+        label = category[0]
+
+    return label.lower().replace(' ', '_')
+
+
+ADMIN_LABELS = {
+  ('', 1): 'Continent',
+  ('', 2): 'Country',
+  ('', 3): 'Region',
+  ('', 4): 'State',
+  ('', 5): 'State District',
+  ('', 6): 'County',
+  ('', 7): 'Municipality',
+  ('', 8): 'City',
+  ('', 9): 'City District',
+  ('', 10): 'Suburb',
+  ('', 11): 'Neighbourhood',
+  ('', 12): 'City Block',
+  ('no', 3): 'State',
+  ('no', 4): 'County',
+  ('se', 3): 'State',
+  ('se', 4): 'County'
+}
+
 
 ICONS = {
     ('boundary', 'administrative'): 'poi_boundary_administrative',

nominatim/api/v1/constants.py

@@ -0,0 +1,43 @@
+# 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.
+"""
+Constants shared by all formats.
+"""
+
+import nominatim.api as napi
+
+# pylint: disable=line-too-long
+OSM_ATTRIBUTION = 'Data © OpenStreetMap contributors, ODbL 1.0. http://www.openstreetmap.org/copyright'
+
+OSM_TYPE_NAME = {
+    'N': 'node',
+    'W': 'way',
+    'R': 'relation'
+}
+
+NODE_EXTENT = [25, 25, 25, 25,
+               7,
+               2.6, 2.6, 2.0, 1.0, 1.0,
+               0.7, 0.7, 0.7,
+               0.16, 0.16, 0.16, 0.16,
+               0.04, 0.04,
+               0.02, 0.02,
+               0.01, 0.01, 0.01, 0.01, 0.01,
+               0.015, 0.015, 0.015, 0.015,
+               0.005]
+
+
+def bbox_from_result(result: napi.ReverseResult) -> napi.Bbox:
+    """ Compute a bounding box for the result. For ways and relations
+        a given boundingbox is used. For all other object, a box is computed
+        around the centroid according to dimensions dereived from the
+        search rank.
+    """
+    if (result.osm_object and result.osm_object[0] == 'N') or result.bbox is None:
+        return napi.Bbox.from_point(result.centroid, NODE_EXTENT[result.rank_search])
+
+    return result.bbox

nominatim/api/v1/format.py

@@ -13,6 +13,7 @@ import collections
 import nominatim.api as napi
 from nominatim.api.result_formatting import FormatDispatcher
 from nominatim.api.v1.classtypes import ICONS
+from nominatim.api.v1 import format_json, format_xml
 from nominatim.utils.json_writer import JsonWriter
 
 dispatch = FormatDispatcher()
@@ -93,7 +94,7 @@ def _add_parent_rows_grouped(writer: JsonWriter, rows: napi.AddressLines,
 
 
 @dispatch.format_func(napi.DetailedResult, 'json')
-def _format_search_json(result: napi.DetailedResult, options: Mapping[str, Any]) -> str:
+def _format_details_json(result: napi.DetailedResult, options: Mapping[str, Any]) -> str:
     locales = options.get('locales', napi.Locales())
     geom = result.geometry.get('geojson')
     centroid = result.centroid.to_geojson()
@@ -161,3 +162,36 @@ def _format_search_json(result: napi.DetailedResult, options: Mapping[str, Any])
     out.end_object()
 
     return out()
+
+
+@dispatch.format_func(napi.ReverseResults, 'xml')
+def _format_reverse_xml(results: napi.ReverseResults, options: Mapping[str, Any]) -> str:
+    return format_xml.format_base_xml(results,
+                                      options, True, 'reversegeocode',
+                                      {'querystring': 'TODO'})
+
+
+@dispatch.format_func(napi.ReverseResults, 'geojson')
+def _format_reverse_geojson(results: napi.ReverseResults,
+                            options: Mapping[str, Any]) -> str:
+    return format_json.format_base_geojson(results, options, True)
+
+
+@dispatch.format_func(napi.ReverseResults, 'geocodejson')
+def _format_reverse_geocodejson(results: napi.ReverseResults,
+                                options: Mapping[str, Any]) -> str:
+    return format_json.format_base_geocodejson(results, options, True)
+
+
+@dispatch.format_func(napi.ReverseResults, 'json')
+def _format_reverse_json(results: napi.ReverseResults,
+                         options: Mapping[str, Any]) -> str:
+    return format_json.format_base_json(results, options, True,
+                                        class_label='class')
+
+
+@dispatch.format_func(napi.ReverseResults, 'jsonv2')
+def _format_reverse_jsonv2(results: napi.ReverseResults,
+                           options: Mapping[str, Any]) -> str:
+    return format_json.format_base_json(results, options, True,
+                                        class_label='category')

nominatim/api/v1/format_json.py

@@ -0,0 +1,283 @@
+# 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.
+"""
+Helper functions for output of results in json formats.
+"""
+from typing import Mapping, Any, Optional, Tuple
+
+import nominatim.api as napi
+from nominatim.api.v1.constants import OSM_ATTRIBUTION, OSM_TYPE_NAME, bbox_from_result
+from nominatim.api.v1.classtypes import ICONS, get_label_tag
+from nominatim.utils.json_writer import JsonWriter
+
+def _write_osm_id(out: JsonWriter, osm_object: Optional[Tuple[str, int]]) -> None:
+    if osm_object is not None:
+        out.keyval_not_none('osm_type', OSM_TYPE_NAME.get(osm_object[0], None))\
+           .keyval('osm_id', osm_object[1])
+
+
+def _write_typed_address(out: JsonWriter, address: Optional[napi.AddressLines],
+                               country_code: Optional[str]) -> None:
+    parts = {}
+    for line in (address or []):
+        if line.isaddress and line.local_name:
+            label = get_label_tag(line.category, line.extratags,
+                                  line.rank_address, country_code)
+            if label not in parts:
+                parts[label] = line.local_name
+
+    for k, v in parts.items():
+        out.keyval(k, v)
+
+    if country_code:
+        out.keyval('country_code', country_code)
+
+
+def _write_geocodejson_address(out: JsonWriter,
+                               address: Optional[napi.AddressLines],
+                               obj_place_id: Optional[int],
+                               country_code: Optional[str]) -> None:
+    extra = {}
+    for line in (address or []):
+        if line.isaddress and line.local_name:
+            if line.category[1] in ('postcode', 'postal_code'):
+                out.keyval('postcode', line.local_name)
+            elif line.category[1] == 'house_number':
+                out.keyval('housenumber', line.local_name)
+            elif (obj_place_id is None or obj_place_id != line.place_id) \
+                 and line.rank_address >= 4 and line.rank_address < 28:
+                extra[GEOCODEJSON_RANKS[line.rank_address]] = line.local_name
+
+    for k, v in extra.items():
+        out.keyval(k, v)
+
+    if country_code:
+        out.keyval('country_code', country_code)
+
+
+def format_base_json(results: napi.ReverseResults, #pylint: disable=too-many-branches
+                     options: Mapping[str, Any], simple: bool,
+                     class_label: str) -> str:
+    """ Return the result list as a simple json string in custom Nominatim format.
+    """
+    locales = options.get('locales', napi.Locales())
+
+    out = JsonWriter()
+
+    if simple:
+        if not results:
+            return '{"error":"Unable to geocode"}'
+    else:
+        out.start_array()
+
+    for result in results:
+        label_parts = result.address_rows.localize(locales) if result.address_rows else []
+
+        out.start_object()\
+             .keyval_not_none('place_id', result.place_id)\
+             .keyval('licence', OSM_ATTRIBUTION)\
+
+        _write_osm_id(out, result.osm_object)
+
+        out.keyval('lat', result.centroid.lat)\
+             .keyval('lon', result.centroid.lon)\
+             .keyval(class_label, result.category[0])\
+             .keyval('type', result.category[1])\
+             .keyval('place_rank', result.rank_search)\
+             .keyval('importance', result.calculated_importance())\
+             .keyval('addresstype', get_label_tag(result.category, result.extratags,
+                                                  result.rank_address,
+                                                  result.country_code))\
+             .keyval('name', locales.display_name(result.names))\
+             .keyval('display_name', ', '.join(label_parts))
+
+
+        if options.get('icon_base_url', None):
+            icon = ICONS.get(result.category)
+            if icon:
+                out.keyval('icon', f"{options['icon_base_url']}/{icon}.p.20.png")
+
+        if options.get('addressdetails', False):
+            out.key('address').start_object()
+            _write_typed_address(out, result.address_rows, result.country_code)
+            out.end_object().next()
+
+        if options.get('extratags', False):
+            out.keyval('extratags', result.extratags)
+
+        if options.get('namedetails', False):
+            out.keyval('namedetails', result.names)
+
+        bbox = bbox_from_result(result)
+        out.key('boundingbox').start_array()\
+             .value(bbox.minlat).next()\
+             .value(bbox.maxlat).next()\
+             .value(bbox.minlon).next()\
+             .value(bbox.maxlon).next()\
+           .end_array().next()
+
+        if result.geometry:
+            for key in ('text', 'kml'):
+                out.keyval_not_none('geo' + key, result.geometry.get(key))
+            if 'geojson' in result.geometry:
+                out.key('geojson').raw(result.geometry['geojson']).next()
+            out.keyval_not_none('svg', result.geometry.get('svg'))
+
+        out.end_object()
+
+        if simple:
+            return out()
+
+        out.next()
+
+    out.end_array()
+
+    return out()
+
+
+def format_base_geojson(results: napi.ReverseResults,
+                        options: Mapping[str, Any],
+                        simple: bool) -> str:
+    """ Return the result list as a geojson string.
+    """
+    if not results and simple:
+        return '{"error":"Unable to geocode"}'
+
+    locales = options.get('locales', napi.Locales())
+
+    out = JsonWriter()
+
+    out.start_object()\
+         .keyval('type', 'FeatureCollection')\
+         .keyval('licence', OSM_ATTRIBUTION)\
+         .key('features').start_array()
+
+    for result in results:
+        if result.address_rows:
+            label_parts = result.address_rows.localize(locales)
+        else:
+            label_parts = []
+
+        out.start_object()\
+             .keyval('type', 'Feature')\
+             .key('properties').start_object()
+
+        out.keyval_not_none('place_id', result.place_id)
+
+        _write_osm_id(out, result.osm_object)
+
+        out.keyval('place_rank', result.rank_search)\
+           .keyval('category', result.category[0])\
+           .keyval('type', result.category[1])\
+           .keyval('importance', result.calculated_importance())\
+           .keyval('addresstype', get_label_tag(result.category, result.extratags,
+                                                result.rank_address,
+                                                result.country_code))\
+           .keyval('name', locales.display_name(result.names))\
+           .keyval('display_name', ', '.join(label_parts))
+
+        if options.get('addressdetails', False):
+            out.key('address').start_object()
+            _write_typed_address(out, result.address_rows, result.country_code)
+            out.end_object().next()
+
+        if options.get('extratags', False):
+            out.keyval('extratags', result.extratags)
+
+        if options.get('namedetails', False):
+            out.keyval('namedetails', result.names)
+
+        out.end_object().next() # properties
+
+        bbox = bbox_from_result(result)
+        out.keyval('bbox', bbox.coords)
+
+        out.key('geometry').raw(result.geometry.get('geojson')
+                                or result.centroid.to_geojson()).next()
+
+        out.end_object().next()
+
+    out.end_array().next().end_object()
+
+    return out()
+
+
+def format_base_geocodejson(results: napi.ReverseResults,
+                            options: Mapping[str, Any], simple: bool) -> str:
+    """ Return the result list as a geocodejson string.
+    """
+    if not results and simple:
+        return '{"error":"Unable to geocode"}'
+
+    locales = options.get('locales', napi.Locales())
+
+    out = JsonWriter()
+
+    out.start_object()\
+         .keyval('type', 'FeatureCollection')\
+         .key('geocoding').start_object()\
+           .keyval('version', '0.1.0')\
+           .keyval('attribution', OSM_ATTRIBUTION)\
+           .keyval('licence', 'ODbL')\
+           .keyval_not_none('query', options.get('query'))\
+           .end_object().next()\
+         .key('features').start_array()
+
+    for result in results:
+        if result.address_rows:
+            label_parts = result.address_rows.localize(locales)
+        else:
+            label_parts = []
+
+        out.start_object()\
+             .keyval('type', 'Feature')\
+             .key('properties').start_object()\
+               .key('geocoding').start_object()
+
+        out.keyval_not_none('place_id', result.place_id)
+
+        _write_osm_id(out, result.osm_object)
+
+        out.keyval('osm_key', result.category[0])\
+           .keyval('osm_value', result.category[1])\
+           .keyval('type', GEOCODEJSON_RANKS[max(3, min(28, result.rank_address))])\
+           .keyval_not_none('accuracy', result.distance)\
+           .keyval('label', ', '.join(label_parts))\
+           .keyval_not_none('name', locales.display_name(result.names))\
+
+        if options.get('addressdetails', False):
+            _write_geocodejson_address(out, result.address_rows, result.place_id,
+                                       result.country_code)
+
+            out.key('admin').start_object()
+            if result.address_rows:
+                for line in result.address_rows:
+                    if line.isaddress and (line.admin_level or 15) < 15 and line.local_name:
+                        out.keyval(f"level{line.admin_level}", line.local_name)
+            out.end_object().next()
+
+        out.end_object().next().end_object().next()
+
+        out.key('geometry').raw(result.geometry.get('geojson')
+                                or result.centroid.to_geojson()).next()
+
+        out.end_object().next()
+
+    out.end_array().next().end_object()
+
+    return out()
+
+
+GEOCODEJSON_RANKS = {
+    3: 'locality',
+    4: 'country',
+    5: 'state', 6: 'state', 7: 'state', 8: 'state', 9: 'state',
+    10: 'county', 11: 'county', 12: 'county',
+    13: 'city', 14: 'city', 15: 'city', 16: 'city',
+    17: 'district', 18: 'district', 19: 'district', 20: 'district', 21: 'district',
+    22: 'locality', 23: 'locality', 24: 'locality',
+    25: 'street', 26: 'street', 27: 'street', 28: 'house'}

nominatim/api/v1/format_xml.py

@@ -0,0 +1,126 @@
+# 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.
+"""
+Helper functions for output of results in XML format.
+"""
+from typing import Mapping, Any, Optional
+import datetime as dt
+import xml.etree.ElementTree as ET
+
+import nominatim.api as napi
+from nominatim.api.v1.constants import OSM_ATTRIBUTION, OSM_TYPE_NAME, bbox_from_result
+from nominatim.api.v1.classtypes import ICONS, get_label_tag
+
+def _write_xml_address(root: ET.Element, address: napi.AddressLines,
+                       country_code: Optional[str]) -> None:
+    parts = {}
+    for line in address:
+        if line.isaddress and line.local_name:
+            label = get_label_tag(line.category, line.extratags,
+                                  line.rank_address, country_code)
+            if label not in parts:
+                parts[label] = line.local_name
+
+    for k,v in parts.items():
+        ET.SubElement(root, k).text = v
+
+    if country_code:
+        ET.SubElement(root, 'country_code').text = country_code
+
+
+def _create_base_entry(result: napi.ReverseResult, #pylint: disable=too-many-branches
+                       root: ET.Element, simple: bool,
+                       locales: napi.Locales) -> ET.Element:
+    if result.address_rows:
+        label_parts = result.address_rows.localize(locales)
+    else:
+        label_parts = []
+
+    place = ET.SubElement(root, 'result' if simple else 'place')
+    if result.place_id is not None:
+        place.set('place_id', str(result.place_id))
+    if result.osm_object:
+        osm_type = OSM_TYPE_NAME.get(result.osm_object[0], None)
+        if osm_type is not None:
+            place.set('osm_type', osm_type)
+        place.set('osm_id', str(result.osm_object[1]))
+    if result.names and 'ref' in result.names:
+        place.set('place_id', result.names['ref'])
+    place.set('lat', str(result.centroid.lat))
+    place.set('lon', str(result.centroid.lon))
+
+    bbox = bbox_from_result(result)
+    place.set('boundingbox', ','.join(map(str, [bbox.minlat, bbox.maxlat,
+                                                bbox.minlon, bbox.maxlon])))
+
+    place.set('place_rank', str(result.rank_search))
+    place.set('address_rank', str(result.rank_address))
+
+    if result.geometry:
+        for key in ('text', 'svg'):
+            if key in result.geometry:
+                place.set('geo' + key, result.geometry[key])
+        if 'kml' in result.geometry:
+            ET.SubElement(root if simple else place, 'geokml')\
+              .append(ET.fromstring(result.geometry['kml']))
+        if 'geojson' in result.geometry:
+            place.set('geojson', result.geometry['geojson'])
+
+    if simple:
+        place.text = ', '.join(label_parts)
+    else:
+        place.set('display_name', ', '.join(label_parts))
+        place.set('class', result.category[0])
+        place.set('type', result.category[1])
+        place.set('importance', str(result.calculated_importance()))
+
+    return place
+
+
+def format_base_xml(results: napi.ReverseResults,
+                    options: Mapping[str, Any],
+                    simple: bool, xml_root_tag: str,
+                    xml_extra_info: Mapping[str, str]) -> str:
+    """ Format the result into an XML response. With 'simple' exactly one
+        result will be output, otherwise a list.
+    """
+    locales = options.get('locales', napi.Locales())
+
+    root = ET.Element(xml_root_tag)
+    root.set('timestamp', dt.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S +00:00'))
+    root.set('attribution', OSM_ATTRIBUTION)
+    for k, v in xml_extra_info.items():
+        root.set(k, v)
+
+    if simple and not results:
+        ET.SubElement(root, 'error').text = 'Unable to geocode'
+
+    for result in results:
+        place = _create_base_entry(result, root, simple, locales)
+
+        if not simple and options.get('icon_base_url', None):
+            icon = ICONS.get(result.category)
+            if icon:
+                place.set('icon', icon)
+
+        if options.get('addressdetails', False) and result.address_rows:
+            _write_xml_address(ET.SubElement(root, 'addressparts') if simple else place,
+                               result.address_rows, result.country_code)
+
+        if options.get('extratags', False):
+            eroot = ET.SubElement(root if simple else place, 'extratags')
+            if result.extratags:
+                for k, v in result.extratags.items():
+                    ET.SubElement(eroot, 'tag', attrib={'key': k, 'value': v})
+
+        if options.get('namedetails', False):
+            eroot = ET.SubElement(root if simple else place, 'namedetails')
+            if result.names:
+                for k,v in result.names.items():
+                    ET.SubElement(eroot, 'name', attrib={'desc': k}).text = v
+
+    return '<?xml version="1.0" encoding="UTF-8" ?>\n' + ET.tostring(root, encoding='unicode')