From 8b08c0df2a12e73161f463e27f23d682b2ffcd15 Mon Sep 17 00:00:00 2001 From: Romain Hugonnet Date: Fri, 1 Sep 2023 18:47:18 -0800 Subject: [PATCH] Linting --- tests/test_coreg/test_base.py | 79 ++++++++++++++++++++++-------- tests/test_coreg/test_biascorr.py | 15 +++--- xdem/coreg/affine.py | 34 +++++++++++-- xdem/coreg/base.py | 80 +++++++++++++++++++++---------- xdem/coreg/biascorr.py | 43 ++++++++++++----- 5 files changed, 182 insertions(+), 69 deletions(-) diff --git a/tests/test_coreg/test_base.py b/tests/test_coreg/test_base.py index e84d566c..c403b98a 100644 --- a/tests/test_coreg/test_base.py +++ b/tests/test_coreg/test_base.py @@ -48,7 +48,7 @@ class TestCoregClass: # Create some 3D coordinates with Z coordinates being 0 to try the apply_pts functions. points = np.array([[1, 2, 3, 4], [1, 2, 3, 4], [0, 0, 0, 0]], dtype="float64").T - def test_init(self): + def test_init(self) -> None: """Test instantiation of Coreg""" c = coreg.Coreg() @@ -465,7 +465,14 @@ def test_pipeline(self) -> None: # Assert that the combined vertical shift is 2 assert pipeline2.to_matrix()[2, 3] == 2.0 - all_coregs = [coreg.VerticalShift(), coreg.NuthKaab(), coreg.ICP(), coreg.Deramp(), coreg.TerrainBias(), coreg.DirectionalBias()] + all_coregs = [ + coreg.VerticalShift(), + coreg.NuthKaab(), + coreg.ICP(), + coreg.Deramp(), + coreg.TerrainBias(), + coreg.DirectionalBias(), + ] @pytest.mark.parametrize("coreg1", all_coregs) # type: ignore @pytest.mark.parametrize("coreg2", all_coregs) # type: ignore @@ -479,12 +486,23 @@ def test_pipeline_combinations__nobiasvar(self, coreg1: Coreg, coreg2: Coreg) -> aligned_dem, _ = pipeline.apply(self.tba.data, transform=self.ref.transform, crs=self.ref.crs) assert aligned_dem.shape == self.ref.data.squeeze().shape - all_coregs = [coreg.VerticalShift(), coreg.NuthKaab(), coreg.ICP(), coreg.Deramp(), coreg.TerrainBias(), - coreg.DirectionalBias()] + all_coregs = [ + coreg.VerticalShift(), + coreg.NuthKaab(), + coreg.ICP(), + coreg.Deramp(), + coreg.TerrainBias(), + coreg.DirectionalBias(), + ] @pytest.mark.parametrize("coreg1", all_coregs) # type: ignore - @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")]) # type: ignore + @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"), + ], + ) # type: ignore def test_pipeline_combinations__biasvar(self, coreg1: Coreg, coreg2: Coreg) -> None: """Test pipelines with all combinations of coregistration subclasses with bias variables""" @@ -493,33 +511,54 @@ def test_pipeline_combinations__biasvar(self, coreg1: Coreg, coreg2: Coreg) -> N bias_vars = {"slope": xdem.terrain.slope(self.ref), "aspect": xdem.terrain.aspect(self.ref)} pipeline.fit(**self.fit_params, bias_vars=bias_vars) - aligned_dem, _ = pipeline.apply(self.tba.data, transform=self.ref.transform, crs=self.ref.crs, bias_vars=bias_vars) + aligned_dem, _ = pipeline.apply( + self.tba.data, transform=self.ref.transform, crs=self.ref.crs, bias_vars=bias_vars + ) assert aligned_dem.shape == self.ref.data.squeeze().shape - def test_pipeline__errors(self): + def test_pipeline__errors(self) -> None: """Test pipeline raises proper errors.""" pipeline = coreg.CoregPipeline([coreg.NuthKaab(), coreg.BiasCorr1D()]) - with pytest.raises(ValueError, match=re.escape("No `bias_vars` passed to .fit() for bias correction step " - " of the pipeline.")): + with pytest.raises( + ValueError, + match=re.escape( + "No `bias_vars` passed to .fit() for bias correction step " + " of the pipeline." + ), + ): pipeline.fit(**self.fit_params) - pipeline2 = coreg.CoregPipeline([coreg.NuthKaab(), coreg.BiasCorr1D(), coreg.BiasCorr1D()]) - with pytest.raises(ValueError, match=re.escape("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']).")): + with pytest.raises( + ValueError, + match=re.escape( + "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'])." + ), + ): pipeline2.fit(**self.fit_params) - with pytest.raises(ValueError, 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']).")): + with pytest.raises( + ValueError, + 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'])." + ), + ): pipeline2.fit(**self.fit_params, bias_vars={"slope": xdem.terrain.slope(self.ref)}) pipeline3 = coreg.CoregPipeline([coreg.NuthKaab(), coreg.BiasCorr1D(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'].")): + 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']." + ), + ): pipeline3.fit(**self.fit_params, bias_vars={"ncc": xdem.terrain.slope(self.ref)}) def test_pipeline_pts(self) -> None: diff --git a/tests/test_coreg/test_biascorr.py b/tests/test_coreg/test_biascorr.py index 372cf6dc..f9538fe9 100644 --- a/tests/test_coreg/test_biascorr.py +++ b/tests/test_coreg/test_biascorr.py @@ -87,7 +87,6 @@ def test_biascorr(self) -> None: bcorr5 = biascorr.BiasCorr(bias_var_names=np.array(["slope", "ncc"])) assert bcorr5._meta["bias_var_names"] == ["slope", "ncc"] - def test_biascorr__errors(self) -> None: """Test the errors that should be raised by BiasCorr.""" @@ -354,8 +353,10 @@ def test_biascorr1d(self) -> None: # 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'].") + 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} @@ -391,8 +392,11 @@ def test_biascorr2d(self) -> None: # 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'].") + 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)} @@ -539,7 +543,6 @@ def test_terrainbias(self) -> None: assert tb._meta["bias_var_names"] == ["maximum_curvature"] - def test_terrainbias__synthetic(self) -> None: """Test the subclass TerrainBias.""" diff --git a/xdem/coreg/affine.py b/xdem/coreg/affine.py index 936a59f1..d9f09613 100644 --- a/xdem/coreg/affine.py +++ b/xdem/coreg/affine.py @@ -298,6 +298,7 @@ def _fit_func( transform: rio.transform.Affine, crs: rio.crs.CRS, weights: NDArrayf | None, + bias_vars: dict[str, NDArrayf] | None = None, verbose: bool = False, **kwargs: Any, ) -> None: @@ -305,7 +306,12 @@ def _fit_func( raise NotImplementedError("This step has to be implemented by subclassing.") def _apply_func( - self, dem: NDArrayf, transform: rio.transform.Affine, crs: rio.crs.CRS, **kwargs: Any + self, + dem: NDArrayf, + transform: rio.transform.Affine, + crs: rio.crs.CRS, + bias_vars: dict[str, NDArrayf] | None = None, + **kwargs: Any, ) -> tuple[NDArrayf, rio.transform.Affine]: # FOR DEVELOPERS: This function is only needed for non-rigid transforms. raise NotImplementedError("This should have been implemented by subclassing") @@ -340,6 +346,7 @@ def _fit_func( transform: rio.transform.Affine, crs: rio.crs.CRS, weights: NDArrayf | None, + bias_vars: dict[str, NDArrayf] | None = None, verbose: bool = False, **kwargs: Any, ) -> None: @@ -368,7 +375,12 @@ def _fit_func( self._meta["vshift"] = vshift def _apply_func( - self, dem: NDArrayf, transform: rio.transform.Affine, crs: rio.crs.CRS, **kwargs: Any + self, + dem: NDArrayf, + transform: rio.transform.Affine, + crs: rio.crs.CRS, + bias_vars: dict[str, NDArrayf] | None = None, + **kwargs: Any, ) -> tuple[NDArrayf, rio.transform.Affine]: """Apply the VerticalShift function to a DEM.""" return dem + self._meta["vshift"], transform @@ -426,6 +438,7 @@ def _fit_func( transform: rio.transform.Affine, crs: rio.crs.CRS, weights: NDArrayf | None, + bias_vars: dict[str, NDArrayf] | None = None, verbose: bool = False, **kwargs: Any, ) -> None: @@ -575,6 +588,7 @@ def _fit_func( transform: rio.transform.Affine, crs: rio.crs.CRS, weights: NDArrayf | None, + bias_vars: dict[str, NDArrayf] | None = None, verbose: bool = False, **kwargs: Any, ) -> None: @@ -589,7 +603,12 @@ def _fit_func( self._meta["func"] = fit_ramp def _apply_func( - self, dem: NDArrayf, transform: rio.transform.Affine, crs: rio.crs.CRS, **kwargs: Any + self, + dem: NDArrayf, + transform: rio.transform.Affine, + crs: rio.crs.CRS, + bias_vars: dict[str, NDArrayf] | None = None, + **kwargs: Any, ) -> tuple[NDArrayf, rio.transform.Affine]: """Apply the deramp function to a DEM.""" x_coords, y_coords = _get_x_and_y_coords(dem.shape, transform) @@ -652,6 +671,7 @@ def _fit_func( transform: rio.transform.Affine, crs: rio.crs.CRS, weights: NDArrayf | None, + bias_vars: dict[str, NDArrayf] | None = None, verbose: bool = False, **kwargs: Any, ) -> None: @@ -928,7 +948,12 @@ def _to_matrix_func(self) -> NDArrayf: return matrix def _apply_func( - self, dem: NDArrayf, transform: rio.transform.Affine, crs: rio.crs.CRS, **kwargs: Any + self, + dem: NDArrayf, + transform: rio.transform.Affine, + crs: rio.crs.CRS, + bias_vars: dict[str, NDArrayf] | None = None, + **kwargs: Any, ) -> tuple[NDArrayf, rio.transform.Affine]: """Apply the Nuth & Kaab shift to a DEM.""" offset_east = self._meta["offset_east_px"] * self._meta["resolution"] @@ -1082,6 +1107,7 @@ def _fit_func( transform: rio.transform.Affine, crs: rio.crs.CRS, weights: NDArrayf | None, + bias_vars: dict[str, NDArrayf] | None = None, verbose: bool = False, **kwargs: Any, ) -> None: diff --git a/xdem/coreg/base.py b/xdem/coreg/base.py index 66f945cf..13b3c4f0 100644 --- a/xdem/coreg/base.py +++ b/xdem/coreg/base.py @@ -768,8 +768,15 @@ def fit( random_state=random_state, ) - main_args = {"ref_dem": ref_dem, "tba_dem": tba_dem, "transform": transform, "crs": crs, "weights": weights, - "verbose": verbose, "random_state": random_state} + main_args = { + "ref_dem": ref_dem, + "tba_dem": tba_dem, + "transform": transform, + "crs": crs, + "weights": weights, + "verbose": verbose, + "random_state": random_state, + } # If bias_vars are defined, update dictionary content to array if bias_vars is not None: @@ -1094,8 +1101,9 @@ def apply( kwargs["resample"] = resample # Run the associated apply function - applied_dem, out_transform = self._apply_func(**main_args, **kwargs - ) # pylint: disable=assignment-from-no-return + applied_dem, out_transform = self._apply_func( + **main_args, **kwargs + ) # pylint: disable=assignment-from-no-return # If it doesn't exist, use apply_matrix() except NotImplementedError: @@ -1309,9 +1317,12 @@ def _fit_func( raise NotImplementedError("This step has to be implemented by subclassing.") def _apply_func( - self, dem: NDArrayf, transform: rio.transform.Affine, crs: rio.crs.CRS, - bias_vars: dict[str, NDArrayf] | None = None, - **kwargs: Any + self, + dem: NDArrayf, + transform: rio.transform.Affine, + crs: rio.crs.CRS, + bias_vars: dict[str, NDArrayf] | None = None, + **kwargs: Any, ) -> tuple[NDArrayf, rio.transform.Affine]: # FOR DEVELOPERS: This function is only needed for non-rigid transforms. raise NotImplementedError("This should have been implemented by subclassing") @@ -1348,7 +1359,7 @@ def copy(self: CoregType) -> CoregType: return new_coreg - def _parse_bias_vars(self, step: int, bias_vars: dict[str, NDArrayf] | None): + def _parse_bias_vars(self, step: int, bias_vars: dict[str, NDArrayf] | None) -> dict[str, NDArrayf]: """Parse bias variables for a pipeline step requiring them.""" # Get number of non-affine coregistration requiring bias variables to be passed @@ -1362,28 +1373,33 @@ def _parse_bias_vars(self, step: int, bias_vars: dict[str, NDArrayf] | None): # Raise error if bias_vars is None if bias_vars is None: - msg = "No `bias_vars` passed to .fit() for bias correction step {} of the pipeline.".format(coreg.__class__) + msg = f"No `bias_vars` passed to .fit() for bias correction step {coreg.__class__} of the pipeline." if nb_needs_vars > 1: - msg += " 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. {}(bias_var_names=['slope']).".format(coreg.__class__.__name__) + msg += ( + " 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. {}(bias_var_names=['slope']).".format(coreg.__class__.__name__) + ) raise ValueError(msg) # Raise error if no variable were explicitly assigned and there is more than 1 step with bias_vars if var_names is None and nb_needs_vars > 1: - raise ValueError("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. {}(bias_var_names=['slope']).".format(coreg.__class__.__name__)) + raise ValueError( + "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. {}(bias_var_names=['slope']).".format(coreg.__class__.__name__) + ) # Raise error if the variables explicitly assigned don't match the ones passed in bias_vars if not all(n in bias_vars.keys() for n in var_names): - raise ValueError("Not all keys of `bias_vars` in .fit() match the `bias_var_names` defined during " - "instantiation of the bias correction step {}: {}.".format(coreg.__class__, var_names)) + raise ValueError( + "Not all keys of `bias_vars` in .fit() match the `bias_var_names` defined during " + "instantiation of the bias correction step {}: {}.".format(coreg.__class__, var_names) + ) # Add subset dict for this pipeline step to args of fit and apply return {n: bias_vars[n] for n in var_names} - def _fit_func( self, ref_dem: NDArrayf, @@ -1402,8 +1418,14 @@ def _fit_func( if verbose: print(f"Running pipeline step: {i + 1} / {len(self.pipeline)}") - main_args_fit = {"ref_dem": ref_dem, "tba_dem": tba_dem_mod, "transform": transform, "crs": crs, - "weights": weights, "verbose": verbose} + main_args_fit = { + "ref_dem": ref_dem, + "tba_dem": tba_dem_mod, + "transform": transform, + "crs": crs, + "weights": weights, + "verbose": verbose, + } main_args_apply = {"dem": tba_dem_mod, "transform": transform, "crs": crs} @@ -1441,11 +1463,11 @@ def _fit_pts_func( def _apply_func( self, - dem: NDArrayf, - transform: rio.transform.Affine, - crs: rio.crs.CRS, - bias_vars: dict[str, NDArrayf] | None = None, - **kwargs: Any + dem: NDArrayf, + transform: rio.transform.Affine, + crs: rio.crs.CRS, + bias_vars: dict[str, NDArrayf] | None = None, + **kwargs: Any, ) -> tuple[NDArrayf, rio.transform.Affine]: """Apply the coregistration steps sequentially to a DEM.""" dem_mod = dem.copy() @@ -1560,6 +1582,7 @@ def _fit_func( transform: rio.transform.Affine, crs: rio.crs.CRS, weights: NDArrayf | None, + bias_vars: dict[str, NDArrayf] | None = None, verbose: bool = False, **kwargs: Any, ) -> None: @@ -1801,7 +1824,12 @@ def subdivide_array(self, shape: tuple[int, ...]) -> NDArrayf: return subdivide_array(shape, count=self.subdivision) def _apply_func( - self, dem: NDArrayf, transform: rio.transform.Affine, crs: rio.crs.CRS, **kwargs: Any + self, + dem: NDArrayf, + transform: rio.transform.Affine, + crs: rio.crs.CRS, + bias_vars: dict[str, NDArrayf] | None = None, + **kwargs: Any, ) -> tuple[NDArrayf, rio.transform.Affine]: if np.count_nonzero(np.isfinite(dem)) == 0: diff --git a/xdem/coreg/biascorr.py b/xdem/coreg/biascorr.py index 7bbe2db2..b312800c 100644 --- a/xdem/coreg/biascorr.py +++ b/xdem/coreg/biascorr.py @@ -28,6 +28,7 @@ BiasCorrType = TypeVar("BiasCorrType", bound="BiasCorr") + class BiasCorr(Coreg): """ Parent class of bias correction methods: non-rigid coregistrations. @@ -110,8 +111,12 @@ def __init__( # For binning elif fit_or_bin == "bin": - meta_bin = {"bin_sizes": bin_sizes, "bin_statistic": bin_statistic, "bin_apply_method": bin_apply_method, - "bias_var_names": list_bias_var_names} + meta_bin = { + "bin_sizes": bin_sizes, + "bin_statistic": bin_statistic, + "bin_apply_method": bin_apply_method, + "bias_var_names": list_bias_var_names, + } super().__init__(meta=meta_bin) # type: ignore # For both @@ -121,7 +126,7 @@ def __init__( "fit_optimizer": fit_optimizer, "bin_sizes": bin_sizes, "bin_statistic": bin_statistic, - "bias_var_names": list_bias_var_names + "bias_var_names": list_bias_var_names, } super().__init__(meta=meta_bin_and_fit) # type: ignore @@ -151,9 +156,11 @@ def _fit_func( # type: ignore # 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(list(bias_vars.keys())) == sorted(self._meta["bias_var_names"]): - raise ValueError("The keys of `bias_vars` do not match the `bias_var_names` defined during " - "instantiation: {}.".format(self._meta["bias_var_names"])) + if not sorted(bias_vars.keys()) == sorted(self._meta["bias_var_names"]): + raise ValueError( + "The keys of `bias_vars` do not match the `bias_var_names` defined during " + "instantiation: {}.".format(self._meta["bias_var_names"]) + ) # Otherwise, store bias variable names from the dictionary else: self._meta["bias_var_names"] = list(bias_vars.keys()) @@ -313,9 +320,11 @@ def _apply_func( # type: ignore raise ValueError("At least one `bias_var` should be passed to the `apply` function, got None.") # Check the bias_vars passed match the ones stored for this bias correction class - if not sorted(list(bias_vars.keys())) == sorted(self._meta["bias_var_names"]): - raise ValueError("The keys of `bias_vars` do not match the `bias_var_names` defined during " - "instantiation or fitting: {}.".format(self._meta["bias_var_names"])) + if not sorted(bias_vars.keys()) == sorted(self._meta["bias_var_names"]): + raise ValueError( + "The keys of `bias_vars` do not match the `bias_var_names` defined during " + "instantiation or fitting: {}.".format(self._meta["bias_var_names"]) + ) # Apply function to get correction (including if binning was done before) if self._fit_or_bin in ["fit", "bin_and_fit"]: @@ -380,7 +389,9 @@ def __init__( 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(). """ - super().__init__(fit_or_bin, fit_func, fit_optimizer, bin_sizes, bin_statistic, bin_apply_method, bias_var_names) + super().__init__( + fit_or_bin, fit_func, fit_optimizer, bin_sizes, bin_statistic, bin_apply_method, bias_var_names + ) def _fit_func( # type: ignore self, @@ -442,7 +453,9 @@ def __init__( 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(). """ - super().__init__(fit_or_bin, fit_func, fit_optimizer, bin_sizes, bin_statistic, bin_apply_method, bias_var_names) + super().__init__( + fit_or_bin, fit_func, fit_optimizer, bin_sizes, bin_statistic, bin_apply_method, bias_var_names + ) def _fit_func( # type: ignore self, @@ -505,7 +518,9 @@ def __init__( 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(). """ - super().__init__(fit_or_bin, fit_func, fit_optimizer, bin_sizes, bin_statistic, bin_apply_method, bias_var_names) + super().__init__( + fit_or_bin, fit_func, fit_optimizer, bin_sizes, bin_statistic, bin_apply_method, bias_var_names + ) def _fit_func( # type: ignore self, @@ -660,7 +675,9 @@ def __init__( between bins, or "per_bin" to apply the statistic for each bin. """ - super().__init__(fit_or_bin, fit_func, fit_optimizer, bin_sizes, bin_statistic, bin_apply_method, [terrain_attribute]) + super().__init__( + fit_or_bin, fit_func, fit_optimizer, bin_sizes, bin_statistic, bin_apply_method, [terrain_attribute] + ) # This is the same as bias_var_names, but let's leave the duplicate for clarity self._meta["terrain_attribute"] = terrain_attribute self._needs_vars = False