X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/aaf2b6032eb3297aeb20b5c98223e9da734f56d4..105597be056bf9cbf3bb4c15f9bc516479e48ee2:/nominatim/clicmd/args.py diff --git a/nominatim/clicmd/args.py b/nominatim/clicmd/args.py index 6a3c8387..8b805496 100644 --- a/nominatim/clicmd/args.py +++ b/nominatim/clicmd/args.py @@ -7,26 +7,184 @@ """ Provides custom functions over command-line arguments. """ +from typing import Optional, List, Dict, Any, Sequence, Tuple +import argparse import logging +from functools import reduce from pathlib import Path from nominatim.errors import UsageError +from nominatim.config import Configuration +from nominatim.typing import Protocol +import nominatim.api as napi LOG = logging.getLogger() +class Subcommand(Protocol): + """ + Interface to be implemented by classes implementing a CLI subcommand. + """ + + def add_args(self, parser: argparse.ArgumentParser) -> None: + """ + Fill the given parser for the subcommand with the appropriate + parameters. + """ + + def run(self, args: 'NominatimArgs') -> int: + """ + Run the subcommand with the given parsed arguments. + """ + + class NominatimArgs: """ Customized namespace class for the nominatim command line tool to receive the command-line arguments. """ + # Basic environment set by root program. + config: Configuration + project_dir: Path + + # Global switches + version: bool + subcommand: Optional[str] + command: Subcommand + + # Shared parameters + osm2pgsql_cache: Optional[int] + socket_timeout: int + + # Arguments added to all subcommands. + verbose: int + threads: Optional[int] + + # Arguments to 'add-data' + file: Optional[str] + diff: Optional[str] + node: Optional[int] + way: Optional[int] + relation: Optional[int] + tiger_data: Optional[str] + use_main_api: bool + + # Arguments to 'admin' + warm: bool + check_database: bool + migrate: bool + collect_os_info: bool + analyse_indexing: bool + target: Optional[str] + osm_id: Optional[str] + place_id: Optional[int] + + # Arguments to 'import' + osm_file: List[str] + continue_at: Optional[str] + reverse_only: bool + no_partitions: bool + no_updates: bool + offline: bool + ignore_errors: bool + index_noanalyse: bool + + # Arguments to 'index' + boundaries_only: bool + no_boundaries: bool + minrank: int + maxrank: int + + # Arguments to 'export' + output_type: str + output_format: str + output_all_postcodes: bool + language: Optional[str] + restrict_to_country: Optional[str] + + # Arguments to 'refresh' + postcodes: bool + word_tokens: bool + word_counts: bool + address_levels: bool + functions: bool + wiki_data: bool + secondary_importance: bool + importance: bool + website: bool + diffs: bool + enable_debug_statements: bool + data_object: Sequence[Tuple[str, int]] + data_area: Sequence[Tuple[str, int]] + + # Arguments to 'replication' + init: bool + update_functions: bool + check_for_updates: bool + once: bool + catch_up: bool + do_index: bool - def osm2pgsql_options(self, default_cache, default_threads): + # Arguments to 'serve' + server: str + engine: str + + # Arguments to 'special-phrases + import_from_wiki: bool + import_from_csv: Optional[str] + no_replace: bool + + # Arguments to all query functions + format: str + addressdetails: bool + extratags: bool + namedetails: bool + lang: Optional[str] + polygon_output: Optional[str] + polygon_threshold: Optional[float] + + # Arguments to 'search' + query: Optional[str] + amenity: Optional[str] + street: Optional[str] + city: Optional[str] + county: Optional[str] + state: Optional[str] + country: Optional[str] + postalcode: Optional[str] + countrycodes: Optional[str] + exclude_place_ids: Optional[str] + limit: int + viewbox: Optional[str] + bounded: bool + dedupe: bool + + # Arguments to 'reverse' + lat: float + lon: float + zoom: Optional[int] + layers: Optional[Sequence[str]] + + # Arguments to 'lookup' + ids: Sequence[str] + + # Arguments to 'details' + object_class: Optional[str] + linkedplaces: bool + hierarchy: bool + keywords: bool + polygon_geojson: bool + group_hierarchy: bool + + + def osm2pgsql_options(self, default_cache: int, + default_threads: int) -> Dict[str, Any]: """ Return the standard osm2pgsql options that can be derived from the command line arguments. The resulting dict can be further customized and then used in `run_osm2pgsql()`. """ - return dict(osm2pgsql=self.config.OSM2PGSQL_BINARY or self.osm2pgsql_path, + return dict(osm2pgsql=self.config.OSM2PGSQL_BINARY or self.config.lib_dir.osm2pgsql, osm2pgsql_cache=self.osm2pgsql_cache or default_cache, osm2pgsql_style=self.config.get_import_style_file(), + osm2pgsql_style_path=self.config.config_dir, threads=self.threads or default_threads, dsn=self.config.get_libpq_dsn(), flatnode_file=str(self.config.get_path('FLATNODE_FILE') or ''), @@ -38,7 +196,7 @@ class NominatimArgs: ) - def get_osm_file_list(self): + def get_osm_file_list(self) -> Optional[List[Path]]: """ Return the --osm-file argument as a list of Paths or None if no argument was given. The function also checks if the files exist and raises a UsageError if one cannot be found. @@ -53,3 +211,45 @@ class NominatimArgs: raise UsageError('Cannot access file.') return files + + + def get_geometry_output(self) -> napi.GeometryFormat: + """ Get the requested geometry output format in a API-compatible + format. + """ + if not self.polygon_output: + return napi.GeometryFormat.NONE + if self.polygon_output == 'geojson': + return napi.GeometryFormat.GEOJSON + if self.polygon_output == 'kml': + return napi.GeometryFormat.KML + if self.polygon_output == 'svg': + return napi.GeometryFormat.SVG + if self.polygon_output == 'text': + return napi.GeometryFormat.TEXT + + try: + return napi.GeometryFormat[self.polygon_output.upper()] + except KeyError as exp: + raise UsageError(f"Unknown polygon output format '{self.polygon_output}'.") from exp + + + def get_locales(self, default: Optional[str]) -> napi.Locales: + """ Get the locales from the language parameter. + """ + if self.lang: + return napi.Locales.from_accept_languages(self.lang) + if default: + return napi.Locales.from_accept_languages(default) + + return napi.Locales() + + + def get_layers(self, default: napi.DataLayer) -> Optional[napi.DataLayer]: + """ Get the list of selected layers as a DataLayer enum. + """ + if not self.layers: + return default + + return reduce(napi.DataLayer.__or__, + (napi.DataLayer[s.upper()] for s in self.layers))