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