From eaac2b818d9ba4360a72d0c9072806d1f619c0c9 Mon Sep 17 00:00:00 2001 From: Stephan Finkensieper Date: Wed, 12 Jun 2024 13:29:03 +0000 Subject: [PATCH] Add generic calibration coefficient selector --- satpy/readers/utils.py | 90 +++++++++++++++++++++++ satpy/tests/reader_tests/test_utils.py | 98 ++++++++++++++++++++++++++ 2 files changed, 188 insertions(+) diff --git a/satpy/readers/utils.py b/satpy/readers/utils.py index c1bf7c7497..aba50c5e2d 100644 --- a/satpy/readers/utils.py +++ b/satpy/readers/utils.py @@ -474,3 +474,93 @@ def remove_earthsun_distance_correction(reflectance, utc_date=None): with xr.set_options(keep_attrs=True): reflectance = reflectance / reflectance.dtype.type(sun_earth_dist * sun_earth_dist) return reflectance + + +class CalibrationCoefficientSelector: + """Helper for choosing coefficients out of multiple options.""" + + def __init__(self, coefs, modes=None, default="nominal", fallback=None, refl_threshold=1): + """Initialize the coefficient selector. + + Args: + coefs (dict): One set of calibration coefficients for each calibration + mode, for example :: + + { + "nominal": { + "ch1": nominal_coefs_ch1, + "ch2": nominal_coefs_ch2 + }, + "gsics": { + "ch2": gsics_coefs_ch2 + } + } + + The actual coefficients can be of any type (reader-specific). + + modes (dict): Desired calibration modes per channel type :: + + { + "reflective": "nominal", + "emissive": "gsics" + } + + or per channel :: + + { + "VIS006": "nominal", + "IR_108": "gsics" + } + + default (str): Default coefficients to be used if no mode has been + specified. Default: "nominal". + fallback (str): Fallback coefficients if the desired coefficients + are not available for some channel. + refl_threshold: Central wavelengths below/above this threshold are + considered reflective/emissive. Default is 1um. + """ + self.coefs = coefs + self.modes = modes or {} + self.default = default + self.fallback = fallback + self.refl_threshold = refl_threshold + if self.default not in self.coefs: + raise KeyError("Need at least default coefficients") + if self.fallback and self.fallback not in self.coefs: + raise KeyError("No fallback coefficients") + + def get_coefs(self, dataset_id): + """Get calibration coefficients for the given dataset. + + Args: + dataset_id (DataID): Desired dataset + """ + mode = self._get_mode(dataset_id) + return self._get_coefs(dataset_id, mode) + + def _get_coefs(self, dataset_id, mode): + ds_name = dataset_id["name"] + try: + return self.coefs[mode][ds_name] + except KeyError: + if self.fallback: + return self.coefs[self.fallback][ds_name] + raise KeyError(f"No calibration coefficients for {ds_name}") + + def _get_mode(self, dataset_id): + try: + return self._get_mode_for_channel(dataset_id) + except KeyError: + return self._get_mode_for_channel_type(dataset_id) + + def _get_mode_for_channel(self, dataset_id): + return self.modes[dataset_id["name"]] + + def _get_mode_for_channel_type(self, dataset_id): + ch_type = self._get_channel_type(dataset_id) + return self.modes.get(ch_type, self.default) + + def _get_channel_type(self, dataset_id): + if dataset_id["wavelength"].central < self.refl_threshold: + return "reflective" + return "emissive" diff --git a/satpy/tests/reader_tests/test_utils.py b/satpy/tests/reader_tests/test_utils.py index ba43688b76..26d709477a 100644 --- a/satpy/tests/reader_tests/test_utils.py +++ b/satpy/tests/reader_tests/test_utils.py @@ -34,6 +34,8 @@ from satpy.readers import FSFile from satpy.readers import utils as hf +from satpy.readers.utils import CalibrationCoefficientSelector +from satpy.tests.utils import make_dataid class TestHelpers(unittest.TestCase): @@ -512,3 +514,99 @@ def test_generic_open_binary(tmp_path, data, filename, mode): read_binary_data = f.read() assert read_binary_data == dummy_data + + +CALIB_COEFS = { + "nominal": { + "ch1": {"slope": 0.1, "offset": 1}, + "ch2": {"slope": 0.2, "offset": 2} + }, + "mode1": { + "ch1": {"slope": 0.3, "offset": 3}, + }, + "mode2": { + "ch2": {"slope": 0.5, "offset": 5}, + } +} + + +class TestCalibrationCoefficientSelector: + """Test selection of calibration coefficients.""" + + @pytest.fixture(name="ch1") + def fixture_ch1(self): + """Make fake data ID.""" + return make_dataid(name="ch1", wavelength=(0.6, 0.7, 0.8)) + + @pytest.fixture(name="ch2") + def fixture_ch2(self): + """Make fake data ID.""" + return make_dataid(name="ch2", wavelength=(10, 11, 12)) + + @pytest.fixture(name="dataset_ids") + def fixture_dataset_ids(self, ch1, ch2): + """Make fake data IDs.""" + return [ch1, ch2] + + @pytest.mark.parametrize( + ("calib_modes", "expected"), + [ + ( + None, + CALIB_COEFS["nominal"] + ), + ( + {"reflective": "mode1"}, + { + "ch1": CALIB_COEFS["mode1"]["ch1"], + "ch2": CALIB_COEFS["nominal"]["ch2"] + } + ), + ( + {"reflective": "mode1", "emissive": "mode2"}, + { + "ch1": CALIB_COEFS["mode1"]["ch1"], + "ch2": CALIB_COEFS["mode2"]["ch2"] + } + ), + ( + {"ch1": "mode1"}, + { + "ch1": CALIB_COEFS["mode1"]["ch1"], + "ch2": CALIB_COEFS["nominal"]["ch2"] + } + ), + ] + ) + def test_get_coefs(self, dataset_ids, calib_modes, expected): + """Test getting calibration coefficients.""" + s = CalibrationCoefficientSelector(CALIB_COEFS, calib_modes) + coefs = { + dataset_id["name"]: s.get_coefs(dataset_id) + for dataset_id in dataset_ids + } + assert coefs == expected + + def test_missing_coefs(self, ch1): + """Test handling of missing coefficients.""" + calib_modes = {"reflective": "mode2"} + s = CalibrationCoefficientSelector(CALIB_COEFS, calib_modes) + with pytest.raises(KeyError, match="No calibration *"): + s.get_coefs(ch1) + + def test_fallback_to_nominal(self, ch1): + """Test falling back to nominal coefficients.""" + calib_modes = {"reflective": "mode2"} + s = CalibrationCoefficientSelector(CALIB_COEFS, calib_modes, fallback="nominal") + coefs = s.get_coefs(ch1) + assert coefs == {"slope": 0.1, "offset": 1} + + def test_no_default_coefs(self): + """Test initialization without default coefficients.""" + with pytest.raises(KeyError, match="Need at least *"): + CalibrationCoefficientSelector({}) + + def test_no_fallback(self): + """Test initialization without fallback coefficients.""" + with pytest.raises(KeyError, match="No fallback coefficients"): + CalibrationCoefficientSelector({"nominal": 123}, fallback="foo")