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 nominatim_core.errors import UsageError
  16 from nominatim_core.config import Configuration
  17 from nominatim_core.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