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