nominatim/clicmd/args.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Provides custom functions over command-line arguments.
   9 """
  10 from typing import Optional, List, Dict, Any, Sequence, Tuple
  11 import argparse
  12 import logging
  13 from pathlib import Path
  14
  15 from nominatim.errors import UsageError
  16 from nominatim.config import Configuration
  17 from nominatim.typing import Protocol
  18
  19 LOG = logging.getLogger()
  20
  21 class Subcommand(Protocol):
  22     """
  23     Interface to be implemented by classes implementing a CLI subcommand.
  24     """
  25
  26     def add_args(self, parser: argparse.ArgumentParser) -> None:
  27         """
  28         Fill the given parser for the subcommand with the appropriate
  29         parameters.
  30         """
  31
  32     def run(self, args: 'NominatimArgs') -> int:
  33         """
  34         Run the subcommand with the given parsed arguments.
  35         """
  36
  37
  38 class NominatimArgs:
  39     """ Customized namespace class for the nominatim command line tool
  40         to receive the command-line arguments.
  41     """
  42     # Basic environment set by root program.
  43     config: Configuration
  44     project_dir: Path
  45     phpcgi_path: Path
  46
  47     # Global switches
  48     version: bool
  49     subcommand: Optional[str]
  50     command: Subcommand
  51
  52     # Shared parameters
  53     osm2pgsql_cache: Optional[int]
  54     socket_timeout: int
  55
  56     # Arguments added to all subcommands.
  57     verbose: int
  58     threads: Optional[int]
  59
  60     # Arguments to 'add-data'
  61     file: Optional[str]
  62     diff: Optional[str]
  63     node: Optional[int]
  64     way: Optional[int]
  65     relation: Optional[int]
  66     tiger_data: Optional[str]
  67     use_main_api: bool
  68
  69     # Arguments to 'admin'
  70     warm: bool
  71     check_database: bool
  72     migrate: bool
  73     collect_os_info: bool
  74     analyse_indexing: bool
  75     target: Optional[str]
  76     osm_id: Optional[str]
  77     place_id: Optional[int]
  78
  79     # Arguments to 'import'
  80     osm_file: List[str]
  81     continue_at: Optional[str]
  82     reverse_only: bool
  83     no_partitions: bool
  84     no_updates: bool
  85     offline: bool
  86     ignore_errors: bool
  87     index_noanalyse: bool
  88
  89     # Arguments to 'index'
  90     boundaries_only: bool
  91     no_boundaries: bool
  92     minrank: int
  93     maxrank: int
  94
  95     # Arguments to 'export'
  96     output_type: str
  97     output_format: str
  98     output_all_postcodes: bool
  99     language: Optional[str]
 100     restrict_to_country: Optional[str]
 101     restrict_to_osm_node: Optional[int]
 102     restrict_to_osm_way: Optional[int]
 103     restrict_to_osm_relation: Optional[int]
 104
 105     # Arguments to 'refresh'
 106     postcodes: bool
 107     word_tokens: bool
 108     word_counts: bool
 109     address_levels: bool
 110     functions: bool
 111     wiki_data: bool
 112     secondary_importance: bool
 113     importance: bool
 114     website: bool
 115     diffs: bool
 116     enable_debug_statements: bool
 117     data_object: Sequence[Tuple[str, int]]
 118     data_area: Sequence[Tuple[str, int]]
 119
 120     # Arguments to 'replication'
 121     init: bool
 122     update_functions: bool
 123     check_for_updates: bool
 124     once: bool
 125     catch_up: bool
 126     do_index: bool
 127
 128     # Arguments to 'serve'
 129     server: str
 130     engine: str
 131
 132     # Arguments to 'special-phrases
 133     import_from_wiki: bool
 134     import_from_csv: Optional[str]
 135     no_replace: bool
 136
 137     # Arguments to all query functions
 138     format: str
 139     addressdetails: bool
 140     extratags: bool
 141     namedetails: bool
 142     lang: Optional[str]
 143     polygon_output: Optional[str]
 144     polygon_threshold: Optional[float]
 145
 146     # Arguments to 'search'
 147     query: Optional[str]
 148     street: Optional[str]
 149     city: Optional[str]
 150     county: Optional[str]
 151     state: Optional[str]
 152     country: Optional[str]
 153     postalcode: Optional[str]
 154     countrycodes: Optional[str]
 155     exclude_place_ids: Optional[str]
 156     limit: Optional[int]
 157     viewbox: Optional[str]
 158     bounded: bool
 159     dedupe: bool
 160
 161     # Arguments to 'reverse'
 162     lat: float
 163     lon: float
 164     zoom: Optional[int]
 165
 166     # Arguments to 'lookup'
 167     ids: Sequence[str]
 168
 169     # Arguments to 'details'
 170     object_class: Optional[str]
 171     linkedplaces: bool
 172     hierarchy: bool
 173     keywords: bool
 174     polygon_geojson: bool
 175     group_hierarchy: bool
 176
 177
 178     def osm2pgsql_options(self, default_cache: int,
 179                           default_threads: int) -> Dict[str, Any]:
 180         """ Return the standard osm2pgsql options that can be derived
 181             from the command line arguments. The resulting dict can be
 182             further customized and then used in `run_osm2pgsql()`.
 183         """
 184         return dict(osm2pgsql=self.config.OSM2PGSQL_BINARY or self.config.lib_dir.osm2pgsql,
 185                     osm2pgsql_cache=self.osm2pgsql_cache or default_cache,
 186                     osm2pgsql_style=self.config.get_import_style_file(),
 187                     osm2pgsql_style_path=self.config.config_dir,
 188                     threads=self.threads or default_threads,
 189                     dsn=self.config.get_libpq_dsn(),
 190                     flatnode_file=str(self.config.get_path('FLATNODE_FILE') or ''),
 191                     tablespaces=dict(slim_data=self.config.TABLESPACE_OSM_DATA,
 192                                      slim_index=self.config.TABLESPACE_OSM_INDEX,
 193                                      main_data=self.config.TABLESPACE_PLACE_DATA,
 194                                      main_index=self.config.TABLESPACE_PLACE_INDEX
 195                                     )
 196                    )
 197
 198
 199     def get_osm_file_list(self) -> Optional[List[Path]]:
 200         """ Return the --osm-file argument as a list of Paths or None
 201             if no argument was given. The function also checks if the files
 202             exist and raises a UsageError if one cannot be found.
 203         """
 204         if not self.osm_file:
 205             return None
 206
 207         files = [Path(f) for f in self.osm_file]
 208         for fname in files:
 209             if not fname.is_file():
 210                 LOG.fatal("OSM file '%s' does not exist.", fname)
 211                 raise UsageError('Cannot access file.')
 212
 213         return files