Utilities

align_rasters(input_images, output_images, *, resampling_method='bilinear', tap=False, resolution='highest', window_size=None, debug_logs=False, cache=None, image_threads=None, io_threads=None, tile_threads=None)

Aligns multiple rasters to a common resolution and grid using specified resampling.

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 $_Local.tif), ["/input/one.tif", "/input/two.tif"].	required
resampling_method	Literal['nearest', 'bilinear', 'cubic']	"nearest" | "bilinear" | "cubic".	'bilinear'
tap	bool	If True, snap output extent to target-aligned pixels (GDAL -tap behavior).	False
resolution	Literal['highest', 'average', 'lowest']	"highest" (min px size), "average", or "lowest" (max px size).	'highest'
window_size	WindowSize	Tile size for output blocks; used for GTiff creation options.	None
debug_logs	DebugLogs	Verbose logging.	False
cache	Cache	Cache for processing.	None
image_threads	Threads	Python-level parallelism over images (e.g., ("process", 4)).	None
io_threads	Threads	Sets GDAL_NUM_THREADS for internal GDAL multithreading (int or str).	None
tile_threads	Threads	Sets GTiff/COG writer NUM_THREADS and Warp’s NUM_THREADS (int or str).	None

Returns:

Type	Description
None	List[str]: Paths to the locally adjusted output raster images.

Source code in spectralmatch/utils.py

def align_rasters(
    input_images: Universal.SearchFolderOrListFiles,
    output_images: Universal.CreateInFolderOrListFiles,
    *,
    resampling_method: Literal["nearest", "bilinear", "cubic"] = "bilinear",
    tap: bool = False,
    resolution: Literal["highest", "average", "lowest"] = "highest",
    window_size: Universal.WindowSize = None,
    debug_logs: Universal.DebugLogs = False,
    cache: Universal.Cache = None,
    image_threads: Universal.Threads = None,
    io_threads: Universal.Threads = None,
    tile_threads: Universal.Threads = None,
) -> None:
    """
    Aligns multiple rasters to a common resolution and grid using specified resampling.

    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 $_Local.tif), ["/input/one.tif", "/input/two.tif"].
        resampling_method: "nearest" | "bilinear" | "cubic".
        tap: If True, snap output extent to target-aligned pixels (GDAL -tap behavior).
        resolution: "highest" (min px size), "average", or "lowest" (max px size).
        window_size: Tile size for output blocks; used for GTiff creation options.
        debug_logs: Verbose logging.
        cache: Cache for processing.
        image_threads: Python-level parallelism over images (e.g., ("process", 4)).
        io_threads: Sets GDAL_NUM_THREADS for internal GDAL multithreading (int or str).
        tile_threads: Sets GTiff/COG writer NUM_THREADS and Warp’s NUM_THREADS (int or str).

    Returns:
        List[str]: Paths to the locally adjusted output raster images.
    """
    if debug_logs:
        print("Start align rasters")

    Universal.validate(
        input_images=input_images,
        output_images=output_images,
        debug_logs=debug_logs,
        window_size=window_size,
        cache=cache,
        image_threads=image_threads,
        io_threads=io_threads,
        tile_threads=tile_threads,
    )

    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": "$_Align.tif",
        },
    )
    input_image_names = [
        os.path.splitext(os.path.basename(p))[0] for p in input_image_paths
    ]

    # Setup gdal
    _set_gdal_cache(cache, debug_logs)
    _set_gdal_workers(io_threads, debug_logs)

    # Setup parallel
    image_backend = "thread" # "process" or "thread"
    image_threads_on, image_thread_workers = _resolve_parallel_config(image_threads)
    tile_thread_on, tile_thread_workers = _resolve_parallel_config(tile_threads)


    if debug_logs:
        print(f"{len(input_image_paths)} rasters to align")

    # Check requirements
    _check_raster_requirements(
        input_image_paths,
        debug_logs,
        check_geotransform=True,
        check_crs=True,
        check_bands=True,
        check_nodata=True,
    )

    # Get target resolution
    target_res = compute_resolution(input_image_paths, resolution)

    if debug_logs:
        print(f"Target resolution: {target_res}")

    # Prepare per-image args
    window_size = _resolve_window_size(window_size, input_image_paths[0], debug_logs)
    args = [
        (
            input_image_names[i],
            input_image_paths[i],
            output_image_paths[i],
            target_res,
            resampling_method,
            tap,
            window_size,
            tile_thread_workers,
            debug_logs,
        )
        for i in range(len(input_image_paths))
    ]

    if image_threads:
        with _get_executor(image_backend, image_thread_workers) as executor:
            futures = [
                executor.submit(_align_process_image, *arg) for arg in args
            ]
            for future in as_completed(futures):
                future.result()
    else:
        for args in args:
            _align_process_image(*args)
    return output_image_paths

compute_overviews(input_images_paths, *, output_image_paths=None, window_scales=(2, 4, 8, 16, 32), cache=None, image_threads=None, io_threads=None, tile_threads=None, debug_logs=False)

Compute and attach GDAL overviews for one or more raster images.

Parameters:

Name	Type	Description	Default
input_images_paths	(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_image_paths	str | List[str] | None	Defines output files as None to update input images or from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_Global.tif), ["/input/one.tif", "/input/two.tif"].	None
window_scales	tuple[int] | None	Overview decimation factors (default: (2, 4, 8, 16, 32)).	(2, 4, 8, 16, 32)
cache	Cache	GDAL cache size configuration.	None
image_threads	Threads	Number of parallel workers for image-level processing.	None
io_threads	Threads	GDAL IO worker configuration.	None
tile_threads	Threads	GDAL internal threads for overview computation.	None
debug_logs	bool	Enable verbose logging.	False

Returns:

Type	Description
	List[str]: Paths of images that received overviews.

Source code in spectralmatch/utils.py

def compute_overviews(
    input_images_paths: Universal.SearchFolderOrListFiles,
    *,
    output_image_paths: Universal.CreateInFolderOrListFiles | None = None,
    window_scales: tuple[int] | None = (2, 4, 8, 16, 32),
    cache: Universal.Cache = None,
    image_threads: Universal.Threads = None,
    io_threads: Universal.Threads = None,
    tile_threads: Universal.Threads = None,
    debug_logs: bool = False,
):
    """
    Compute and attach GDAL overviews for one or more raster images.

    Args:
        input_images_paths (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_image_paths (str | List[str] | None): Defines output files as None to update input images or from a template path, folder, or list of paths (with the same length as the input). Specify like: "/input/files/$.tif", "/input/folder" (assumes $_Global.tif), ["/input/one.tif", "/input/two.tif"].
        window_scales: Overview decimation factors (default: (2, 4, 8, 16, 32)).
        cache: GDAL cache size configuration.
        image_threads: Number of parallel workers for image-level processing.
        io_threads: GDAL IO worker configuration.
        tile_threads: GDAL internal threads for overview computation.
        debug_logs: Enable verbose logging.

    Returns:
        List[str]: Paths of images that received overviews.
    """
    print("Start overviews computation")
    if debug_logs: print(f"Input images: {input_images_paths}")
    if debug_logs and output_image_paths: print(f"Output images: {output_image_paths}")
    if debug_logs: print(f"Window scales: {window_scales}")

    Universal.validate(
        input_images=input_images_paths,
        cache=cache,
        image_threads=image_threads,
        io_threads=io_threads,
        tile_threads=tile_threads,
    )

    # Worker
    def _process_image_overview(path: str):
        ds = gdal.Open(path, gdal.GA_Update)
        if ds is None:
            raise RuntimeError(f"Cannot open {path}")

        opts = []
        if tile_thread_on:
            opts.append(f"NUM_THREADS={tile_workers}")

        ds.BuildOverviews(
            "AVERAGE",
            window_scales,
            options=opts,
        )
        ds = None

        if debug_logs:
            print(f"Overviews built for: {path}")

    def _copy_files_if_needed(
            src_paths: List[str],
            dst_paths: List[str],
        ) -> List[str]:
        """
        Copy src to dst.
        """
        out: List[str] = []

        for src, dst in zip(src_paths, dst_paths):
            os.makedirs(os.path.dirname(dst), exist_ok=True)
            ds = gdal.Translate(dst, src)
            if ds is None:
                raise RuntimeError(f"Failed copying {src} to {dst}")
            ds = None
            out.append(dst)

        return out


    # Paths
    input_paths = _resolve_paths(
        "search",
        input_images_paths,
        kwargs={"default_file_pattern": "*.tif"},
    )

    if output_image_paths is None:
        target_paths = input_paths
    else:
        target_paths = _resolve_paths(
            "create",
            output_image_paths,
            kwargs={
                "paths_or_bases": input_paths,
                "default_file_pattern": "$.tif",
            },
        )
        _copy_files_if_needed(input_paths, target_paths)

    # GDAL config
    _set_gdal_cache(cache, debug_logs)
    _set_gdal_workers(io_threads, debug_logs)

    image_backend = "thread"
    image_threads_on, image_workers = _resolve_parallel_config(image_threads)
    tile_thread_on, tile_workers = _resolve_parallel_config(tile_threads)


    # Execute
    if image_threads_on:
        with _get_executor(image_backend, image_workers) as ex:
            futures = [ex.submit(_process_image_overview, p) for p in target_paths]
            for f in as_completed(futures):
                f.result()
    else:
        for p in target_paths:
            _process_image_overview(p)

    return target_paths

create_masked_vrts(input_image_path_pairs, *, vector_mask=None, out_dir=None, debug_logs=False)

For each (name -> image_path), write: - mask_{name}.geojson (cutline: include polys OR exclude complement) - vrt_{name}.vrt (alpha = band1 nodata U cutline-outside) Returns: dict[name, vrt_path]

Source code in spectralmatch/utils.py

def create_masked_vrts(
    input_image_path_pairs: Dict[str, str],
    *,
    vector_mask: Universal.VectorMask = None,
    out_dir: Optional[str] = None,
    debug_logs: bool = False,
) -> Dict[str, str]:
    """
    For each (name -> image_path), write:
      - mask_{name}.geojson  (cutline: include polys OR exclude complement)
      - vrt_{name}.vrt       (alpha = band1 nodata U cutline-outside)
    Returns: dict[name, vrt_path]
    """
    # temp dir for all masks+VRTs
    workdir = out_dir or tempfile.mkdtemp(prefix="spectralmatch_masks_")
    if debug_logs: print(f"Creating VRTs: {workdir}")

    out_vrts: Dict[str, str] = {}

    for image_name, image_path in input_image_path_pairs.items():
        if debug_logs:
            print(f"    {image_name}")

        src = gdal.Open(image_path, gdal.GA_ReadOnly)
        if src is None:
            raise RuntimeError(f"Could not open {image_path}")

        # raster geo + nodata
        nodata = src.GetRasterBand(1).GetNoDataValue()
        dst_wkt = src.GetProjectionRef() or ""
        dst_srs = osr.SpatialReference(); dst_srs.ImportFromWkt(dst_wkt)
        gt = src.GetGeoTransform()
        xmin, xmax = gt[0], gt[0] + gt[1] * src.RasterXSize
        ymax, ymin = gt[3], gt[3] + gt[5] * src.RasterYSize

        # build cutline only if requested
        cutline_ds = None
        if vector_mask:
            mode, vpath, *field = vector_mask
            field_name = field[0] if field else None

            vds = ogr.Open(vpath)
            if vds is None:
                raise RuntimeError(f"Could not open vector: {vpath}")
            lyr = vds.GetLayer(0)
            lyr.SetSpatialFilterRect(xmin, ymin, xmax, ymax)

            src_srs = lyr.GetSpatialRef()
            tx = osr.CoordinateTransformation(src_srs, dst_srs) if (src_srs and not src_srs.IsSame(dst_srs)) else None

            # extent polygon once
            ring = ogr.Geometry(ogr.wkbLinearRing)
            ring.AddPoint(xmin, ymin); ring.AddPoint(xmin, ymax)
            ring.AddPoint(xmax, ymax); ring.AddPoint(xmax, ymin); ring.AddPoint(xmin, ymin)
            extent_poly = ogr.Geometry(ogr.wkbPolygon); extent_poly.AddGeometry(ring)

            # collect selected geoms in raster CRS
            selected = []
            for feat in lyr:
                if field_name:
                    val = feat.GetField(field_name)
                    if val is None or (image_name not in str(val)):
                        continue
                g = feat.GetGeometryRef()
                if not g:
                    continue
                gc = g.Clone()
                if tx: gc.Transform(tx)
                if gc.GetGeometryType() not in (ogr.wkbPolygon, ogr.wkbMultiPolygon):
                    gc = gc.Buffer(0)
                gc = gc.Intersection(extent_poly)
                if gc and not gc.IsEmpty():
                    selected.append(gc)

            # decide what to write as the cutline geometry
            if mode == "include":
                geoms_to_write = selected
                suffix = "include"
            else:
                if selected:
                    mp = ogr.Geometry(ogr.wkbMultiPolygon)
                    for g in selected: mp.AddGeometry(g)
                    union_geom = mp.UnionCascaded()
                else:
                    union_geom = None
                complement = extent_poly if union_geom is None else extent_poly.Difference(union_geom)
                geoms_to_write = [complement]
                suffix = "exclude_complement"

            if geoms_to_write:
                cutline_path = os.path.join(workdir, f"mask_{image_name}.geojson")
                drv = ogr.GetDriverByName("GeoJSON")
                ods = drv.CreateDataSource(cutline_path)
                ol = ods.CreateLayer("cut", srs=dst_srs, geom_type=ogr.wkbPolygon)
                defn = ol.GetLayerDefn()
                for g in geoms_to_write:
                    if g and not g.IsEmpty():
                        of = ogr.Feature(defn); of.SetGeometry(g)
                        ol.CreateFeature(of); of = None
                ol = None; ods = None
                cutline_ds = cutline_path

            lyr = None; vds = None

        # build the VRT with alpha (nodata + cutline-outside)
        vrt_path = os.path.join(workdir, f"vrt_{image_name}.vrt")
        warp_opts = gdal.WarpOptions(
            format="VRT",
            dstSRS=dst_wkt or None,
            dstAlpha=True,
            dstNodata=None,
            cutlineDSName=cutline_ds,
            cropToCutline=False,
            resampleAlg=gdal.GRA_NearestNeighbour,
            multithread=True,
            warpOptions=[
                "SKIP_NOSOURCE=YES",
                "NUM_THREADS=ALL_CPUS",
                "UNIFIED_SRC_NODATA=YES",
            ],
        )
        out_ds = gdal.Warp(vrt_path, image_path, options=warp_opts)
        if out_ds is None:
            raise RuntimeError(f"Failed to build masked VRT for {image_name}")
        out_ds = None
        src = None

        out_vrts[image_name] = vrt_path
    return out_vrts

mask_rasters(input_images, output_images, vector_mask=None, window_size=None, debug_logs=False, cache=None, image_threads=None, io_threads=None, tile_threads=None, include_touched_pixels=False, custom_nodata_value=None)

Applies a vector-based mask to one or more rasters using GDAL Warp.

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 $_Local.tif), ["/input/one.tif", "/input/two.tif"].	required
vector_mask	VectorMask	Tuple ('include'|'exclude', vector_path, optional field name).	None
window_size	int | None	Tile size for processing tiles. Defaults to None.	None
debug_logs	bool	If True, prints progress. Defaults to False.	False
cache	int | Tuple[int, str] | None	Controls GDAL cache size. Examples: 2048 (MB), (2, "GB"). Set None to use GDAL’s default. Applied via GDAL_CACHEMAX. window_parallel_workers (Tuple[Literal["process"], Literal["cpu"] | int] | None = None): Parallelization strategy at the window level within each image. Same format as image_parallel_workers. Threads are not supported. Set to None to disable.	None
image_threads	Literal['cpu'] | int | None	Parallelism for per-image operations. "cpu" to get number of cores, int to assign number, and None to disable image level parallelism.	None
io_threads	Literal['cpu'] | int | None	Parallelism for IO operations. "cpu" to get number of cores, int to assign number, and None to disable io level parallelism.	None
tile_threads	Literal['cpu'] | int | None	"cpu" to get number of cores, int to assign number, and None to disable tile level parallelism.	None
include_touched_pixels	bool	If True, uses all touched pixels for cutline mask.	False
custom_nodata_value	float | int | None	Overrides detected NoData value. Defaults to None.	None

Returns:

Name	Type	Description
list	list	Output image paths after masking.

Source code in spectralmatch/utils.py

def mask_rasters(
    input_images: Universal.SearchFolderOrListFiles,
    output_images: Universal.CreateInFolderOrListFiles,
    vector_mask: Universal.VectorMask = None,
    window_size: Universal.WindowSize = None,
    debug_logs: Universal.DebugLogs = False,
    cache: Universal.Cache = None,
    image_threads: Universal.Threads = None,
    io_threads: Universal.Threads = None,
    tile_threads: Universal.Threads = None,
    include_touched_pixels: bool = False,
    custom_nodata_value: Universal.CustomNodataValue = None,
    ) -> list:
    """
    Applies a vector-based mask to one or more rasters using GDAL Warp.

    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 $_Local.tif), ["/input/one.tif", "/input/two.tif"].
        vector_mask (Universal.VectorMask, optional): Tuple ('include'|'exclude', vector_path, optional field name).
        window_size (int | None): Tile size for processing tiles. Defaults to None.
        debug_logs (bool, optional): If True, prints progress. Defaults to False.
        cache (int | Tuple[int, str] | None, optional): Controls GDAL cache size. Examples: 2048 (MB), (2, "GB"). Set None to use GDAL’s default. Applied via GDAL_CACHEMAX.        window_parallel_workers (Tuple[Literal["process"], Literal["cpu"] | int] | None = None): Parallelization strategy at the window level within each image. Same format as image_parallel_workers. Threads are not supported. Set to None to disable.
        image_threads (Literal["cpu"] | int | None): Parallelism for per-image operations. "cpu" to get number of cores, int to assign number, and None to disable image level parallelism.
        io_threads (Literal["cpu"] | int | None): Parallelism for IO operations. "cpu" to get number of cores, int to assign number, and None to disable io level parallelism.
        tile_threads (Literal["cpu"] | int | None): "cpu" to get number of cores, int to assign number, and None to disable tile level parallelism.
        include_touched_pixels (bool, optional): If True, uses all touched pixels for cutline mask.
        custom_nodata_value (float | int | None, optional): Overrides detected NoData value. Defaults to None.

    Returns:
        list: Output image paths after masking.
    """

    if debug_logs:
        print("Start mask rasters")

    Universal.validate(
        input_images=input_images,
        output_images=output_images,
        debug_logs=debug_logs,
        vector_mask=vector_mask,
        window_size=window_size,
        image_threads=image_threads,
        io_threads=io_threads,
        tile_threads=tile_threads,
        custom_nodata_value=custom_nodata_value,
        cache=cache,
    )

    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": "$_Mask.tif"},
    )

    input_image_names = [
        os.path.splitext(os.path.basename(p))[0] for p in input_image_paths
    ]

    _set_gdal_cache(cache, debug_logs)
    _set_gdal_workers(io_threads, debug_logs)

    # Determine multiprocessing and worker count
    image_backend = "thread" # "thread" or "process"
    image_threads_on, image_thread_workers = _resolve_parallel_config(image_threads)
    tile_thread_on, tile_thread_workers = _resolve_parallel_config(tile_threads)

    mode, per_image_cutlines, original_vector_path, field_given = _prepare_cutline_sources(
        vector_mask, input_image_names, debug_logs
    )

    args = [
        (
            input_image_paths[i],
            output_image_paths[i],
            input_image_names[i],
            mode,
            (per_image_cutlines[input_image_names[i]] if field_given else original_vector_path),
            field_given,
            debug_logs,
            include_touched_pixels,
            custom_nodata_value,
            tile_thread_workers,
            tile_thread_on,
        )
        for i in range(len(input_image_paths))
    ]

    if image_threads_on:
        with _get_executor(image_backend, image_thread_workers) as executor:
            futures = [executor.submit(_mask_raster_process_image, *arg) for arg in args]
            for future in as_completed(futures):
                future.result()
    else:
        for arg in args:
            _mask_raster_process_image(*arg)

    return output_image_paths

merge_rasters(input_images, output_image_path, *, cache=None, io_threads=None, tile_threads=None, debug_logs=False, output_dtype=None, custom_nodata_value=None, resolution='highest', window_size=None, build_overviews=False)

Merges multiple rasters into a single output.

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_image_path	str	Path to output mosaic.	required
cache	int | Tuple[int, str] | None	Controls GDAL cache size. Examples: 2048 (MB), (2, "GB"). Set None to use GDAL’s default. Applied via GDAL_CACHEMAX. window_parallel_workers (Tuple[Literal["process"], Literal["cpu"] | int] | None = None): Parallelization strategy at the window level within each image. Same format as image_parallel_workers. Threads are not supported. Set to None to disable.	None
io_threads	Literal['cpu'] | int | None	Parallelism for IO operations. "cpu" to get number of cores, int to assign number, and None to disable io level parallelism.	None
tile_threads	Literal['cpu'] | int | None	"cpu" to get number of cores, int to assign number, and None to disable tile level parallelism.	None
debug_logs	bool	If True, prints progress. Defaults to False.	False
output_dtype	str | None	Data type for output rasters. Defaults to input image dtype.	None
custom_nodata_value	float | int | None	Overrides detected NoData value. Defaults to None.	None
resolution	highest | average | lowest	Strategy for computing merge resolution.	'highest'
window_size	int | None	Tile size for processing tiles. Defaults to None.	None
build_overviews	bool	If True, computes overviews. Defaults to False.	False

Returns:

Name	Type	Description
str	str	Path of the merged raster.

Source code in spectralmatch/utils.py

def merge_rasters(
    input_images: Universal.SearchFolderOrListFiles,
    output_image_path: str,
    *,
    cache: Universal.Cache = None,
    io_threads: Universal.Threads = None,
    tile_threads: Universal.Threads = None,
    debug_logs: Universal.DebugLogs = False,
    output_dtype: Universal.CustomOutputDtype = None,
    custom_nodata_value: Universal.CustomNodataValue = None,
    resolution: Literal["highest", "average", "lowest"] = "highest",
    window_size: Universal.WindowSize = None,
    build_overviews: bool = False,
) -> str:
    """
    Merges multiple rasters into a single output.

    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_image_path (str): Path to output mosaic.
        cache (int | Tuple[int, str] | None, optional): Controls GDAL cache size. Examples: 2048 (MB), (2, "GB"). Set None to use GDAL’s default. Applied via GDAL_CACHEMAX.        window_parallel_workers (Tuple[Literal["process"], Literal["cpu"] | int] | None = None): Parallelization strategy at the window level within each image. Same format as image_parallel_workers. Threads are not supported. Set to None to disable.
        io_threads (Literal["cpu"] | int | None): Parallelism for IO operations. "cpu" to get number of cores, int to assign number, and None to disable io level parallelism.
        tile_threads (Literal["cpu"] | int | None): "cpu" to get number of cores, int to assign number, and None to disable tile level parallelism.
        debug_logs (bool, optional): If True, prints progress. Defaults to False.
        output_dtype (str | None, optional): Data type for output rasters. Defaults to input image dtype.
        custom_nodata_value (float | int | None, optional): Overrides detected NoData value. Defaults to None.
        resolution ("highest" | "average" | "lowest", optional): Strategy for computing merge resolution.
        window_size (int | None): Tile size for processing tiles. Defaults to None.
        build_overviews (bool, optional): If True, computes overviews. Defaults to False.

    Returns:
        str: Path of the merged raster.

    """

    Universal.validate(
        input_images=input_images,
        debug_logs=debug_logs,
        cache=cache,
        io_threads=io_threads,
        tile_threads=tile_threads,
        output_dtype=output_dtype,
        window_size=window_size,
        custom_nodata_value=custom_nodata_value,
    )

    # Setup parallel
    tile_thread_on, tile_thread_workers = _resolve_parallel_config(tile_threads)

    input_image_paths = _resolve_paths(
        "search", input_images, kwargs={"default_file_pattern": "*.tif"}
    )

    # Dtype
    output_dtype = _gdal_dtype_str_to_enum(_resolve_gdal_dtype(output_dtype, input_image_paths[0]))

    _set_gdal_cache(cache, debug_logs)
    _set_gdal_workers(io_threads, debug_logs)

    if debug_logs:
        print(f"Building VRT from {len(input_image_paths)} rasters")

    vrt_opts = gdal.BuildVRTOptions(
        resolution=resolution,
        srcNodata=custom_nodata_value,
        VRTNodata=custom_nodata_value,
    )

    vrt_ds = gdal.BuildVRT("", input_image_paths, options=vrt_opts)

    creation_options = [
        "TILED=YES",
        "BIGTIFF=YES",
        "COMPRESS=ZSTD",
    ]

    if window_size:
        creation_options += [
            f"BLOCKXSIZE={window_size}",
            f"BLOCKYSIZE={window_size}",
        ]

    if tile_thread_workers is not None and str(tile_thread_workers).strip():
        creation_options.append(f"NUM_THREADS={tile_thread_workers}")

    translate_opts = gdal.TranslateOptions(
        format="GTiff",
        outputType=output_dtype,
        noData=custom_nodata_value,
        creationOptions=creation_options,
    )

    gdal.Translate(
        destName=output_image_path,
        srcDS=vrt_ds,
        options=translate_opts,
    )

    vrt_ds = None

    if build_overviews: compute_overviews(
        input_images_paths=output_image_path,
        cache=cache,
        io_threads=io_threads,
        tile_threads=tile_threads,
        debug_logs=debug_logs,
        )
    return output_image_path

merge_vectors(input_vectors, merged_vector_path, method, debug_logs=False, create_name_attribute=None)

Merge multiple vector files using the specified geometric method.

Parameters:

Name	Type	Description	Default
input_vectors	str | List[str]	Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/.gpkg", "/input/folder" (assumes .gpkg), ["/input/one.tif", "/input/two.tif"].	required
merged_vector_path	str	Path to save merged output.	required
method	Literal['intersection', 'union', 'keep']	Merge strategy.	required
debug_logs	bool	If True, print debug information.	False
create_name_attribute	Optional[Tuple[str, str]]	Tuple of (field_name, separator) to add a combined name field.	None

Returns:

Type	Description
None	None

Source code in spectralmatch/utils.py

def merge_vectors(
    input_vectors: Universal.SearchFolderOrListFiles,
    merged_vector_path: str,
    method: Literal["intersection", "union", "keep"],
    debug_logs: bool = False,
    create_name_attribute: Optional[Tuple[str, str]] = None,
) -> None:
    """
    Merge multiple vector files using the specified geometric method.

    Args:
        input_vectors (str | List[str]): Defines input files from a glob path, folder, or list of paths. Specify like: "/input/files/*.gpkg", "/input/folder" (assumes *.gpkg), ["/input/one.tif", "/input/two.tif"].
        merged_vector_path (str): Path to save merged output.
        method (Literal["intersection", "union", "keep"]): Merge strategy.
        debug_logs (bool): If True, print debug information.
        create_name_attribute (Optional[Tuple[str, str]]): Tuple of (field_name, separator) to add a combined name field.

    Returns:
        None
    """
    print("Start vector merge")

    os.makedirs(os.path.dirname(merged_vector_path), exist_ok=True)
    input_vector_paths = _resolve_paths(
        "search", input_vectors, kwargs={"default_file_pattern": "*.gpkg"}
    )

    geoms = []
    input_names = []

    for path in input_vector_paths:
        gdf = gpd.read_file(path)
        if create_name_attribute:
            name = os.path.splitext(os.path.basename(path))[0]
            input_names.append(name)
        geoms.append(gdf)

    combined_name_value = None
    if create_name_attribute:
        field_name, sep = create_name_attribute
        combined_name_value = sep.join(input_names)

    if method == "keep":
        merged_dfs = []
        field_name = create_name_attribute[0] if create_name_attribute else None
        for path in input_vector_paths:
            gdf = gpd.read_file(path)
            if field_name:
                name = os.path.splitext(os.path.basename(path))[0]
                gdf[field_name] = name
            merged_dfs.append(gdf)
        merged = gpd.GeoDataFrame(
            pd.concat(merged_dfs, ignore_index=True), crs=merged_dfs[0].crs
        )

    elif method == "union":
        merged = gpd.GeoDataFrame(pd.concat(geoms, ignore_index=True), crs=geoms[0].crs)
        if create_name_attribute:
            merged[field_name] = combined_name_value

    elif method == "intersection":
        merged = geoms[0]
        for gdf in geoms[1:]:
            shared_cols = set(merged.columns).intersection(gdf.columns) - {"geometry"}
            gdf = gdf.drop(columns=shared_cols)
            merged = gpd.overlay(merged, gdf, how="intersection", keep_geom_type=True)
        if create_name_attribute:
            merged[field_name] = combined_name_value

    else:
        raise ValueError(f"Unsupported merge method: {method}")

    merged.to_file(merged_vector_path)