Add generic calibration coefficient selector

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"

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")