]> git.openstreetmap.org Git - nominatim.git/blobdiff - nominatim/api/v1/format.py
Merge remote-tracking branch 'upstream/master'
[nominatim.git] / nominatim / api / v1 / format.py
index 116e2ae634e671a7ac191431e705dfd19d46da38..53156fdd41132ada43139546d36ec0e470590858 100644 (file)
@@ -7,22 +7,32 @@
 """
 Output formatters for API version v1.
 """
 """
 Output formatters for API version v1.
 """
+from typing import List, Dict, Mapping, Any
+import collections
+import datetime as dt
+
+import nominatim.api as napi
 from nominatim.api.result_formatting import FormatDispatcher
 from nominatim.api.result_formatting import FormatDispatcher
-from nominatim.api import StatusResult
+from nominatim.api.v1.classtypes import ICONS
+from nominatim.api.v1 import format_json, format_xml
 from nominatim.utils.json_writer import JsonWriter
 
 from nominatim.utils.json_writer import JsonWriter
 
+class RawDataList(List[Dict[str, Any]]):
+    """ Data type for formatting raw data lists 'as is' in json.
+    """
+
 dispatch = FormatDispatcher()
 
 dispatch = FormatDispatcher()
 
-@dispatch.format_func(StatusResult, 'text')
-def _format_status_text(result: StatusResult) -> str:
+@dispatch.format_func(napi.StatusResult, 'text')
+def _format_status_text(result: napi.StatusResult, _: Mapping[str, Any]) -> str:
     if result.status:
         return f"ERROR: {result.message}"
 
     return 'OK'
 
 
     if result.status:
         return f"ERROR: {result.message}"
 
     return 'OK'
 
 
-@dispatch.format_func(StatusResult, 'json')
-def _format_status_json(result: StatusResult) -> str:
+@dispatch.format_func(napi.StatusResult, 'json')
+def _format_status_json(result: napi.StatusResult, _: Mapping[str, Any]) -> str:
     out = JsonWriter()
 
     out.start_object()\
     out = JsonWriter()
 
     out.start_object()\
@@ -35,3 +45,212 @@ def _format_status_json(result: StatusResult) -> str:
        .end_object()
 
     return out()
        .end_object()
 
     return out()
+
+
+def _add_address_row(writer: JsonWriter, row: napi.AddressLine,
+                     locales: napi.Locales) -> None:
+    writer.start_object()\
+            .keyval('localname', locales.display_name(row.names))\
+            .keyval_not_none('place_id', row.place_id)
+
+    if row.osm_object is not None:
+        writer.keyval('osm_id', row.osm_object[1])\
+              .keyval('osm_type', row.osm_object[0])
+
+    if row.extratags:
+        writer.keyval_not_none('place_type', row.extratags.get('place_type'))
+
+    writer.keyval('class', row.category[0])\
+          .keyval('type', row.category[1])\
+          .keyval_not_none('admin_level', row.admin_level)\
+          .keyval('rank_address', row.rank_address)\
+          .keyval('distance', row.distance)\
+          .keyval('isaddress', row.isaddress)\
+        .end_object()
+
+
+def _add_address_rows(writer: JsonWriter, section: str, rows: napi.AddressLines,
+                      locales: napi.Locales) -> None:
+    writer.key(section).start_array()
+    for row in rows:
+        _add_address_row(writer, row, locales)
+        writer.next()
+    writer.end_array().next()
+
+
+def _add_parent_rows_grouped(writer: JsonWriter, rows: napi.AddressLines,
+                             locales: napi.Locales) -> None:
+    # group by category type
+    data = collections.defaultdict(list)
+    for row in rows:
+        sub = JsonWriter()
+        _add_address_row(sub, row, locales)
+        data[row.category[1]].append(sub())
+
+    writer.key('hierarchy').start_object()
+    for group, grouped in data.items():
+        writer.key(group).start_array()
+        grouped.sort() # sorts alphabetically by local name
+        for line in grouped:
+            writer.raw(line).next()
+        writer.end_array().next()
+
+    writer.end_object().next()
+
+
+@dispatch.format_func(napi.DetailedResult, 'json')
+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()
+
+    out = JsonWriter()
+    out.start_object()\
+         .keyval_not_none('place_id', result.place_id)\
+         .keyval_not_none('parent_place_id', result.parent_place_id)
+
+    if result.osm_object is not None:
+        out.keyval('osm_type', result.osm_object[0])\
+           .keyval('osm_id', result.osm_object[1])
+
+    out.keyval('category', result.category[0])\
+         .keyval('type', result.category[1])\
+         .keyval('admin_level', result.admin_level)\
+         .keyval('localname', result.locale_name or '')\
+         .keyval('names', result.names or {})\
+         .keyval('addresstags', result.address or {})\
+         .keyval_not_none('housenumber', result.housenumber)\
+         .keyval_not_none('calculated_postcode', result.postcode)\
+         .keyval_not_none('country_code', result.country_code)\
+         .keyval_not_none('indexed_date', result.indexed_date, lambda v: v.isoformat())\
+         .keyval_not_none('importance', result.importance)\
+         .keyval('calculated_importance', result.calculated_importance())\
+         .keyval('extratags', result.extratags or {})\
+         .keyval_not_none('calculated_wikipedia', result.wikipedia)\
+         .keyval('rank_address', result.rank_address)\
+         .keyval('rank_search', result.rank_search)\
+         .keyval('isarea', 'Polygon' in (geom or result.geometry.get('type') or ''))\
+         .key('centroid').raw(centroid).next()\
+         .key('geometry').raw(geom or centroid).next()
+
+    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 result.address_rows is not None:
+        _add_address_rows(out, 'address', result.address_rows, locales)
+
+    if result.linked_rows:
+        _add_address_rows(out, 'linked_places', result.linked_rows, locales)
+
+    if result.name_keywords is not None or result.address_keywords is not None:
+        out.key('keywords').start_object()
+
+        for sec, klist in (('name', result.name_keywords), ('address', result.address_keywords)):
+            out.key(sec).start_array()
+            for word in (klist or []):
+                out.start_object()\
+                     .keyval('id', word.word_id)\
+                     .keyval('token', word.word_token)\
+                   .end_object().next()
+            out.end_array().next()
+
+        out.end_object().next()
+
+    if result.parented_rows is not None:
+        if options.get('group_hierarchy', False):
+            _add_parent_rows_grouped(out, result.parented_rows, locales)
+        else:
+            _add_address_rows(out, 'hierarchy', result.parented_rows, locales)
+
+    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': options.get('query', '')})
+
+
+@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')
+
+
+@dispatch.format_func(napi.SearchResults, 'xml')
+def _format_search_xml(results: napi.SearchResults, options: Mapping[str, Any]) -> str:
+    extra = {'querystring': options.get('query', '')}
+    for attr in ('more_url', 'exclude_place_ids', 'viewbox'):
+        if options.get(attr):
+            extra[attr] = options[attr]
+    return format_xml.format_base_xml(results, options, False, 'searchresults',
+                                      extra)
+
+
+
+@dispatch.format_func(napi.SearchResults, 'geojson')
+def _format_search_geojson(results: napi.SearchResults,
+                            options: Mapping[str, Any]) -> str:
+    return format_json.format_base_geojson(results, options, False)
+
+
+@dispatch.format_func(napi.SearchResults, 'geocodejson')
+def _format_search_geocodejson(results: napi.SearchResults,
+                                options: Mapping[str, Any]) -> str:
+    return format_json.format_base_geocodejson(results, options, False)
+
+
+@dispatch.format_func(napi.SearchResults, 'json')
+def _format_search_json(results: napi.SearchResults,
+                         options: Mapping[str, Any]) -> str:
+    return format_json.format_base_json(results, options, False,
+                                        class_label='class')
+
+
+@dispatch.format_func(napi.SearchResults, 'jsonv2')
+def _format_search_jsonv2(results: napi.SearchResults,
+                           options: Mapping[str, Any]) -> str:
+    return format_json.format_base_json(results, options, False,
+                                        class_label='category')
+
+@dispatch.format_func(RawDataList, 'json')
+def _format_raw_data_json(results: RawDataList,  _: Mapping[str, Any]) -> str:
+    out = JsonWriter()
+    out.start_array()
+    for res in results:
+        out.start_object()
+        for k, v in res.items():
+            if isinstance(v, dt.datetime):
+                out.keyval(k, v.isoformat(sep= ' ', timespec='seconds'))
+            else:
+                out.keyval(k, v)
+        out.end_object().next()
+
+    out.end_array()
+
+    return out()