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 functools import reduce
  14 from pathlib import Path
  15
  16 from nominatim.errors import UsageError
  17 from nominatim.config import Configuration
  18 from nominatim.typing import Protocol
  19 import nominatim.api as napi
  20
  21 LOG = logging.getLogger()
  22
  23 class Subcommand(Protocol):
  24     """
  25     Interface to be implemented by classes implementing a CLI subcommand.
  26     """
  27
  28     def add_args(self, parser: argparse.ArgumentParser) -> None:
  29         """
  30         Fill the given parser for the subcommand with the appropriate
  31         parameters.
  32         """
  33
  34     def run(self, args: 'NominatimArgs') -> int:
  35         """
  36         Run the subcommand with the given parsed arguments.
  37         """
  38
  39
  40 class NominatimArgs:
  41     """ Customized namespace class for the nominatim command line tool
  42         to receive the command-line arguments.
  43     """
  44     # Basic environment set by root program.
  45     config: Configuration
  46     project_dir: Path
  47
  48     # Global switches
  49     version: bool
  50     subcommand: Optional[str]
  51     command: Subcommand
  52
  53     # Shared parameters
  54     osm2pgsql_cache: Optional[int]
  55     socket_timeout: int
  56
  57     # Arguments added to all subcommands.
  58     verbose: int
  59     threads: Optional[int]
  60
  61     # Arguments to 'add-data'
  62     file: Optional[str]
  63     diff: Optional[str]
  64     node: Optional[int]
  65     way: Optional[int]
  66     relation: Optional[int]
  67     tiger_data: Optional[str]
  68     use_main_api: bool
  69
  70     # Arguments to 'admin'
  71     warm: bool
  72     check_database: bool
  73     migrate: bool
  74     collect_os_info: bool
  75     clean_deleted: str
  76     analyse_indexing: bool
  77     target: Optional[str]
  78     osm_id: Optional[str]
  79     place_id: Optional[int]
  80
  81     # Arguments to 'import'
  82     osm_file: List[str]
  83     continue_at: Optional[str]
  84     reverse_only: bool
  85     no_partitions: bool
  86     no_updates: bool
  87     offline: bool
  88     ignore_errors: bool
  89     index_noanalyse: bool
  90     only_import_data: bool
  91     only_prepare_database: bool
  92
  93     # Arguments to 'index'
  94     boundaries_only: bool
  95     no_boundaries: bool
  96     minrank: int
  97     maxrank: int
  98
  99     # Arguments to 'export'
 100     output_type: str
 101     output_format: str
 102     output_all_postcodes: bool
 103     language: Optional[str]
 104     restrict_to_country: Optional[str]
 105
 106     # Arguments to 'convert'
 107     output: Path
 108
 109     # Arguments to 'refresh'
 110     postcodes: bool
 111     word_tokens: bool
 112     word_counts: bool
 113     address_levels: bool
 114     functions: bool
 115     wiki_data: bool
 116     secondary_importance: bool
 117     importance: bool
 118     website: bool
 119     diffs: bool
 120     enable_debug_statements: bool
 121     data_object: Sequence[Tuple[str, int]]
 122     data_area: Sequence[Tuple[str, int]]
 123
 124     # Arguments to 'replication'
 125     init: bool
 126     update_functions: bool
 127     check_for_updates: bool
 128     once: bool
 129     catch_up: bool
 130     do_index: bool
 131
 132     # Arguments to 'serve'
 133     server: str
 134     engine: str
 135
 136     # Arguments to 'special-phrases
 137     import_from_wiki: bool
 138     import_from_csv: Optional[str]
 139     no_replace: bool
 140
 141     # Arguments to all query functions
 142     format: str
 143     addressdetails: bool
 144     extratags: bool
 145     namedetails: bool
 146     lang: Optional[str]
 147     polygon_output: Optional[str]
 148     polygon_threshold: Optional[float]
 149
 150     # Arguments to 'search'
 151     query: Optional[str]
 152     amenity: Optional[str]
 153     street: Optional[str]
 154     city: Optional[str]
 155     county: Optional[str]
 156     state: Optional[str]
 157     country: Optional[str]
 158     postalcode: Optional[str]
 159     countrycodes: Optional[str]
 160     exclude_place_ids: Optional[str]
 161     limit: int
 162     viewbox: Optional[str]
 163     bounded: bool
 164     dedupe: bool
 165
 166     # Arguments to 'reverse'
 167     lat: float
 168     lon: float
 169     zoom: Optional[int]
 170     layers: Optional[Sequence[str]]
 171
 172     # Arguments to 'lookup'
 173     ids: Sequence[str]
 174
 175     # Arguments to 'details'
 176     object_class: Optional[str]
 177     linkedplaces: bool
 178     hierarchy: bool
 179     keywords: bool
 180     polygon_geojson: bool
 181     group_hierarchy: bool
 182
 183
 184     def osm2pgsql_options(self, default_cache: int,
 185                           default_threads: int) -> Dict[str, Any]:
 186         """ Return the standard osm2pgsql options that can be derived
 187             from the command line arguments. The resulting dict can be
 188             further customized and then used in `run_osm2pgsql()`.
 189         """
 190         return dict(osm2pgsql=self.config.OSM2PGSQL_BINARY or self.config.lib_dir.osm2pgsql,
 191                     osm2pgsql_cache=self.osm2pgsql_cache or default_cache,
 192                     osm2pgsql_style=self.config.get_import_style_file(),
 193                     osm2pgsql_style_path=self.config.config_dir,
 194                     threads=self.threads or default_threads,
 195                     dsn=self.config.get_libpq_dsn(),
 196                     flatnode_file=str(self.config.get_path('FLATNODE_FILE') or ''),
 197                     tablespaces=dict(slim_data=self.config.TABLESPACE_OSM_DATA,
 198                                      slim_index=self.config.TABLESPACE_OSM_INDEX,
 199                                      main_data=self.config.TABLESPACE_PLACE_DATA,
 200                                      main_index=self.config.TABLESPACE_PLACE_INDEX
 201                                     )
 202                    )
 203
 204
 205     def get_osm_file_list(self) -> Optional[List[Path]]:
 206         """ Return the --osm-file argument as a list of Paths or None
 207             if no argument was given. The function also checks if the files
 208             exist and raises a UsageError if one cannot be found.
 209         """
 210         if not self.osm_file:
 211             return None
 212
 213         files = [Path(f) for f in self.osm_file]
 214         for fname in files:
 215             if not fname.is_file():
 216                 LOG.fatal("OSM file '%s' does not exist.", fname)
 217                 raise UsageError('Cannot access file.')
 218
 219         return files
 220
 221
 222     def get_geometry_output(self) -> napi.GeometryFormat:
 223         """ Get the requested geometry output format in a API-compatible
 224             format.
 225         """
 226         if not self.polygon_output:
 227             return napi.GeometryFormat.NONE
 228         if self.polygon_output == 'geojson':
 229             return napi.GeometryFormat.GEOJSON
 230         if self.polygon_output == 'kml':
 231             return napi.GeometryFormat.KML
 232         if self.polygon_output == 'svg':
 233             return napi.GeometryFormat.SVG
 234         if self.polygon_output == 'text':
 235             return napi.GeometryFormat.TEXT
 236
 237         try:
 238             return napi.GeometryFormat[self.polygon_output.upper()]
 239         except KeyError as exp:
 240             raise UsageError(f"Unknown polygon output format '{self.polygon_output}'.") from exp
 241
 242
 243     def get_locales(self, default: Optional[str]) -> napi.Locales:
 244         """ Get the locales from the language parameter.
 245         """
 246         if self.lang:
 247             return napi.Locales.from_accept_languages(self.lang)
 248         if default:
 249             return napi.Locales.from_accept_languages(default)
 250
 251         return napi.Locales()
 252
 253
 254     def get_layers(self, default: napi.DataLayer) -> Optional[napi.DataLayer]:
 255         """ Get the list of selected layers as a DataLayer enum.
 256         """
 257         if not self.layers:
 258             return default
 259
 260         return reduce(napi.DataLayer.__or__,
 261                       (napi.DataLayer[s.upper()] for s in self.layers))