Merge pull request #371 from GispoCoding/370-add-raster-utility-tools

370 add raster utility tools

docs/utilities/file_io.md

@@ -0,0 +1,3 @@
+# File I/O utilities
+
+::: eis_toolkit.utilities.file_io

docs/utilities/nodata.md

@@ -0,0 +1,3 @@
+# Nodata utilities
+
+::: eis_toolkit.utilities.nodata

docs/utilities/raster.md

@@ -0,0 +1,3 @@
+# Raster data utilities
+
+::: eis_toolkit.utilities.raster

eis_toolkit/cli.py

@@ -2888,6 +2888,123 @@ def winsorize_transform_cli(
 # TODO
 
 
+# --- UTILITIES ---
+@app.command()
+def split_raster_bands_cli(input_raster: INPUT_FILE_OPTION, output_dir: OUTPUT_DIR_OPTION):  # CHECK
+    """Split multiband raster into singleband rasters."""
+    from eis_toolkit.utilities.file_io import get_output_paths_from_common_name
+    from eis_toolkit.utilities.raster import split_raster_bands
+
+    typer.echo("Progress: 10%")
+
+    with rasterio.open(input_raster) as raster:
+        typer.echo("Progress: 25%")
+        output_singleband_rasters = split_raster_bands(raster)
+    typer.echo("Progress: 70%")
+
+    name = os.path.splitext(os.path.basename(input_raster))[0]
+    output_paths = get_output_paths_from_common_name(output_singleband_rasters, output_dir, f"{name}_split", ".tif")
+    for output_path, (out_image, out_profile) in zip(output_paths, output_singleband_rasters):
+        with rasterio.open(output_path, "w", **out_profile) as dst:
+            dst.write(out_image, 1)
+    typer.echo("Progress: 100%")
+
+    typer.echo(f"Splitting raster completed, writing rasters to directory {output_dir}.")
+
+
+@app.command()
+def combine_raster_bands_cli(input_rasters: INPUT_FILES_ARGUMENT, output_raster: OUTPUT_FILE_OPTION):
+    """Combine multiple rasters into one multiband raster."""
+    from eis_toolkit.utilities.raster import combine_raster_bands
+
+    typer.echo("Progress: 10%")
+
+    rasters = [rasterio.open(raster) for raster in input_rasters]  # Open all rasters to be combined
+    typer.echo("Progress: 25%")
+
+    out_image, out_meta = combine_raster_bands(rasters)
+    [raster.close() for raster in rasters]
+    typer.echo("Progress: 70%")
+
+    with rasterio.open(output_raster, "w", **out_meta) as dst:
+        dst.write(out_image)
+    typer.echo("Progress: 100%")
+
+    typer.echo(f"Combining rasters completed, writing raster to {output_raster}.")
+
+
+@app.command()
+def unify_raster_nodata_cli(
+    input_rasters: INPUT_FILES_ARGUMENT, output_dir: OUTPUT_DIR_OPTION, new_nodata: float = -9999  # CHECK
+):
+    """Unifies nodata for the input rasters."""
+    from eis_toolkit.utilities.file_io import get_output_paths_from_inputs
+    from eis_toolkit.utilities.nodata import unify_raster_nodata
+
+    typer.echo("Progress: 10%")
+
+    rasters = [rasterio.open(raster) for raster in input_rasters]  # Open all rasters to be unified
+    typer.echo("Progress: 25%")
+
+    unified = unify_raster_nodata(rasters, new_nodata)
+    [raster.close() for raster in rasters]
+    typer.echo("Progress: 70%")
+
+    output_paths = get_output_paths_from_inputs(input_rasters, output_dir, "nodata_unified", ".tif")
+    for output_path, (out_image, out_profile) in zip(output_paths, unified):
+        with rasterio.open(output_path, "w", **out_profile) as dst:
+            dst.write(out_image)
+    typer.echo("Progress: 100%")
+
+    typer.echo(f"Unifying nodata completed, writing rasters to directory {output_dir}.")
+
+
+@app.command()
+def convert_raster_nodata_cli(
+    input_raster: INPUT_FILE_OPTION,
+    output_raster: OUTPUT_FILE_OPTION,
+    old_nodata: float = None,
+    new_nodata: float = -9999,
+):
+    """Convert existing nodata values with a new nodata value for a raster."""
+    from eis_toolkit.utilities.nodata import convert_raster_nodata
+
+    typer.echo("Progress: 10%")
+
+    with rasterio.open(input_raster) as raster:
+        typer.echo("Progress: 25%")
+        out_image, out_meta = convert_raster_nodata(raster, old_nodata, new_nodata)
+    typer.echo("Progress: 70%")
+
+    with rasterio.open(output_raster, "w", **out_meta) as dst:
+        dst.write(out_image)
+    typer.echo("Progress: 100%")
+
+    typer.echo(f"Converting nodata completed, writing raster to {output_raster}.")
+
+
+@app.command()
+def set_raster_nodata_cli(
+    input_raster: INPUT_FILE_OPTION, output_raster: OUTPUT_FILE_OPTION, new_nodata: float = typer.Option()
+):
+    """Set new nodata value for raster profile."""
+    from eis_toolkit.utilities.nodata import set_raster_nodata
+
+    typer.echo("Progress: 10%")
+
+    with rasterio.open(input_raster) as raster:
+        out_image = raster.read()
+        typer.echo("Progress: 25%")
+        out_meta = set_raster_nodata(raster.meta, new_nodata)
+    typer.echo("Progress: 70%")
+
+    with rasterio.open(output_raster, "w", **out_meta) as dst:
+        dst.write(out_image)
+    typer.echo("Progress: 100%")
+
+    typer.echo(f"Setting nodata completed, writing raster to {output_raster}.")
+
+
 # if __name__ == "__main__":
 def cli():
     """CLI app."""

eis_toolkit/prediction/fuzzy_overlay.py

@@ -3,7 +3,7 @@
 from beartype.typing import Sequence, Union
 
 from eis_toolkit.exceptions import InvalidDatasetException, InvalidParameterValueException
-from eis_toolkit.utilities.miscellaneous import stack_raster_arrays
+from eis_toolkit.utilities.raster import stack_raster_arrays
 
 
 def _prepare_data_for_fuzzy_overlay(data: Union[Sequence[np.ndarray], np.ndarray]) -> np.ndarray:

eis_toolkit/utilities/file_io.py

@@ -1,11 +1,12 @@
+import os
 from pathlib import Path
 
 import geopandas as gpd
 import numpy as np
 import pandas as pd
 import rasterio
 from beartype import beartype
-from beartype.typing import Literal, Sequence, Tuple, Union
+from beartype.typing import Any, Literal, Sequence, Tuple, Union
 
 from eis_toolkit import exceptions
 from eis_toolkit.utilities.checks.raster import check_raster_grids
@@ -162,3 +163,95 @@ def read_tabular(file_path: Path) -> pd.DataFrame:
     except pd.errors.ParserError:
         raise exceptions.FileReadError(f"Failed to read tabular data from {file_path}.")
     return data
+
+
+def get_output_paths_from_inputs(
+    input_paths: Sequence[Path], directory: Path, suffix: str, extension: str
+) -> Sequence[Path]:
+    """
+    Get output paths using input paths to extract file name bases.
+
+    Combines directory, file name extracted from input path, suffix and extension.
+    Include dot in the extension, for example '.tif'.
+
+    This tool is designed mainly for convenience in CLI functions.
+
+    Args:
+        input_paths: Input paths.
+        directory: Path of the output directory.
+        suffix: Common suffix added to the end of each output file name, for example "nodata_unified".
+        extension: The extension used for the output path, for example ".tif".
+
+    Returns:
+        List of output paths.
+    """
+    output_paths = []
+    for input_path in input_paths:
+        input_file_name_with_extension = os.path.split(input_path)[1]
+        input_file_name = os.path.splitext(input_file_name_with_extension)[0]
+        output_file_name = f"{input_file_name}_{suffix}"
+        output_path = directory.joinpath(output_file_name + extension)
+        output_paths.append(output_path)
+
+    return output_paths
+
+
+def get_output_paths_from_names(
+    file_names: Sequence[str], directory: Path, suffix: str, extension: str
+) -> Sequence[Path]:
+    """
+    Get output paths directly from given file names.
+
+    Combines directory, file name, suffix and extension.
+    Include dot in the extension, for example '.tif'.
+
+    This tool is designed mainly for convenience in CLI functions.
+
+    Args:
+        input_paths: Raw file names.
+        directory: Path of the output directory.
+        suffix: Common suffix added to the end of each output file name, for example "nodata_unified".
+        extension: The extension used for the output path, for example ".tif".
+
+    Returns:
+        List of output paths.
+    """
+    output_paths = []
+    for name in file_names:
+        output_file_name = f"{name}_{suffix}"
+        output_path = directory.joinpath(output_file_name + extension)
+        output_paths.append(output_path)
+
+    return output_paths
+
+
+def get_output_paths_from_common_name(
+    outputs: Sequence[Any], directory: Path, common_name: str, extension: str, first_num: int = 1
+) -> Sequence[Path]:
+    """
+    Get output paths for cases where outputs should be just numbered.
+
+    Combines directory, given common file name, number and extension. Outputs are used
+    to get the number used as suffix.
+    Include dot in the extension, for example '.tif'.
+
+    This tool is designed mainly for convenience in CLI functions.
+
+    Args:
+        input_paths: Outputs. Used just to iterate and get numbers for suffixes.
+        directory: Path of the output directory.
+        common_name: Common name used as the basis of each output file name. A number is appended to this.
+        extension: The extension used for the output path, for example ".tif".
+        first_num: The first number used as a suffix.
+
+    Returns:
+        List of output paths.
+    """
+    output_paths = []
+    for i in range(first_num, len(outputs) + first_num):
+        output_path = directory.joinpath(common_name + f"_{i}" + extension)
+        output_paths.append(output_path)
+
+        output_paths.append
+
+    return output_paths

eis_toolkit/utilities/miscellaneous.py

@@ -9,46 +9,11 @@
 from rasterio import transform
 from shapely.geometry import Point
 
-from eis_toolkit.exceptions import InvalidColumnException, InvalidColumnIndexException, InvalidDataShapeException
+from eis_toolkit.exceptions import InvalidColumnException, InvalidColumnIndexException
 from eis_toolkit.utilities.checks.dataframe import check_columns_valid
 from eis_toolkit.utilities.checks.parameter import check_dtype_for_int
 
 
-@beartype
-def stack_raster_arrays(arrays: Sequence[np.ndarray]) -> np.ndarray:
-    """
-    Stack 2D and 3D NumPy arrays (each representing a raster with one or multiple bands) along the bands axis.
-
-    Parameters:
-        arrays: List of 2D and 3D NumPy arrays. Each 2D array should have shape (height, width).
-            and 3D array shape (bands, height, width).
-
-    Returns:
-        A single 3D NumPy array where the first dimension size equals the total number of bands.
-    """
-    processed_arrays = []
-    for array in arrays:
-        # Add a new axis if the array is 2D
-        if array.ndim == 2:
-            array = array[np.newaxis, :]
-            print(array)
-            print(array.ndim)
-        elif array.ndim != 3:
-            raise InvalidDataShapeException("All raster arrays must be 2D or 3D for stacking.")
-        processed_arrays.append(array)
-
-    shape_set = {arr.shape[1:] for arr in processed_arrays}
-    if len(shape_set) != 1:
-        raise InvalidDataShapeException(
-            "All raster arrays must have the same shape in 2 last dimensions (height, width)."
-        )
-
-    # Stack along the first axis
-    stacked_array = np.concatenate(processed_arrays, axis=0)
-
-    return stacked_array
-
-
 @beartype
 def reduce_ndim(
     data: np.ndarray,