nominatim/clicmd/api.py

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