diff --git a/doc/source/api.md b/doc/source/api.md index cf4df17d..b0414521 100644 --- a/doc/source/api.md +++ b/doc/source/api.md @@ -244,17 +244,6 @@ To build and pass your coregistration pipeline to {func}`~xdem.DEM.coregister_3d xdem.coreg.BiasCorr ``` -**Classes for any 1-, 2- and N-D biases:** - -```{eval-rst} -.. autosummary:: - :toctree: gen_modules/ - - xdem.coreg.BiasCorr1D - xdem.coreg.BiasCorr2D - xdem.coreg.BiasCorrND -``` - **Convenience classes for specific corrections:** ```{eval-rst} diff --git a/doc/source/biascorr.md b/doc/source/biascorr.md index ef83b8a8..44e3ac3a 100644 --- a/doc/source/biascorr.md +++ b/doc/source/biascorr.md @@ -158,16 +158,3 @@ terbias.fit(ref_dem, tba_dem, inlier_mask=inlier_mask) # Apply the transformation corrected_dem = terbias.apply(tba_dem) ``` - -## Generic 1-D, 2-D and N-D classes - -All bias-corrections methods are inherited from generic classes that perform corrections in 1-, 2- or N-D. Having these -separate helps the user navigating the dimensionality of the functions, optimizer, binning or variables used. - -{class}`xdem.coreg.BiasCorr1D` -{class}`xdem.coreg.BiasCorr2D` -{class}`xdem.coreg.BiasCorrND` - -- **Performs:** Correct biases with any function and optimizer, or any binning, in 1-, 2- or N-D. -- **Supports weights** Yes. -- **Recommended for:** Anything. diff --git a/tests/test_coreg/test_base.py b/tests/test_coreg/test_base.py index 15537b0a..264e26a5 100644 --- a/tests/test_coreg/test_base.py +++ b/tests/test_coreg/test_base.py @@ -568,8 +568,8 @@ def test_pipeline_combinations__nobiasvar(self, coreg1: Coreg, coreg2: Coreg) -> @pytest.mark.parametrize( "coreg2", [ - coreg.BiasCorr1D(bias_var_names=["slope"], fit_or_bin="bin"), - coreg.BiasCorr2D(bias_var_names=["slope", "aspect"], fit_or_bin="bin"), + coreg.BiasCorr(bias_var_names=["slope"], fit_or_bin="bin"), + coreg.BiasCorr(bias_var_names=["slope", "aspect"], fit_or_bin="bin"), ], ) # type: ignore def test_pipeline_combinations__biasvar(self, coreg1: Coreg, coreg2: Coreg) -> None: @@ -588,24 +588,24 @@ def test_pipeline_combinations__biasvar(self, coreg1: Coreg, coreg2: Coreg) -> N def test_pipeline__errors(self) -> None: """Test pipeline raises proper errors.""" - pipeline = coreg.CoregPipeline([coreg.NuthKaab(), coreg.BiasCorr1D()]) + pipeline = coreg.CoregPipeline([coreg.NuthKaab(), coreg.BiasCorr()]) with pytest.raises( ValueError, match=re.escape( "No `bias_vars` passed to .fit() for bias correction step " - " of the pipeline." + " of the pipeline." ), ): pipeline.fit(**self.fit_params) - pipeline2 = coreg.CoregPipeline([coreg.NuthKaab(), coreg.BiasCorr1D(), coreg.BiasCorr1D()]) + pipeline2 = coreg.CoregPipeline([coreg.NuthKaab(), coreg.BiasCorr(), coreg.BiasCorr()]) with pytest.raises( ValueError, match=re.escape( - "No `bias_vars` passed to .fit() for bias correction step " + "No `bias_vars` passed to .fit() for bias correction step " "of the pipeline. As you are using several bias correction steps requiring" " `bias_vars`, don't forget to explicitly define their `bias_var_names` " - "during instantiation, e.g. BiasCorr1D(bias_var_names=['slope'])." + "during instantiation, e.g. BiasCorr(bias_var_names=['slope'])." ), ): pipeline2.fit(**self.fit_params) @@ -615,17 +615,17 @@ def test_pipeline__errors(self) -> None: match=re.escape( "When using several bias correction steps requiring `bias_vars` in a pipeline," "the `bias_var_names` need to be explicitly defined at each step's " - "instantiation, e.g. BiasCorr1D(bias_var_names=['slope'])." + "instantiation, e.g. BiasCorr(bias_var_names=['slope'])." ), ): pipeline2.fit(**self.fit_params, bias_vars={"slope": xdem.terrain.slope(self.ref)}) - pipeline3 = coreg.CoregPipeline([coreg.NuthKaab(), coreg.BiasCorr1D(bias_var_names=["slope"])]) + pipeline3 = coreg.CoregPipeline([coreg.NuthKaab(), coreg.BiasCorr(bias_var_names=["slope"])]) with pytest.raises( ValueError, match=re.escape( "Not all keys of `bias_vars` in .fit() match the `bias_var_names` defined during " - "instantiation of the bias correction step : ['slope']." + "instantiation of the bias correction step : ['slope']." ), ): pipeline3.fit(**self.fit_params, bias_vars={"ncc": xdem.terrain.slope(self.ref)}) diff --git a/tests/test_coreg/test_biascorr.py b/tests/test_coreg/test_biascorr.py index 6baac537..fda108a0 100644 --- a/tests/test_coreg/test_biascorr.py +++ b/tests/test_coreg/test_biascorr.py @@ -156,6 +156,48 @@ def test_biascorr__errors(self) -> None: ): biascorr.BiasCorr(fit_or_bin="bin", bin_apply_method=1) # type: ignore + # When wrong number of parameters are passed + + # Copy fit parameters + fit_args = self.fit_args_rst_rst.copy() + with pytest.raises( + ValueError, + match=re.escape("A number of 1 variable(s) has to be provided through the argument 'bias_vars', " "got 2."), + ): + bias_vars_dict = {"elevation": self.ref, "slope": xdem.terrain.slope(self.ref)} + bcorr1d = biascorr.BiasCorr(bias_var_names=["elevation"]) + bcorr1d.fit(**fit_args, bias_vars=bias_vars_dict) + + with pytest.raises( + ValueError, + match=re.escape("A number of 2 variable(s) has to be provided through the argument " "'bias_vars', got 1."), + ): + bias_vars_dict = {"elevation": self.ref} + bcorr2d = biascorr.BiasCorr(bias_var_names=["elevation", "slope"]) + bcorr2d.fit(**fit_args, bias_vars=bias_vars_dict) + + # When variables don't match + with pytest.raises( + ValueError, + match=re.escape( + "The keys of `bias_vars` do not match the `bias_var_names` defined during " "instantiation: ['ncc']." + ), + ): + bcorr1d2 = biascorr.BiasCorr(bias_var_names=["ncc"]) + bias_vars_dict = {"elevation": self.ref} + bcorr1d2.fit(**fit_args, bias_vars=bias_vars_dict) + + with pytest.raises( + ValueError, + match=re.escape( + "The keys of `bias_vars` do not match the `bias_var_names` defined during " + "instantiation: ['elevation', 'ncc']." + ), + ): + bcorr2d2 = biascorr.BiasCorr(bias_var_names=["elevation", "ncc"]) + bias_vars_dict = {"elevation": self.ref, "slope": xdem.terrain.slope(self.ref)} + bcorr2d2.fit(**fit_args, bias_vars=bias_vars_dict) + @pytest.mark.parametrize("fit_args", all_fit_args) # type: ignore @pytest.mark.parametrize( "fit_func", ("norder_polynomial", "nfreq_sumsin", lambda x, a, b: x[0] * a + b) @@ -354,87 +396,6 @@ def test_biascorr__bin_and_fit_2d(self, fit_args, fit_func, fit_optimizer, bin_s # Apply the correction bcorr.apply(elev=self.tba, bias_vars=bias_vars_dict) - @pytest.mark.parametrize("fit_args", [fit_args_rst_pts, fit_args_rst_rst]) # type: ignore - def test_biascorr1d(self, fit_args) -> None: - """ - Test the subclass BiasCorr1D, which defines default parameters for 1D. - The rest is already tested in test_biascorr. - """ - - # Try default "fit" parameters instantiation - bcorr1d = biascorr.BiasCorr1D() - - assert bcorr1d._meta["fit_func"] == biascorr.fit_workflows["norder_polynomial"]["func"] - assert bcorr1d._meta["fit_optimizer"] == biascorr.fit_workflows["norder_polynomial"]["optimizer"] - assert bcorr1d._needs_vars is True - - # Try default "bin" parameter instantiation - bcorr1d = biascorr.BiasCorr1D(fit_or_bin="bin") - - assert bcorr1d._meta["bin_sizes"] == 10 - assert bcorr1d._meta["bin_statistic"] == np.nanmedian - assert bcorr1d._meta["bin_apply_method"] == "linear" - - elev_fit_args = fit_args.copy() - # Raise error when wrong number of parameters are passed - with pytest.raises( - ValueError, match="A single variable has to be provided through the argument 'bias_vars', " "got 2." - ): - bias_vars_dict = {"elevation": self.ref, "slope": xdem.terrain.slope(self.ref)} - bcorr1d.fit(**elev_fit_args, bias_vars=bias_vars_dict) - - # Raise error when variables don't match - with pytest.raises( - ValueError, - match=re.escape( - "The keys of `bias_vars` do not match the `bias_var_names` defined during " "instantiation: ['ncc']." - ), - ): - bcorr1d2 = biascorr.BiasCorr1D(bias_var_names=["ncc"]) - bias_vars_dict = {"elevation": self.ref} - bcorr1d2.fit(**elev_fit_args, bias_vars=bias_vars_dict) - - @pytest.mark.parametrize("fit_args", all_fit_args) # type: ignore - def test_biascorr2d(self, fit_args) -> None: - """ - Test the subclass BiasCorr2D, which defines default parameters for 2D. - The rest is already tested in test_biascorr. - """ - - # Try default "fit" parameters instantiation - bcorr2d = biascorr.BiasCorr2D() - - assert bcorr2d._meta["fit_func"] == polynomial_2d - assert bcorr2d._meta["fit_optimizer"] == scipy.optimize.curve_fit - assert bcorr2d._needs_vars is True - - # Try default "bin" parameter instantiation - bcorr2d = biascorr.BiasCorr2D(fit_or_bin="bin") - - assert bcorr2d._meta["bin_sizes"] == 10 - assert bcorr2d._meta["bin_statistic"] == np.nanmedian - assert bcorr2d._meta["bin_apply_method"] == "linear" - - elev_fit_args = fit_args.copy() - # Raise error when wrong number of parameters are passed - with pytest.raises( - ValueError, match="Exactly two variables have to be provided through the argument " "'bias_vars', got 1." - ): - bias_vars_dict = {"elevation": self.ref} - bcorr2d.fit(**elev_fit_args, bias_vars=bias_vars_dict) - - # Raise error when variables don't match - with pytest.raises( - ValueError, - match=re.escape( - "The keys of `bias_vars` do not match the `bias_var_names` defined during " - "instantiation: ['elevation', 'ncc']." - ), - ): - bcorr2d2 = biascorr.BiasCorr2D(bias_var_names=["elevation", "ncc"]) - bias_vars_dict = {"elevation": self.ref, "slope": xdem.terrain.slope(self.ref)} - bcorr2d2.fit(**elev_fit_args, bias_vars=bias_vars_dict) - def test_directionalbias(self) -> None: """Test the subclass DirectionalBias.""" diff --git a/xdem/coreg/__init__.py b/xdem/coreg/__init__.py index 3776e6ba..eac0f29a 100644 --- a/xdem/coreg/__init__.py +++ b/xdem/coreg/__init__.py @@ -17,13 +17,5 @@ apply_matrix, invert_matrix, ) -from xdem.coreg.biascorr import ( # noqa - BiasCorr, - BiasCorr1D, - BiasCorr2D, - BiasCorrND, - Deramp, - DirectionalBias, - TerrainBias, -) +from xdem.coreg.biascorr import BiasCorr, Deramp, DirectionalBias, TerrainBias # noqa from xdem.coreg.workflows import dem_coregistration # noqa diff --git a/xdem/coreg/base.py b/xdem/coreg/base.py index 39c34972..af33ee5c 100644 --- a/xdem/coreg/base.py +++ b/xdem/coreg/base.py @@ -1026,6 +1026,7 @@ class CoregDict(TypedDict, total=False): bin_statistic: Callable[[NDArrayf], np.floating[Any]] bin_apply_method: Literal["linear"] | Literal["per_bin"] bias_var_names: list[str] + nd: int | None # 2/ Outputs fit_params: NDArrayf diff --git a/xdem/coreg/biascorr.py b/xdem/coreg/biascorr.py index 416dc8f4..93a7d654 100644 --- a/xdem/coreg/biascorr.py +++ b/xdem/coreg/biascorr.py @@ -135,6 +135,9 @@ def __init__( # Add subsample attribute self._meta["subsample"] = subsample + # Add number of dimensions attribute (length of bias_var_names, counted generically for iterator) + self._meta["nd"] = sum(1 for _ in bias_var_names) if bias_var_names is not None else None + # Update attributes self._fit_or_bin = fit_or_bin self._is_affine = False @@ -162,6 +165,14 @@ def _fit_biascorr( # type: ignore if bias_vars is None: raise ValueError("At least one `bias_var` should be passed to the fitting function, got None.") + # Check number of variables + nd = self._meta["nd"] + if nd is not None and len(bias_vars) != nd: + raise ValueError( + "A number of {} variable(s) has to be provided through the argument 'bias_vars', " + "got {}.".format(nd, len(bias_vars)) + ) + # If bias var names were explicitly passed at instantiation, check that they match the one from the dict if self._meta["bias_var_names"] is not None: if not sorted(bias_vars.keys()) == sorted(self._meta["bias_var_names"]): @@ -478,239 +489,7 @@ def _apply_rst( # type: ignore return dem_corr, transform -class BiasCorr1D(BiasCorr): - """ - Bias-correction along a single variable (e.g., angle, terrain attribute). - - The correction can be done by fitting a function along the variable, or binning with that variable. - """ - - def __init__( - self, - fit_or_bin: Literal["bin_and_fit"] | Literal["fit"] | Literal["bin"] = "fit", - fit_func: Callable[..., NDArrayf] - | Literal["norder_polynomial"] - | Literal["nfreq_sumsin"] = "norder_polynomial", - fit_optimizer: Callable[..., tuple[NDArrayf, Any]] = scipy.optimize.curve_fit, - bin_sizes: int | dict[str, int | Iterable[float]] = 10, - bin_statistic: Callable[[NDArrayf], np.floating[Any]] = np.nanmedian, - bin_apply_method: Literal["linear"] | Literal["per_bin"] = "linear", - bias_var_names: Iterable[str] = None, - subsample: float | int = 1.0, - ): - """ - Instantiate a 1D bias correction. - - :param fit_or_bin: Whether to fit or bin. Use "fit" to correct by optimizing a function or - "bin" to correct with a statistic of central tendency in defined bins. - :param fit_func: Function to fit to the bias with variables later passed in .fit(). - :param fit_optimizer: Optimizer to minimize the function. - :param bin_sizes: Size (if integer) or edges (if iterable) for binning variables later passed in .fit(). - :param bin_statistic: Statistic of central tendency (e.g., mean) to apply during the binning. - :param bin_apply_method: Method to correct with the binned statistics, either "linear" to interpolate linearly - between bins, or "per_bin" to apply the statistic for each bin. - :param bias_var_names: (Optional) For pipelines, explicitly define bias variables names to use during .fit(). - :param subsample: Subsample the input for speed-up. <1 is parsed as a fraction. >1 is a pixel count. - """ - super().__init__( - fit_or_bin, - fit_func, - fit_optimizer, - bin_sizes, - bin_statistic, - bin_apply_method, - bias_var_names, - subsample, - ) - - def _fit_biascorr( # type: ignore - self, - ref_elev: NDArrayf, - tba_elev: NDArrayf, - inlier_mask: NDArrayb, - bias_vars: dict[str, NDArrayf], - transform: rio.transform.Affine, # Never None thanks to Coreg.fit() pre-process - crs: rio.crs.CRS, # Never None thanks to Coreg.fit() pre-process - z_name: str, - weights: None | NDArrayf = None, - verbose: bool = False, - **kwargs, - ) -> None: - """Estimate the bias along the single provided variable using the bias function.""" - - # Check number of variables - if len(bias_vars) != 1: - raise ValueError( - "A single variable has to be provided through the argument 'bias_vars', " - "got {}.".format(len(bias_vars)) - ) - - super()._fit_biascorr( - ref_elev=ref_elev, - tba_elev=tba_elev, - inlier_mask=inlier_mask, - bias_vars=bias_vars, - transform=transform, - crs=crs, - z_name=z_name, - weights=weights, - verbose=verbose, - **kwargs, - ) - - -class BiasCorr2D(BiasCorr): - """ - Bias-correction along two variables (e.g., X/Y coordinates, slope and curvature simultaneously). - """ - - def __init__( - self, - fit_or_bin: Literal["bin_and_fit"] | Literal["fit"] | Literal["bin"] = "fit", - fit_func: Callable[..., NDArrayf] = polynomial_2d, - fit_optimizer: Callable[..., tuple[NDArrayf, Any]] = scipy.optimize.curve_fit, - bin_sizes: int | dict[str, int | Iterable[float]] = 10, - bin_statistic: Callable[[NDArrayf], np.floating[Any]] = np.nanmedian, - bin_apply_method: Literal["linear"] | Literal["per_bin"] = "linear", - bias_var_names: Iterable[str] = None, - subsample: float | int = 1.0, - ): - """ - Instantiate a 2D bias correction. - - :param fit_or_bin: Whether to fit or bin. Use "fit" to correct by optimizing a function or - "bin" to correct with a statistic of central tendency in defined bins. - :param fit_func: Function to fit to the bias with variables later passed in .fit(). - :param fit_optimizer: Optimizer to minimize the function. - :param bin_sizes: Size (if integer) or edges (if iterable) for binning variables later passed in .fit(). - :param bin_statistic: Statistic of central tendency (e.g., mean) to apply during the binning. - :param bin_apply_method: Method to correct with the binned statistics, either "linear" to interpolate linearly - between bins, or "per_bin" to apply the statistic for each bin. - :param bias_var_names: (Optional) For pipelines, explicitly define bias variables names to use during .fit(). - :param subsample: Subsample the input for speed-up. <1 is parsed as a fraction. >1 is a pixel count. - """ - super().__init__( - fit_or_bin, - fit_func, - fit_optimizer, - bin_sizes, - bin_statistic, - bin_apply_method, - bias_var_names, - subsample, - ) - - def _fit_biascorr( # type: ignore - self, - ref_elev: NDArrayf, - tba_elev: NDArrayf, - inlier_mask: NDArrayb, - bias_vars: dict[str, NDArrayf], - transform: rio.transform.Affine, # Never None thanks to Coreg.fit() pre-process - crs: rio.crs.CRS, # Never None thanks to Coreg.fit() pre-process - z_name: str, - weights: None | NDArrayf = None, - verbose: bool = False, - **kwargs, - ) -> None: - - # Check number of variables - if len(bias_vars) != 2: - raise ValueError( - "Exactly two variables have to be provided through the argument 'bias_vars'" - ", got {}.".format(len(bias_vars)) - ) - - super()._fit_biascorr( - ref_elev=ref_elev, - tba_elev=tba_elev, - inlier_mask=inlier_mask, - bias_vars=bias_vars, - transform=transform, - crs=crs, - z_name=z_name, - weights=weights, - verbose=verbose, - **kwargs, - ) - - -class BiasCorrND(BiasCorr): - """ - Bias-correction along N variables (e.g., simultaneously slope, curvature, aspect and elevation). - """ - - def __init__( - self, - fit_or_bin: Literal["bin_and_fit"] | Literal["fit"] | Literal["bin"] = "bin", - fit_func: Callable[..., NDArrayf] - | Literal["norder_polynomial"] - | Literal["nfreq_sumsin"] = "norder_polynomial", - fit_optimizer: Callable[..., tuple[NDArrayf, Any]] = scipy.optimize.curve_fit, - bin_sizes: int | dict[str, int | Iterable[float]] = 10, - bin_statistic: Callable[[NDArrayf], np.floating[Any]] = np.nanmedian, - bin_apply_method: Literal["linear"] | Literal["per_bin"] = "linear", - bias_var_names: Iterable[str] = None, - subsample: float | int = 1.0, - ): - """ - Instantiate an N-D bias correction. - - :param fit_or_bin: Whether to fit or bin. Use "fit" to correct by optimizing a function or - "bin" to correct with a statistic of central tendency in defined bins. - :param fit_func: Function to fit to the bias with variables later passed in .fit(). - :param fit_optimizer: Optimizer to minimize the function. - :param bin_sizes: Size (if integer) or edges (if iterable) for binning variables later passed in .fit(). - :param bin_statistic: Statistic of central tendency (e.g., mean) to apply during the binning. - :param bin_apply_method: Method to correct with the binned statistics, either "linear" to interpolate linearly - between bins, or "per_bin" to apply the statistic for each bin. - :param bias_var_names: (Optional) For pipelines, explicitly define bias variables names to use during .fit(). - :param subsample: Subsample the input for speed-up. <1 is parsed as a fraction. >1 is a pixel count. - """ - super().__init__( - fit_or_bin, - fit_func, - fit_optimizer, - bin_sizes, - bin_statistic, - bin_apply_method, - bias_var_names, - subsample, - ) - - def _fit_biascorr( # type: ignore - self, - ref_elev: NDArrayf, - tba_elev: NDArrayf, - inlier_mask: NDArrayb, - bias_vars: dict[str, NDArrayf], # Never None thanks to BiasCorr.fit() pre-process - transform: rio.transform.Affine, # Never None thanks to Coreg.fit() pre-process - crs: rio.crs.CRS, # Never None thanks to Coreg.fit() pre-process - z_name: str, - weights: None | NDArrayf = None, - verbose: bool = False, - **kwargs, - ) -> None: - - # Check bias variable - if bias_vars is None or len(bias_vars) <= 2: - raise ValueError('At least three variables have to be provided through the argument "bias_vars".') - - super()._fit_biascorr( - ref_elev=ref_elev, - tba_elev=tba_elev, - inlier_mask=inlier_mask, - bias_vars=bias_vars, - transform=transform, - crs=crs, - z_name=z_name, - weights=weights, - verbose=verbose, - **kwargs, - ) - - -class DirectionalBias(BiasCorr1D): +class DirectionalBias(BiasCorr): """ Bias correction for directional biases, for example along- or across-track of satellite angle. """ @@ -849,7 +628,7 @@ def _apply_rst( return super()._apply_rst(elev=elev, transform=transform, crs=crs, bias_vars={"angle": x}, **kwargs) -class TerrainBias(BiasCorr1D): +class TerrainBias(BiasCorr): """ Correct a bias according to terrain, such as elevation or curvature. @@ -1016,7 +795,7 @@ def _apply_rst( return super()._apply_rst(elev=elev, transform=transform, crs=crs, bias_vars=bias_vars, **kwargs) -class Deramp(BiasCorr2D): +class Deramp(BiasCorr): """ Correct for a 2D polynomial along X/Y coordinates, for example from residual camera model deformations. """