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