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                                           }
 200
 201                 if args.query:
 202                     results = api.search(args.query, **params)
 203                 else:
 204                     results = api.search_address(amenity=args.amenity,
 205                                                  street=args.street,
 206                                                  city=args.city,
 207                                                  county=args.county,
 208                                                  state=args.state,
 209                                                  postalcode=args.postalcode,
 210                                                  country=args.country,
 211                                                  **params)
 212         except napi.UsageError as ex:
 213             raise UsageError(ex) from ex
 214
 215         locales = _get_locales(args, api.config.DEFAULT_LANGUAGE)
 216         locales.localize_results(results)
 217
 218         if args.dedupe and len(results) > 1:
 219             results = deduplicate_results(results, args.limit)
 220
 221         if args.format == 'debug':
 222             print(loglib.get_and_disable())
 223             return 0
 224
 225         _print_output(formatter, results, args.format,
 226                       {'extratags': args.extratags,
 227                        'namedetails': args.namedetails,
 228                        'addressdetails': args.addressdetails})
 229         return 0
 230
 231
 232 class APIReverse:
 233     """\
 234     Execute API reverse query.
 235
 236     This command works exactly the same as if calling the /reverse endpoint on
 237     the web API. See the online documentation for more details on the
 238     various parameters:
 239     https://nominatim.org/release-docs/latest/api/Reverse/
 240     """
 241
 242     def add_args(self, parser: argparse.ArgumentParser) -> None:
 243         group = parser.add_argument_group('Query arguments')
 244         group.add_argument('--lat', type=float,
 245                            help='Latitude of coordinate to look up (in WGS84)')
 246         group.add_argument('--lon', type=float,
 247                            help='Longitude of coordinate to look up (in WGS84)')
 248         group.add_argument('--zoom', type=int,
 249                            help='Level of detail required for the address')
 250         group.add_argument('--layer', metavar='LAYER',
 251                            choices=[n.name.lower() for n in napi.DataLayer if n.name],
 252                            action='append', required=False, dest='layers',
 253                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 254
 255         _add_api_output_arguments(parser)
 256         _add_list_format(parser)
 257
 258     def run(self, args: NominatimArgs) -> int:
 259         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 260
 261         if args.list_formats:
 262             return _list_formats(formatter, napi.ReverseResults)
 263
 264         if args.format in ('debug', 'raw'):
 265             loglib.set_log_output('text')
 266         elif not formatter.supports_format(napi.ReverseResults, args.format):
 267             raise UsageError(f"Unsupported format '{args.format}'. "
 268                              'Use --list-formats to see supported formats.')
 269
 270         if args.lat is None or args.lon is None:
 271             raise UsageError("lat' and 'lon' parameters are required.")
 272
 273         layers = _get_layers(args, napi.DataLayer.ADDRESS | napi.DataLayer.POI)
 274
 275         try:
 276             with napi.NominatimAPI(args.project_dir) as api:
 277                 result = api.reverse(napi.Point(args.lon, args.lat),
 278                                      max_rank=zoom_to_rank(args.zoom or 18),
 279                                      layers=layers,
 280                                      address_details=True,  # needed for display name
 281                                      geometry_output=_get_geometry_output(args),
 282                                      geometry_simplification=args.polygon_threshold)
 283         except napi.UsageError as ex:
 284             raise UsageError(ex) from ex
 285
 286         if result is not None:
 287             locales = _get_locales(args, api.config.DEFAULT_LANGUAGE)
 288             locales.localize_results([result])
 289
 290         if args.format == 'debug':
 291             print(loglib.get_and_disable())
 292             return 0
 293
 294         if result:
 295             _print_output(formatter, napi.ReverseResults([result]), args.format,
 296                           {'extratags': args.extratags,
 297                            'namedetails': args.namedetails,
 298                            'addressdetails': args.addressdetails})
 299
 300             return 0
 301
 302         LOG.error("Unable to geocode.")
 303         return 42
 304
 305
 306 class APILookup:
 307     """\
 308     Execute API lookup query.
 309
 310     This command works exactly the same as if calling the /lookup endpoint on
 311     the web API. See the online documentation for more details on the
 312     various parameters:
 313     https://nominatim.org/release-docs/latest/api/Lookup/
 314     """
 315
 316     def add_args(self, parser: argparse.ArgumentParser) -> None:
 317         group = parser.add_argument_group('Query arguments')
 318         group.add_argument('--id', metavar='OSMID',
 319                            action='append', dest='ids',
 320                            help='OSM id to lookup in format <NRW><id> (may be repeated)')
 321
 322         _add_api_output_arguments(parser)
 323         _add_list_format(parser)
 324
 325     def run(self, args: NominatimArgs) -> int:
 326         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 327
 328         if args.list_formats:
 329             return _list_formats(formatter, napi.ReverseResults)
 330
 331         if args.format in ('debug', 'raw'):
 332             loglib.set_log_output('text')
 333         elif not formatter.supports_format(napi.ReverseResults, args.format):
 334             raise UsageError(f"Unsupported format '{args.format}'. "
 335                              'Use --list-formats to see supported formats.')
 336
 337         if args.ids is None:
 338             raise UsageError("'id' parameter required.")
 339
 340         places = [napi.OsmID(o[0], int(o[1:])) for o in args.ids]
 341
 342         try:
 343             with napi.NominatimAPI(args.project_dir) as api:
 344                 results = api.lookup(places,
 345                                      address_details=True,  # needed for display name
 346                                      geometry_output=_get_geometry_output(args),
 347                                      geometry_simplification=args.polygon_threshold or 0.0)
 348         except napi.UsageError as ex:
 349             raise UsageError(ex) from ex
 350
 351         locales = _get_locales(args, api.config.DEFAULT_LANGUAGE)
 352         locales.localize_results(results)
 353
 354         if args.format == 'debug':
 355             print(loglib.get_and_disable())
 356             return 0
 357
 358         _print_output(formatter, results, args.format,
 359                       {'extratags': args.extratags,
 360                        'namedetails': args.namedetails,
 361                        'addressdetails': args.addressdetails})
 362         return 0
 363
 364
 365 class APIDetails:
 366     """\
 367     Execute API details query.
 368
 369     This command works exactly the same as if calling the /details endpoint on
 370     the web API. See the online documentation for more details on the
 371     various parameters:
 372     https://nominatim.org/release-docs/latest/api/Details/
 373     """
 374
 375     def add_args(self, parser: argparse.ArgumentParser) -> None:
 376         group = parser.add_argument_group('Query arguments')
 377         group.add_argument('--node', '-n', type=int,
 378                            help="Look up the OSM node with the given ID.")
 379         group.add_argument('--way', '-w', type=int,
 380                            help="Look up the OSM way with the given ID.")
 381         group.add_argument('--relation', '-r', type=int,
 382                            help="Look up the OSM relation with the given ID.")
 383         group.add_argument('--place_id', '-p', type=int,
 384                            help='Database internal identifier of the OSM object to look up')
 385         group.add_argument('--class', dest='object_class',
 386                            help=("Class type to disambiguated multiple entries "
 387                                  "of the same object."))
 388
 389         group = parser.add_argument_group('Output arguments')
 390         group.add_argument('--format', type=str, default='json',
 391                            help='Format of result (use --list-formats to see supported formats)')
 392         group.add_argument('--addressdetails', action='store_true',
 393                            help='Include a breakdown of the address into elements')
 394         group.add_argument('--keywords', action='store_true',
 395                            help='Include a list of name keywords and address keywords')
 396         group.add_argument('--linkedplaces', action='store_true',
 397                            help='Include a details of places that are linked with this one')
 398         group.add_argument('--hierarchy', action='store_true',
 399                            help='Include details of places lower in the address hierarchy')
 400         group.add_argument('--group_hierarchy', action='store_true',
 401                            help='Group the places by type')
 402         group.add_argument('--polygon_geojson', action='store_true',
 403                            help='Include geometry of result')
 404         group.add_argument('--lang', '--accept-language', metavar='LANGS',
 405                            help='Preferred language order for presenting search results')
 406         _add_list_format(parser)
 407
 408     def run(self, args: NominatimArgs) -> int:
 409         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 410
 411         if args.list_formats:
 412             return _list_formats(formatter, napi.DetailedResult)
 413
 414         if args.format in ('debug', 'raw'):
 415             loglib.set_log_output('text')
 416         elif not formatter.supports_format(napi.DetailedResult, args.format):
 417             raise UsageError(f"Unsupported format '{args.format}'. "
 418                              'Use --list-formats to see supported formats.')
 419
 420         place: napi.PlaceRef
 421         if args.node:
 422             place = napi.OsmID('N', args.node, args.object_class)
 423         elif args.way:
 424             place = napi.OsmID('W', args.way, args.object_class)
 425         elif args.relation:
 426             place = napi.OsmID('R', args.relation, args.object_class)
 427         elif args.place_id is not None:
 428             place = napi.PlaceID(args.place_id)
 429         else:
 430             raise UsageError('One of the arguments --node/-n --way/-w '
 431                              '--relation/-r --place_id/-p is required/')
 432
 433         try:
 434             with napi.NominatimAPI(args.project_dir) as api:
 435                 result = api.details(place,
 436                                      address_details=args.addressdetails,
 437                                      linked_places=args.linkedplaces,
 438                                      parented_places=args.hierarchy,
 439                                      keywords=args.keywords,
 440                                      geometry_output=(napi.GeometryFormat.GEOJSON
 441                                                       if args.polygon_geojson
 442                                                       else napi.GeometryFormat.NONE))
 443         except napi.UsageError as ex:
 444             raise UsageError(ex) from ex
 445
 446         if result is not None:
 447             locales = _get_locales(args, api.config.DEFAULT_LANGUAGE)
 448             locales.localize_results([result])
 449
 450         if args.format == 'debug':
 451             print(loglib.get_and_disable())
 452             return 0
 453
 454         if result:
 455             _print_output(formatter, result, args.format or 'json',
 456                           {'group_hierarchy': args.group_hierarchy})
 457             return 0
 458
 459         LOG.error("Object not found in database.")
 460         return 42
 461
 462
 463 class APIStatus:
 464     """
 465     Execute API status query.
 466
 467     This command works exactly the same as if calling the /status endpoint on
 468     the web API. See the online documentation for more details on the
 469     various parameters:
 470     https://nominatim.org/release-docs/latest/api/Status/
 471     """
 472
 473     def add_args(self, parser: argparse.ArgumentParser) -> None:
 474         group = parser.add_argument_group('API parameters')
 475         group.add_argument('--format', type=str, default='text',
 476                            help='Format of result (use --list-formats to see supported formats)')
 477         _add_list_format(parser)
 478
 479     def run(self, args: NominatimArgs) -> int:
 480         formatter = napi.load_format_dispatcher('v1', args.project_dir)
 481
 482         if args.list_formats:
 483             return _list_formats(formatter, napi.StatusResult)
 484
 485         if args.format in ('debug', 'raw'):
 486             loglib.set_log_output('text')
 487         elif not formatter.supports_format(napi.StatusResult, args.format):
 488             raise UsageError(f"Unsupported format '{args.format}'. "
 489                              'Use --list-formats to see supported formats.')
 490
 491         try:
 492             with napi.NominatimAPI(args.project_dir) as api:
 493                 status = api.status()
 494         except napi.UsageError as ex:
 495             raise UsageError(ex) from ex
 496
 497         if args.format == 'debug':
 498             print(loglib.get_and_disable())
 499             return 0
 500
 501         _print_output(formatter, status, args.format, {})
 502
 503         return 0