src/nominatim_api/v1/format_json.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2024 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Helper functions for output of results in json formats.
   9 """
  10 from typing import Mapping, Any, Optional, Tuple, Union, List
  11
  12 from ..utils.json_writer import JsonWriter
  13 from ..results import AddressLines, ReverseResults, SearchResults
  14 from . import classtypes as cl
  15 from ..types import EntranceDetails
  16
  17
  18 def _write_osm_id(out: JsonWriter, osm_object: Optional[Tuple[str, int]]) -> None:
  19     if osm_object is not None:
  20         out.keyval_not_none('osm_type', cl.OSM_TYPE_NAME.get(osm_object[0], None))\
  21            .keyval('osm_id', osm_object[1])
  22
  23
  24 def _write_typed_address(out: JsonWriter, address: Optional[AddressLines],
  25                          country_code: Optional[str]) -> None:
  26     parts = {}
  27     for line in (address or []):
  28         if line.isaddress:
  29             if line.local_name:
  30                 label = cl.get_label_tag(line.category, line.extratags,
  31                                          line.rank_address, country_code)
  32                 if label not in parts:
  33                     parts[label] = line.local_name
  34             if line.names and 'ISO3166-2' in line.names and line.admin_level:
  35                 parts[f"ISO3166-2-lvl{line.admin_level}"] = line.names['ISO3166-2']
  36
  37     for k, v in parts.items():
  38         out.keyval(k, v)
  39
  40     if country_code:
  41         out.keyval('country_code', country_code)
  42
  43
  44 def _write_geocodejson_address(out: JsonWriter,
  45                                address: Optional[AddressLines],
  46                                obj_place_id: Optional[int],
  47                                country_code: Optional[str]) -> None:
  48     extra = {}
  49     for line in (address or []):
  50         if line.isaddress and line.local_name:
  51             if line.category[1] in ('postcode', 'postal_code'):
  52                 out.keyval('postcode', line.local_name)
  53             elif line.category[1] == 'house_number':
  54                 out.keyval('housenumber', line.local_name)
  55             elif ((obj_place_id is None or obj_place_id != line.place_id)
  56                   and line.rank_address >= 4 and line.rank_address < 28):
  57                 rank_name = GEOCODEJSON_RANKS[line.rank_address]
  58                 if rank_name not in extra:
  59                     extra[rank_name] = line.local_name
  60
  61     for k, v in extra.items():
  62         out.keyval(k, v)
  63
  64     if country_code:
  65         out.keyval('country_code', country_code)
  66
  67
  68 def write_entrances(out: JsonWriter, entrances: Optional[List[EntranceDetails]]) -> None:
  69     if entrances is None:
  70         out.keyval('entrances', None)
  71         return
  72
  73     out.key('entrances')\
  74        .start_array()
  75
  76     for entrance in entrances:
  77         out.start_object()\
  78            .keyval('osm_id', entrance.osm_id)\
  79            .keyval('type', entrance.type)\
  80            .keyval('lat', f"{entrance.location.lat:0.7f}")\
  81            .keyval('lon', f"{entrance.location.lon:0.7f}")
  82
  83         if entrance.extratags:
  84             out.keyval('extratags', entrance.extratags)
  85         out.end_object().next()
  86
  87     out.end_array().next()
  88
  89
  90 def format_base_json(results: Union[ReverseResults, SearchResults],
  91                      options: Mapping[str, Any], simple: bool,
  92                      class_label: str) -> str:
  93     """ Return the result list as a simple json string in custom Nominatim format.
  94     """
  95     out = JsonWriter()
  96
  97     if simple:
  98         if not results:
  99             return '{"error":"Unable to geocode"}'
 100     else:
 101         out.start_array()
 102
 103     for result in results:
 104         out.start_object()\
 105              .keyval_not_none('place_id', result.place_id)\
 106              .keyval('licence', cl.OSM_ATTRIBUTION)\
 107
 108         _write_osm_id(out, result.osm_object)
 109
 110         # lat and lon must be string values
 111         out.keyval('lat', f"{result.centroid.lat:0.7f}")\
 112            .keyval('lon', f"{result.centroid.lon:0.7f}")\
 113            .keyval(class_label, result.category[0])\
 114            .keyval('type', result.category[1])\
 115            .keyval('place_rank', result.rank_search)\
 116            .keyval('importance', result.calculated_importance())\
 117            .keyval('addresstype', cl.get_label_tag(result.category, result.extratags,
 118                                                    result.rank_address,
 119                                                    result.country_code))\
 120            .keyval('name', result.locale_name or '')\
 121            .keyval('display_name', result.display_name or '')
 122
 123         if options.get('icon_base_url', None):
 124             icon = cl.ICONS.get(result.category)
 125             if icon:
 126                 out.keyval('icon', f"{options['icon_base_url']}/{icon}.p.20.png")
 127
 128         if options.get('addressdetails', False):
 129             out.key('address').start_object()
 130             _write_typed_address(out, result.address_rows, result.country_code)
 131             out.end_object().next()
 132
 133         if options.get('entrances', False):
 134             write_entrances(out, result.entrances)
 135
 136         if options.get('extratags', False):
 137             out.keyval('extratags', result.extratags)
 138
 139         if options.get('namedetails', False):
 140             out.keyval('namedetails', result.names)
 141
 142         # must be string values
 143         bbox = cl.bbox_from_result(result)
 144         out.key('boundingbox').start_array()\
 145            .value(f"{bbox.minlat:0.7f}").next()\
 146            .value(f"{bbox.maxlat:0.7f}").next()\
 147            .value(f"{bbox.minlon:0.7f}").next()\
 148            .value(f"{bbox.maxlon:0.7f}").next()\
 149            .end_array().next()
 150
 151         if result.geometry:
 152             for key in ('text', 'kml'):
 153                 out.keyval_not_none('geo' + key, result.geometry.get(key))
 154             if 'geojson' in result.geometry:
 155                 out.key('geojson').raw(result.geometry['geojson']).next()
 156             out.keyval_not_none('svg', result.geometry.get('svg'))
 157
 158         out.end_object()
 159
 160         if simple:
 161             return out()
 162
 163         out.next()
 164
 165     out.end_array()
 166
 167     return out()
 168
 169
 170 def format_base_geojson(results: Union[ReverseResults, SearchResults],
 171                         options: Mapping[str, Any],
 172                         simple: bool) -> str:
 173     """ Return the result list as a geojson string.
 174     """
 175     if not results and simple:
 176         return '{"error":"Unable to geocode"}'
 177
 178     out = JsonWriter()
 179
 180     out.start_object()\
 181        .keyval('type', 'FeatureCollection')\
 182        .keyval('licence', cl.OSM_ATTRIBUTION)\
 183        .key('features').start_array()
 184
 185     for result in results:
 186         out.start_object()\
 187              .keyval('type', 'Feature')\
 188              .key('properties').start_object()
 189
 190         out.keyval_not_none('place_id', result.place_id)
 191
 192         _write_osm_id(out, result.osm_object)
 193
 194         out.keyval('place_rank', result.rank_search)\
 195            .keyval('category', result.category[0])\
 196            .keyval('type', result.category[1])\
 197            .keyval('importance', result.calculated_importance())\
 198            .keyval('addresstype', cl.get_label_tag(result.category, result.extratags,
 199                                                    result.rank_address,
 200                                                    result.country_code))\
 201            .keyval('name', result.locale_name or '')\
 202            .keyval('display_name', result.display_name or '')
 203
 204         if options.get('addressdetails', False):
 205             out.key('address').start_object()
 206             _write_typed_address(out, result.address_rows, result.country_code)
 207             out.end_object().next()
 208
 209         if options.get('entrances', False):
 210             write_entrances(out, result.entrances)
 211
 212         if options.get('extratags', False):
 213             out.keyval('extratags', result.extratags)
 214
 215         if options.get('namedetails', False):
 216             out.keyval('namedetails', result.names)
 217
 218         out.end_object().next()  # properties
 219
 220         out.key('bbox').start_array()
 221         for coord in cl.bbox_from_result(result).coords:
 222             out.float(coord, 7).next()
 223         out.end_array().next()
 224
 225         out.key('geometry').raw(result.geometry.get('geojson')
 226                                 or result.centroid.to_geojson()).next()
 227
 228         out.end_object().next()
 229
 230     out.end_array().next().end_object()
 231
 232     return out()
 233
 234
 235 def format_base_geocodejson(results: Union[ReverseResults, SearchResults],
 236                             options: Mapping[str, Any], simple: bool) -> str:
 237     """ Return the result list as a geocodejson string.
 238     """
 239     if not results and simple:
 240         return '{"error":"Unable to geocode"}'
 241
 242     out = JsonWriter()
 243
 244     out.start_object()\
 245        .keyval('type', 'FeatureCollection')\
 246        .key('geocoding').start_object()\
 247                         .keyval('version', '0.1.0')\
 248                         .keyval('attribution', cl.OSM_ATTRIBUTION)\
 249                         .keyval('licence', 'ODbL')\
 250                         .keyval_not_none('query', options.get('query'))\
 251                         .end_object().next()\
 252        .key('features').start_array()
 253
 254     for result in results:
 255         out.start_object()\
 256              .keyval('type', 'Feature')\
 257              .key('properties').start_object()\
 258                                .key('geocoding').start_object()
 259
 260         out.keyval_not_none('place_id', result.place_id)
 261
 262         _write_osm_id(out, result.osm_object)
 263
 264         out.keyval('osm_key', result.category[0])\
 265            .keyval('osm_value', result.category[1])\
 266            .keyval('type', GEOCODEJSON_RANKS[max(3, min(28, result.rank_address))])\
 267            .keyval_not_none('accuracy', getattr(result, 'distance', None), transform=int)\
 268            .keyval('label', result.display_name or '')\
 269            .keyval_not_none('name', result.locale_name or None)\
 270
 271         if options.get('addressdetails', False):
 272             _write_geocodejson_address(out, result.address_rows, result.place_id,
 273                                        result.country_code)
 274
 275             out.key('admin').start_object()
 276             if result.address_rows:
 277                 for line in result.address_rows:
 278                     if line.isaddress and (line.admin_level or 15) < 15 and line.local_name \
 279                        and line.category[0] == 'boundary' and line.category[1] == 'administrative':
 280                         out.keyval(f"level{line.admin_level}", line.local_name)
 281             out.end_object().next()
 282
 283         if options.get('entrances', False):
 284             write_entrances(out, result.entrances)
 285
 286         if options.get('extratags', False):
 287             out.keyval('extra', result.extratags)
 288
 289         out.end_object().next().end_object().next()
 290
 291         out.key('geometry').raw(result.geometry.get('geojson')
 292                                 or result.centroid.to_geojson()).next()
 293
 294         out.end_object().next()
 295
 296     out.end_array().next().end_object()
 297
 298     return out()
 299
 300
 301 GEOCODEJSON_RANKS = {
 302     3: 'locality',
 303     4: 'country',
 304     5: 'state', 6: 'state', 7: 'state', 8: 'state', 9: 'state',
 305     10: 'county', 11: 'county', 12: 'county',
 306     13: 'city', 14: 'city', 15: 'city', 16: 'city',
 307     17: 'district', 18: 'district', 19: 'district', 20: 'district', 21: 'district',
 308     22: 'locality', 23: 'locality', 24: 'locality',
 309     25: 'street', 26: 'street', 27: 'street', 28: 'house'}