Create Masks

band_math(input_images, output_images, custom_math, *, debug_logs=False, custom_nodata_value=None, image_parallel_workers=None, window_parallel_workers=None, window_size=None, custom_output_dtype=None, calculation_dtype=None)

Applies custom band math expression to a list of input images and writes the results.

Parameters:

Name	Type	Description	Default
input_images	(str | List[str], required)	Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/.tif", "/input/folder" (assumes .tif), ["/input/one.tif", "/input/two.tif"].	required
output_images	(str | List[str], required)	Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_Math.tif), ["/input/one.tif", "/input/two.tif"].	required
custom_math	str	Python-compatible math expression using bands (e.g., "b1 + b2 / 2").	required
debug_logs	bool	If True, prints debug messages.	False
custom_nodata_value	Any	Override nodata value in source image.	None
image_parallel_workers	int | str | None	Controls image-level parallelism.	None
window_parallel_workers	int | str | None	Controls window-level parallelism.	None
window_size	tuple[int, int] | None	Size of processing windows (width, height).	None
custom_output_dtype	str | None	Output image data type (e.g., "uint16").	None
calculation_dtype	str | None	Computation data type (e.g., "float32").	None

Source code in spectralmatch/mask/mask.py

def band_math(
    input_images: Universal.SearchFolderOrListFiles,
    output_images: Universal.CreateInFolderOrListFiles,
    custom_math: str,
    *,
    debug_logs: Universal.DebugLogs = False,
    custom_nodata_value: Universal.CustomNodataValue = None,
    image_parallel_workers: Universal.ImageParallelWorkers = None,
    window_parallel_workers: Universal.WindowParallelWorkers = None,
    window_size: Universal.WindowSize = None,
    custom_output_dtype: Universal.CustomOutputDtype = None,
    calculation_dtype: Universal.CalculationDtype = None,
):
    """
    Applies custom band math expression to a list of input images and writes the results.

    Args:
        input_images (str | List[str], required): Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/*.tif", "/input/folder" (assumes *.tif), ["/input/one.tif", "/input/two.tif"].
        output_images (str | List[str], required): Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_Math.tif), ["/input/one.tif", "/input/two.tif"].
        custom_math (str): Python-compatible math expression using bands (e.g., "b1 + b2 / 2").
        debug_logs (bool, optional): If True, prints debug messages.
        custom_nodata_value (Any, optional): Override nodata value in source image.
        image_parallel_workers (int | str | None, optional): Controls image-level parallelism.
        window_parallel_workers (int | str | None, optional): Controls window-level parallelism.
        window_size (tuple[int, int] | None, optional): Size of processing windows (width, height).
        custom_output_dtype (str | None, optional): Output image data type (e.g., "uint16").
        calculation_dtype (str | None, optional): Computation data type (e.g., "float32").
    """

    input_image_paths = _resolve_paths(
        "search", input_images, kwargs={"default_file_pattern": "*.tif"}
    )
    output_image_paths = _resolve_paths(
        "create",
        output_images,
        kwargs={
            "paths_or_bases": input_image_paths,
            "default_file_pattern": "$_Math.tif",
        },
    )
    image_names = _resolve_paths("name", input_image_paths)

    with rasterio.open(input_image_paths[0]) as ds:
        nodata_value = _resolve_nodata_value(ds, custom_nodata_value)
        output_dtype = _resolve_output_dtype(ds, custom_output_dtype)

    image_parallel, image_backend, image_max_workers = _resolve_parallel_config(
        image_parallel_workers
    )

    # Extract referenced bands from custom_math (e.g., b1, b2, ...)
    band_indices = sorted(
        {int(match[1:]) for match in re.findall(r"\bb\d+\b", custom_math)}
    )

    image_args = [
        (
            in_path,
            out_path,
            name,
            custom_math,
            debug_logs,
            nodata_value,
            window_parallel_workers,
            window_size,
            band_indices,
            output_dtype,
            calculation_dtype,
        )
        for in_path, out_path, name in zip(
            input_image_paths, output_image_paths, image_names
        )
    ]

    if image_parallel:
        with _get_executor(image_backend, image_max_workers) as executor:
            futures = [
                executor.submit(_band_math_process_image, *arg) for arg in image_args
            ]
            for future in as_completed(futures):
                future.result()
    else:
        for arg in image_args:
            _band_math_process_image(*arg)

create_cloud_mask_with_omnicloudmask(input_images, output_images, red_band_index, green_band_index, nir_band_index, *, down_sample_m=None, debug_logs=False, image_parallel_workers=None, omnicloud_kwargs=None)

Generates cloud masks from input images using OmniCloudMask, with optional downsampling and multiprocessing.

Parameters:

Name	Type	Description	Default
input_images	(str | List[str], required)	Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/.tif", "/input/folder" (assumes .tif), ["/input/one.tif", "/input/two.tif"].	required
output_images	(str | List[str], required)	Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_CloudMask.tif), ["/input/one.tif", "/input/two.tif"].	required
red_band_index	int	Index of red band in the image.	required
green_band_index	int	Index of green band in the image.	required
nir_band_index	int	Index of NIR band in the image.	required
down_sample_m	float	If set, resamples input to this resolution in meters.	None
debug_logs	bool	If True, prints progress and debug info.	False
image_parallel_workers	ImageParallelWorkers	Enables parallel execution. Note: "process" does not work on macOS due to PyTorch MPS limitations.	None
omnicloud_kwargs	dict | None	Additional arguments forwarded to predict_from_array.	None

Raises:

Type	Description
Exception	Propagates any error from processing individual images.

Source code in spectralmatch/mask/mask.py

def create_cloud_mask_with_omnicloudmask(
    input_images: Universal.SearchFolderOrListFiles,
    output_images: Universal.CreateInFolderOrListFiles,
    red_band_index: int,
    green_band_index: int,
    nir_band_index: int,
    *,
    down_sample_m: float = None,
    debug_logs: Universal.DebugLogs = False,
    image_parallel_workers: Universal.ImageParallelWorkers = None,
    omnicloud_kwargs: dict | None = None,
):
    """
    Generates cloud masks from input images using OmniCloudMask, with optional downsampling and multiprocessing.

    Args:
        input_images (str | List[str], required): Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/*.tif", "/input/folder" (assumes *.tif), ["/input/one.tif", "/input/two.tif"].
        output_images (str | List[str], required): Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_CloudMask.tif), ["/input/one.tif", "/input/two.tif"].
        red_band_index (int): Index of red band in the image.
        green_band_index (int): Index of green band in the image.
        nir_band_index (int): Index of NIR band in the image.
        down_sample_m (float, optional): If set, resamples input to this resolution in meters.
        debug_logs (bool, optional): If True, prints progress and debug info.
        image_parallel_workers (ImageParallelWorkers, optional): Enables parallel execution. Note: "process" does not work on macOS due to PyTorch MPS limitations.
        omnicloud_kwargs (dict | None): Additional arguments forwarded to predict_from_array.

    Raises:
        Exception: Propagates any error from processing individual images.
    """

    print("Start omnicloudmask")
    Universal.validate(
        input_images=input_images, output_images=output_images, debug_logs=debug_logs
    )

    input_image_paths = _resolve_paths(
        "search", input_images, kwargs={"default_file_pattern": "*.tif"}
    )
    output_image_paths = _resolve_paths(
        "create",
        output_images,
        kwargs={
            "paths_or_bases": input_image_paths,
            "default_file_pattern": "$_CloudClip.tif",
        },
    )
    image_parallel, image_backend, image_max_workers = _resolve_parallel_config(
        image_parallel_workers
    )

    if debug_logs:
        print(f"Input images: {input_image_paths}")
    if debug_logs:
        print(f"Output images: {output_image_paths}")

    image_args = [
        (
            input_path,
            output_path,
            red_band_index,
            green_band_index,
            nir_band_index,
            down_sample_m,
            debug_logs,
            omnicloud_kwargs,
        )
        for input_path, output_path in zip(input_image_paths, output_image_paths)
    ]

    if image_parallel:
        with _get_executor(image_backend, image_max_workers) as executor:
            futures = [
                executor.submit(_process_cloud_mask_image, *args) for args in image_args
            ]
            for future in as_completed(futures):
                future.result()
    else:
        for args in image_args:
            _process_cloud_mask_image(*args)

create_ndvi_raster(input_images, output_images, nir_band_index, red_band_index, *, custom_output_dtype='float32', window_size=None, debug_logs=False, image_parallel_workers=None, window_parallel_workers=None)

Computes NDVI masks for one or more images and writes them to disk.

Parameters:

Name	Type	Description	Default
input_images	(str | List[str], required)	Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/.tif", "/input/folder" (assumes .tif), ["/input/one.tif", "/input/two.tif"].	required
output_images	(str | List[str], required)	Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_Vegetation.tif), ["/input/one.tif", "/input/two.tif"].	required
nir_band_index	int	Band index for NIR (1-based).	required
red_band_index	int	Band index for Red (1-based).	required
custom_output_dtype	CustomOutputDtype	Optional output data type (e.g., "float32").	'float32'
window_size	WindowSize	Tile size or mode for window-based processing.	None
debug_logs	DebugLogs	Whether to print debug messages.	False
image_parallel_workers	ImageParallelWorkers	Parallelism strategy for image-level processing.	None
window_parallel_workers	WindowParallelWorkers	Parallelism strategy for window-level processing.	None

Output

NDVI raster saved to output_images.

Source code in spectralmatch/mask/mask.py

def create_ndvi_raster(
    input_images: Universal.SearchFolderOrListFiles,
    output_images: Universal.CreateInFolderOrListFiles,
    nir_band_index: int,
    red_band_index: int,
    *,
    custom_output_dtype: Universal.CustomOutputDtype = "float32",
    window_size: Universal.WindowSize = None,
    debug_logs: Universal.DebugLogs = False,
    image_parallel_workers: Universal.ImageParallelWorkers = None,
    window_parallel_workers: Universal.WindowParallelWorkers = None,
) -> None:
    """Computes NDVI masks for one or more images and writes them to disk.

    Args:
        input_images (str | List[str], required): Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/*.tif", "/input/folder" (assumes *.tif), ["/input/one.tif", "/input/two.tif"].
        output_images (str | List[str], required): Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_Vegetation.tif), ["/input/one.tif", "/input/two.tif"].
        nir_band_index: Band index for NIR (1-based).
        red_band_index: Band index for Red (1-based).
        custom_output_dtype: Optional output data type (e.g., "float32").
        window_size: Tile size or mode for window-based processing.
        debug_logs: Whether to print debug messages.
        image_parallel_workers: Parallelism strategy for image-level processing.
        window_parallel_workers: Parallelism strategy for window-level processing.

    Output:
        NDVI raster saved to output_images.
    """

    print("Start create NDVI rasters")
    Universal.validate(
        input_images=input_images,
        output_images=output_images,
        custom_output_dtype=custom_output_dtype,
        window_size=window_size,
        debug_logs=debug_logs,
        image_parallel_workers=image_parallel_workers,
        window_parallel_workers=window_parallel_workers,
    )

    input_image_paths = _resolve_paths(
        "search", input_images, kwargs={"default_file_pattern": "*.tif"}
    )
    output_image_paths = _resolve_paths(
        "create",
        output_images,
        kwargs={
            "paths_or_bases": input_image_paths,
            "default_file_pattern": "$_Vegetation.tif",
        },
    )
    image_names = _resolve_paths("name", input_image_paths)

    image_parallel, image_backend, image_max_workers = _resolve_parallel_config(
        image_parallel_workers
    )

    image_args = [
        (
            in_path,
            out_path,
            image_name,
            nir_band_index,
            red_band_index,
            custom_output_dtype,
            window_size,
            debug_logs,
            window_parallel_workers,
        )
        for in_path, out_path, image_name in zip(
            input_image_paths, output_image_paths, image_names
        )
    ]

    if image_parallel:
        with _get_executor(image_backend, image_max_workers) as executor:
            futures = [
                executor.submit(_ndvi_process_image, *args) for args in image_args
            ]
            for f in as_completed(futures):
                f.result()
    else:
        for args in image_args:
            _ndvi_process_image(*args)

process_raster_values_to_vector_polygons(input_images, output_vectors, *, custom_nodata_value=None, custom_output_dtype=None, image_parallel_workers=None, window_parallel_workers=None, window_size=None, debug_logs=False, extraction_expression, filter_by_polygon_size=None, polygon_buffer=0.0, value_mapping=None)

Converts raster values into vector polygons based on an expression and optional filtering logic.

Parameters:

Name	Type	Description	Default
input_images	(str | List[str], required)	Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/.tif", "/input/folder" (assumes .tif), ["/input/one.tif", "/input/two.tif"].	required
output_vectors	(str | List[str], required)	Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.gpkg", "/input/folder" (assumes $_Vectorized.gpkg), ["/input/one.gpkg", "/input/two.gpkg"].	required
custom_nodata_value	CustomNodataValue	Custom NoData value to override the default from the raster metadata.	None
custom_output_dtype	CustomOutputDtype	Desired output data type. If not set, defaults to raster’s dtype.	None
image_parallel_workers	ImageParallelWorkers	Controls parallelism across input images. Can be an integer, executor string, or boolean.	None
window_parallel_workers	WindowParallelWorkers	Controls parallelism within a single image by processing windows in parallel.	None
window_size	WindowSizeWithBlock	Size of each processing block (width, height), or a strategy string such as "block" or "whole".	None
debug_logs	DebugLogs	Whether to print debug logs to the console.	False
extraction_expression	str	Logical expression to identify pixels of interest using band references (e.g., "b1 > 10 & b2 < 50").	required
filter_by_polygon_size	str	Area filter for resulting polygons. Can be a number (e.g., ">100") or percentile (e.g., ">95%").	None
polygon_buffer	float	Distance in coordinate units to buffer the resulting polygons. Default is 0.	0.0
value_mapping	dict	Mapping from original raster values to new values. Use None to convert to NoData.	None

Source code in spectralmatch/mask/utils_mask.py

def process_raster_values_to_vector_polygons(
    input_images: Universal.SearchFolderOrListFiles,
    output_vectors: Universal.CreateInFolderOrListFiles,
    *,
    custom_nodata_value: Universal.CustomNodataValue = None,
    custom_output_dtype: Universal.CustomOutputDtype = None,
    image_parallel_workers: Universal.ImageParallelWorkers = None,
    window_parallel_workers: Universal.WindowParallelWorkers = None,
    window_size: Universal.WindowSizeWithBlock = None,
    debug_logs: Universal.DebugLogs = False,
    extraction_expression: str,
    filter_by_polygon_size: str = None,
    polygon_buffer: float = 0.0,
    value_mapping: dict = None,
):
    """
    Converts raster values into vector polygons based on an expression and optional filtering logic.

    Args:
        input_images (str | List[str], required): Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/*.tif", "/input/folder" (assumes *.tif), ["/input/one.tif", "/input/two.tif"].
        output_vectors (str | List[str], required): Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.gpkg", "/input/folder" (assumes $_Vectorized.gpkg), ["/input/one.gpkg", "/input/two.gpkg"].
        custom_nodata_value (Universal.CustomNodataValue, optional): Custom NoData value to override the default from the raster metadata.
        custom_output_dtype (Universal.CustomOutputDtype, optional): Desired output data type. If not set, defaults to raster’s dtype.
        image_parallel_workers (Universal.ImageParallelWorkers, optional): Controls parallelism across input images. Can be an integer, executor string, or boolean.
        window_parallel_workers (Universal.WindowParallelWorkers, optional): Controls parallelism within a single image by processing windows in parallel.
        window_size (Universal.WindowSizeWithBlock, optional): Size of each processing block (width, height), or a strategy string such as "block" or "whole".
        debug_logs (Universal.DebugLogs, optional): Whether to print debug logs to the console.
        extraction_expression (str): Logical expression to identify pixels of interest using band references (e.g., "b1 > 10 & b2 < 50").
        filter_by_polygon_size (str, optional): Area filter for resulting polygons. Can be a number (e.g., ">100") or percentile (e.g., ">95%").
        polygon_buffer (float, optional): Distance in coordinate units to buffer the resulting polygons. Default is 0.
        value_mapping (dict, optional): Mapping from original raster values to new values. Use `None` to convert to NoData.

    """

    print("Start raster value extraction to polygons")

    Universal.validate(
        input_images=input_images,
        output_images=output_vectors,
        custom_nodata_value=custom_nodata_value,
        custom_output_dtype=custom_output_dtype,
        image_parallel_workers=image_parallel_workers,
        window_parallel_workers=window_parallel_workers,
        window_size=window_size,
        debug_logs=debug_logs,
    )

    input_image_paths = _resolve_paths(
        "search", input_images, kwargs={"default_file_pattern": "*.tif"}
    )
    output_image_paths = _resolve_paths(
        "create",
        output_vectors,
        kwargs={
            "paths_or_bases": input_image_paths,
            "default_file_pattern": "$_Vectorized.gpkg",
        },
    )

    image_parallel, image_backend, image_max_workers = _resolve_parallel_config(
        image_parallel_workers
    )
    window_parallel, window_backend, window_max_workers = _resolve_parallel_config(
        window_parallel_workers
    )

    image_args = [
        (
            in_path,
            out_path,
            extraction_expression,
            filter_by_polygon_size,
            polygon_buffer,
            value_mapping,
            custom_nodata_value,
            custom_output_dtype,
            window_parallel,
            window_backend,
            window_max_workers,
            window_size,
            debug_logs,
        )
        for in_path, out_path in zip(input_image_paths, output_image_paths)
    ]

    if image_parallel:
        with _get_executor(image_backend, image_max_workers) as executor:
            futures = [
                executor.submit(_process_image_to_polygons, *args)
                for args in image_args
            ]
            for future in as_completed(futures):
                future.result()
    else:
        for args in image_args:
            _process_image_to_polygons(*args)

threshold_raster(input_images, output_images, threshold_math, *, debug_logs=False, custom_nodata_value=None, image_parallel_workers=None, window_parallel_workers=None, window_size=None, custom_output_dtype=None, calculation_dtype='float32')

Applies a thresholding operation to input raster images using a mathematical expression string.

Parameters:

Name	Type	Description	Default
input_images	(str | List[str], required)	Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/.tif", "/input/folder" (assumes .tif), ["/input/one.tif", "/input/two.tif"].	required
output_images	(str | List[str], required)	Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_Threshold.tif), ["/input/one.tif", "/input/two.tif"].	required
threshold_math	str	A logical expression string using bands (e.g., "b1 > 5", "b1 > 5 & b2 < 10"). Supports: Band references: b1, b2, ...; Operators: >, <, >=, <=, ==, !=, &, |, ~, and (); Percentile-based thresholds: use e.g. "5%b1" to use the 5th percentile of band 1.	required
debug_logs	bool	If True, prints debug messages.	False
custom_nodata_value	float | int | None	Override the dataset's nodata value.	None
image_parallel_workers	ImageParallelWorkers	Parallelism config for image-level processing.	None
window_parallel_workers	WindowParallelWorkers	Parallelism config for window-level processing.	None
window_size	WindowSize	Window tiling strategy for memory-efficient processing.	None
custom_output_dtype	CustomOutputDtype	Output data type override.	None
calculation_dtype	CalculationDtype	Internal computation dtype.	'float32'

Source code in spectralmatch/mask/utils_mask.py

def threshold_raster(
    input_images: Universal.SearchFolderOrListFiles,
    output_images: Universal.CreateInFolderOrListFiles,
    threshold_math: str,
    *,
    debug_logs: Universal.DebugLogs = False,
    custom_nodata_value: Universal.CustomNodataValue = None,
    image_parallel_workers: Universal.ImageParallelWorkers = None,
    window_parallel_workers: Universal.WindowParallelWorkers = None,
    window_size: Universal.WindowSize = None,
    custom_output_dtype: Universal.CustomOutputDtype = None,
    calculation_dtype: Universal.CalculationDtype = "float32",
):
    """
    Applies a thresholding operation to input raster images using a mathematical expression string.

    Args:
        input_images (str | List[str], required): Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/*.tif", "/input/folder" (assumes *.tif), ["/input/one.tif", "/input/two.tif"].
        output_images (str | List[str], required): Defines output files from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_Threshold.tif), ["/input/one.tif", "/input/two.tif"].
        threshold_math (str): A logical expression string using bands (e.g., "b1 > 5", "b1 > 5 & b2 < 10"). Supports: Band references: b1, b2, ...; Operators: >, <, >=, <=, ==, !=, &, |, ~, and (); Percentile-based thresholds: use e.g. "5%b1" to use the 5th percentile of band 1.
        debug_logs (bool, optional): If True, prints debug messages.
        custom_nodata_value (float | int | None, optional): Override the dataset's nodata value.
        image_parallel_workers (ImageParallelWorkers, optional): Parallelism config for image-level processing.
        window_parallel_workers (WindowParallelWorkers, optional): Parallelism config for window-level processing.
        window_size (WindowSize, optional): Window tiling strategy for memory-efficient processing.
        custom_output_dtype (CustomOutputDtype, optional): Output data type override.
        calculation_dtype (CalculationDtype, optional): Internal computation dtype.
    """

    Universal.validate(
        input_images=input_images,
        output_images=output_images,
        debug_logs=debug_logs,
        custom_nodata_value=custom_nodata_value,
        image_parallel_workers=image_parallel_workers,
        window_parallel_workers=window_parallel_workers,
        window_size=window_size,
        custom_output_dtype=custom_output_dtype,
    )
    input_image_paths = _resolve_paths(
        "search", input_images, kwargs={"default_file_pattern": "*.tif"}
    )
    output_image_paths = _resolve_paths(
        "create",
        output_images,
        kwargs={
            "paths_or_bases": input_image_paths,
            "default_file_pattern": "$_Threshold.tif",
        },
    )
    image_names = _resolve_paths("name", input_image_paths)

    with rasterio.open(input_image_paths[0]) as ds:
        nodata_value = _resolve_nodata_value(ds, custom_nodata_value)
        output_dtype = _resolve_output_dtype(ds, custom_output_dtype)

    image_parallel, image_backend, image_max_workers = _resolve_parallel_config(
        image_parallel_workers
    )

    image_args = [
        (
            in_path,
            out_path,
            name,
            threshold_math,
            debug_logs,
            nodata_value,
            window_parallel_workers,
            window_size,
            output_dtype,
            calculation_dtype,
        )
        for in_path, out_path, name in zip(
            input_image_paths, output_image_paths, image_names
        )
    ]

    if image_parallel:
        with _get_executor(image_backend, image_max_workers) as executor:
            futures = [
                executor.submit(_threshold_process_image, *arg) for arg in image_args
            ]
            for future in as_completed(futures):
                future.result()
    else:
        for arg in image_args:
            _threshold_process_image(*arg)