]> git.openstreetmap.org Git - nominatim.git/blob - src/nominatim_api/v1/format_json.py
split code into submodules
[nominatim.git] / 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
11
12 from nominatim_core.utils.json_writer import JsonWriter
13 from ..results import AddressLines, ReverseResults, SearchResults
14 from . import classtypes as cl
15
16 #pylint: disable=too-many-branches
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
62     for k, v in extra.items():
63         out.keyval(k, v)
64
65     if country_code:
66         out.keyval('country_code', country_code)
67
68
69 def format_base_json(results: Union[ReverseResults, SearchResults],
70                      options: Mapping[str, Any], simple: bool,
71                      class_label: str) -> str:
72     """ Return the result list as a simple json string in custom Nominatim format.
73     """
74     out = JsonWriter()
75
76     if simple:
77         if not results:
78             return '{"error":"Unable to geocode"}'
79     else:
80         out.start_array()
81
82     for result in results:
83         out.start_object()\
84              .keyval_not_none('place_id', result.place_id)\
85              .keyval('licence', cl.OSM_ATTRIBUTION)\
86
87         _write_osm_id(out, result.osm_object)
88
89         out.keyval('lat', f"{result.centroid.lat}")\
90              .keyval('lon', f"{result.centroid.lon}")\
91              .keyval(class_label, result.category[0])\
92              .keyval('type', result.category[1])\
93              .keyval('place_rank', result.rank_search)\
94              .keyval('importance', result.calculated_importance())\
95              .keyval('addresstype', cl.get_label_tag(result.category, result.extratags,
96                                                      result.rank_address,
97                                                      result.country_code))\
98              .keyval('name', result.locale_name or '')\
99              .keyval('display_name', result.display_name or '')
100
101
102         if options.get('icon_base_url', None):
103             icon = cl.ICONS.get(result.category)
104             if icon:
105                 out.keyval('icon', f"{options['icon_base_url']}/{icon}.p.20.png")
106
107         if options.get('addressdetails', False):
108             out.key('address').start_object()
109             _write_typed_address(out, result.address_rows, result.country_code)
110             out.end_object().next()
111
112         if options.get('extratags', False):
113             out.keyval('extratags', result.extratags)
114
115         if options.get('namedetails', False):
116             out.keyval('namedetails', result.names)
117
118         bbox = cl.bbox_from_result(result)
119         out.key('boundingbox').start_array()\
120              .value(f"{bbox.minlat:0.7f}").next()\
121              .value(f"{bbox.maxlat:0.7f}").next()\
122              .value(f"{bbox.minlon:0.7f}").next()\
123              .value(f"{bbox.maxlon:0.7f}").next()\
124            .end_array().next()
125
126         if result.geometry:
127             for key in ('text', 'kml'):
128                 out.keyval_not_none('geo' + key, result.geometry.get(key))
129             if 'geojson' in result.geometry:
130                 out.key('geojson').raw(result.geometry['geojson']).next()
131             out.keyval_not_none('svg', result.geometry.get('svg'))
132
133         out.end_object()
134
135         if simple:
136             return out()
137
138         out.next()
139
140     out.end_array()
141
142     return out()
143
144
145 def format_base_geojson(results: Union[ReverseResults, SearchResults],
146                         options: Mapping[str, Any],
147                         simple: bool) -> str:
148     """ Return the result list as a geojson string.
149     """
150     if not results and simple:
151         return '{"error":"Unable to geocode"}'
152
153     out = JsonWriter()
154
155     out.start_object()\
156          .keyval('type', 'FeatureCollection')\
157          .keyval('licence', cl.OSM_ATTRIBUTION)\
158          .key('features').start_array()
159
160     for result in results:
161         out.start_object()\
162              .keyval('type', 'Feature')\
163              .key('properties').start_object()
164
165         out.keyval_not_none('place_id', result.place_id)
166
167         _write_osm_id(out, result.osm_object)
168
169         out.keyval('place_rank', result.rank_search)\
170            .keyval('category', result.category[0])\
171            .keyval('type', result.category[1])\
172            .keyval('importance', result.calculated_importance())\
173            .keyval('addresstype', cl.get_label_tag(result.category, result.extratags,
174                                                    result.rank_address,
175                                                    result.country_code))\
176            .keyval('name', result.locale_name or '')\
177            .keyval('display_name', result.display_name or '')
178
179         if options.get('addressdetails', False):
180             out.key('address').start_object()
181             _write_typed_address(out, result.address_rows, result.country_code)
182             out.end_object().next()
183
184         if options.get('extratags', False):
185             out.keyval('extratags', result.extratags)
186
187         if options.get('namedetails', False):
188             out.keyval('namedetails', result.names)
189
190         out.end_object().next() # properties
191
192         out.key('bbox').start_array()
193         for coord in cl.bbox_from_result(result).coords:
194             out.float(coord, 7).next()
195         out.end_array().next()
196
197         out.key('geometry').raw(result.geometry.get('geojson')
198                                 or result.centroid.to_geojson()).next()
199
200         out.end_object().next()
201
202     out.end_array().next().end_object()
203
204     return out()
205
206
207 def format_base_geocodejson(results: Union[ReverseResults, SearchResults],
208                             options: Mapping[str, Any], simple: bool) -> str:
209     """ Return the result list as a geocodejson string.
210     """
211     if not results and simple:
212         return '{"error":"Unable to geocode"}'
213
214     out = JsonWriter()
215
216     out.start_object()\
217          .keyval('type', 'FeatureCollection')\
218          .key('geocoding').start_object()\
219            .keyval('version', '0.1.0')\
220            .keyval('attribution', cl.OSM_ATTRIBUTION)\
221            .keyval('licence', 'ODbL')\
222            .keyval_not_none('query', options.get('query'))\
223            .end_object().next()\
224          .key('features').start_array()
225
226     for result in results:
227         out.start_object()\
228              .keyval('type', 'Feature')\
229              .key('properties').start_object()\
230                .key('geocoding').start_object()
231
232         out.keyval_not_none('place_id', result.place_id)
233
234         _write_osm_id(out, result.osm_object)
235
236         out.keyval('osm_key', result.category[0])\
237            .keyval('osm_value', result.category[1])\
238            .keyval('type', GEOCODEJSON_RANKS[max(3, min(28, result.rank_address))])\
239            .keyval_not_none('accuracy', getattr(result, 'distance', None), transform=int)\
240            .keyval('label', result.display_name or '')\
241            .keyval_not_none('name', result.locale_name or None)\
242
243         if options.get('addressdetails', False):
244             _write_geocodejson_address(out, result.address_rows, result.place_id,
245                                        result.country_code)
246
247             out.key('admin').start_object()
248             if result.address_rows:
249                 for line in result.address_rows:
250                     if line.isaddress and (line.admin_level or 15) < 15 and line.local_name \
251                        and line.category[0] == 'boundary' and line.category[1] == 'administrative':
252                         out.keyval(f"level{line.admin_level}", line.local_name)
253             out.end_object().next()
254
255         out.end_object().next().end_object().next()
256
257         out.key('geometry').raw(result.geometry.get('geojson')
258                                 or result.centroid.to_geojson()).next()
259
260         out.end_object().next()
261
262     out.end_array().next().end_object()
263
264     return out()
265
266
267 GEOCODEJSON_RANKS = {
268     3: 'locality',
269     4: 'country',
270     5: 'state', 6: 'state', 7: 'state', 8: 'state', 9: 'state',
271     10: 'county', 11: 'county', 12: 'county',
272     13: 'city', 14: 'city', 15: 'city', 16: 'city',
273     17: 'district', 18: 'district', 19: 'district', 20: 'district', 21: 'district',
274     22: 'locality', 23: 'locality', 24: 'locality',
275     25: 'street', 26: 'street', 27: 'street', 28: 'house'}