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 functools import reduce
  14 from pathlib import Path
  15
  16 from nominatim_core.errors import UsageError
  17 from nominatim_core.config import Configuration
  18 from nominatim_core.typing import Protocol
  19
  20 LOG = logging.getLogger()
  21
  22 class Subcommand(Protocol):
  23     """
  24     Interface to be implemented by classes implementing a CLI subcommand.
  25     """
  26
  27     def add_args(self, parser: argparse.ArgumentParser) -> None:
  28         """
  29         Fill the given parser for the subcommand with the appropriate
  30         parameters.
  31         """
  32
  33     def run(self, args: 'NominatimArgs') -> int:
  34         """
  35         Run the subcommand with the given parsed arguments.
  36         """
  37
  38
  39 class NominatimArgs:
  40     """ Customized namespace class for the nominatim command line tool
  41         to receive the command-line arguments.
  42     """
  43     # Basic environment set by root program.
  44     config: Configuration
  45     project_dir: 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     clean_deleted: str
  75     analyse_indexing: bool
  76     target: Optional[str]
  77     osm_id: Optional[str]
  78     place_id: Optional[int]
  79
  80     # Arguments to 'import'
  81     osm_file: List[str]
  82     continue_at: Optional[str]
  83     reverse_only: bool
  84     no_partitions: bool
  85     no_updates: bool
  86     offline: bool
  87     ignore_errors: bool
  88     index_noanalyse: bool
  89     prepare_database: bool
  90
  91     # Arguments to 'index'
  92     boundaries_only: bool
  93     no_boundaries: bool
  94     minrank: int
  95     maxrank: int
  96
  97     # Arguments to 'export'
  98     output_type: str
  99     output_format: str
 100     output_all_postcodes: bool
 101     language: Optional[str]
 102     restrict_to_country: Optional[str]
 103
 104     # Arguments to 'convert'
 105     output: Path
 106
 107     # Arguments to 'refresh'
 108     postcodes: bool
 109     word_tokens: bool
 110     word_counts: bool
 111     address_levels: bool
 112     functions: bool
 113     wiki_data: bool
 114     secondary_importance: bool
 115     importance: bool
 116     website: bool
 117     diffs: bool
 118     enable_debug_statements: bool
 119     data_object: Sequence[Tuple[str, int]]
 120     data_area: Sequence[Tuple[str, int]]
 121
 122     # Arguments to 'replication'
 123     init: bool
 124     update_functions: bool
 125     check_for_updates: bool
 126     once: bool
 127     catch_up: bool
 128     do_index: bool
 129
 130     # Arguments to 'serve'
 131     server: str
 132     engine: str
 133
 134     # Arguments to 'special-phrases
 135     import_from_wiki: bool
 136     import_from_csv: Optional[str]
 137     no_replace: bool
 138
 139     # Arguments to all query functions
 140     format: str
 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