Function sliding_window_view

dask/array/_array_expr/_overlap.py:838–909

(x, window_shape, axis=None, automatic_rechunk=True)

@derived_from(np.lib.stride_tricks)
def sliding_window_view(x, window_shape, axis=None, automatic_rechunk=True):
    window_shape = tuple(window_shape) if np.iterable(window_shape) else (window_shape,)

    window_shape_array = np.array(window_shape)
    if np.any(window_shape_array <= 0):
        raise ValueError("`window_shape` must contain values > 0")

    if axis is None:
        axis = tuple(range(x.ndim))
        if len(window_shape) != len(axis):
            raise ValueError(
                f"Since axis is `None`, must provide "
                f"window_shape for all dimensions of `x`; "
                f"got {len(window_shape)} window_shape elements "
                f"and `x.ndim` is {x.ndim}."
            )
    else:
        axis = normalize_axis_tuple(axis, x.ndim, allow_duplicate=True)
        if len(window_shape) != len(axis):
            raise ValueError(
                f"Must provide matching length window_shape and "
                f"axis; got {len(window_shape)} window_shape "
                f"elements and {len(axis)} axes elements."
            )

    depths = [0] * x.ndim
    for ax, window in zip(axis, window_shape):
        depths[ax] += window - 1

    # Ensure that each chunk is big enough to leave at least a size-1 chunk
    # after windowing (this is only really necessary for the last chunk).
    safe_chunks = list(
        ensure_minimum_chunksize(d + 1, c) for d, c in zip(depths, x.chunks)
    )
    if automatic_rechunk:
        safe_chunks = [
            s if d != 0 else c for d, c, s in zip(depths, x.chunks, safe_chunks)
        ]
        # safe chunks is our output chunks, so add the new dimensions
        safe_chunks.extend([(w,) for w in window_shape])
        max_chunk = reduce(mul, map(max, x.chunks))
        new_chunks = _calculate_new_chunksizes(
            x.chunks,
            safe_chunks.copy(),
            {i for i, d in enumerate(depths) if d == 0},
            max_chunk,
        )
        x = x.rechunk(tuple(new_chunks))
    else:
        x = x.rechunk(tuple(safe_chunks))

    # result.shape = x_shape_trimmed + window_shape,
    # where x_shape_trimmed is x.shape with every entry
    # reduced by one less than the corresponding window size.
    # trim chunks to match x_shape_trimmed
    newchunks = tuple(c[:-1] + (c[-1] - d,) for d, c in zip(depths, x.chunks)) + tuple(
        (window,) for window in window_shape
    )