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         json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
 126     else:
 127         sys.stdout.write(output)
 128     sys.stdout.write('\n')
 129
 130 class APISearch:
 131     """\
 132     Execute a search query.
 133
 134     This command works exactly the same as if calling the /search endpoint on
 135     the web API. See the online documentation for more details on the
 136     various parameters:
 137     https://nominatim.org/release-docs/latest/api/Search/
 138     """
 139
 140     def add_args(self, parser: argparse.ArgumentParser) -> None:
 141         group = parser.add_argument_group('Query arguments')
 142         group.add_argument('--query',
 143                            help='Free-form query string')
 144         for name, desc in STRUCTURED_QUERY:
 145             group.add_argument('--' + name, help='Structured query: ' + desc)
 146
 147         _add_api_output_arguments(parser)
 148
 149         group = parser.add_argument_group('Result limitation')
 150         group.add_argument('--countrycodes', metavar='CC,..',
 151                            help='Limit search results to one or more countries')
 152         group.add_argument('--exclude_place_ids', metavar='ID,..',
 153                            help='List of search object to be excluded')
 154         group.add_argument('--limit', type=int, default=10,
 155                            help='Limit the number of returned results')
 156         group.add_argument('--viewbox', metavar='X1,Y1,X2,Y2',
 157                            help='Preferred area to find search results')
 158         group.add_argument('--bounded', action='store_true',
 159                            help='Strictly restrict results to viewbox area')
 160         group.add_argument('--no-dedupe', action='store_false', dest='dedupe',
 161                            help='Do not remove duplicates from the result list')
 162         _add_list_format(parser)
 163
 164
 165     def run(self, args: NominatimArgs) -> int:
 166         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 167
 168         if args.list_formats:
 169             return _list_formats(formatter, napi.SearchResults)
 170
 171         if args.format == 'debug':
 172             loglib.set_log_output('text')
 173         elif not formatter.supports_format(napi.SearchResults, args.format):
 174             raise UsageError(f"Unsupported format '{args.format}'. "
 175                              'Use --list-formats to see supported formats.')
 176
 177         api = napi.NominatimAPI(args.project_dir)
 178         params: Dict[str, Any] = {'max_results': args.limit + min(args.limit, 10),
 179                                   'address_details': True, # needed for display name
 180                                   'geometry_output': _get_geometry_output(args),
 181                                   'geometry_simplification': args.polygon_threshold,
 182                                   'countries': args.countrycodes,
 183                                   'excluded': args.exclude_place_ids,
 184                                   'viewbox': args.viewbox,
 185                                   'bounded_viewbox': args.bounded,
 186                                   'locales': _get_locales(args, api.config.DEFAULT_LANGUAGE)
 187                                  }
 188
 189         if args.query:
 190             results = api.search(args.query, **params)
 191         else:
 192             results = api.search_address(amenity=args.amenity,
 193                                          street=args.street,
 194                                          city=args.city,
 195                                          county=args.county,
 196                                          state=args.state,
 197                                          postalcode=args.postalcode,
 198                                          country=args.country,
 199                                          **params)
 200
 201         if args.dedupe and len(results) > 1:
 202             results = deduplicate_results(results, args.limit)
 203
 204         if args.format == 'debug':
 205             print(loglib.get_and_disable())
 206             return 0
 207
 208         _print_output(formatter, results, args.format,
 209                       {'extratags': args.extratags,
 210                        'namedetails': args.namedetails,
 211                        'addressdetails': args.addressdetails})
 212         return 0
 213
 214
 215 class APIReverse:
 216     """\
 217     Execute API reverse query.
 218
 219     This command works exactly the same as if calling the /reverse endpoint on
 220     the web API. See the online documentation for more details on the
 221     various parameters:
 222     https://nominatim.org/release-docs/latest/api/Reverse/
 223     """
 224
 225     def add_args(self, parser: argparse.ArgumentParser) -> None:
 226         group = parser.add_argument_group('Query arguments')
 227         group.add_argument('--lat', type=float,
 228                            help='Latitude of coordinate to look up (in WGS84)')
 229         group.add_argument('--lon', type=float,
 230                            help='Longitude of coordinate to look up (in WGS84)')
 231         group.add_argument('--zoom', type=int,
 232                            help='Level of detail required for the address')
 233         group.add_argument('--layer', metavar='LAYER',
 234                            choices=[n.name.lower() for n in napi.DataLayer if n.name],
 235                            action='append', required=False, dest='layers',
 236                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 237
 238         _add_api_output_arguments(parser)
 239         _add_list_format(parser)
 240
 241
 242     def run(self, args: NominatimArgs) -> int:
 243         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 244
 245         if args.list_formats:
 246             return _list_formats(formatter, napi.ReverseResults)
 247
 248         if args.format == 'debug':
 249             loglib.set_log_output('text')
 250         elif not formatter.supports_format(napi.ReverseResults, args.format):
 251             raise UsageError(f"Unsupported format '{args.format}'. "
 252                              'Use --list-formats to see supported formats.')
 253
 254         if args.lat is None or args.lon is None:
 255             raise UsageError("lat' and 'lon' parameters are required.")
 256
 257         api = napi.NominatimAPI(args.project_dir)
 258         result = api.reverse(napi.Point(args.lon, args.lat),
 259                              max_rank=zoom_to_rank(args.zoom or 18),
 260                              layers=_get_layers(args, napi.DataLayer.ADDRESS | napi.DataLayer.POI),
 261                              address_details=True, # needed for display name
 262                              geometry_output=_get_geometry_output(args),
 263                              geometry_simplification=args.polygon_threshold,
 264                              locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
 265
 266         if args.format == 'debug':
 267             print(loglib.get_and_disable())
 268             return 0
 269
 270         if result:
 271             _print_output(formatter, napi.ReverseResults([result]), args.format,
 272                           {'extratags': args.extratags,
 273                            'namedetails': args.namedetails,
 274                            'addressdetails': args.addressdetails})
 275
 276             return 0
 277
 278         LOG.error("Unable to geocode.")
 279         return 42
 280
 281
 282
 283 class APILookup:
 284     """\
 285     Execute API lookup query.
 286
 287     This command works exactly the same as if calling the /lookup endpoint on
 288     the web API. See the online documentation for more details on the
 289     various parameters:
 290     https://nominatim.org/release-docs/latest/api/Lookup/
 291     """
 292
 293     def add_args(self, parser: argparse.ArgumentParser) -> None:
 294         group = parser.add_argument_group('Query arguments')
 295         group.add_argument('--id', metavar='OSMID',
 296                            action='append', dest='ids',
 297                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 298
 299         _add_api_output_arguments(parser)
 300         _add_list_format(parser)
 301
 302
 303     def run(self, args: NominatimArgs) -> int:
 304         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 305
 306         if args.list_formats:
 307             return _list_formats(formatter, napi.ReverseResults)
 308
 309         if args.format == 'debug':
 310             loglib.set_log_output('text')
 311         elif not formatter.supports_format(napi.ReverseResults, args.format):
 312             raise UsageError(f"Unsupported format '{args.format}'. "
 313                              'Use --list-formats to see supported formats.')
 314
 315         if args.ids is None:
 316             raise UsageError("'id' parameter required.")
 317
 318         places = [napi.OsmID(o[0], int(o[1:])) for o in args.ids]
 319
 320         api = napi.NominatimAPI(args.project_dir)
 321         results = api.lookup(places,
 322                              address_details=True, # needed for display name
 323                              geometry_output=_get_geometry_output(args),
 324                              geometry_simplification=args.polygon_threshold or 0.0,
 325                              locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
 326
 327         if args.format == 'debug':
 328             print(loglib.get_and_disable())
 329             return 0
 330
 331         _print_output(formatter, results, args.format,
 332                       {'extratags': args.extratags,
 333                        'namedetails': args.namedetails,
 334                        'addressdetails': args.addressdetails})
 335         return 0
 336
 337
 338 class APIDetails:
 339     """\
 340     Execute API details query.
 341
 342     This command works exactly the same as if calling the /details endpoint on
 343     the web API. See the online documentation for more details on the
 344     various parameters:
 345     https://nominatim.org/release-docs/latest/api/Details/
 346     """
 347
 348     def add_args(self, parser: argparse.ArgumentParser) -> None:
 349         group = parser.add_argument_group('Query arguments')
 350         group.add_argument('--node', '-n', type=int,
 351                            help="Look up the OSM node with the given ID.")
 352         group.add_argument('--way', '-w', type=int,
 353                            help="Look up the OSM way with the given ID.")
 354         group.add_argument('--relation', '-r', type=int,
 355                            help="Look up the OSM relation with the given ID.")
 356         group.add_argument('--place_id', '-p', type=int,
 357                            help='Database internal identifier of the OSM object to look up')
 358         group.add_argument('--class', dest='object_class',
 359                            help=("Class type to disambiguated multiple entries "
 360                                  "of the same object."))
 361
 362         group = parser.add_argument_group('Output arguments')
 363         group.add_argument('--format', type=str, default='json',
 364                            help='Format of result (use --list-formats to see supported formats)')
 365         group.add_argument('--addressdetails', action='store_true',
 366                            help='Include a breakdown of the address into elements')
 367         group.add_argument('--keywords', action='store_true',
 368                            help='Include a list of name keywords and address keywords')
 369         group.add_argument('--linkedplaces', action='store_true',
 370                            help='Include a details of places that are linked with this one')
 371         group.add_argument('--hierarchy', action='store_true',
 372                            help='Include details of places lower in the address hierarchy')
 373         group.add_argument('--group_hierarchy', action='store_true',
 374                            help='Group the places by type')
 375         group.add_argument('--polygon_geojson', action='store_true',
 376                            help='Include geometry of result')
 377         group.add_argument('--lang', '--accept-language', metavar='LANGS',
 378                            help='Preferred language order for presenting search results')
 379         _add_list_format(parser)
 380
 381
 382     def run(self, args: NominatimArgs) -> int:
 383         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 384
 385         if args.list_formats:
 386             return _list_formats(formatter, napi.DetailedResult)
 387
 388         if args.format == 'debug':
 389             loglib.set_log_output('text')
 390         elif not formatter.supports_format(napi.DetailedResult, args.format):
 391             raise UsageError(f"Unsupported format '{args.format}'. "
 392                              'Use --list-formats to see supported formats.')
 393
 394         place: napi.PlaceRef
 395         if args.node:
 396             place = napi.OsmID('N', args.node, args.object_class)
 397         elif args.way:
 398             place = napi.OsmID('W', args.way, args.object_class)
 399         elif args.relation:
 400             place = napi.OsmID('R', args.relation, args.object_class)
 401         elif  args.place_id is not None:
 402             place = napi.PlaceID(args.place_id)
 403         else:
 404             raise UsageError('One of the arguments --node/-n --way/-w '
 405                              '--relation/-r --place_id/-p is required/')
 406
 407         api = napi.NominatimAPI(args.project_dir)
 408         locales = _get_locales(args, api.config.DEFAULT_LANGUAGE)
 409         result = api.details(place,
 410                              address_details=args.addressdetails,
 411                              linked_places=args.linkedplaces,
 412                              parented_places=args.hierarchy,
 413                              keywords=args.keywords,
 414                              geometry_output=napi.GeometryFormat.GEOJSON
 415                                              if args.polygon_geojson
 416                                              else napi.GeometryFormat.NONE,
 417                             locales=locales)
 418
 419         if args.format == 'debug':
 420             print(loglib.get_and_disable())
 421             return 0
 422
 423         if result:
 424             _print_output(formatter, result, args.format or 'json',
 425                           {'locales': locales,
 426                            'group_hierarchy': args.group_hierarchy})
 427             return 0
 428
 429         LOG.error("Object not found in database.")
 430         return 42
 431
 432
 433 class APIStatus:
 434     """
 435     Execute API status query.
 436
 437     This command works exactly the same as if calling the /status endpoint on
 438     the web API. See the online documentation for more details on the
 439     various parameters:
 440     https://nominatim.org/release-docs/latest/api/Status/
 441     """
 442
 443     def add_args(self, parser: argparse.ArgumentParser) -> None:
 444         group = parser.add_argument_group('API parameters')
 445         group.add_argument('--format', type=str, default='text',
 446                            help='Format of result (use --list-formats to see supported formats)')
 447         _add_list_format(parser)
 448
 449
 450     def run(self, args: NominatimArgs) -> int:
 451         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 452
 453         if args.list_formats:
 454             return _list_formats(formatter, napi.StatusResult)
 455
 456         if args.format == 'debug':
 457             loglib.set_log_output('text')
 458         elif not formatter.supports_format(napi.StatusResult, args.format):
 459             raise UsageError(f"Unsupported format '{args.format}'. "
 460                              'Use --list-formats to see supported formats.')
 461
 462         status = napi.NominatimAPI(args.project_dir).status()
 463
 464         if args.format == 'debug':
 465             print(loglib.get_and_disable())
 466             return 0
 467
 468         _print_output(formatter, status, args.format, {})
 469
 470         return 0