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