src/nominatim_db/clicmd/api.py

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