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