Skip to content

blockutils.windows

Utility class to handle windowed read and write rasterio operations.

WindowsUtil

Utility class to handle raster IO in windows. Can do regular windows, buffered windows and transform windows.

__init__(self, rio_ds) special

Initialize the instance with a rasterio dataset (read or write).

Parameters:

Name Type Description Default
rio_ds Union[rasterio.io.DatasetReader, rasterio.io.DatasetWriter]

An open rasterio dataset.

required
Source code in blockutils/windows.py
def __init__(self, rio_ds: Union[rio.io.DatasetReader, rio.io.DatasetWriter]):
    """
    Initialize the instance with a rasterio dataset (read or write).

    Arguments:
        rio_ds: An open rasterio dataset.
    """
    self.rio_ds = rio_ds
    self.windows = rio_ds.block_windows(1)

buffer_window(self, window, buffer)

Buffers a window with a set number of pixels (buffer) in every possible direction given a shape of the overall raster file where window is derived from.

For instance, if window matches the shape of the source raster file, the shape of the source raster file is returned in window format.

Parameters:

Name Type Description Default
window Window

Original window (for instance from block).

required
buffer int

Number of pixels to buffer by where possible.

required

Returns:

Type Description
Window

A buffered window.

Source code in blockutils/windows.py
def buffer_window(self, window: Window, buffer: int) -> Window:
    """
    Buffers a window with a set number of pixels (buffer) in every possible direction
    given a shape of the overall raster file where window is derived from.

    For instance, if window matches the shape of the source raster file, the shape
    of the source raster file is returned in window format.

    Arguments:
        window: Original window (for instance from block).
        buffer: Number of pixels to buffer by where possible.

    Returns:
        A buffered window.
    """
    row_slice, col_slice = window.toslices()

    can_row_start = row_slice.start - buffer
    can_row_stop = row_slice.stop + buffer

    can_col_start = col_slice.start - buffer
    can_col_stop = col_slice.stop + buffer

    out_row_slice = slice(can_row_start, can_row_stop)
    out_col_slice = slice(can_col_start, can_col_stop)

    return self.limit_window_to_raster_bounds(
        Window.from_slices(out_row_slice, out_col_slice, boundless=True)
    )

crop_array_to_window(self, buffered_array, window, window_buffer)

Crops an array created with the windows_buffered to the extent of window. Makes use of the a higher res Affine to "reproject" the original window to the extent of window_buffer.

Parameters:

Name Type Description Default
buffered_array <built-in function array>

Buffered array read with window_buffer. Has same shape as buffered window.

required
window Window

Original window. Buffered window cropped with a given buffer.

required
window_buffer Window

Buffered window.

required

Returns:

Type Description
<built-in function array>

A cropped numpy array.

Source code in blockutils/windows.py
def crop_array_to_window(
    self, buffered_array: np.array, window: Window, window_buffer: Window
) -> np.array:
    """
    Crops an array created with the windows_buffered to the extent of
    window. Makes use of the a higher res Affine to "reproject" the original
    window to the extent of window_buffer.

    Arguments:
        buffered_array: Buffered array read with window_buffer.
            Has same shape as buffered window.
        window: Original window.
            Buffered window cropped with a given buffer.
        window_buffer: Buffered window.

    Returns:
        A cropped numpy array.
    """
    buffer_transform = rio.windows.transform(window_buffer, self.rio_ds.transform)
    window_transformed = rio.windows.from_bounds(
        *rio.windows.bounds(window, transform=self.rio_ds.transform),
        transform=buffer_transform
    )

    # row, col
    (
        window_buffer_slices_row,
        window_buffer_slices_col,
    ) = window_transformed.toslices()

    slice_col_start = int(round(window_buffer_slices_col.start))
    slice_row_start = int(round(window_buffer_slices_row.start))

    slice_col_stop = int(round(window_buffer_slices_col.stop))
    slice_row_stop = int(round(window_buffer_slices_row.stop))

    cropped_array = buffered_array[
        :, slice_row_start:slice_row_stop, slice_col_start:slice_col_stop
    ]
    return cropped_array

limit_window_to_raster_bounds(self, window, dst_height=None, dst_width=None)

Make sure the window fits in the dst raster. If not "clips" window to the bounds of the raster. This method is required because when applying a transform in the Windows precision in the Affine transformation can cause inconsistencies on the size of the windows in relation to the final output file.

Parameters:

Name Type Description Default
window Window

Original window.

required
dst_height int

Destination dataset height.

None
dst_width int

Destination dataset width.

None

Returns:

Type Description
Window

The original window limited to the destination dataset.

Source code in blockutils/windows.py
def limit_window_to_raster_bounds(
    self, window: Window, dst_height: int = None, dst_width: int = None
) -> Window:
    """
    Make sure the window fits in the dst raster. If not "clips" window to the
    bounds of the raster.
    This method is required because when applying a transform in the Windows
    precision in the Affine transformation can cause inconsistencies on the
    size of the windows in relation to the final output file.

    Arguments:
        window: Original window.
        dst_height: Destination dataset height.
        dst_width: Destination dataset width.

    Returns:
        The original window limited to the destination dataset.
    """
    if dst_height is None:
        dst_height = int(self.rio_ds.height)
    if dst_width is None:
        dst_width = int(self.rio_ds.width)

    window_slices_row, window_slices_col = window.toslices()
    result_row = [int(window_slices_row.start), int(window_slices_row.stop)]
    result_col = [int(window_slices_col.start), int(window_slices_col.stop)]

    if window_slices_row.start < 0:
        result_row[0] = 0
    if window_slices_row.stop > dst_height:
        result_row[1] = dst_height

    if window_slices_col.start < 0:
        result_col[0] = 0
    if window_slices_col.stop > dst_width:
        result_col[1] = dst_width

    return Window.from_slices(result_row, result_col)

transform_window(self, window, out_transform)

Transforms a window using the geopgraphical bounds to generate a new window given the Affine transform.

Parameters:

Name Type Description Default
window Window

Original window.

required
out_transform Affine

Transformation of the destination dataset.

required

Returns:

Type Description
Window

The original window transformed into the destination dataset.

Source code in blockutils/windows.py
def transform_window(self, window: Window, out_transform: rio.Affine) -> Window:
    """
    Transforms a window using the geopgraphical bounds to generate a new window
    given the Affine transform.

    Arguments:
        window: Original window.
        out_transform: Transformation of the destination dataset.

    Returns:
        The original window transformed into the destination dataset.
    """
    out_window = rio.windows.from_bounds(
        *rio.windows.bounds(window, transform=self.rio_ds.transform),
        transform=out_transform
    ).round_shape("ceil")
    return out_window

upsample_window_array(self, low_res_window, high_res_window, resampling_method=<Resampling.bilinear: 1>)

Method that returns an upsampled array to a given window, with an adapatable resampling method.

Parameters:

Name Type Description Default
low_res_window Window

The low resolution window.

required
high_res_window Window

The upsampled high resolution target window.

required
resampling_method Resampling

The resampling method to use.

<Resampling.bilinear: 1>

Returns:

Type Description
<built-in function array>

An upsampled window numpy array.

Source code in blockutils/windows.py
def upsample_window_array(
    self,
    low_res_window: Window,
    high_res_window: Window,
    resampling_method: Resampling = Resampling.bilinear,
) -> np.array:
    """
    Method that returns an upsampled array to a given window, with an adapatable
    resampling method.

    Arguments:
        low_res_window: The low resolution window.
        high_res_window: The upsampled high resolution target window.
        resampling_method: The resampling method to use.

    Returns:
        An upsampled window numpy array.
    """
    up_ar = self.rio_ds.read(
        window=low_res_window,
        out_shape=(
            self.rio_ds.count,
            high_res_window.height,
            high_res_window.width,
        ),
        resampling=resampling_method,
    )
    return up_ar

windows_buffered(self, buffer=0)

Method that returns buffered windows with a given int buffer.

Parameters:

Name Type Description Default
buffer int

A buffer size in pixels.

0

Returns:

Type Description
Iterable[rasterio.windows.Window]

An iterable of rasterio windows.

Source code in blockutils/windows.py
def windows_buffered(self, buffer: int = 0) -> Iterable[Window]:
    """
    Method that returns buffered windows with a given int buffer.

    Arguments:
        buffer: A buffer size in pixels.

    Returns:
        An iterable of rasterio windows.
    """
    for _, window in self.windows:
        buffered_window = self.buffer_window(window, buffer)
        yield window, buffered_window

windows_regular(self)

Method that returns regular rasterio windows.

Returns:

Type Description
Iterable[rasterio.windows.Window]

An iterable of rasterio windows.

Source code in blockutils/windows.py
def windows_regular(self) -> Iterable[Window]:
    """
    Method that returns regular rasterio windows.

    Returns:
        An iterable of rasterio windows.
    """
    for _, window in self.windows:
        yield window

windows_transformed(self, dst_transform, dst_px_height, dst_px_width)

Method that returns windows with a transformed into a given rasterio Dataset Affine transform.

Parameters:

Name Type Description Default
dst_transform Affine

A target transform to use to generate windows.

required
dst_px_height int

The destination dataset pixel height.

required
dst_px_width int

The destination dataset pixel width.

required

Returns:

Type Description
Iterable[rasterio.windows.Window]

An iterable of rasterio windows.

Source code in blockutils/windows.py
def windows_transformed(
    self, dst_transform: rio.Affine, dst_px_height: int, dst_px_width: int
) -> Iterable[Window]:
    """
    Method that returns windows with a transformed into a given rasterio Dataset
    Affine transform.

    Arguments:
        dst_transform: A target transform to use to generate windows.
        dst_px_height: The destination dataset pixel height.
        dst_px_width: The destination dataset pixel width.

    Returns:
        An iterable of rasterio windows.
    """
    for _, window in self.windows:
        transformed_window = self.transform_window(window, dst_transform)
        transformed_window = self.limit_window_to_raster_bounds(
            transformed_window, dst_px_height, dst_px_width
        )
        yield window, transformed_window