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
  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 'refresh'
 105     postcodes: bool
 106     word_tokens: bool
 107     word_counts: bool
 108     address_levels: bool
 109     functions: bool
 110     wiki_data: bool
 111     secondary_importance: bool
 112     importance: bool
 113     website: bool
 114     diffs: bool
 115     enable_debug_statements: bool
 116     data_object: Sequence[Tuple[str, int]]
 117     data_area: Sequence[Tuple[str, int]]
 118
 119     # Arguments to 'replication'
 120     init: bool
 121     update_functions: bool
 122     check_for_updates: bool
 123     once: bool
 124     catch_up: bool
 125     do_index: bool
 126
 127     # Arguments to 'serve'
 128     server: str
 129     engine: str
 130
 131     # Arguments to 'special-phrases
 132     import_from_wiki: bool
 133     import_from_csv: Optional[str]
 134     no_replace: bool
 135
 136     # Arguments to all query functions
 137     format: str
 138     addressdetails: bool
 139     extratags: bool
 140     namedetails: bool
 141     lang: Optional[str]
 142     polygon_output: Optional[str]
 143     polygon_threshold: Optional[float]
 144
 145     # Arguments to 'search'
 146     query: Optional[str]
 147     amenity: Optional[str]
 148     street: Optional[str]
 149     city: Optional[str]
 150     county: Optional[str]
 151     state: Optional[str]
 152     country: Optional[str]
 153     postalcode: Optional[str]
 154     countrycodes: Optional[str]
 155     exclude_place_ids: Optional[str]
 156     limit: int
 157     viewbox: Optional[str]
 158     bounded: bool
 159     dedupe: bool
 160
 161     # Arguments to 'reverse'
 162     lat: float
 163     lon: float
 164     zoom: Optional[int]
 165     layers: Optional[Sequence[str]]
 166
 167     # Arguments to 'lookup'
 168     ids: Sequence[str]
 169
 170     # Arguments to 'details'
 171     object_class: Optional[str]
 172     linkedplaces: bool
 173     hierarchy: bool
 174     keywords: bool
 175     polygon_geojson: bool
 176     group_hierarchy: bool
 177
 178
 179     def osm2pgsql_options(self, default_cache: int,
 180                           default_threads: int) -> Dict[str, Any]:
 181         """ Return the standard osm2pgsql options that can be derived
 182             from the command line arguments. The resulting dict can be
 183             further customized and then used in `run_osm2pgsql()`.
 184         """
 185         return dict(osm2pgsql=self.config.OSM2PGSQL_BINARY or self.config.lib_dir.osm2pgsql,
 186                     osm2pgsql_cache=self.osm2pgsql_cache or default_cache,
 187                     osm2pgsql_style=self.config.get_import_style_file(),
 188                     osm2pgsql_style_path=self.config.config_dir,
 189                     threads=self.threads or default_threads,
 190                     dsn=self.config.get_libpq_dsn(),
 191                     flatnode_file=str(self.config.get_path('FLATNODE_FILE') or ''),
 192                     tablespaces=dict(slim_data=self.config.TABLESPACE_OSM_DATA,
 193                                      slim_index=self.config.TABLESPACE_OSM_INDEX,
 194                                      main_data=self.config.TABLESPACE_PLACE_DATA,
 195                                      main_index=self.config.TABLESPACE_PLACE_INDEX
 196                                     )
 197                    )
 198
 199
 200     def get_osm_file_list(self) -> Optional[List[Path]]:
 201         """ Return the --osm-file argument as a list of Paths or None
 202             if no argument was given. The function also checks if the files
 203             exist and raises a UsageError if one cannot be found.
 204         """
 205         if not self.osm_file:
 206             return None
 207
 208         files = [Path(f) for f in self.osm_file]
 209         for fname in files:
 210             if not fname.is_file():
 211                 LOG.fatal("OSM file '%s' does not exist.", fname)
 212                 raise UsageError('Cannot access file.')
 213
 214         return files
 215
 216
 217     def get_geometry_output(self) -> napi.GeometryFormat:
 218         """ Get the requested geometry output format in a API-compatible
 219             format.
 220         """
 221         if not self.polygon_output:
 222             return napi.GeometryFormat.NONE
 223         if self.polygon_output == 'geojson':
 224             return napi.GeometryFormat.GEOJSON
 225         if self.polygon_output == 'kml':
 226             return napi.GeometryFormat.KML
 227         if self.polygon_output == 'svg':
 228             return napi.GeometryFormat.SVG
 229         if self.polygon_output == 'text':
 230             return napi.GeometryFormat.TEXT
 231
 232         try:
 233             return napi.GeometryFormat[self.polygon_output.upper()]
 234         except KeyError as exp:
 235             raise UsageError(f"Unknown polygon output format '{self.polygon_output}'.") from exp
 236
 237
 238     def get_locales(self, default: Optional[str]) -> napi.Locales:
 239         """ Get the locales from the language parameter.
 240         """
 241         if self.lang:
 242             return napi.Locales.from_accept_languages(self.lang)
 243         if default:
 244             return napi.Locales.from_accept_languages(default)
 245
 246         return napi.Locales()
 247
 248
 249     def get_layers(self, default: napi.DataLayer) -> Optional[napi.DataLayer]:
 250         """ Get the list of selected layers as a DataLayer enum.
 251         """
 252         if not self.layers:
 253             return default
 254
 255         return reduce(napi.DataLayer.__or__,
 256                       (napi.DataLayer[s.upper()] for s in self.layers))