nominatim/clicmd/api.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2023 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Subcommand definitions for API calls from the command line.
   9 """
  10 from typing import Dict, Any
  11 import argparse
  12 import logging
  13 import json
  14 import sys
  15
  16 from nominatim.clicmd.args import NominatimArgs
  17 import nominatim.api as napi
  18 import nominatim.api.v1 as api_output
  19 from nominatim.api.v1.helpers import zoom_to_rank, deduplicate_results
  20 import nominatim.api.logging as loglib
  21
  22 # Do not repeat documentation of subcommand classes.
  23 # pylint: disable=C0111
  24
  25 LOG = logging.getLogger()
  26
  27 STRUCTURED_QUERY = (
  28     ('amenity', 'name and/or type of POI'),
  29     ('street', 'housenumber and street'),
  30     ('city', 'city, town or village'),
  31     ('county', 'county'),
  32     ('state', 'state'),
  33     ('country', 'country'),
  34     ('postalcode', 'postcode')
  35 )
  36
  37 EXTRADATA_PARAMS = (
  38     ('addressdetails', 'Include a breakdown of the address into elements'),
  39     ('extratags', ("Include additional information if available "
  40                    "(e.g. wikipedia link, opening hours)")),
  41     ('namedetails', 'Include a list of alternative names')
  42 )
  43
  44 def _add_api_output_arguments(parser: argparse.ArgumentParser) -> None:
  45     group = parser.add_argument_group('Output arguments')
  46     group.add_argument('--format', default='jsonv2',
  47                        choices=['xml', 'json', 'jsonv2', 'geojson', 'geocodejson', 'debug'],
  48                        help='Format of result')
  49     for name, desc in EXTRADATA_PARAMS:
  50         group.add_argument('--' + name, action='store_true', help=desc)
  51
  52     group.add_argument('--lang', '--accept-language', metavar='LANGS',
  53                        help='Preferred language order for presenting search results')
  54     group.add_argument('--polygon-output',
  55                        choices=['geojson', 'kml', 'svg', 'text'],
  56                        help='Output geometry of results as a GeoJSON, KML, SVG or WKT')
  57     group.add_argument('--polygon-threshold', type=float, default = 0.0,
  58                        metavar='TOLERANCE',
  59                        help=("Simplify output geometry."
  60                              "Parameter is difference tolerance in degrees."))
  61
  62
  63 class APISearch:
  64     """\
  65     Execute a search query.
  66
  67     This command works exactly the same as if calling the /search endpoint on
  68     the web API. See the online documentation for more details on the
  69     various parameters:
  70     https://nominatim.org/release-docs/latest/api/Search/
  71     """
  72
  73     def add_args(self, parser: argparse.ArgumentParser) -> None:
  74         group = parser.add_argument_group('Query arguments')
  75         group.add_argument('--query',
  76                            help='Free-form query string')
  77         for name, desc in STRUCTURED_QUERY:
  78             group.add_argument('--' + name, help='Structured query: ' + desc)
  79
  80         _add_api_output_arguments(parser)
  81
  82         group = parser.add_argument_group('Result limitation')
  83         group.add_argument('--countrycodes', metavar='CC,..',
  84                            help='Limit search results to one or more countries')
  85         group.add_argument('--exclude_place_ids', metavar='ID,..',
  86                            help='List of search object to be excluded')
  87         group.add_argument('--limit', type=int, default=10,
  88                            help='Limit the number of returned results')
  89         group.add_argument('--viewbox', metavar='X1,Y1,X2,Y2',
  90                            help='Preferred area to find search results')
  91         group.add_argument('--bounded', action='store_true',
  92                            help='Strictly restrict results to viewbox area')
  93
  94         group = parser.add_argument_group('Other arguments')
  95         group.add_argument('--no-dedupe', action='store_false', dest='dedupe',
  96                            help='Do not remove duplicates from the result list')
  97
  98
  99     def run(self, args: NominatimArgs) -> int:
 100         if args.format == 'debug':
 101             loglib.set_log_output('text')
 102
 103         api = napi.NominatimAPI(args.project_dir)
 104
 105         params: Dict[str, Any] = {'max_results': args.limit + min(args.limit, 10),
 106                                   'address_details': True, # needed for display name
 107                                   'geometry_output': args.get_geometry_output(),
 108                                   'geometry_simplification': args.polygon_threshold,
 109                                   'countries': args.countrycodes,
 110                                   'excluded': args.exclude_place_ids,
 111                                   'viewbox': args.viewbox,
 112                                   'bounded_viewbox': args.bounded
 113                                  }
 114
 115         if args.query:
 116             results = api.search(args.query, **params)
 117         else:
 118             results = api.search_address(amenity=args.amenity,
 119                                          street=args.street,
 120                                          city=args.city,
 121                                          county=args.county,
 122                                          state=args.state,
 123                                          postalcode=args.postalcode,
 124                                          country=args.country,
 125                                          **params)
 126
 127         for result in results:
 128             result.localize(args.get_locales(api.config.DEFAULT_LANGUAGE))
 129
 130         if args.dedupe and len(results) > 1:
 131             results = deduplicate_results(results, args.limit)
 132
 133         if args.format == 'debug':
 134             print(loglib.get_and_disable())
 135             return 0
 136
 137         output = api_output.format_result(
 138                     results,
 139                     args.format,
 140                     {'extratags': args.extratags,
 141                      'namedetails': args.namedetails,
 142                      'addressdetails': args.addressdetails})
 143         if args.format != 'xml':
 144             # reformat the result, so it is pretty-printed
 145             json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
 146         else:
 147             sys.stdout.write(output)
 148         sys.stdout.write('\n')
 149
 150         return 0
 151
 152
 153 class APIReverse:
 154     """\
 155     Execute API reverse query.
 156
 157     This command works exactly the same as if calling the /reverse endpoint on
 158     the web API. See the online documentation for more details on the
 159     various parameters:
 160     https://nominatim.org/release-docs/latest/api/Reverse/
 161     """
 162
 163     def add_args(self, parser: argparse.ArgumentParser) -> None:
 164         group = parser.add_argument_group('Query arguments')
 165         group.add_argument('--lat', type=float, required=True,
 166                            help='Latitude of coordinate to look up (in WGS84)')
 167         group.add_argument('--lon', type=float, required=True,
 168                            help='Longitude of coordinate to look up (in WGS84)')
 169         group.add_argument('--zoom', type=int,
 170                            help='Level of detail required for the address')
 171         group.add_argument('--layer', metavar='LAYER',
 172                            choices=[n.name.lower() for n in napi.DataLayer if n.name],
 173                            action='append', required=False, dest='layers',
 174                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 175
 176         _add_api_output_arguments(parser)
 177
 178
 179     def run(self, args: NominatimArgs) -> int:
 180         if args.format == 'debug':
 181             loglib.set_log_output('text')
 182
 183         api = napi.NominatimAPI(args.project_dir)
 184
 185         result = api.reverse(napi.Point(args.lon, args.lat),
 186                              max_rank=zoom_to_rank(args.zoom or 18),
 187                              layers=args.get_layers(napi.DataLayer.ADDRESS | napi.DataLayer.POI),
 188                              address_details=True, # needed for display name
 189                              geometry_output=args.get_geometry_output(),
 190                              geometry_simplification=args.polygon_threshold)
 191
 192         if args.format == 'debug':
 193             print(loglib.get_and_disable())
 194             return 0
 195
 196         if result:
 197             result.localize(args.get_locales(api.config.DEFAULT_LANGUAGE))
 198             output = api_output.format_result(
 199                         napi.ReverseResults([result]),
 200                         args.format,
 201                         {'extratags': args.extratags,
 202                          'namedetails': args.namedetails,
 203                          'addressdetails': args.addressdetails})
 204             if args.format != 'xml':
 205                 # reformat the result, so it is pretty-printed
 206                 json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
 207             else:
 208                 sys.stdout.write(output)
 209             sys.stdout.write('\n')
 210
 211             return 0
 212
 213         LOG.error("Unable to geocode.")
 214         return 42
 215
 216
 217
 218 class APILookup:
 219     """\
 220     Execute API lookup query.
 221
 222     This command works exactly the same as if calling the /lookup endpoint on
 223     the web API. See the online documentation for more details on the
 224     various parameters:
 225     https://nominatim.org/release-docs/latest/api/Lookup/
 226     """
 227
 228     def add_args(self, parser: argparse.ArgumentParser) -> None:
 229         group = parser.add_argument_group('Query arguments')
 230         group.add_argument('--id', metavar='OSMID',
 231                            action='append', required=True, dest='ids',
 232                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 233
 234         _add_api_output_arguments(parser)
 235
 236
 237     def run(self, args: NominatimArgs) -> int:
 238         if args.format == 'debug':
 239             loglib.set_log_output('text')
 240
 241         api = napi.NominatimAPI(args.project_dir)
 242
 243         if args.format == 'debug':
 244             print(loglib.get_and_disable())
 245             return 0
 246
 247         places = [napi.OsmID(o[0], int(o[1:])) for o in args.ids]
 248
 249         results = api.lookup(places,
 250                              address_details=True, # needed for display name
 251                              geometry_output=args.get_geometry_output(),
 252                              geometry_simplification=args.polygon_threshold or 0.0)
 253
 254         for result in results:
 255             result.localize(args.get_locales(api.config.DEFAULT_LANGUAGE))
 256
 257         output = api_output.format_result(
 258                     results,
 259                     args.format,
 260                     {'extratags': args.extratags,
 261                      'namedetails': args.namedetails,
 262                      'addressdetails': args.addressdetails})
 263         if args.format != 'xml':
 264             # reformat the result, so it is pretty-printed
 265             json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
 266         else:
 267             sys.stdout.write(output)
 268         sys.stdout.write('\n')
 269
 270         return 0
 271
 272
 273 class APIDetails:
 274     """\
 275     Execute API details query.
 276
 277     This command works exactly the same as if calling the /details endpoint on
 278     the web API. See the online documentation for more details on the
 279     various parameters:
 280     https://nominatim.org/release-docs/latest/api/Details/
 281     """
 282
 283     def add_args(self, parser: argparse.ArgumentParser) -> None:
 284         group = parser.add_argument_group('Query arguments')
 285         objs = group.add_mutually_exclusive_group(required=True)
 286         objs.add_argument('--node', '-n', type=int,
 287                           help="Look up the OSM node with the given ID.")
 288         objs.add_argument('--way', '-w', type=int,
 289                           help="Look up the OSM way with the given ID.")
 290         objs.add_argument('--relation', '-r', type=int,
 291                           help="Look up the OSM relation with the given ID.")
 292         objs.add_argument('--place_id', '-p', type=int,
 293                           help='Database internal identifier of the OSM object to look up')
 294         group.add_argument('--class', dest='object_class',
 295                            help=("Class type to disambiguated multiple entries "
 296                                  "of the same object."))
 297
 298         group = parser.add_argument_group('Output arguments')
 299         group.add_argument('--addressdetails', action='store_true',
 300                            help='Include a breakdown of the address into elements')
 301         group.add_argument('--keywords', action='store_true',
 302                            help='Include a list of name keywords and address keywords')
 303         group.add_argument('--linkedplaces', action='store_true',
 304                            help='Include a details of places that are linked with this one')
 305         group.add_argument('--hierarchy', action='store_true',
 306                            help='Include details of places lower in the address hierarchy')
 307         group.add_argument('--group_hierarchy', action='store_true',
 308                            help='Group the places by type')
 309         group.add_argument('--polygon_geojson', action='store_true',
 310                            help='Include geometry of result')
 311         group.add_argument('--lang', '--accept-language', metavar='LANGS',
 312                            help='Preferred language order for presenting search results')
 313
 314
 315     def run(self, args: NominatimArgs) -> int:
 316         place: napi.PlaceRef
 317         if args.node:
 318             place = napi.OsmID('N', args.node, args.object_class)
 319         elif args.way:
 320             place = napi.OsmID('W', args.way, args.object_class)
 321         elif args.relation:
 322             place = napi.OsmID('R', args.relation, args.object_class)
 323         else:
 324             assert args.place_id is not None
 325             place = napi.PlaceID(args.place_id)
 326
 327         api = napi.NominatimAPI(args.project_dir)
 328
 329         result = api.details(place,
 330                              address_details=args.addressdetails,
 331                              linked_places=args.linkedplaces,
 332                              parented_places=args.hierarchy,
 333                              keywords=args.keywords,
 334                              geometry_output=napi.GeometryFormat.GEOJSON
 335                                              if args.polygon_geojson
 336                                              else napi.GeometryFormat.NONE)
 337
 338
 339         if result:
 340             locales = args.get_locales(api.config.DEFAULT_LANGUAGE)
 341             result.localize(locales)
 342
 343             output = api_output.format_result(
 344                         result,
 345                         'json',
 346                         {'locales': locales,
 347                          'group_hierarchy': args.group_hierarchy})
 348             # reformat the result, so it is pretty-printed
 349             json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
 350             sys.stdout.write('\n')
 351
 352             return 0
 353
 354         LOG.error("Object not found in database.")
 355         return 42
 356
 357
 358 class APIStatus:
 359     """
 360     Execute API status query.
 361
 362     This command works exactly the same as if calling the /status endpoint on
 363     the web API. See the online documentation for more details on the
 364     various parameters:
 365     https://nominatim.org/release-docs/latest/api/Status/
 366     """
 367
 368     def add_args(self, parser: argparse.ArgumentParser) -> None:
 369         formats = api_output.list_formats(napi.StatusResult)
 370         group = parser.add_argument_group('API parameters')
 371         group.add_argument('--format', default=formats[0], choices=formats,
 372                            help='Format of result')
 373
 374
 375     def run(self, args: NominatimArgs) -> int:
 376         status = napi.NominatimAPI(args.project_dir).status()
 377         print(api_output.format_result(status, args.format, {}))
 378         return 0