X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/68bd9c6091521a53206fa2996b26ca26834f8df8..424ebd7fe9c63f85e8d94449715d52062a27d43d:/nominatim/clicmd/args.py diff --git a/nominatim/clicmd/args.py b/nominatim/clicmd/args.py index ee194187..433435bc 100644 --- a/nominatim/clicmd/args.py +++ b/nominatim/clicmd/args.py @@ -1,27 +1,260 @@ +# SPDX-License-Identifier: GPL-2.0-only +# +# This file is part of Nominatim. (https://nominatim.org) +# +# Copyright (C) 2022 by the Nominatim developer community. +# For a full list of authors see the git log. """ 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 + clean_deleted: str + 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 + prepare_database: 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 'convert' + output: Path + + # 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 + + # Arguments to 'serve' + server: str + engine: str + + # Arguments to 'special-phrases + import_from_wiki: bool + import_from_csv: Optional[str] + no_replace: bool - def osm2pgsql_options(self, default_cache, default_threads): + # 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=self.config.FLATNODE_FILE, + flatnode_file=str(self.config.get_path('FLATNODE_FILE') or ''), tablespaces=dict(slim_data=self.config.TABLESPACE_OSM_DATA, slim_index=self.config.TABLESPACE_OSM_INDEX, main_data=self.config.TABLESPACE_PLACE_DATA, main_index=self.config.TABLESPACE_PLACE_INDEX ) - ) + ) + + + 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. + """ + if not self.osm_file: + return None + + files = [Path(f) for f in self.osm_file] + for fname in files: + if not fname.is_file(): + LOG.fatal("OSM file '%s' does not exist.", fname) + 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))