src/nominatim_db/clicmd/args.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2024 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_core.errors import UsageError
  17 from nominatim_core.config import Configuration
  18 from nominatim_core.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))