src/nominatim_db/clicmd/args.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2024 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 ..errors import UsageError
  16 from ..config import Configuration
  17 from ..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
  46     # Global switches
  47     version: bool
  48     subcommand: Optional[str]
  49     command: Subcommand
  50
  51     # Shared parameters
  52     osm2pgsql_cache: Optional[int]
  53     socket_timeout: int
  54
  55     # Arguments added to all subcommands.
  56     verbose: int
  57     threads: Optional[int]
  58
  59     # Arguments to 'add-data'
  60     file: Optional[str]
  61     diff: Optional[str]
  62     node: Optional[int]
  63     way: Optional[int]
  64     relation: Optional[int]
  65     tiger_data: Optional[str]
  66     use_main_api: bool
  67
  68     # Arguments to 'admin'
  69     warm: bool
  70     check_database: bool
  71     migrate: bool
  72     collect_os_info: bool
  73     clean_deleted: str
  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     prepare_database: bool
  89
  90     # Arguments to 'index'
  91     boundaries_only: bool
  92     no_boundaries: bool
  93     minrank: int
  94     maxrank: int
  95
  96     # Arguments to 'export'
  97     output_type: str
  98     output_format: str
  99     output_all_postcodes: bool
 100     language: Optional[str]
 101     restrict_to_country: Optional[str]
 102
 103     # Arguments to 'convert'
 104     output: Path
 105
 106     # Arguments to 'refresh'
 107     postcodes: bool
 108     word_tokens: bool
 109     word_counts: bool
 110     address_levels: bool
 111     functions: bool
 112     wiki_data: bool
 113     secondary_importance: bool
 114     importance: bool
 115     website: bool
 116     diffs: bool
 117     enable_debug_statements: bool
 118     data_object: Sequence[Tuple[str, int]]
 119     data_area: Sequence[Tuple[str, int]]
 120
 121     # Arguments to 'replication'
 122     init: bool
 123     update_functions: bool
 124     check_for_updates: bool
 125     once: bool
 126     catch_up: bool
 127     do_index: bool
 128
 129     # Arguments to 'serve'
 130     server: str
 131     engine: str
 132
 133     # Arguments to 'special-phrases
 134     import_from_wiki: bool
 135     import_from_csv: Optional[str]
 136     no_replace: bool
 137
 138     # Arguments to all query functions
 139     format: str
 140     list_formats: bool
 141     addressdetails: bool
 142     extratags: bool
 143     namedetails: bool
 144     lang: Optional[str]
 145     polygon_output: Optional[str]
 146     polygon_threshold: Optional[float]
 147
 148     # Arguments to 'search'
 149     query: Optional[str]
 150     amenity: Optional[str]
 151     street: Optional[str]
 152     city: Optional[str]
 153     county: Optional[str]
 154     state: Optional[str]
 155     country: Optional[str]
 156     postalcode: Optional[str]
 157     countrycodes: Optional[str]
 158     exclude_place_ids: Optional[str]
 159     limit: int
 160     viewbox: Optional[str]
 161     bounded: bool
 162     dedupe: bool
 163
 164     # Arguments to 'reverse'
 165     lat: float
 166     lon: float
 167     zoom: Optional[int]
 168     layers: Optional[Sequence[str]]
 169
 170     # Arguments to 'lookup'
 171     ids: Sequence[str]
 172
 173     # Arguments to 'details'
 174     object_class: Optional[str]
 175     linkedplaces: bool
 176     hierarchy: bool
 177     keywords: bool
 178     polygon_geojson: bool
 179     group_hierarchy: bool
 180
 181
 182     def osm2pgsql_options(self, default_cache: int,
 183                           default_threads: int) -> Dict[str, Any]:
 184         """ Return the standard osm2pgsql options that can be derived
 185             from the command line arguments. The resulting dict can be
 186             further customized and then used in `run_osm2pgsql()`.
 187         """
 188         return dict(osm2pgsql=self.config.OSM2PGSQL_BINARY or self.config.lib_dir.osm2pgsql,
 189                     osm2pgsql_cache=self.osm2pgsql_cache or default_cache,
 190                     osm2pgsql_style=self.config.get_import_style_file(),
 191                     osm2pgsql_style_path=self.config.config_dir,
 192                     threads=self.threads or default_threads,
 193                     dsn=self.config.get_libpq_dsn(),
 194                     flatnode_file=str(self.config.get_path('FLATNODE_FILE') or ''),
 195                     tablespaces=dict(slim_data=self.config.TABLESPACE_OSM_DATA,
 196                                      slim_index=self.config.TABLESPACE_OSM_INDEX,
 197                                      main_data=self.config.TABLESPACE_PLACE_DATA,
 198                                      main_index=self.config.TABLESPACE_PLACE_INDEX
 199                                     )
 200                    )
 201
 202
 203     def get_osm_file_list(self) -> Optional[List[Path]]:
 204         """ Return the --osm-file argument as a list of Paths or None
 205             if no argument was given. The function also checks if the files
 206             exist and raises a UsageError if one cannot be found.
 207         """
 208         if not self.osm_file:
 209             return None
 210
 211         files = [Path(f) for f in self.osm_file]
 212         for fname in files:
 213             if not fname.is_file():
 214                 LOG.fatal("OSM file '%s' does not exist.", fname)
 215                 raise UsageError('Cannot access file.')
 216
 217         return files