Source code for polar2grid.resample.resample_decisions
#!/usr/bin/env python
# encoding: utf-8
# Copyright (C) 2021 Space Science and Engineering Center (SSEC),
# University of Wisconsin-Madison.
#
# This program 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.
#
# This program 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 this program. If not, see <http://www.gnu.org/licenses/>.
#
# This file is part of the polar2grid software package. Polar2grid takes
# satellite observation data, remaps it, and writes it to a file format for
# input into another program.
# Documentation: http://www.ssec.wisc.edu/software/polar2grid/
"""Decision utilties for resampling."""
import logging
import os
import yaml
from pyresample.geometry import SwathDefinition
from satpy._config import config_search_paths
from satpy.utils import recursive_dict_update
from satpy.writers import DecisionTree
from polar2grid.utils.legacy_compat import get_sensor_alias
logger = logging.getLogger(__name__)
[docs]
class ResamplerDecisionTree(DecisionTree):
"""Helper class to determine resampler algorithm and other options."""
def __init__(self, *decision_dicts, **kwargs):
"""Init the decision tree."""
match_keys = kwargs.pop(
"match_keys",
(
"name",
"reader",
"platform_name",
"sensor",
"area_type",
"standard_name",
"units",
),
)
self.prefix = kwargs.pop("config_section", "resampling")
multival_keys = kwargs.pop("multival_keys", ["sensor"])
super(ResamplerDecisionTree, self).__init__(decision_dicts, match_keys, multival_keys)
[docs]
@classmethod
def from_configs(cls, config_filename="resampling.yaml"):
config_paths = config_search_paths(config_filename)
return cls(*config_paths)
[docs]
def add_config_to_tree(self, *config_files):
"""Add configuration to tree."""
conf = {}
for config_file in config_files:
if os.path.isfile(config_file):
with open(config_file) as fd:
resample_config = yaml.load(fd, Loader=yaml.UnsafeLoader)
if resample_config is None:
# empty file
continue
resampling_section = resample_config.get(self.prefix, {})
if not resampling_section:
logging.debug("Config '{}' has no '{}' section or it is empty".format(config_file, self.prefix))
continue
conf = recursive_dict_update(conf, resampling_section)
elif isinstance(config_file, dict):
conf = recursive_dict_update(conf, config_file)
else:
logger.debug("Loading resampling config string")
d = yaml.load(config_file, Loader=yaml.UnsafeLoader)
if not isinstance(d, dict):
raise ValueError("YAML file doesn't exist or string is not YAML dict: {}".format(config_file))
conf = recursive_dict_update(conf, d)
self._build_tree(conf)
[docs]
def find_match(self, **query_dict):
"""Find a match."""
query_dict["area_type"] = "swath" if isinstance(query_dict["area"], SwathDefinition) else "area"
query_dict["sensor"] = get_sensor_alias(query_dict.get("sensor"))
try:
return super().find_match(**query_dict)
except KeyError as err:
# give a more understandable error message
raise KeyError(
f"No resampling configuration found for {query_dict['area_type']=} | {query_dict['name']=}"
) from err