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