/
config.py
218 lines (178 loc) · 9.96 KB
/
config.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
"""
Module for managing configuration data from `config.toml`
"""
from __future__ import annotations
import copy
import json
import os
import warnings
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Any
import tomli
import tomli_w
from .exceptions import SHDeprecationWarning
DEFAULT_PROFILE = "default-profile"
SH_PROFILE_ENV_VAR = "SH_PROFILE"
SH_CLIENT_ID_ENV_VAR = "SH_CLIENT_ID"
SH_CLIENT_SECRET_ENV_VAR = "SH_CLIENT_SECRET"
@dataclass(repr=False)
class _SHConfig:
instance_id: str = ""
sh_client_id: str = ""
sh_client_secret: str = ""
sh_base_url: str = "https://services.sentinel-hub.com"
sh_auth_base_url: str | None = None
sh_token_url: str = "https://services.sentinel-hub.com/auth/realms/main/protocol/openid-connect/token"
geopedia_wms_url: str = "https://service.geopedia.world"
geopedia_rest_url: str = "https://www.geopedia.world/rest"
aws_access_key_id: str = ""
aws_secret_access_key: str = ""
aws_session_token: str = ""
aws_metadata_url: str = "https://roda.sentinel-hub.com"
aws_s3_l1c_bucket: str = "sentinel-s2-l1c"
aws_s3_l2a_bucket: str = "sentinel-s2-l2a"
opensearch_url: str = "http://opensearch.sentinel-hub.com/resto/api/collections/Sentinel2"
max_wfs_records_per_query: int = 100
max_opensearch_records_per_query: int = 500 # pylint: disable=invalid-name
max_download_attempts: int = 4
download_sleep_time: float = 5.0
download_timeout_seconds: float = 120.0
number_of_download_processes: int = 1
def __post_init__(self) -> None:
if self.sh_auth_base_url is not None:
self.sh_token_url = self.sh_auth_base_url + "/oauth/token"
warnings.warn(
"The parameter `sh_auth_base_url` of `SHConfig` has been replaced with `sh_token_url`. Please"
" update your configuration, for now the parameters were automatically adjusted to `sh_token_url ="
" sh_auth_base_url + '/oauth/token'`.",
category=SHDeprecationWarning,
)
if self.max_wfs_records_per_query > 100:
raise ValueError("Value of config parameter `max_wfs_records_per_query` must be at most 100")
if self.max_opensearch_records_per_query > 500:
raise ValueError("Value of config parameter `max_opensearch_records_per_query` must be at most 500")
class SHConfig(_SHConfig):
"""A sentinelhub-py package configuration class.
The class reads the configurable settings from ``config.toml`` file on initialization:
- `instance_id`: An instance ID for Sentinel Hub service used for OGC requests.
- `sh_client_id`: User's OAuth client ID for Sentinel Hub service. Can be set via SH_CLIENT_ID environment
variable. The environment variable has precedence.
- `sh_client_secret`: User's OAuth client secret for Sentinel Hub service. Can be set via SH_CLIENT_SECRET
environment variable. The environment variable has precedence.
- `sh_base_url`: There exist multiple deployed instances of Sentinel Hub service, this parameter defines the
location of a specific service instance.
- `sh_token_url`: Url for Sentinel Hub Authentication service. Authentication is typically sent to the main
service deployment even if `sh_base_url` points to another deployment.
- `geopedia_wms_url`: Base url for Geopedia WMS services.
- `geopedia_rest_url`: Base url for Geopedia REST services.
- `aws_access_key_id`: Access key for AWS Requester Pays buckets.
- `aws_secret_access_key`: Secret access key for AWS Requester Pays buckets.
- `aws_session_token`: A session token for your AWS account. It is only needed when you are using temporary
credentials.
- `aws_metadata_url`: Base url for publicly available metadata files
- `aws_s3_l1c_bucket`: Name of Sentinel-2 L1C bucket at AWS s3 service.
- `aws_s3_l2a_bucket`: Name of Sentinel-2 L2A bucket at AWS s3 service.
- `opensearch_url`: Base url for Sentinelhub Opensearch service.
- `max_wfs_records_per_query`: Maximum number of records returned for each WFS query.
- `max_opensearch_records_per_query`: Maximum number of records returned for each Opensearch query.
- `max_download_attempts`: Maximum number of download attempts from a single URL until an error will be raised.
- `download_sleep_time`: Number of seconds to sleep between the first failed attempt and the next. Every next
attempt this number exponentially increases with factor `3`.
- `download_timeout_seconds`: Maximum number of seconds before download attempt is canceled.
- `number_of_download_processes`: Number of download processes, used to calculate rate-limit sleep time.
The location of `config.toml` for manual modification can be found with `SHConfig.get_config_location()`.
"""
CREDENTIALS = (
"instance_id",
"sh_client_id",
"sh_client_secret",
"aws_access_key_id",
"aws_secret_access_key",
"aws_session_token",
)
def __init__(self, profile: str | None = None, *, use_defaults: bool = False, **kwargs: Any):
"""
:param profile: Specifies which profile to load from the configuration file. Has precedence over the environment
variable `SH_USER_PROFILE`.
:param use_defaults: Does not load the configuration file, returns config object with defaults only.
:param kwargs: Any fields of `SHConfig` to be updated. Overrides settings from `config.toml` and environment.
"""
profile = self._get_profile(profile)
if not use_defaults:
env_kwargs = {
"sh_client_id": os.environ.get(SH_CLIENT_ID_ENV_VAR),
"sh_client_secret": os.environ.get(SH_CLIENT_SECRET_ENV_VAR),
}
env_kwargs = {k: v for k, v in env_kwargs.items() if v is not None}
# load from config.toml
loaded_kwargs = SHConfig.load(profile=profile).to_dict(mask_credentials=False)
kwargs = {**loaded_kwargs, **env_kwargs, **kwargs} # precedence: init params > env > loaded
super().__init__(**kwargs)
def __str__(self) -> str:
"""Content of `SHConfig` in json schema. Credentials are masked for safety."""
return json.dumps(self.to_dict(mask_credentials=True), indent=2)
def __repr__(self) -> str:
"""Representation of `SHConfig`. Credentials are masked for safety."""
config_dict = self.to_dict(mask_credentials=True)
content = ",\n ".join(f"{key}={value!r}" for key, value in config_dict.items())
return f"{self.__class__.__name__}(\n {content},\n)"
@staticmethod
def _get_profile(profile: str | None) -> str:
return profile if profile is not None else os.environ.get(SH_PROFILE_ENV_VAR, default=DEFAULT_PROFILE)
@classmethod
def load(cls, profile: str | None = None) -> SHConfig:
"""Loads configuration parameters from the config file at `SHConfig.get_config_location()`.
:param profile: Which profile to load from the configuration file.
"""
profile = cls._get_profile(profile)
filename = cls.get_config_location()
if not os.path.exists(filename):
cls(use_defaults=True).save() # store default configuration to standard location
with open(filename, "rb") as cfg_file:
configurations_dict = tomli.load(cfg_file)
if profile not in configurations_dict:
raise KeyError(f"Profile `{profile}` not found in configuration file.")
return cls(use_defaults=True, **configurations_dict[profile])
def save(self, profile: str | None = None) -> None:
"""Saves configuration parameters to the config file at `SHConfig.get_config_location()`.
:param profile: Under which profile to save the configuration.
"""
profile = self._get_profile(profile)
file_path = Path(self.get_config_location())
file_path.parent.mkdir(parents=True, exist_ok=True)
if file_path.exists():
with open(file_path, "rb") as cfg_file:
current_configuration = tomli.load(cfg_file)
else:
current_configuration = {}
current_configuration[profile] = self._get_dict_of_diffs_from_defaults()
with open(file_path, "wb") as cfg_file:
tomli_w.dump(current_configuration, cfg_file)
def _get_dict_of_diffs_from_defaults(self) -> dict[str, str | float]:
"""Returns a dictionary containing key: value pairs for parameters that have values different from defaults."""
current_profile_config = self.to_dict(mask_credentials=False)
default_values = SHConfig(use_defaults=True).to_dict(mask_credentials=False)
return {key: value for key, value in current_profile_config.items() if default_values[key] != value}
def copy(self) -> SHConfig:
"""Makes a copy of an instance of `SHConfig`"""
return copy.copy(self)
def to_dict(self, mask_credentials: bool = True) -> dict[str, str | float]:
"""Get a dictionary representation of the `SHConfig` class.
:param mask_credentials: Wether to mask fields containing credentials.
:return: A dictionary with configuration parameters
"""
config_params = asdict(self)
if mask_credentials:
for param in self.CREDENTIALS:
config_params[param] = self._mask_credentials(config_params[param])
return config_params
def _mask_credentials(self, value: str) -> str:
"""In case a parameter that holds credentials is given it will mask its value"""
hide_size = min(max(len(value) - 4, 10), len(value))
return "*" * hide_size + value[hide_size:]
@classmethod
def get_config_location(cls) -> str:
"""Returns the default location of the user configuration file on disk."""
user_folder = os.path.expanduser("~")
return os.path.join(user_folder, ".config", "sentinelhub", "config.toml")