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