Merge pull request #382 from GispoCoding/354-distance-computation-criteria

354 distance computation criteria

Author: nmaarnio

eis_toolkit/cli.py

@@ -1194,6 +1194,7 @@ def distance_to_anomaly_cli(
     threshold_criteria: Annotated[ThresholdCriteria, typer.Option(case_sensitive=False)],
     first_threshold_criteria_value: float = typer.Option(),
     second_threshold_criteria_value: float = None,
+    max_distance: float = None,
 ):
     """
     Calculate distance from each raster cell to nearest anomaly cell.
@@ -1216,6 +1217,7 @@ def distance_to_anomaly_cli(
             anomaly_raster_data=raster.read(1),
             threshold_criteria_value=threshold_criteria_value,
             threshold_criteria=get_enum_values(threshold_criteria),
+            max_distance=max_distance,
         )
 
     typer.echo("Progress: 75%")
@@ -1979,6 +1981,7 @@ def distance_computation_cli(
     base_raster: INPUT_FILE_OPTION = None,
     pixel_size: float = None,
     extent: Tuple[float, float, float, float] = (None, None, None, None),
+    max_distance: float = None,
 ):
     """Calculate distance from raster cell to nearest geometry."""
     from eis_toolkit.exceptions import InvalidParameterValueException
@@ -2004,7 +2007,7 @@ def distance_computation_cli(
         with rasterio.open(base_raster) as raster:
             profile = raster.profile.copy()
 
-    out_image = distance_computation(geodataframe=geodataframe, raster_profile=profile)
+    out_image = distance_computation(geodataframe=geodataframe, raster_profile=profile, max_distance=max_distance)
     profile["count"] = 1
     typer.echo("Progress: 75%")
 

eis_toolkit/raster_processing/distance_to_anomaly.py

@@ -40,6 +40,7 @@ def distance_to_anomaly(
     anomaly_raster_data: np.ndarray,
     threshold_criteria_value: Union[Tuple[Number, Number], Number],
     threshold_criteria: Literal["lower", "higher", "in_between", "outside"],
+    max_distance: Optional[Number] = None,
 ) -> Tuple[np.ndarray, Union[profiles.Profile, dict]]:
     """Calculate distance from raster cell to nearest anomaly.
 
@@ -59,6 +60,7 @@ def distance_to_anomaly(
             the first value should be the minimum and the second
             the maximum value.
         threshold_criteria: Method to define anomalous.
+        max_distance: The maximum distance in the output array.
 
     Returns:
         A 2D numpy array with the distances to anomalies computed
@@ -75,6 +77,7 @@ def distance_to_anomaly(
         anomaly_raster_data=anomaly_raster_data,
         threshold_criteria=threshold_criteria,
         threshold_criteria_value=threshold_criteria_value,
+        max_distance=max_distance,
     )
     return out_image, anomaly_raster_profile
 
@@ -201,6 +204,7 @@ def _distance_to_anomaly(
     anomaly_raster_data: np.ndarray,
     threshold_criteria_value: Union[Tuple[Number, Number], Number],
     threshold_criteria: Literal["lower", "higher", "in_between", "outside"],
+    max_distance: Optional[Number],
 ) -> np.ndarray:
     data_fits_criteria = _fits_criteria(
         threshold_criteria=threshold_criteria,
@@ -230,6 +234,8 @@ def _distance_to_anomaly(
     all_points = list(chain(*all_points_by_rows))
     all_points_gdf = gpd.GeoDataFrame(geometry=all_points, crs=anomaly_raster_profile["crs"])
 
-    distance_array = distance_computation(raster_profile=anomaly_raster_profile, geodataframe=all_points_gdf)
+    distance_array = distance_computation(
+        raster_profile=anomaly_raster_profile, geodataframe=all_points_gdf, max_distance=max_distance
+    )
 
     return distance_array

eis_toolkit/vector_processing/distance_computation.py

@@ -1,23 +1,28 @@
+from numbers import Number
+
 import geopandas as gpd
 import numpy as np
 from beartype import beartype
-from beartype.typing import Union
+from beartype.typing import Optional, Union
 from rasterio import profiles, transform
 from shapely.geometry.base import BaseGeometry, BaseMultipartGeometry
 
-from eis_toolkit.exceptions import EmptyDataFrameException, NonMatchingCrsException
+from eis_toolkit.exceptions import EmptyDataFrameException, NonMatchingCrsException, NumericValueSignException
 from eis_toolkit.utilities.checks.raster import check_raster_profile
 from eis_toolkit.utilities.miscellaneous import row_points
 
 
 @beartype
-def distance_computation(geodataframe: gpd.GeoDataFrame, raster_profile: Union[profiles.Profile, dict]) -> np.ndarray:
+def distance_computation(
+    geodataframe: gpd.GeoDataFrame, raster_profile: Union[profiles.Profile, dict], max_distance: Optional[Number] = None
+) -> np.ndarray:
     """Calculate distance from raster cell to nearest geometry.
 
     Args:
         geodataframe: The GeoDataFrame with geometries to determine distance to.
         raster_profile: The raster profile of the raster in which the distances
             to the nearest geometry are determined.
+        max_distance: The maximum distance in the output array.
 
     Returns:
         A 2D numpy array with the distances computed.
@@ -30,19 +35,25 @@ def distance_computation(geodataframe: gpd.GeoDataFrame, raster_profile: Union[p
         raise NonMatchingCrsException("Expected coordinate systems to match between raster and GeoDataFrame.")
     if geodataframe.shape[0] == 0:
         raise EmptyDataFrameException("Expected GeoDataFrame to not be empty.")
+    if max_distance is not None and max_distance <= 0:
+        raise NumericValueSignException("Expected max distance to be a positive number.")
 
     check_raster_profile(raster_profile=raster_profile)
 
     raster_width = raster_profile.get("width")
     raster_height = raster_profile.get("height")
     raster_transform = raster_profile.get("transform")
 
-    return _distance_computation(
+    distance_matrix = _distance_computation(
         raster_width=raster_width,
         raster_height=raster_height,
         raster_transform=raster_transform,
         geodataframe=geodataframe,
     )
+    if max_distance is not None:
+        distance_matrix[distance_matrix > max_distance] = max_distance
+
+    return distance_matrix
 
 
 def _calculate_row_distances(