Ensure env_params intake for calibration (#985)

* add get_env_params_EK60 to CalibrateEK60.__init__, make sure CalibrateEK80.__init__ env param intake uses self.echodata and self.env_params

* add a narrow integration test to ensure env_params intake for CalibrateEK60

* add a narrow integration test to ensure env_params intake for CalibrateAZFP

* add sanitize_user_env_dict, revise get_env_params_AZFP to use new func

* revise get_env_params_EK60, add sound speed and absorption formula sources as env params, fix tests

* fix EK60 env params intake and tests

* use get_env_params_EK for EK60 intake, add EK60 integration test without input

* use get_env_params_EK in CalibrateEK80, add test_env_params_intake_EK80_no_input

* add test_env_params_intake_EK80_with_input, fix condition to use values from EK80 data

* add sound speed to FG absorption input in get_env_params_EK, shorten formula name

* fix pre-commit errors

* change to include pH in terms of whether to recompute sound speed/absorption, add input data array test case

* remove get_env_params_EK80 completely from codebase

* add test_sanitize_user_env_dict, fix typing for sanitize_user_*_dict

* add test_get_env_params_AZFP

* add test_get_env_params_EK60_calculate/from_data

* add partial test_get_env_params_EK80_from_data

* fix test_get_env_params_EK80_from_data, revise get_env_params_EK comments

* change ed_group to ed_beam_group

* fix azfp tests involving mean temperature

* remove redundancy in passing in extra ed_beam_group in _cal_power_samples and _cal_complex_samples

* add missed test cases in test_sanitize_user_env_dict

* change list to List in typing

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Emilio Mayorga <[email protected]>

Author: 3 people

echopype/calibrate/cal_params.py

@@ -49,10 +49,6 @@
 }
 
 
-# TODO: need a function (something like "_check_param_freq_dep")
-# to check user input cal_params and env_params
-
-
 def param2da(p_val: Union[int, float, list], channel: Union[list, xr.DataArray]) -> xr.DataArray:
     """
     Organize individual parameter in scalar or list to xr.DataArray with channel coordinate.
@@ -88,7 +84,7 @@ def param2da(p_val: Union[int, float, list], channel: Union[list, xr.DataArray])
 
 def sanitize_user_cal_dict(
     sonar_type: Literal["EK60", "EK80", "AZFP"],
-    user_dict: Dict[str, Union[int, float, xr.DataArray]],
+    user_dict: Dict[str, Union[int, float, list, xr.DataArray]],
     channel: Union[List, xr.DataArray],
 ) -> Dict[str, Union[int, float, xr.DataArray]]:
     """
@@ -126,7 +122,7 @@ def sanitize_user_cal_dict(
 
     # Screen parameters: only retain those defined in CAL_PARAMS
     #  -- transform params in scalar or list to xr.DataArray
-    #  -- directly pass through those that are xr.DataArray
+    #  -- directly pass through those that are xr.DataArray and pass the check for coordinates
     out_dict = dict.fromkeys(CAL_PARAMS[sonar_type])
     for p_name, p_val in user_dict.items():
         if p_name in out_dict:

echopype/calibrate/calibrate_azfp.py

@@ -15,7 +15,7 @@ def __init__(self, echodata: EchoData, env_params=None, cal_params=None, **kwarg
         self.sonar_type = "AZFP"
 
         # load env and cal parameters
-        self.env_params = get_env_params_AZFP(echodata=self.echodata, user_env_dict=self.env_params)
+        self.env_params = get_env_params_AZFP(echodata=self.echodata, user_dict=self.env_params)
         self.cal_params = get_cal_params_AZFP(
             beam=self.echodata["Sonar/Beam_group1"],
             vend=self.echodata["Vendor_specific"],

echopype/calibrate/calibrate_ek.py

@@ -9,7 +9,7 @@
 from .cal_params import get_cal_params_EK
 from .calibrate_base import CalibrateBase
 from .ek80_complex import compress_pulse, get_filter_coeff, get_tau_effective, get_transmit_signal
-from .env_params import get_env_params_EK60, get_env_params_EK80
+from .env_params import get_env_params_EK
 from .range import compute_range_EK, range_mod_TVG_EK
 
 logger = _init_logger(__name__)
@@ -19,6 +19,8 @@ class CalibrateEK(CalibrateBase):
     def __init__(self, echodata: EchoData, env_params, cal_params):
         super().__init__(echodata, env_params, cal_params)
 
+        self.ed_beam_group = None  # will be assigned in child class
+
     def compute_echo_range(self, chan_sel: xr.DataArray = None):
         """
         Compute echo range for EK echosounders.
@@ -36,24 +38,22 @@ def compute_echo_range(self, chan_sel: xr.DataArray = None):
             chan_sel=chan_sel,
         )
 
-    def _cal_power_samples(self, cal_type: str, power_ed_group: str = None) -> xr.Dataset:
+    def _cal_power_samples(self, cal_type: str) -> xr.Dataset:
         """Calibrate power data from EK60 and EK80.
 
         Parameters
         ----------
         cal_type: str
             'Sv' for calculating volume backscattering strength, or
             'TS' for calculating target strength
-        power_ed_group:
-            The ``EchoData`` beam group path containing the power data
 
         Returns
         -------
         xr.Dataset
             The calibrated dataset containing Sv or TS
         """
         # Select source of backscatter data
-        beam = self.echodata[power_ed_group]
+        beam = self.echodata[self.ed_beam_group]
 
         # Derived params
         wavelength = self.env_params["sound_speed"] / beam["frequency_nominal"]  # wavelength
@@ -63,7 +63,7 @@ def _cal_power_samples(self, cal_type: str, power_ed_group: str = None) -> xr.Da
         sound_speed = self.env_params["sound_speed"]
         absorption = self.env_params["sound_absorption"]
         tvg_mod_range = range_mod_TVG_EK(
-            self.echodata, self.ed_group, self.range_meter, sound_speed
+            self.echodata, self.ed_beam_group, self.range_meter, sound_speed
         )
         tvg_mod_range = tvg_mod_range.where(tvg_mod_range > 0, np.nan)
 
@@ -130,26 +130,31 @@ class CalibrateEK60(CalibrateEK):
     def __init__(self, echodata, env_params, cal_params, **kwargs):
         super().__init__(echodata, env_params, cal_params)
 
-        # Set sonar_type
+        # Set sonar_type and waveform/encode mode
         self.sonar_type = "EK60"
-
-        # Get env_params
-        self.env_params = get_env_params_EK60(echodata=echodata, user_env_dict=self.env_params)
         self.waveform_mode = "CW"
         self.encode_mode = "power"
 
-        # Compute range
-        self.compute_echo_range()
-
-        # Get the right ed_group for CW power samples
-        self.ed_group = retrieve_correct_beam_group(
+        # Get the right ed_beam_group for CW power samples
+        self.ed_beam_group = retrieve_correct_beam_group(
             echodata=self.echodata, waveform_mode=self.waveform_mode, encode_mode=self.encode_mode
         )
 
+        # Get env_params
+        self.env_params = get_env_params_EK(
+            sonar_type=self.sonar_type,
+            beam=self.echodata[self.ed_beam_group],
+            env=self.echodata["Environment"],
+            user_dict=self.env_params,
+        )
+
+        # Compute range
+        self.compute_echo_range()
+
         # Set the channels to calibrate
         # For EK60 this is all channels
-        self.chan_sel = self.echodata[self.ed_group]["channel"]
-        beam = self.echodata[self.ed_group]
+        self.chan_sel = self.echodata[self.ed_beam_group]["channel"]
+        beam = self.echodata[self.ed_beam_group]
 
         # Get cal_params
         self.cal_params = get_cal_params_EK(
@@ -162,10 +167,10 @@ def __init__(self, echodata, env_params, cal_params, **kwargs):
         )
 
     def compute_Sv(self, **kwargs):
-        return self._cal_power_samples(cal_type="Sv", power_ed_group=self.ed_group)
+        return self._cal_power_samples(cal_type="Sv")
 
     def compute_TS(self, **kwargs):
-        return self._cal_power_samples(cal_type="TS", power_ed_group=self.ed_group)
+        return self._cal_power_samples(cal_type="TS")
 
 
 class CalibrateEK80(CalibrateEK):
@@ -197,23 +202,23 @@ def __init__(self, echodata, env_params, cal_params, waveform_mode, encode_mode)
         self.encode_mode = encode_mode
         self.echodata = echodata
 
-        # Get the right ed_group given waveform and encode mode
-        self.ed_group = retrieve_correct_beam_group(
+        # Get the right ed_beam_group given waveform and encode mode
+        self.ed_beam_group = retrieve_correct_beam_group(
             echodata=self.echodata, waveform_mode=self.waveform_mode, encode_mode=self.encode_mode
         )
 
         # Select the channels to calibrate
         if self.encode_mode == "power":
             # Power sample only possible under CW mode,
             # and all power samples will live in the same group
-            self.chan_sel = self.echodata[self.ed_group]["channel"]
+            self.chan_sel = self.echodata[self.ed_beam_group]["channel"]
         else:
             # Complex samples can be CW or BB, so select based on waveform mode
-            chan_dict = self._get_chan_dict(self.echodata[self.ed_group])
+            chan_dict = self._get_chan_dict(self.echodata[self.ed_beam_group])
             self.chan_sel = chan_dict[self.waveform_mode]
 
         # Subset of the right Sonar/Beam_groupX group given the selected channels
-        beam = self.echodata[self.ed_group].sel(channel=self.chan_sel)
+        beam = self.echodata[self.ed_beam_group].sel(channel=self.chan_sel)
 
         # Use center frequency if in BB mode, else use nominal channel frequency
         if self.waveform_mode == "BB":
@@ -226,11 +231,12 @@ def __init__(self, echodata, env_params, cal_params, waveform_mode, encode_mode)
             self.freq_center = beam["frequency_nominal"].sel(channel=self.chan_sel)
 
         # Get env_params: depends on waveform mode
-        self.env_params = get_env_params_EK80(
-            echodata=echodata,
+        self.env_params = get_env_params_EK(
+            sonar_type=self.sonar_type,
+            beam=beam,
+            env=self.echodata["Environment"],
+            user_dict=self.env_params,
             freq=self.freq_center,
-            user_env_dict=env_params,
-            ed_group=self.ed_group,
         )
 
         # Get cal_params: depends on waveform and encode mode
@@ -333,24 +339,22 @@ def _get_B_theta_phi_m(self):
 
         return B_theta_phi_m
 
-    def _cal_complex_samples(self, cal_type: str, complex_ed_group: str) -> xr.Dataset:
+    def _cal_complex_samples(self, cal_type: str) -> xr.Dataset:
         """Calibrate complex data from EK80.
 
         Parameters
         ----------
         cal_type : str
             'Sv' for calculating volume backscattering strength, or
             'TS' for calculating target strength
-        complex_ed_group : str
-            The ``EchoData`` beam group path containing complex data
 
         Returns
         -------
         xr.Dataset
             The calibrated dataset containing Sv or TS
         """
         # Select source of backscatter data
-        beam = self.echodata[complex_ed_group].sel(channel=self.chan_sel)
+        beam = self.echodata[self.ed_beam_group].sel(channel=self.chan_sel)
         vend = self.echodata["Vendor_specific"].sel(channel=self.chan_sel)
 
         # Get transmit signal
@@ -376,7 +380,9 @@ def _cal_complex_samples(self, cal_type: str, complex_ed_group: str) -> xr.Datas
         transmit_power = beam["transmit_power"]
 
         # TVG compensation with modified range
-        tvg_mod_range = range_mod_TVG_EK(self.echodata, self.ed_group, range_meter, sound_speed)
+        tvg_mod_range = range_mod_TVG_EK(
+            self.echodata, self.ed_beam_group, range_meter, sound_speed
+        )
         tvg_mod_range = tvg_mod_range.where(tvg_mod_range > 0, np.nan)
 
         spreading_loss = 20 * np.log10(tvg_mod_range)
@@ -476,10 +482,10 @@ def _compute_cal(self, cal_type) -> xr.Dataset:
 
         if flag_complex:
             # Complex samples can be BB or CW
-            ds_cal = self._cal_complex_samples(cal_type=cal_type, complex_ed_group=self.ed_group)
+            ds_cal = self._cal_complex_samples(cal_type=cal_type)
         else:
             # Power samples only make sense for CW mode data
-            ds_cal = self._cal_power_samples(cal_type=cal_type, power_ed_group=self.ed_group)
+            ds_cal = self._cal_power_samples(cal_type=cal_type)
 
         return ds_cal