Add CLI functions for Points to raster and Generate negatives tools

eis_toolkit/cli.py

@@ -15,6 +15,7 @@
 import numpy as np
 import pandas as pd
 import rasterio
+import rasterio.profiles
 import typer
 from beartype.typing import List, Optional, Sequence, Tuple, Union
 from typing_extensions import Annotated
@@ -355,6 +356,14 @@ class ReplaceCondition(str, Enum):
     greater_than_or_equal = "greater_than_or_equal"
 
 
+class BufferOption(str, Enum):
+    """Buffer options."""
+
+    avg = "avg"
+    min = "min"
+    max = "max"
+
+
 INPUT_FILE_OPTION = Annotated[
     Path,
     typer.Option(
@@ -438,7 +447,8 @@ def saving_output_files(savepath: Union[str, Sequence[str]]):  # noqa: D102
         yield
         if isinstance(savepath, Sequence):
             for file in savepath:
-                typer.echo(f"✅ Output file(s) saved to {file}\n")
+                typer.echo(f"✅ Output file(s) saved to {file}")
+            typer.echo(" ")
         else:
             typer.echo(f"✅ Output file(s) saved to {savepath}\n")
 
@@ -1963,23 +1973,14 @@ def idw_interpolation_cli(
     search_radius: Optional[float] = None,
 ):
     """Apply inverse distance weighting (IDW) interpolation to input vector file."""
-    from eis_toolkit.exceptions import InvalidParameterValueException
-    from eis_toolkit.utilities.raster import profile_from_extent_and_pixel_size
+    from eis_toolkit.utilities.raster import base_profile
     from eis_toolkit.vector_processing.idw_interpolation import idw
 
     with ProgressLog.reading_input_files():
         geodataframe = gpd.read_file(input_vector)
 
         if base_raster is None or base_raster == "":
-            if any(bound is None for bound in extent) or pixel_size is None or pixel_size <= 0:
-                raise InvalidParameterValueException(
-                    "Expected positive pixel size and defined extent in absence of base raster. "
-                    + f"Pixel size: {pixel_size}, extent: {extent}."
-                )
-            profile = profile_from_extent_and_pixel_size(extent, (pixel_size, pixel_size))
-            profile["crs"] = geodataframe.crs
-            profile["driver"] = "GTiff"
-            profile["dtype"] = "float32"
+            profile = base_profile(extent, pixel_size, geodataframe.crs)
         else:
             with rasterio.open(base_raster) as raster:
                 profile = raster.profile.copy()
@@ -1993,7 +1994,6 @@ def idw_interpolation_cli(
             search_radius=search_radius,
         )
 
-    profile["count"] = 1
     with ProgressLog.saving_output_files(output_raster):
         with rasterio.open(output_raster, "w", **profile) as dst:
             dst.write(out_image, 1)
@@ -2015,23 +2015,14 @@ def kriging_interpolation_cli(
     method: Annotated[KrigingMethod, typer.Option(case_sensitive=False)] = KrigingMethod.ordinary,
 ):
     """Apply kriging interpolation to input vector file."""
-    from eis_toolkit.exceptions import InvalidParameterValueException
-    from eis_toolkit.utilities.raster import profile_from_extent_and_pixel_size
+    from eis_toolkit.utilities.raster import base_profile
     from eis_toolkit.vector_processing.kriging_interpolation import kriging
 
     with ProgressLog.reading_input_files():
         geodataframe = gpd.read_file(input_vector)
 
         if base_raster is None or base_raster == "":
-            if any(bound is None for bound in extent) or pixel_size is None or pixel_size <= 0:
-                raise InvalidParameterValueException(
-                    "Expected positive pixel size and defined extent in absence of base raster. "
-                    + f"Pixel size: {pixel_size}, extent: {extent}."
-                )
-            profile = profile_from_extent_and_pixel_size(extent, (pixel_size, pixel_size))
-            profile["crs"] = geodataframe.crs
-            profile["driver"] = "GTiff"
-            profile["dtype"] = "float32"
+            profile = base_profile(extent, pixel_size, geodataframe.crs)
         else:
             with rasterio.open(base_raster) as raster:
                 profile = raster.profile.copy()
@@ -2046,7 +2037,6 @@ def kriging_interpolation_cli(
             method=get_enum_values(method),
         )
 
-    profile["count"] = 1
     with ProgressLog.saving_output_files(output_raster):
         with rasterio.open(output_raster, "w", **profile) as dst:
             dst.write(out_image, 1)
@@ -2073,23 +2063,14 @@ def rasterize_cli(
 
     Either base raster or pixel size + extent must be provided.
     """
-    from eis_toolkit.exceptions import InvalidParameterValueException
-    from eis_toolkit.utilities.raster import profile_from_extent_and_pixel_size
+    from eis_toolkit.utilities.raster import base_profile
     from eis_toolkit.vector_processing.rasterize_vector import rasterize_vector
 
     with ProgressLog.reading_input_files():
         geodataframe = gpd.read_file(input_vector)
 
         if base_raster is None or base_raster == "":
-            if any(bound is None for bound in extent) or pixel_size is None or pixel_size <= 0:
-                raise InvalidParameterValueException(
-                    "Expected positive pixel size and defined extent in absence of base raster. "
-                    + f"Pixel size: {pixel_size}, extent: {extent}."
-                )
-            profile = profile_from_extent_and_pixel_size(extent, (pixel_size, pixel_size))
-            profile["crs"] = geodataframe.crs
-            profile["driver"] = "GTiff"
-            profile["dtype"] = "float32"
+            profile = base_profile(extent, pixel_size, geodataframe.crs)
         else:
             with rasterio.open(base_raster) as raster:
                 profile = raster.profile.copy()
@@ -2105,7 +2086,6 @@ def rasterize_cli(
             get_enum_values(merge_strategy),
         )
 
-    profile["count"] = 1
     with ProgressLog.saving_output_files(output_raster):
         with rasterio.open(output_raster, "w", **profile) as dst:
             dst.write(out_image, 1)
@@ -2151,23 +2131,14 @@ def vector_density_cli(
 
     Either base raster or pixel size + extent must be provided.
     """
-    from eis_toolkit.exceptions import InvalidParameterValueException
-    from eis_toolkit.utilities.raster import profile_from_extent_and_pixel_size
+    from eis_toolkit.utilities.raster import base_profile
     from eis_toolkit.vector_processing.vector_density import vector_density
 
     with ProgressLog.reading_input_files():
         geodataframe = gpd.read_file(input_vector)
 
         if base_raster is None or base_raster == "":
-            if any(bound is None for bound in extent) or pixel_size is None or pixel_size <= 0:
-                raise InvalidParameterValueException(
-                    "Expected positive pixel size and defined extent in absence of base raster. "
-                    + f"Pixel size: {pixel_size}, extent: {extent}."
-                )
-            profile = profile_from_extent_and_pixel_size(extent, (pixel_size, pixel_size))
-            profile["crs"] = geodataframe.crs
-            profile["driver"] = "GTiff"
-            profile["dtype"] = "float32"
+            profile = base_profile(extent, pixel_size, geodataframe.crs)
         else:
             with rasterio.open(base_raster) as raster:
                 profile = raster.profile.copy()
@@ -2180,7 +2151,6 @@ def vector_density_cli(
             statistic=get_enum_values(statistic),
         )
 
-    profile["count"] = 1
     with ProgressLog.saving_output_files(output_raster):
         with rasterio.open(output_raster, "w", **profile) as dst:
             dst.write(out_image, 1)
@@ -2199,23 +2169,14 @@ def distance_computation_cli(
     max_distance: float = None,
 ):
     """Calculate distance from raster cell to nearest geometry."""
-    from eis_toolkit.exceptions import InvalidParameterValueException
-    from eis_toolkit.utilities.raster import profile_from_extent_and_pixel_size
+    from eis_toolkit.utilities.raster import base_profile
     from eis_toolkit.vector_processing.distance_computation import distance_computation
 
     with ProgressLog.reading_input_files():
         geodataframe = gpd.read_file(input_vector)
 
         if base_raster is None or base_raster == "":
-            if any(bound is None for bound in extent) or pixel_size is None or pixel_size <= 0:
-                raise InvalidParameterValueException(
-                    "Expected positive pixel size and defined extent in absence of base raster. "
-                    + f"Pixel size: {pixel_size}, extent: {extent}."
-                )
-            profile = profile_from_extent_and_pixel_size(extent, (pixel_size, pixel_size))
-            profile["crs"] = geodataframe.crs
-            profile["driver"] = "GTiff"
-            profile["dtype"] = "float32"
+            profile = base_profile(extent, pixel_size, geodataframe.crs)
             mask = None
         else:
             with rasterio.open(base_raster) as raster:
@@ -2289,23 +2250,14 @@ def proximity_computation_cli(
     extent: Tuple[float, float, float, float] = (None, None, None, None),
 ):
     """Calculate proximity from raster cell to nearest geometry."""
-    from eis_toolkit.exceptions import InvalidParameterValueException
-    from eis_toolkit.utilities.raster import profile_from_extent_and_pixel_size
+    from eis_toolkit.utilities.raster import base_profile
     from eis_toolkit.vector_processing.proximity_computation import proximity_computation
 
     with ProgressLog.reading_input_files():
         geodataframe = gpd.read_file(input_vector)
 
         if base_raster is None or base_raster == "":
-            if any(bound is None for bound in extent) or pixel_size is None or pixel_size <= 0:
-                raise InvalidParameterValueException(
-                    "Expected positive pixel size and defined extent in absence of base raster. "
-                    + f"Pixel size: {pixel_size}, extent: {extent}."
-                )
-            profile = profile_from_extent_and_pixel_size(extent, (pixel_size, pixel_size))
-            profile["crs"] = geodataframe.crs
-            profile["driver"] = "GTiff"
-            profile["dtype"] = "float32"
+            profile = base_profile(extent, pixel_size, geodataframe.crs)
             mask = None
         else:
             with rasterio.open(base_raster) as raster:
@@ -2332,6 +2284,84 @@ def proximity_computation_cli(
     ProgressLog.finish()
 
 
+# --- TRAINING DATA TOOLS ---
+
+# POINTS TO RASTER
+@app.command()
+def points_to_raster_cli(
+    input_vector: INPUT_FILE_OPTION,
+    output_raster: OUTPUT_FILE_OPTION,
+    base_raster: INPUT_FILE_OPTION = None,
+    pixel_size: float = None,
+    extent: Tuple[float, float, float, float] = (None, None, None, None),
+    attribute: Optional[str] = None,
+    radius: Optional[int] = None,
+    buffer: Annotated[BufferOption, typer.Option(case_sensitive=False)] = None,
+):
+    """Convert a point data set into a binary raster."""
+    from eis_toolkit.training_data_tools.points_to_raster import points_to_raster
+    from eis_toolkit.utilities.raster import base_profile
+
+    with ProgressLog.reading_input_files():
+        geodataframe = gpd.read_file(input_vector)
+
+        if base_raster is None or base_raster == "":
+            profile = base_profile(extent, pixel_size, geodataframe.crs)
+            mask = None
+        else:
+            with rasterio.open(base_raster) as raster:
+                profile = raster.profile.copy()
+                raster_array = raster.read(1)
+                mask = (raster_array == profile["nodata"]) | np.isnan(raster_array)
+
+    with ProgressLog.running_algorithm():
+        out_image, out_profile = points_to_raster(
+            geodataframe=geodataframe, raster_profile=profile, attribute=attribute, radius=radius, buffer=buffer
+        )
+
+    # Apply nodata mask
+    if mask is not None:
+        out_image[mask] = out_profile["nodata"]
+
+    with ProgressLog.saving_output_files(output_raster):
+        with rasterio.open(output_raster, "w", **out_profile) as dst:
+            dst.write(out_image, 1)
+
+    ProgressLog.finish()
+
+
+# GENERATE NEGATIVES
+@app.command()
+def generate_negatives_cli(
+    input_raster: INPUT_FILE_OPTION,
+    output_raster: OUTPUT_FILE_OPTION,
+    output_vector: OUTPUT_FILE_OPTION,
+    sample_number: Annotated[int, typer.Option()],
+    random_seed: int = 48,
+):
+    """Generate probable negatives from raster array with marked positives."""
+    from eis_toolkit.training_data_tools.random_sampling import generate_negatives
+
+    with ProgressLog.reading_input_files():
+        with rasterio.open(input_raster) as raster:
+            raster_array = raster.read(1)
+            profile = raster.profile.copy()
+            mask = (raster_array == profile["nodata"]) | np.isnan(raster_array)
+
+    with ProgressLog.running_algorithm():
+        out_points, out_image, out_profile = generate_negatives(
+            raster_array=raster_array, raster_profile=profile, sample_number=sample_number, random_seed=random_seed
+        )
+    out_image[mask] = out_profile["nodata"]
+
+    with ProgressLog.saving_output_files([output_raster, output_vector]):
+        with rasterio.open(output_raster, "w", **out_profile) as dst:
+            dst.write(out_image, 1)
+        out_points.to_file(output_vector)
+
+    ProgressLog.finish()
+
+
 # --- PREDICTION ---
 
 

eis_toolkit/training_data_tools/points_to_raster.py

@@ -124,6 +124,7 @@ def points_to_raster(
 
     Args:
         geodataframe: The geodataframe points set to be converted into raster.
+        raster_profile: The raster profile determining the output raster grid properties.
         attribute: Values to be be assigned to the geodataframe.
         radius: Radius to be applied around the geodataframe in [m].
         buffer: Buffers the matrix value when two or more radii with different attribute value overlap.

eis_toolkit/training_data_tools/random_sampling.py

@@ -33,23 +33,23 @@ def _random_sampling(
 @beartype
 def generate_negatives(
     raster_array: np.ndarray,
-    raster_meta: Union[profiles.Profile, dict],
+    raster_profile: Union[profiles.Profile, dict],
     sample_number: Number,
     random_seed: int = 48,
 ) -> Tuple[gpd.GeoDataFrame, np.ndarray, Union[profiles.Profile, dict]]:
     """Generate probable negatives from raster array with marked positives.
 
     Args:
         raster_array: Raster array with marked positives.
-        raster_meta: Raster metadata.
-        sample_number: maximum number of negatives to be generated.
+        raster_profile: The raster profile determining the output raster grid properties.
+        sample_number: Maximum number of negatives to be generated.
         random_seed: Seed for generating random negatives.
 
     Returns:
         A tuple containing the shapely points, output raster as a NumPy array and updated metadata.
 
     Raises:
-        EmptyDataException:  The raster array is empty.
+        EmptyDataException: The raster array is empty.
     """
 
     if raster_array.size == 0:
@@ -80,11 +80,11 @@ def generate_negatives(
 
     out_array[row, col] = -1
 
-    x, y = rasterio.transform.xy(raster_meta["transform"], row, col)
+    x, y = rasterio.transform.xy(raster_profile["transform"], row, col)
 
     points = [Point(x[i], y[i]) for i in range(len(x))]
 
     sample_negative = gpd.GeoDataFrame(geometry=points)
-    sample_negative.set_crs(raster_meta["crs"], allow_override=True, inplace=True)
+    sample_negative.set_crs(raster_profile["crs"], allow_override=True, inplace=True)
 
-    return sample_negative, out_array, raster_meta
+    return sample_negative, out_array, raster_profile