diff --git a/satpy/readers/seviri_base.py b/satpy/readers/seviri_base.py index c9ad563899..a262b2ff58 100644 --- a/satpy/readers/seviri_base.py +++ b/satpy/readers/seviri_base.py @@ -201,14 +201,15 @@ import datetime as dt import warnings +from collections import namedtuple import dask.array as da import numpy as np import pyproj from numpy.polynomial.chebyshev import Chebyshev +import satpy.readers.utils as utils from satpy.readers.eum_base import issue_revision, time_cds_short -from satpy.readers.utils import apply_earthsun_distance_correction from satpy.utils import get_legacy_chunk_size CHUNK_SIZE = get_legacy_chunk_size() @@ -436,41 +437,6 @@ } -def get_meirink_slope(meirink_coefs, acquisition_time): - """Compute the slope for the visible channel calibration according to Meirink 2013. - - S = A + B * 1.e-3* Day - - S is here in µW m-2 sr-1 (cm-1)-1 - - EUMETSAT calibration is given in mW m-2 sr-1 (cm-1)-1, so an extra factor of 1/1000 must - be applied. - """ - A = meirink_coefs[0] - B = meirink_coefs[1] - delta_t = (acquisition_time - MEIRINK_EPOCH).total_seconds() - S = A + B * delta_t / (3600*24) / 1000. - return S/1000 - - -def should_apply_meirink(calib_mode, channel_name): - """Decide whether to use the Meirink calibration coefficients.""" - return "MEIRINK" in calib_mode and channel_name in ["VIS006", "VIS008", "IR_016"] - - -class MeirinkCalibrationHandler: - """Re-calibration of the SEVIRI visible channels slope (see Meirink 2013).""" - - def __init__(self, calib_mode): - """Initialize the calibration handler.""" - self.coefs = MEIRINK_COEFS[calib_mode.split("-")[1]] - - def get_slope(self, platform, channel, time): - """Return the slope using the provided calibration coefficients.""" - coefs = self.coefs[platform][channel] - return get_meirink_slope(coefs, time) - - def get_cds_time(days, msecs): """Compute timestamp given the days since epoch and milliseconds of the day. @@ -663,7 +629,11 @@ def vis_calibrate(self, data, solar_irradiance): reflectances for SEVIRI warm channels: https://www-cdn.eumetsat.int/files/2020-04/pdf_msg_seviri_rad2refl.pdf """ reflectance = np.pi * data * 100.0 / solar_irradiance - return apply_earthsun_distance_correction(reflectance, self._scan_time) + return utils.apply_earthsun_distance_correction(reflectance, self._scan_time) + + +CalibParams = namedtuple("CalibParams", ["mode", "internal_coefs", "external_coefs", "radiance_type"]) +ScanParams = namedtuple("ScanParams", ["platform_id", "channel_name", "scan_time"]) class SEVIRICalibrationHandler: @@ -673,23 +643,22 @@ class SEVIRICalibrationHandler: calibration algorithm. """ - def __init__(self, platform_id, channel_name, coefs, calib_mode, scan_time): + def __init__(self, calib_params, scan_params): """Initialize the calibration handler.""" - self._platform_id = platform_id - self._channel_name = channel_name - self._coefs = coefs - self._calib_mode = calib_mode.upper() - self._scan_time = scan_time + self._calib_params = calib_params + self._scan_params = scan_params self._algo = SEVIRICalibrationAlgorithm( - platform_id=self._platform_id, - scan_time=self._scan_time + platform_id=scan_params.platform_id, + scan_time=scan_params.scan_time ) + self._check_calib_mode(calib_params.mode) + def _check_calib_mode(self, calib_mode): valid_modes = ("NOMINAL", "GSICS", "MEIRINK-2023") - if self._calib_mode not in valid_modes: + if calib_mode not in valid_modes: raise ValueError( "Invalid calibration mode: {}. Choose one of {}".format( - self._calib_mode, valid_modes) + calib_mode, valid_modes) ) def calibrate(self, data, calibration): @@ -698,57 +667,42 @@ def calibrate(self, data, calibration): res = data elif calibration in ["radiance", "reflectance", "brightness_temperature"]: - gain, offset = self.get_gain_offset() + coefs = self.get_coefs() res = self._algo.convert_to_radiance( - data.astype(np.float32), np.float32(gain), np.float32(offset) + data.astype(np.float32), + np.float32(coefs["coefs"]["gain"]), + np.float32(coefs["coefs"]["offset"]) ) else: raise ValueError( "Invalid calibration {} for channel {}".format( - calibration, self._channel_name + calibration, self._scan_params.channel_name ) ) if calibration == "reflectance": - solar_irradiance = CALIB[self._platform_id][self._channel_name]["F"] + solar_irradiance = CALIB[self._scan_params.platform_id][self._scan_params.channel_name]["F"] res = self._algo.vis_calibrate(res, solar_irradiance) elif calibration == "brightness_temperature": res = self._algo.ir_calibrate( - res, self._channel_name, self._coefs["radiance_type"] + res, self._scan_params.channel_name, self._calib_params.radiance_type ) - return res - def get_gain_offset(self): - """Get gain & offset for calibration from counts to radiance. + def get_coefs(self): + """Get calibration coefficients.""" + picker = utils.CalibrationCoefficientPicker(self._calib_params.internal_coefs, + self._get_calib_wishlist(), + default="NOMINAL", + fallback="NOMINAL") + return picker.get_coefs(self._scan_params.channel_name) - Choices for internal coefficients are nominal or GSICS. If no - GSICS coefficients are available for a certain channel, fall back to - nominal coefficients. External coefficients take precedence over - internal coefficients. - """ - coefs = self._coefs["coefs"] - - # Select internal coefficients for the given calibration mode - internal_gain = coefs["NOMINAL"]["gain"] - internal_offset = coefs["NOMINAL"]["offset"] - if self._calib_mode == "GSICS": - gsics_gain = coefs["GSICS"]["gain"] - gsics_offset = coefs["GSICS"]["offset"] * gsics_gain - if gsics_gain != 0 and gsics_offset != 0: - # If no GSICS coefficients are available for a certain channel, - # they are set to zero in the file. - internal_gain = gsics_gain - internal_offset = gsics_offset - - if should_apply_meirink(self._calib_mode, self._channel_name): - meirink = MeirinkCalibrationHandler(calib_mode=self._calib_mode) - internal_gain = meirink.get_slope(self._platform_id, self._channel_name, self._scan_time) - - # Override with external coefficients, if any. - gain = coefs["EXTERNAL"].get("gain", internal_gain) - offset = coefs["EXTERNAL"].get("offset", internal_offset) - return gain, offset + def _get_calib_wishlist(self): + ext_coefs = self._calib_params.external_coefs or {} + wishlist = { + ch: self._calib_params.mode for ch in CHANNEL_NAMES.values() + } + return wishlist | ext_coefs def chebyshev(coefs, time, domain): @@ -1006,23 +960,117 @@ def calculate_area_extent(area_dict): return (ll_c, ll_l, ur_c, ur_l) -def create_coef_dict(coefs_nominal, coefs_gsics, radiance_type, ext_coefs): +def create_coef_dict(nominal_coefs, gsics_coefs=None, meirink_coefs=None): """Create coefficient dictionary expected by calibration class.""" - return { - "coefs": { + coefs = nominal_coefs.get_coefs() + if gsics_coefs: + coefs.update(gsics_coefs.get_coefs()) + if meirink_coefs: + coefs.update(meirink_coefs.get_coefs(nominal_coefs.offset)) + return coefs + + +class NominalCoefficients: + """Nominal calibration coefficients.""" + def __init__(self, channel_name, gain, offset): + """Initialize coefficients.""" + self.channel_name = channel_name + self.gain = gain + self.offset = offset + + def get_coefs(self): + """Get coefficient dictionary.""" + return { "NOMINAL": { - "gain": coefs_nominal[0], - "offset": coefs_nominal[1], - }, - "GSICS": { - "gain": coefs_gsics[0], - "offset": coefs_gsics[1] - }, - "EXTERNAL": ext_coefs - }, - "radiance_type": radiance_type - } + self.channel_name: { + "gain": self.gain, + "offset": self.offset + } + } + } + + +class GsicsCoefficients: + """GSICS calibration coefficients.""" + def __init__(self, channel_name, gain, offset): + """Initialize coefficients.""" + self.channel_name = channel_name + self.gain = gain + self.offset = offset + + def get_coefs(self): + """Get coefficient dictionary.""" + coefs = {"GSICS": {}} + if self._is_available(): + coefs["GSICS"][self.channel_name] = { + "gain": self.gain, + "offset": self.offset * self.gain + } + return coefs + + def _is_available(self): + # If no GSICS coefficients are available they are set to zero in + # the file. + return self.gain != 0 and self.offset != 0 + + +class MeirinkCoefficients: + """Re-calibration of the SEVIRI visible channels slope (see Meirink 2013).""" + + def __init__(self, platform_id, channel_name, scan_time): + """Initialize coefficients.""" + self.platform_id = platform_id + self.channel_name = channel_name + self.scan_time = scan_time + + def get_coefs(self, offset): + """Get coefficient dictionary. + + Args: + offset: Nominal calibration offset. + """ + gain = self._get_gain() + return self._combine_gain_and_offset(gain, offset) + + def _get_gain(self): + res = {} + for version, coefs in MEIRINK_COEFS.items(): + gain = self._get_gain_single_channel(coefs) + if gain: + res[f"MEIRINK-{version}"] = gain + return res + + def _get_gain_single_channel(self, coefs): + try: + coefs_ch = coefs[self.platform_id][self.channel_name] + return self.get_slope(coefs_ch, self.scan_time) + except KeyError: + return None + @staticmethod + def get_slope(coefs_single_channel, acquisition_time): + """Compute the slope for the visible channel calibration according to Meirink 2013. + + S = A + B * 1.e-3* Day + + S is here in µW m-2 sr-1 (cm-1)-1 + + EUMETSAT calibration is given in mW m-2 sr-1 (cm-1)-1, so an extra factor of 1/1000 must + be applied. + """ + A = coefs_single_channel[0] + B = coefs_single_channel[1] + delta_t = (acquisition_time - MEIRINK_EPOCH).total_seconds() + S = A + B * delta_t / (3600*24) / 1000. + return S/1000 + + def _combine_gain_and_offset(self, gain, offset): + return { + calib_mode: { + self.channel_name: {"gain": gain_, "offset": offset} + } + for calib_mode, gain_ in gain.items() + } def get_padding_area(shape, dtype): """Create a padding area filled with no data.""" diff --git a/satpy/readers/seviri_l1b_hrit.py b/satpy/readers/seviri_l1b_hrit.py index f65faa8ecc..7cb4997782 100644 --- a/satpy/readers/seviri_l1b_hrit.py +++ b/satpy/readers/seviri_l1b_hrit.py @@ -238,8 +238,13 @@ HRV_NUM_COLUMNS, REPEAT_CYCLE_DURATION, SATNUM, + CalibParams, + GsicsCoefficients, + MeirinkCoefficients, + NominalCoefficients, NoValidOrbitParams, OrbitPolynomialFinder, + ScanParams, SEVIRICalibrationHandler, add_scanline_acq_time, create_coef_dict, @@ -722,16 +727,21 @@ def pad_hrv_data(self, res): def calibrate(self, data, calibration): """Calibrate the data.""" - calib = SEVIRICalibrationHandler( - platform_id=self.platform_id, - channel_name=self.channel_name, - coefs=self._get_calib_coefs(self.channel_name), - calib_mode=self.calib_mode, - scan_time=self.observation_start_time - ) + calib = self._get_calibration_handler() res = calib.calibrate(data, calibration) return res + def _get_calibration_handler(self): + calib_params = CalibParams( + mode=self.calib_mode.upper(), + internal_coefs=self._get_calib_coefs(), + external_coefs=self.ext_calib_coefs, + radiance_type=self._get_radiance_type() + ) + scan_params = ScanParams(self.platform_id, self.channel_name, + self.observation_start_time) + return SEVIRICalibrationHandler(calib_params, scan_params) + def _mask_bad_quality(self, data): """Mask scanlines with bad quality.""" line_validity = self.mda["image_segment_line_quality"]["line_validity"] @@ -783,27 +793,35 @@ def _update_attrs(self, res, info): if self.include_raw_metadata: res.attrs["raw_metadata"] = self._get_raw_mda() - def _get_calib_coefs(self, channel_name): + def _get_calib_coefs(self): """Get coefficients for calibration from counts to radiance.""" - band_idx = self.mda["spectral_channel_id"] - 1 + band_idx = self._get_band_index() coefs_nominal = self.prologue["RadiometricProcessing"][ "Level15ImageCalibration"] coefs_gsics = self.prologue["RadiometricProcessing"]["MPEFCalFeedback"] - radiance_types = self.prologue["ImageDescription"][ - "Level15ImageProduction"]["PlannedChanProcessing"] return create_coef_dict( - coefs_nominal=( + nominal_coefs=NominalCoefficients( + self.channel_name, coefs_nominal["CalSlope"][band_idx], coefs_nominal["CalOffset"][band_idx] ), - coefs_gsics=( + gsics_coefs=GsicsCoefficients( + self.channel_name, coefs_gsics["GSICSCalCoeff"][band_idx], coefs_gsics["GSICSOffsetCount"][band_idx] ), - ext_coefs=self.ext_calib_coefs.get(channel_name, {}), - radiance_type=radiance_types[band_idx] + meirink_coefs=MeirinkCoefficients(self.platform_id, self.channel_name, self.observation_start_time) ) + def _get_radiance_type(self): + band_idx = self._get_band_index() + radiance_types = self.prologue["ImageDescription"][ + "Level15ImageProduction"]["PlannedChanProcessing"] + return radiance_types[band_idx] + + def _get_band_index(self): + return self.mda["spectral_channel_id"] - 1 + def pad_data(data, final_size, east_bound, west_bound): """Pad the data given east and west bounds and the desired size.""" diff --git a/satpy/readers/seviri_l1b_native.py b/satpy/readers/seviri_l1b_native.py index 3eaa9b4dfd..18c6d9f711 100644 --- a/satpy/readers/seviri_l1b_native.py +++ b/satpy/readers/seviri_l1b_native.py @@ -119,8 +119,13 @@ SATNUM, VISIR_NUM_COLUMNS, VISIR_NUM_LINES, + CalibParams, + GsicsCoefficients, + MeirinkCoefficients, + NominalCoefficients, NoValidOrbitParams, OrbitPolynomialFinder, + ScanParams, SEVIRICalibrationHandler, add_scanline_acq_time, calculate_area_extent, @@ -618,44 +623,57 @@ def _get_hrv_channel(self): def calibrate(self, data, dataset_id): """Calibrate the data.""" tic = dt.datetime.now() - channel_name = dataset_id["name"] - calib = SEVIRICalibrationHandler( - platform_id=self.platform_id, - channel_name=channel_name, - coefs=self._get_calib_coefs(channel_name), - calib_mode=self.calib_mode, - scan_time=self.observation_start_time - ) + calib = self._get_calibration_handler(dataset_id) res = calib.calibrate(data, dataset_id["calibration"]) logger.debug("Calibration time " + str(dt.datetime.now() - tic)) return res + def _get_calibration_handler(self, dataset_id): + channel_name = dataset_id["name"] + calib_params = CalibParams( + mode=self.calib_mode.upper(), + internal_coefs=self._get_calib_coefs(channel_name), + external_coefs=self.ext_calib_coefs, + radiance_type=self._get_radiance_type(channel_name) + ) + scan_params = ScanParams(self.platform_id, channel_name, + self.observation_start_time) + return SEVIRICalibrationHandler(calib_params, scan_params) + def _get_calib_coefs(self, channel_name): """Get coefficients for calibration from counts to radiance.""" - # even though all the channels may not be present in the file, - # the header does have calibration coefficients for all the channels - # hence, this channel index needs to refer to full channel list - band_idx = list(CHANNEL_NAMES.values()).index(channel_name) - + band_idx = self._get_band_index(channel_name) coefs_nominal = self.header["15_DATA_HEADER"][ "RadiometricProcessing"]["Level15ImageCalibration"] coefs_gsics = self.header["15_DATA_HEADER"][ "RadiometricProcessing"]["MPEFCalFeedback"] - radiance_types = self.header["15_DATA_HEADER"]["ImageDescription"][ - "Level15ImageProduction"]["PlannedChanProcessing"] + nominal_coefs = NominalCoefficients( + channel_name, coefs_nominal["CalSlope"][band_idx], coefs_nominal["CalOffset"][band_idx] + ) + gsics_coefs = GsicsCoefficients( + channel_name, coefs_gsics["GSICSCalCoeff"][band_idx], coefs_gsics["GSICSOffsetCount"][band_idx] + ) + meirink_coefs = MeirinkCoefficients( + self.platform_id, channel_name, self.observation_start_time + ) return create_coef_dict( - coefs_nominal=( - coefs_nominal["CalSlope"][band_idx], - coefs_nominal["CalOffset"][band_idx] - ), - coefs_gsics=( - coefs_gsics["GSICSCalCoeff"][band_idx], - coefs_gsics["GSICSOffsetCount"][band_idx] - ), - ext_coefs=self.ext_calib_coefs.get(channel_name, {}), - radiance_type=radiance_types[band_idx] + nominal_coefs, + gsics_coefs, + meirink_coefs ) + def _get_band_index(self, channel_name): + # even though all the channels may not be present in the file, + # the header does have calibration coefficients for all the channels + # hence, this channel index needs to refer to full channel list + return list(CHANNEL_NAMES.values()).index(channel_name) + + def _get_radiance_type(self, channel_name): + band_idx = self._get_band_index(channel_name) + radiance_types = self.header["15_DATA_HEADER"]["ImageDescription"][ + "Level15ImageProduction"]["PlannedChanProcessing"] + return radiance_types[band_idx] + def _add_scanline_acq_time(self, dataset, dataset_id): """Add scanline acquisition time to the given dataset.""" if dataset_id["name"] == "HRV": diff --git a/satpy/readers/seviri_l1b_nc.py b/satpy/readers/seviri_l1b_nc.py index fd19634fda..726caf83a1 100644 --- a/satpy/readers/seviri_l1b_nc.py +++ b/satpy/readers/seviri_l1b_nc.py @@ -29,10 +29,14 @@ from satpy.readers.seviri_base import ( CHANNEL_NAMES, SATNUM, + CalibParams, + NominalCoefficients, NoValidOrbitParams, OrbitPolynomialFinder, + ScanParams, SEVIRICalibrationHandler, add_scanline_acq_time, + create_coef_dict, get_cds_time, get_satpos, mask_bad_quality, @@ -185,32 +189,36 @@ def calibrate(self, dataset, dataset_id): if dataset_id["calibration"] == "counts": dataset.attrs["_FillValue"] = 0 - calib = SEVIRICalibrationHandler( - platform_id=int(self.platform_id), - channel_name=channel, - coefs=self._get_calib_coefs(dataset, channel), - calib_mode="NOMINAL", - scan_time=self.observation_start_time + calib = self._get_calibration_handler(dataset, channel) + return calib.calibrate(dataset, calibration) + + def _get_calibration_handler(self, dataset, channel): + calib_params = CalibParams( + mode="NOMINAL", + internal_coefs=self._get_calib_coefs(dataset, channel), + external_coefs=self.ext_calib_coefs, + radiance_type=self._get_radiance_type(channel) + ) + scan_params = ScanParams( + int(self.platform_id), + channel, + self.observation_start_time ) + return SEVIRICalibrationHandler(calib_params, scan_params) - return calib.calibrate(dataset, calibration) + def _get_radiance_type(self, channel): + band_idx = list(CHANNEL_NAMES.values()).index(channel) + return self.nc["planned_chan_processing"].values[band_idx] def _get_calib_coefs(self, dataset, channel): - """Get coefficients for calibration from counts to radiance.""" - band_idx = list(CHANNEL_NAMES.values()).index(channel) + """Get coefficients for calibration from counts to radiance. + + Only nominal calibration coefficients are available in netCDF files. + """ offset = dataset.attrs["add_offset"].astype("float32") gain = dataset.attrs["scale_factor"].astype("float32") - # Only one calibration available here - return { - "coefs": { - "NOMINAL": { - "gain": gain, - "offset": offset - }, - "EXTERNAL": self.ext_calib_coefs.get(channel, {}) - }, - "radiance_type": self.nc["planned_chan_processing"].values[band_idx] - } + nominal_coefs = NominalCoefficients(channel, gain, offset) + return create_coef_dict(nominal_coefs) def _mask_bad_quality(self, dataset, dataset_info): """Mask scanlines with bad quality.""" diff --git a/satpy/readers/utils.py b/satpy/readers/utils.py index 983225acd5..37e3e9366f 100644 --- a/satpy/readers/utils.py +++ b/satpy/readers/utils.py @@ -593,7 +593,7 @@ class CalibrationCoefficientPicker: calib_wishlist = { "ch1": "meirink", - ("ch2", "ch3"): "gsics" + ("ch2", "ch3"): "gsics", "ch4": {"mygain": 123}, } # Also possible: Same mode for all channels via diff --git a/satpy/tests/reader_tests/test_seviri_base.py b/satpy/tests/reader_tests/test_seviri_base.py index 1b81aa1599..83b0b24b3d 100644 --- a/satpy/tests/reader_tests/test_seviri_base.py +++ b/satpy/tests/reader_tests/test_seviri_base.py @@ -28,10 +28,10 @@ from satpy.readers.seviri_base import ( MEIRINK_COEFS, MEIRINK_EPOCH, + MeirinkCoefficients, NoValidOrbitParams, OrbitPolynomial, OrbitPolynomialFinder, - SEVIRICalibrationHandler, chebyshev, dec10216, get_cds_time, @@ -371,11 +371,9 @@ class TestMeirinkSlope: @pytest.mark.parametrize("channel_name", ["VIS006", "VIS008", "IR_016"]) def test_get_meirink_slope_epoch(self, platform_id, channel_name): """Test the value of the slope of the Meirink calibration on 2000-01-01.""" - coefs = {"coefs": {}} - coefs["coefs"]["NOMINAL"] = {"gain": -1, "offset": -1} - coefs["coefs"]["EXTERNAL"] = {} - calibration_handler = SEVIRICalibrationHandler(platform_id, channel_name, coefs, "MEIRINK-2023", MEIRINK_EPOCH) - assert calibration_handler.get_gain_offset()[0] == MEIRINK_COEFS["2023"][platform_id][channel_name][0]/1000. + comp = MeirinkCoefficients(platform_id, channel_name, MEIRINK_EPOCH) + coefs = comp.get_coefs("dummy_offset") + assert coefs["MEIRINK-2023"][channel_name]["gain"] == MEIRINK_COEFS["2023"][platform_id][channel_name][0]/1000. @pytest.mark.parametrize(("platform_id", "time", "expected"), [ (321, dt.datetime(2005, 1, 18, 0, 0), [0.0250354716, 0.0315626684, 0.022880986]), @@ -389,9 +387,7 @@ def test_get_meirink_slope_epoch(self, platform_id, channel_name): ]) def test_get_meirink_slope_2020(self, platform_id, time, expected): """Test the value of the slope of the Meirink calibration.""" - coefs = {"coefs": {}} - coefs["coefs"]["NOMINAL"] = {"gain": -1, "offset": -1} - coefs["coefs"]["EXTERNAL"] = {} for i, channel_name in enumerate(["VIS006", "VIS008", "IR_016"]): - calibration_handler = SEVIRICalibrationHandler(platform_id, channel_name, coefs, "MEIRINK-2023", time) - assert abs(calibration_handler.get_gain_offset()[0] - expected[i]) < 1e-6 + comp = MeirinkCoefficients(platform_id, channel_name, time) + coefs = comp.get_coefs("dummy_offset") + assert abs(coefs["MEIRINK-2023"][channel_name]["gain"] - expected[i]) < 1e-6 diff --git a/satpy/tests/reader_tests/test_seviri_l1b_calibration.py b/satpy/tests/reader_tests/test_seviri_l1b_calibration.py index 8eaf2b83da..ad35d6494f 100644 --- a/satpy/tests/reader_tests/test_seviri_l1b_calibration.py +++ b/satpy/tests/reader_tests/test_seviri_l1b_calibration.py @@ -23,8 +23,9 @@ import numpy as np import pytest import xarray as xr +from pytest_lazy_fixtures.lazy_fixture import lf -from satpy.readers.seviri_base import SEVIRICalibrationAlgorithm, SEVIRICalibrationHandler +import satpy.readers.seviri_base as sev COUNTS_INPUT = xr.DataArray( np.array([[377., 377., 377., 376., 375.], @@ -108,7 +109,7 @@ class TestSEVIRICalibrationAlgorithm(unittest.TestCase): def setUp(self): """Set up the SEVIRI Calibration algorithm for testing.""" - self.algo = SEVIRICalibrationAlgorithm( + self.algo = sev.SEVIRICalibrationAlgorithm( platform_id=PLATFORM_ID, scan_time=dt.datetime(2020, 8, 15, 13, 0, 40) ) @@ -148,57 +149,87 @@ class TestSeviriCalibrationHandler: def test_init(self): """Test initialization of the calibration handler.""" with pytest.raises(ValueError, match="Invalid calibration mode: INVALID. Choose one of (.*)"): - SEVIRICalibrationHandler( - platform_id=None, - channel_name=None, - coefs=None, - calib_mode="invalid", - scan_time=None - ) - - def _get_calibration_handler(self, calib_mode="NOMINAL", ext_coefs=None): + self._get_calibration_handler("IR_108", "INVALID") + + def _get_calibration_handler(self, channel, calib_mode="NOMINAL", ext_coefs=None): """Provide a calibration handler.""" - return SEVIRICalibrationHandler( - platform_id=324, - channel_name="IR_108", - coefs={ - "coefs": { - "NOMINAL": { + int_coefs = { + "NOMINAL": { + "IR_108": { "gain": 10, "offset": -1 }, - "GSICS": { + "VIS006": { "gain": 20, "offset": -2 }, - "EXTERNAL": ext_coefs or {} }, - "radiance_type": 1 - }, - calib_mode=calib_mode, + "GSICS": { + "IR_108": { + "gain": 30, + "offset": -3 + }, + }, + } + calib_params = sev.CalibParams( + mode=calib_mode, + internal_coefs=int_coefs, + external_coefs=ext_coefs, + radiance_type=1) + scan_params = sev.ScanParams( + platform_id=324, + channel_name=channel, scan_time=None ) + return sev.SEVIRICalibrationHandler(calib_params, scan_params) def test_calibrate_exceptions(self): """Test exceptions raised by the calibration handler.""" - calib = self._get_calibration_handler() + calib = self._get_calibration_handler("IR_108") with pytest.raises(ValueError, match="Invalid calibration invalid for channel IR_108"): calib.calibrate(None, "invalid") + @pytest.fixture + def external_coefs(self): + """Get external coefficients.""" + return {"IR_108": {"gain": 40, "offset": -4}} + + @pytest.fixture + def coefs_ir108_nominal_exp(self): + """Get expected IR coefficients in nominal calib mode.""" + return {"coefs": {"gain": 10, "offset": -1}, "mode": "NOMINAL"} + + @pytest.fixture + def coefs_vis006_exp(self): + """Get expected VIS coefficients.""" + return {"coefs": {"gain": 20, "offset": -2}, "mode": "NOMINAL"} + + @pytest.fixture + def coefs_ir108_gsics_exp(self): + """Get expected IR coefficients in GSICS calib mode.""" + return {"coefs": {"gain": 30, "offset": -3}, "mode": "GSICS"} + + @pytest.fixture + def coefs_ir108_external_exp(self): + """Get expected IR coefficients in the presence of external coefficients.""" + return {"coefs": {"gain": 40, "offset": -4}, "mode": "external"} + @pytest.mark.parametrize( - ("calib_mode", "ext_coefs", "expected"), + ("channel", "calib_mode", "ext_coefs", "expected"), [ - ("NOMINAL", {}, (10, -1)), - ("GSICS", {}, (20, -40)), - ("GSICS", {"gain": 30, "offset": -3}, (30, -3)), - ("NOMINAL", {"gain": 30, "offset": -3}, (30, -3)) + ("IR_108", "NOMINAL", None, lf("coefs_ir108_nominal_exp")), + ("IR_108", "GSICS", None, lf("coefs_ir108_gsics_exp")), + ("IR_108", "NOMINAL", lf("external_coefs"), lf("coefs_ir108_external_exp")), + # For VIS006 there's only nominal coefficients in this example + ("VIS006", "NOMINAL", None, lf("coefs_vis006_exp")), + ("VIS006", "GSICS", None, lf("coefs_vis006_exp")), + ("VIS006", "NOMINAL", lf("external_coefs"), lf("coefs_vis006_exp")) ] ) - def test_get_gain_offset(self, calib_mode, ext_coefs, expected): + def test_get_coefs(self, channel, calib_mode, ext_coefs, expected): """Test selection of gain and offset.""" - calib = self._get_calibration_handler(calib_mode=calib_mode, - ext_coefs=ext_coefs) - coefs = calib.get_gain_offset() + calib = self._get_calibration_handler(channel, calib_mode, ext_coefs) + coefs = calib.get_coefs() assert coefs == expected