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