forked from pytroll/satpy
/
config_loader.py
318 lines (255 loc) · 12.1 KB
/
config_loader.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
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2020-2021 Satpy developers
#
# This file is part of satpy.
#
# satpy is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# satpy is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# satpy. If not, see <http://www.gnu.org/licenses/>.
"""Classes for loading compositor and modifier configuration files."""
from __future__ import annotations
import logging
import os
import warnings
from functools import lru_cache, update_wrapper
from typing import Callable, Iterable
import yaml
from yaml import UnsafeLoader
import satpy
from satpy import DataID, DataQuery
from satpy._config import config_search_paths, get_entry_points_config_dirs, glob_config
from satpy.dataset.dataid import minimal_default_keys_config
from satpy.utils import recursive_dict_update
logger = logging.getLogger(__name__)
def _convert_dep_info_to_data_query(dep_info):
key_item = dep_info.copy()
key_item.pop('prerequisites', None)
key_item.pop('optional_prerequisites', None)
if 'modifiers' in key_item:
key_item['modifiers'] = tuple(key_item['modifiers'])
key = DataQuery.from_dict(key_item)
return key
class _CompositeConfigHelper:
"""Helper class for parsing composite configurations.
The provided `loaded_compositors` dictionary is updated inplace.
"""
def __init__(self, loaded_compositors, sensor_id_keys):
self.loaded_compositors = loaded_compositors
self.sensor_id_keys = sensor_id_keys
def _create_comp_from_info(self, composite_info, loader):
key = DataID(self.sensor_id_keys, **composite_info)
comp = loader(_satpy_id=key, **composite_info)
return key, comp
def _handle_inline_comp_dep(self, dep_info, dep_num, parent_name):
# Create an unique temporary name for the composite
sub_comp_name = '_' + parent_name + '_dep_{}'.format(dep_num)
dep_info['name'] = sub_comp_name
self._load_config_composite(dep_info)
@staticmethod
def _get_compositor_loader_from_config(composite_name, composite_info):
try:
loader = composite_info.pop('compositor')
except KeyError:
raise ValueError("'compositor' key missing or empty for '{}'. Option keys = {}".format(
composite_name, str(composite_info.keys())))
return loader
def _process_composite_deps(self, composite_info):
dep_num = -1
for prereq_type in ['prerequisites', 'optional_prerequisites']:
prereqs = []
for dep_info in composite_info.get(prereq_type, []):
dep_num += 1
if not isinstance(dep_info, dict):
prereqs.append(dep_info)
continue
elif 'compositor' in dep_info:
self._handle_inline_comp_dep(
dep_info, dep_num, composite_info['name'])
prereq_key = _convert_dep_info_to_data_query(dep_info)
prereqs.append(prereq_key)
composite_info[prereq_type] = prereqs
def _load_config_composite(self, composite_info):
composite_name = composite_info['name']
loader = self._get_compositor_loader_from_config(composite_name, composite_info)
self._process_composite_deps(composite_info)
key, comp = self._create_comp_from_info(composite_info, loader)
self.loaded_compositors[key] = comp
def _load_config_composites(self, configured_composites):
for composite_name, composite_info in configured_composites.items():
composite_info['name'] = composite_name
self._load_config_composite(composite_info)
def parse_config(self, configured_composites, composite_configs):
"""Parse composite configuration dictionary."""
try:
self._load_config_composites(configured_composites)
except (ValueError, KeyError):
raise RuntimeError("Failed to load composites from configs "
"'{}'".format(composite_configs))
class _ModifierConfigHelper:
"""Helper class for parsing modifier configurations.
The provided `loaded_modifiers` dictionary is updated inplace.
"""
def __init__(self, loaded_modifiers, sensor_id_keys):
self.loaded_modifiers = loaded_modifiers
self.sensor_id_keys = sensor_id_keys
@staticmethod
def _get_modifier_loader_from_config(modifier_name, modifier_info):
try:
loader = modifier_info.pop('modifier', None)
if loader is None:
loader = modifier_info.pop('compositor')
warnings.warn("Modifier '{}' uses deprecated 'compositor' "
"key to point to Python class, replace "
"with 'modifier'.".format(modifier_name))
except KeyError:
raise ValueError("'modifier' key missing or empty for '{}'. Option keys = {}".format(
modifier_name, str(modifier_info.keys())))
return loader
def _process_modifier_deps(self, modifier_info):
for prereq_type in ['prerequisites', 'optional_prerequisites']:
prereqs = []
for dep_info in modifier_info.get(prereq_type, []):
if not isinstance(dep_info, dict):
prereqs.append(dep_info)
continue
prereq_key = _convert_dep_info_to_data_query(dep_info)
prereqs.append(prereq_key)
modifier_info[prereq_type] = prereqs
def _load_config_modifier(self, modifier_info):
modifier_name = modifier_info['name']
loader = self._get_modifier_loader_from_config(modifier_name, modifier_info)
self._process_modifier_deps(modifier_info)
self.loaded_modifiers[modifier_name] = (loader, modifier_info)
def _load_config_modifiers(self, configured_modifiers):
for modifier_name, modifier_info in configured_modifiers.items():
modifier_info['name'] = modifier_name
self._load_config_modifier(modifier_info)
def parse_config(self, configured_modifiers, composite_configs):
"""Parse modifier configuration dictionary."""
try:
self._load_config_modifiers(configured_modifiers)
except (ValueError, KeyError):
raise RuntimeError("Failed to load modifiers from configs "
"'{}'".format(composite_configs))
def _load_config(composite_configs):
if not isinstance(composite_configs, (list, tuple)):
composite_configs = [composite_configs]
conf = {}
for composite_config in composite_configs:
with open(composite_config, 'r', encoding='utf-8') as conf_file:
conf = recursive_dict_update(conf, yaml.load(conf_file, Loader=UnsafeLoader))
try:
sensor_name = conf['sensor_name']
except KeyError:
logger.debug('No "sensor_name" tag found in %s, skipping.',
composite_configs)
return {}, {}, {}
sensor_compositors = {}
sensor_modifiers = {}
dep_id_keys = None
sensor_deps = sensor_name.split('/')[:-1]
if sensor_deps:
# get dependent
for sensor_dep in sensor_deps:
dep_comps, dep_mods, dep_id_keys = load_compositor_configs_for_sensor(sensor_dep)
# the last parent should include all of its parents so only add the last one
sensor_compositors.update(dep_comps)
sensor_modifiers.update(dep_mods)
id_keys = _get_sensor_id_keys(conf, dep_id_keys)
mod_config_helper = _ModifierConfigHelper(sensor_modifiers, id_keys)
configured_modifiers = conf.get('modifiers', {})
mod_config_helper.parse_config(configured_modifiers, composite_configs)
comp_config_helper = _CompositeConfigHelper(sensor_compositors, id_keys)
configured_composites = conf.get('composites', {})
comp_config_helper.parse_config(configured_composites, composite_configs)
return sensor_compositors, sensor_modifiers, id_keys
def _get_sensor_id_keys(conf, parent_id_keys):
try:
id_keys = conf['composite_identification_keys']
except KeyError:
id_keys = parent_id_keys
if not id_keys:
id_keys = minimal_default_keys_config
return id_keys
def _lru_cache_with_config_path(func: Callable):
"""Use lru_cache but include satpy's current config_path."""
@lru_cache()
def _call_without_config_path_wrapper(sensor_name, _):
return func(sensor_name)
def _add_config_path_wrapper(sensor_name: str):
config_path = satpy.config.get("config_path")
# make sure config_path is hashable, but keep original order since it matters
config_path = tuple(config_path)
return _call_without_config_path_wrapper(sensor_name, config_path)
wrapper = update_wrapper(_add_config_path_wrapper, func)
wrapper = _update_cached_wrapper(wrapper, _call_without_config_path_wrapper)
return wrapper
def _update_cached_wrapper(wrapper, cached_func):
for meth_name in ("cache_clear", "cache_parameters", "cache_info"):
if hasattr(cached_func, meth_name):
setattr(wrapper, meth_name, getattr(cached_func, meth_name))
return wrapper
@_lru_cache_with_config_path
def load_compositor_configs_for_sensor(sensor_name: str) -> tuple[dict[str, dict], dict[str, dict], dict]:
"""Load compositor, modifier, and DataID key information from configuration files for the specified sensor.
Args:
sensor_name: Sensor name that has matching ``sensor_name.yaml``
config files.
Returns:
(comps, mods, data_id_keys): Where `comps` is a dictionary:
composite ID -> compositor object
And `mods` is a dictionary:
modifier name -> (modifier class, modifiers options)
Add `data_id_keys` is a dictionary:
DataID key -> key properties
"""
config_filename = sensor_name + ".yaml"
logger.debug("Looking for composites config file %s", config_filename)
paths = get_entry_points_config_dirs('satpy.composites')
composite_configs = config_search_paths(
os.path.join("composites", config_filename),
search_dirs=paths, check_exists=True)
if not composite_configs:
logger.debug("No composite config found called %s",
config_filename)
return {}, {}, minimal_default_keys_config
return _load_config(composite_configs)
def load_compositor_configs_for_sensors(sensor_names: Iterable[str]) -> tuple[dict[str, dict], dict[str, dict]]:
"""Load compositor and modifier configuration files for the specified sensors.
Args:
sensor_names (list of strings): Sensor names that have matching
``sensor_name.yaml`` config files.
Returns:
(comps, mods): Where `comps` is a dictionary:
sensor_name -> composite ID -> compositor object
And `mods` is a dictionary:
sensor_name -> modifier name -> (modifier class,
modifiers options)
"""
comps = {}
mods = {}
for sensor_name in sensor_names:
sensor_comps, sensor_mods = load_compositor_configs_for_sensor(sensor_name)[:2]
comps[sensor_name] = sensor_comps
mods[sensor_name] = sensor_mods
return comps, mods
def all_composite_sensors():
"""Get all sensor names from available composite configs."""
paths = get_entry_points_config_dirs('satpy.composites')
composite_configs = glob_config(
os.path.join("composites", "*.yaml"),
search_dirs=paths)
yaml_names = set([os.path.splitext(os.path.basename(fn))[0]
for fn in composite_configs])
non_sensor_yamls = ('visir',)
sensor_names = [x for x in yaml_names if x not in non_sensor_yamls]
return sensor_names