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         with napi.NominatimAPI(args.project_dir) as api:
 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         with napi.NominatimAPI(args.project_dir) as api:
 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,
 267                                                     napi.DataLayer.ADDRESS | napi.DataLayer.POI),
 268                                  address_details=True, # needed for display name
 269                                  geometry_output=_get_geometry_output(args),
 270                                  geometry_simplification=args.polygon_threshold,
 271                                  locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
 272
 273         if args.format == 'debug':
 274             print(loglib.get_and_disable())
 275             return 0
 276
 277         if result:
 278             _print_output(formatter, napi.ReverseResults([result]), args.format,
 279                           {'extratags': args.extratags,
 280                            'namedetails': args.namedetails,
 281                            'addressdetails': args.addressdetails})
 282
 283             return 0
 284
 285         LOG.error("Unable to geocode.")
 286         return 42
 287
 288
 289
 290 class APILookup:
 291     """\
 292     Execute API lookup query.
 293
 294     This command works exactly the same as if calling the /lookup endpoint on
 295     the web API. See the online documentation for more details on the
 296     various parameters:
 297     https://nominatim.org/release-docs/latest/api/Lookup/
 298     """
 299
 300     def add_args(self, parser: argparse.ArgumentParser) -> None:
 301         group = parser.add_argument_group('Query arguments')
 302         group.add_argument('--id', metavar='OSMID',
 303                            action='append', dest='ids',
 304                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 305
 306         _add_api_output_arguments(parser)
 307         _add_list_format(parser)
 308
 309
 310     def run(self, args: NominatimArgs) -> int:
 311         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 312
 313         if args.list_formats:
 314             return _list_formats(formatter, napi.ReverseResults)
 315
 316         if args.format == 'debug':
 317             loglib.set_log_output('text')
 318         elif not formatter.supports_format(napi.ReverseResults, args.format):
 319             raise UsageError(f"Unsupported format '{args.format}'. "
 320                              'Use --list-formats to see supported formats.')
 321
 322         if args.ids is None:
 323             raise UsageError("'id' parameter required.")
 324
 325         places = [napi.OsmID(o[0], int(o[1:])) for o in args.ids]
 326
 327         with napi.NominatimAPI(args.project_dir) as api:
 328             results = api.lookup(places,
 329                                  address_details=True, # needed for display name
 330                                  geometry_output=_get_geometry_output(args),
 331                                  geometry_simplification=args.polygon_threshold or 0.0,
 332                                  locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
 333
 334         if args.format == 'debug':
 335             print(loglib.get_and_disable())
 336             return 0
 337
 338         _print_output(formatter, results, args.format,
 339                       {'extratags': args.extratags,
 340                        'namedetails': args.namedetails,
 341                        'addressdetails': args.addressdetails})
 342         return 0
 343
 344
 345 class APIDetails:
 346     """\
 347     Execute API details query.
 348
 349     This command works exactly the same as if calling the /details endpoint on
 350     the web API. See the online documentation for more details on the
 351     various parameters:
 352     https://nominatim.org/release-docs/latest/api/Details/
 353     """
 354
 355     def add_args(self, parser: argparse.ArgumentParser) -> None:
 356         group = parser.add_argument_group('Query arguments')
 357         group.add_argument('--node', '-n', type=int,
 358                            help="Look up the OSM node with the given ID.")
 359         group.add_argument('--way', '-w', type=int,
 360                            help="Look up the OSM way with the given ID.")
 361         group.add_argument('--relation', '-r', type=int,
 362                            help="Look up the OSM relation with the given ID.")
 363         group.add_argument('--place_id', '-p', type=int,
 364                            help='Database internal identifier of the OSM object to look up')
 365         group.add_argument('--class', dest='object_class',
 366                            help=("Class type to disambiguated multiple entries "
 367                                  "of the same object."))
 368
 369         group = parser.add_argument_group('Output arguments')
 370         group.add_argument('--format', type=str, default='json',
 371                            help='Format of result (use --list-formats to see supported formats)')
 372         group.add_argument('--addressdetails', action='store_true',
 373                            help='Include a breakdown of the address into elements')
 374         group.add_argument('--keywords', action='store_true',
 375                            help='Include a list of name keywords and address keywords')
 376         group.add_argument('--linkedplaces', action='store_true',
 377                            help='Include a details of places that are linked with this one')
 378         group.add_argument('--hierarchy', action='store_true',
 379                            help='Include details of places lower in the address hierarchy')
 380         group.add_argument('--group_hierarchy', action='store_true',
 381                            help='Group the places by type')
 382         group.add_argument('--polygon_geojson', action='store_true',
 383                            help='Include geometry of result')
 384         group.add_argument('--lang', '--accept-language', metavar='LANGS',
 385                            help='Preferred language order for presenting search results')
 386         _add_list_format(parser)
 387
 388
 389     def run(self, args: NominatimArgs) -> int:
 390         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 391
 392         if args.list_formats:
 393             return _list_formats(formatter, napi.DetailedResult)
 394
 395         if args.format == 'debug':
 396             loglib.set_log_output('text')
 397         elif not formatter.supports_format(napi.DetailedResult, args.format):
 398             raise UsageError(f"Unsupported format '{args.format}'. "
 399                              'Use --list-formats to see supported formats.')
 400
 401         place: napi.PlaceRef
 402         if args.node:
 403             place = napi.OsmID('N', args.node, args.object_class)
 404         elif args.way:
 405             place = napi.OsmID('W', args.way, args.object_class)
 406         elif args.relation:
 407             place = napi.OsmID('R', args.relation, args.object_class)
 408         elif  args.place_id is not None:
 409             place = napi.PlaceID(args.place_id)
 410         else:
 411             raise UsageError('One of the arguments --node/-n --way/-w '
 412                              '--relation/-r --place_id/-p is required/')
 413
 414         with napi.NominatimAPI(args.project_dir) as api:
 415             locales = _get_locales(args, api.config.DEFAULT_LANGUAGE)
 416             result = api.details(place,
 417                                  address_details=args.addressdetails,
 418                                  linked_places=args.linkedplaces,
 419                                  parented_places=args.hierarchy,
 420                                  keywords=args.keywords,
 421                                  geometry_output=napi.GeometryFormat.GEOJSON
 422                                                  if args.polygon_geojson
 423                                                  else napi.GeometryFormat.NONE,
 424                                 locales=locales)
 425
 426         if args.format == 'debug':
 427             print(loglib.get_and_disable())
 428             return 0
 429
 430         if result:
 431             _print_output(formatter, result, args.format or 'json',
 432                           {'locales': locales,
 433                            'group_hierarchy': args.group_hierarchy})
 434             return 0
 435
 436         LOG.error("Object not found in database.")
 437         return 42
 438
 439
 440 class APIStatus:
 441     """
 442     Execute API status query.
 443
 444     This command works exactly the same as if calling the /status endpoint on
 445     the web API. See the online documentation for more details on the
 446     various parameters:
 447     https://nominatim.org/release-docs/latest/api/Status/
 448     """
 449
 450     def add_args(self, parser: argparse.ArgumentParser) -> None:
 451         group = parser.add_argument_group('API parameters')
 452         group.add_argument('--format', type=str, default='text',
 453                            help='Format of result (use --list-formats to see supported formats)')
 454         _add_list_format(parser)
 455
 456
 457     def run(self, args: NominatimArgs) -> int:
 458         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 459
 460         if args.list_formats:
 461             return _list_formats(formatter, napi.StatusResult)
 462
 463         if args.format == 'debug':
 464             loglib.set_log_output('text')
 465         elif not formatter.supports_format(napi.StatusResult, args.format):
 466             raise UsageError(f"Unsupported format '{args.format}'. "
 467                              'Use --list-formats to see supported formats.')
 468
 469         with napi.NominatimAPI(args.project_dir) as api:
 470             status = api.status()
 471
 472         if args.format == 'debug':
 473             print(loglib.get_and_disable())
 474             return 0
 475
 476         _print_output(formatter, status, args.format, {})
 477
 478         return 0