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     addressdetails: bool
 141     extratags: bool
 142     namedetails: bool
 143     lang: Optional[str]
 144     polygon_output: Optional[str]
 145     polygon_threshold: Optional[float]
 146
 147     # Arguments to 'search'
 148     query: Optional[str]
 149     amenity: Optional[str]
 150     street: Optional[str]
 151     city: Optional[str]
 152     county: Optional[str]
 153     state: Optional[str]
 154     country: Optional[str]
 155     postalcode: Optional[str]
 156     countrycodes: Optional[str]
 157     exclude_place_ids: Optional[str]
 158     limit: int
 159     viewbox: Optional[str]
 160     bounded: bool
 161     dedupe: bool
 162
 163     # Arguments to 'reverse'
 164     lat: float
 165     lon: float
 166     zoom: Optional[int]
 167     layers: Optional[Sequence[str]]
 168
 169     # Arguments to 'lookup'
 170     ids: Sequence[str]
 171
 172     # Arguments to 'details'
 173     object_class: Optional[str]
 174     linkedplaces: bool
 175     hierarchy: bool
 176     keywords: bool
 177     polygon_geojson: bool
 178     group_hierarchy: bool
 179
 180
 181     def osm2pgsql_options(self, default_cache: int,
 182                           default_threads: int) -> Dict[str, Any]:
 183         """ Return the standard osm2pgsql options that can be derived
 184             from the command line arguments. The resulting dict can be
 185             further customized and then used in `run_osm2pgsql()`.
 186         """
 187         return dict(osm2pgsql=self.config.OSM2PGSQL_BINARY or self.config.lib_dir.osm2pgsql,
 188                     osm2pgsql_cache=self.osm2pgsql_cache or default_cache,
 189                     osm2pgsql_style=self.config.get_import_style_file(),
 190                     osm2pgsql_style_path=self.config.config_dir,
 191                     threads=self.threads or default_threads,
 192                     dsn=self.config.get_libpq_dsn(),
 193                     flatnode_file=str(self.config.get_path('FLATNODE_FILE') or ''),
 194                     tablespaces=dict(slim_data=self.config.TABLESPACE_OSM_DATA,
 195                                      slim_index=self.config.TABLESPACE_OSM_INDEX,
 196                                      main_data=self.config.TABLESPACE_PLACE_DATA,
 197                                      main_index=self.config.TABLESPACE_PLACE_INDEX
 198                                     )
 199                    )
 200
 201
 202     def get_osm_file_list(self) -> Optional[List[Path]]:
 203         """ Return the --osm-file argument as a list of Paths or None
 204             if no argument was given. The function also checks if the files
 205             exist and raises a UsageError if one cannot be found.
 206         """
 207         if not self.osm_file:
 208             return None
 209
 210         files = [Path(f) for f in self.osm_file]
 211         for fname in files:
 212             if not fname.is_file():
 213                 LOG.fatal("OSM file '%s' does not exist.", fname)
 214                 raise UsageError('Cannot access file.')
 215
 216         return files