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