Merge remote-tracking branch 'upstream/master'

[nominatim.git] / nominatim / clicmd / args.py

diff --git a/nominatim/clicmd/args.py b/nominatim/clicmd/args.py
index f9b94ef260670a7ca15169c2b8534220afe57e15..8b805496fffd750bec95f9e25303e17c5eca6305 100644 (file)
--- a/nominatim/clicmd/args.py
+++ b/nominatim/clicmd/args.py
@@ -1,22 +1,255 @@
+# 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.
 """
 """
 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.
     """
 
 
 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
+
+    # 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 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_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(),
                     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))