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