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