"""
Subcommand definitions for API calls from the command line.
"""
-from typing import Dict, Any
+from typing import Dict, Any, Optional, Type, Mapping
import argparse
import logging
import json
import sys
+import pprint
+from functools import reduce
import nominatim_api as napi
-import nominatim_api.v1 as api_output
from nominatim_api.v1.helpers import zoom_to_rank, deduplicate_results
-from nominatim_api.v1.format import dispatch as formatting
+from nominatim_api.server.content_types import CONTENT_JSON
import nominatim_api.logging as loglib
+from ..errors import UsageError
from .args import NominatimArgs
-# Do not repeat documentation of subcommand classes.
-# pylint: disable=C0111
LOG = logging.getLogger()
+
STRUCTURED_QUERY = (
('amenity', 'name and/or type of POI'),
('street', 'housenumber and street'),
('postalcode', 'postcode')
)
+
EXTRADATA_PARAMS = (
('addressdetails', 'Include a breakdown of the address into elements'),
('extratags', ("Include additional information if available "
('namedetails', 'Include a list of alternative names')
)
+
+def _add_list_format(parser: argparse.ArgumentParser) -> None:
+ group = parser.add_argument_group('Other options')
+ group.add_argument('--list-formats', action='store_true',
+ help='List supported output formats and exit.')
+
+
def _add_api_output_arguments(parser: argparse.ArgumentParser) -> None:
- group = parser.add_argument_group('Output arguments')
- group.add_argument('--format', default='jsonv2',
- choices=formatting.list_formats(napi.SearchResults) + ['debug'],
- help='Format of result')
+ group = parser.add_argument_group('Output formatting')
+ group.add_argument('--format', type=str, default='jsonv2',
+ help='Format of result (use --list-format to see supported formats)')
for name, desc in EXTRADATA_PARAMS:
group.add_argument('--' + name, action='store_true', help=desc)
group.add_argument('--polygon-output',
choices=['geojson', 'kml', 'svg', 'text'],
help='Output geometry of results as a GeoJSON, KML, SVG or WKT')
- group.add_argument('--polygon-threshold', type=float, default = 0.0,
+ group.add_argument('--polygon-threshold', type=float, default=0.0,
metavar='TOLERANCE',
help=("Simplify output geometry."
"Parameter is difference tolerance in degrees."))
-def _get_geometry_output(args) -> napi.GeometryFormat:
+def _get_geometry_output(args: NominatimArgs) -> napi.GeometryFormat:
""" Get the requested geometry output format in a API-compatible
format.
"""
raise UsageError(f"Unknown polygon output format '{args.polygon_output}'.") from exp
-def _get_locales(args, default: Optional[str]) -> napi.Locales:
+def _get_locales(args: NominatimArgs, default: Optional[str]) -> napi.Locales:
""" Get the locales from the language parameter.
"""
if args.lang:
return napi.Locales()
-def _get_layers(args, default: napi.DataLayer) -> Optional[napi.DataLayer]:
+def _get_layers(args: NominatimArgs, default: napi.DataLayer) -> Optional[napi.DataLayer]:
""" Get the list of selected layers as a DataLayer enum.
"""
if not args.layers:
(napi.DataLayer[s.upper()] for s in args.layers))
+def _list_formats(formatter: napi.FormatDispatcher, rtype: Type[Any]) -> int:
+ for fmt in formatter.list_formats(rtype):
+ print(fmt)
+ print('debug')
+ print('raw')
+
+ return 0
+
+
+def _print_output(formatter: napi.FormatDispatcher, result: Any,
+ fmt: str, options: Mapping[str, Any]) -> None:
+
+ if fmt == 'raw':
+ pprint.pprint(result)
+ else:
+ output = formatter.format_result(result, fmt, options)
+ if formatter.get_content_type(fmt) == CONTENT_JSON:
+ # reformat the result, so it is pretty-printed
+ try:
+ json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
+ except json.decoder.JSONDecodeError as err:
+ # Catch the error here, so that data can be debugged,
+ # when people are developping custom result formatters.
+ LOG.fatal("Parsing json failed: %s\nUnformatted output:\n%s", err, output)
+ else:
+ sys.stdout.write(output)
+ sys.stdout.write('\n')
+
+
class APISearch:
"""\
Execute a search query.
help='Preferred area to find search results')
group.add_argument('--bounded', action='store_true',
help='Strictly restrict results to viewbox area')
-
- group = parser.add_argument_group('Other arguments')
group.add_argument('--no-dedupe', action='store_false', dest='dedupe',
help='Do not remove duplicates from the result list')
-
+ _add_list_format(parser)
def run(self, args: NominatimArgs) -> int:
- if args.format == 'debug':
- loglib.set_log_output('text')
+ formatter = napi.load_format_dispatcher('v1', args.project_dir)
- api = napi.NominatimAPI(args.project_dir)
-
- params: Dict[str, Any] = {'max_results': args.limit + min(args.limit, 10),
- 'address_details': True, # needed for display name
- 'geometry_output': _get_geometry_output(args),
- 'geometry_simplification': args.polygon_threshold,
- 'countries': args.countrycodes,
- 'excluded': args.exclude_place_ids,
- 'viewbox': args.viewbox,
- 'bounded_viewbox': args.bounded,
- 'locales': _get_locales(args, api.config.DEFAULT_LANGUAGE)
- }
-
- if args.query:
- results = api.search(args.query, **params)
- else:
- results = api.search_address(amenity=args.amenity,
- street=args.street,
- city=args.city,
- county=args.county,
- state=args.state,
- postalcode=args.postalcode,
- country=args.country,
- **params)
+ if args.list_formats:
+ return _list_formats(formatter, napi.SearchResults)
+
+ if args.format in ('debug', 'raw'):
+ loglib.set_log_output('text')
+ elif not formatter.supports_format(napi.SearchResults, args.format):
+ raise UsageError(f"Unsupported format '{args.format}'. "
+ 'Use --list-formats to see supported formats.')
+
+ try:
+ with napi.NominatimAPI(args.project_dir) as api:
+ params: Dict[str, Any] = {'max_results': args.limit + min(args.limit, 10),
+ 'address_details': True, # needed for display name
+ 'geometry_output': _get_geometry_output(args),
+ 'geometry_simplification': args.polygon_threshold,
+ 'countries': args.countrycodes,
+ 'excluded': args.exclude_place_ids,
+ 'viewbox': args.viewbox,
+ 'bounded_viewbox': args.bounded,
+ 'locales': _get_locales(args, api.config.DEFAULT_LANGUAGE)
+ }
+
+ if args.query:
+ results = api.search(args.query, **params)
+ else:
+ results = api.search_address(amenity=args.amenity,
+ street=args.street,
+ city=args.city,
+ county=args.county,
+ state=args.state,
+ postalcode=args.postalcode,
+ country=args.country,
+ **params)
+ except napi.UsageError as ex:
+ raise UsageError(ex) from ex
if args.dedupe and len(results) > 1:
results = deduplicate_results(results, args.limit)
print(loglib.get_and_disable())
return 0
- output = api_output.format_result(
- results,
- args.format,
- {'extratags': args.extratags,
- 'namedetails': args.namedetails,
- 'addressdetails': args.addressdetails})
- if args.format != 'xml':
- # reformat the result, so it is pretty-printed
- json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
- else:
- sys.stdout.write(output)
- sys.stdout.write('\n')
-
+ _print_output(formatter, results, args.format,
+ {'extratags': args.extratags,
+ 'namedetails': args.namedetails,
+ 'addressdetails': args.addressdetails})
return 0
def add_args(self, parser: argparse.ArgumentParser) -> None:
group = parser.add_argument_group('Query arguments')
- group.add_argument('--lat', type=float, required=True,
+ group.add_argument('--lat', type=float,
help='Latitude of coordinate to look up (in WGS84)')
- group.add_argument('--lon', type=float, required=True,
+ group.add_argument('--lon', type=float,
help='Longitude of coordinate to look up (in WGS84)')
group.add_argument('--zoom', type=int,
help='Level of detail required for the address')
help='OSM id to lookup in format <NRW><id> (may be repeated)')
_add_api_output_arguments(parser)
-
+ _add_list_format(parser)
def run(self, args: NominatimArgs) -> int:
- if args.format == 'debug':
- loglib.set_log_output('text')
+ formatter = napi.load_format_dispatcher('v1', args.project_dir)
- api = napi.NominatimAPI(args.project_dir)
+ if args.list_formats:
+ return _list_formats(formatter, napi.ReverseResults)
- result = api.reverse(napi.Point(args.lon, args.lat),
- max_rank=zoom_to_rank(args.zoom or 18),
- layers=_get_layers(args, napi.DataLayer.ADDRESS | napi.DataLayer.POI),
- address_details=True, # needed for display name
- geometry_output=_get_geometry_output(args),
- geometry_simplification=args.polygon_threshold,
- locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
+ if args.format in ('debug', 'raw'):
+ loglib.set_log_output('text')
+ elif not formatter.supports_format(napi.ReverseResults, args.format):
+ raise UsageError(f"Unsupported format '{args.format}'. "
+ 'Use --list-formats to see supported formats.')
+
+ if args.lat is None or args.lon is None:
+ raise UsageError("lat' and 'lon' parameters are required.")
+
+ layers = _get_layers(args, napi.DataLayer.ADDRESS | napi.DataLayer.POI)
+
+ try:
+ with napi.NominatimAPI(args.project_dir) as api:
+ result = api.reverse(napi.Point(args.lon, args.lat),
+ max_rank=zoom_to_rank(args.zoom or 18),
+ layers=layers,
+ address_details=True, # needed for display name
+ geometry_output=_get_geometry_output(args),
+ geometry_simplification=args.polygon_threshold,
+ locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
+ except napi.UsageError as ex:
+ raise UsageError(ex) from ex
if args.format == 'debug':
print(loglib.get_and_disable())
return 0
if result:
- output = api_output.format_result(
- napi.ReverseResults([result]),
- args.format,
- {'extratags': args.extratags,
- 'namedetails': args.namedetails,
- 'addressdetails': args.addressdetails})
- if args.format != 'xml':
- # reformat the result, so it is pretty-printed
- json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
- else:
- sys.stdout.write(output)
- sys.stdout.write('\n')
+ _print_output(formatter, napi.ReverseResults([result]), args.format,
+ {'extratags': args.extratags,
+ 'namedetails': args.namedetails,
+ 'addressdetails': args.addressdetails})
return 0
return 42
-
class APILookup:
"""\
Execute API lookup query.
def add_args(self, parser: argparse.ArgumentParser) -> None:
group = parser.add_argument_group('Query arguments')
group.add_argument('--id', metavar='OSMID',
- action='append', required=True, dest='ids',
+ action='append', dest='ids',
help='OSM id to lookup in format <NRW><id> (may be repeated)')
_add_api_output_arguments(parser)
-
+ _add_list_format(parser)
def run(self, args: NominatimArgs) -> int:
- if args.format == 'debug':
+ formatter = napi.load_format_dispatcher('v1', args.project_dir)
+
+ if args.list_formats:
+ return _list_formats(formatter, napi.ReverseResults)
+
+ if args.format in ('debug', 'raw'):
loglib.set_log_output('text')
+ elif not formatter.supports_format(napi.ReverseResults, args.format):
+ raise UsageError(f"Unsupported format '{args.format}'. "
+ 'Use --list-formats to see supported formats.')
+
+ if args.ids is None:
+ raise UsageError("'id' parameter required.")
- api = napi.NominatimAPI(args.project_dir)
+ places = [napi.OsmID(o[0], int(o[1:])) for o in args.ids]
+
+ try:
+ with napi.NominatimAPI(args.project_dir) as api:
+ results = api.lookup(places,
+ address_details=True, # needed for display name
+ geometry_output=_get_geometry_output(args),
+ geometry_simplification=args.polygon_threshold or 0.0,
+ locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
+ except napi.UsageError as ex:
+ raise UsageError(ex) from ex
if args.format == 'debug':
print(loglib.get_and_disable())
return 0
- places = [napi.OsmID(o[0], int(o[1:])) for o in args.ids]
-
- results = api.lookup(places,
- address_details=True, # needed for display name
- geometry_output=_get_geometry_output(args),
- geometry_simplification=args.polygon_threshold or 0.0,
- locales=_get_locales(args, api.config.DEFAULT_LANGUAGE))
-
- output = api_output.format_result(
- results,
- args.format,
- {'extratags': args.extratags,
- 'namedetails': args.namedetails,
- 'addressdetails': args.addressdetails})
- if args.format != 'xml':
- # reformat the result, so it is pretty-printed
- json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
- else:
- sys.stdout.write(output)
- sys.stdout.write('\n')
-
+ _print_output(formatter, results, args.format,
+ {'extratags': args.extratags,
+ 'namedetails': args.namedetails,
+ 'addressdetails': args.addressdetails})
return 0
def add_args(self, parser: argparse.ArgumentParser) -> None:
group = parser.add_argument_group('Query arguments')
- objs = group.add_mutually_exclusive_group(required=True)
- objs.add_argument('--node', '-n', type=int,
- help="Look up the OSM node with the given ID.")
- objs.add_argument('--way', '-w', type=int,
- help="Look up the OSM way with the given ID.")
- objs.add_argument('--relation', '-r', type=int,
- help="Look up the OSM relation with the given ID.")
- objs.add_argument('--place_id', '-p', type=int,
- help='Database internal identifier of the OSM object to look up')
+ group.add_argument('--node', '-n', type=int,
+ help="Look up the OSM node with the given ID.")
+ group.add_argument('--way', '-w', type=int,
+ help="Look up the OSM way with the given ID.")
+ group.add_argument('--relation', '-r', type=int,
+ help="Look up the OSM relation with the given ID.")
+ group.add_argument('--place_id', '-p', type=int,
+ help='Database internal identifier of the OSM object to look up')
group.add_argument('--class', dest='object_class',
help=("Class type to disambiguated multiple entries "
"of the same object."))
group = parser.add_argument_group('Output arguments')
+ group.add_argument('--format', type=str, default='json',
+ help='Format of result (use --list-formats to see supported formats)')
group.add_argument('--addressdetails', action='store_true',
help='Include a breakdown of the address into elements')
group.add_argument('--keywords', action='store_true',
help='Include geometry of result')
group.add_argument('--lang', '--accept-language', metavar='LANGS',
help='Preferred language order for presenting search results')
-
+ _add_list_format(parser)
def run(self, args: NominatimArgs) -> int:
+ formatter = napi.load_format_dispatcher('v1', args.project_dir)
+
+ if args.list_formats:
+ return _list_formats(formatter, napi.DetailedResult)
+
+ if args.format in ('debug', 'raw'):
+ loglib.set_log_output('text')
+ elif not formatter.supports_format(napi.DetailedResult, args.format):
+ raise UsageError(f"Unsupported format '{args.format}'. "
+ 'Use --list-formats to see supported formats.')
+
place: napi.PlaceRef
if args.node:
place = napi.OsmID('N', args.node, args.object_class)
place = napi.OsmID('W', args.way, args.object_class)
elif args.relation:
place = napi.OsmID('R', args.relation, args.object_class)
- else:
- assert args.place_id is not None
+ elif args.place_id is not None:
place = napi.PlaceID(args.place_id)
+ else:
+ raise UsageError('One of the arguments --node/-n --way/-w '
+ '--relation/-r --place_id/-p is required/')
+
+ try:
+ with napi.NominatimAPI(args.project_dir) as api:
+ locales = _get_locales(args, api.config.DEFAULT_LANGUAGE)
+ result = api.details(place,
+ address_details=args.addressdetails,
+ linked_places=args.linkedplaces,
+ parented_places=args.hierarchy,
+ keywords=args.keywords,
+ geometry_output=(napi.GeometryFormat.GEOJSON
+ if args.polygon_geojson
+ else napi.GeometryFormat.NONE),
+ locales=locales)
+ except napi.UsageError as ex:
+ raise UsageError(ex) from ex
- api = napi.NominatimAPI(args.project_dir)
-
- locales = _get_locales(args, api.config.DEFAULT_LANGUAGE)
- result = api.details(place,
- address_details=args.addressdetails,
- linked_places=args.linkedplaces,
- parented_places=args.hierarchy,
- keywords=args.keywords,
- geometry_output=napi.GeometryFormat.GEOJSON
- if args.polygon_geojson
- else napi.GeometryFormat.NONE,
- locales=locales)
-
+ if args.format == 'debug':
+ print(loglib.get_and_disable())
+ return 0
if result:
- output = api_output.format_result(
- result,
- 'json',
- {'locales': locales,
- 'group_hierarchy': args.group_hierarchy})
- # reformat the result, so it is pretty-printed
- json.dump(json.loads(output), sys.stdout, indent=4, ensure_ascii=False)
- sys.stdout.write('\n')
-
+ _print_output(formatter, result, args.format or 'json',
+ {'locales': locales,
+ 'group_hierarchy': args.group_hierarchy})
return 0
LOG.error("Object not found in database.")
"""
def add_args(self, parser: argparse.ArgumentParser) -> None:
- formats = api_output.list_formats(napi.StatusResult)
group = parser.add_argument_group('API parameters')
- group.add_argument('--format', default=formats[0], choices=formats,
- help='Format of result')
-
+ group.add_argument('--format', type=str, default='text',
+ help='Format of result (use --list-formats to see supported formats)')
+ _add_list_format(parser)
def run(self, args: NominatimArgs) -> int:
- status = napi.NominatimAPI(args.project_dir).status()
- print(api_output.format_result(status, args.format, {}))
+ formatter = napi.load_format_dispatcher('v1', args.project_dir)
+
+ if args.list_formats:
+ return _list_formats(formatter, napi.StatusResult)
+
+ if args.format in ('debug', 'raw'):
+ loglib.set_log_output('text')
+ elif not formatter.supports_format(napi.StatusResult, args.format):
+ raise UsageError(f"Unsupported format '{args.format}'. "
+ 'Use --list-formats to see supported formats.')
+
+ try:
+ with napi.NominatimAPI(args.project_dir) as api:
+ status = api.status()
+ except napi.UsageError as ex:
+ raise UsageError(ex) from ex
+
+ if args.format == 'debug':
+ print(loglib.get_and_disable())
+ return 0
+
+ _print_output(formatter, status, args.format, {})
+
return 0
projects / nominatim.git / blobdiff