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