(x, window_shape, axis=None, automatic_rechunk=True)
| 820 | |
| 821 | @derived_from(np.lib.stride_tricks) |
| 822 | def sliding_window_view(x, window_shape, axis=None, automatic_rechunk=True): |
| 823 | window_shape = tuple(window_shape) if np.iterable(window_shape) else (window_shape,) |
| 824 | |
| 825 | window_shape_array = np.array(window_shape) |
| 826 | if np.any(window_shape_array <= 0): |
| 827 | raise ValueError("`window_shape` must contain values > 0") |
| 828 | |
| 829 | if axis is None: |
| 830 | axis = tuple(range(x.ndim)) |
| 831 | if len(window_shape) != len(axis): |
| 832 | raise ValueError( |
| 833 | f"Since axis is `None`, must provide " |
| 834 | f"window_shape for all dimensions of `x`; " |
| 835 | f"got {len(window_shape)} window_shape elements " |
| 836 | f"and `x.ndim` is {x.ndim}." |
| 837 | ) |
| 838 | else: |
| 839 | axis = normalize_axis_tuple(axis, x.ndim, allow_duplicate=True) |
| 840 | if len(window_shape) != len(axis): |
| 841 | raise ValueError( |
| 842 | f"Must provide matching length window_shape and " |
| 843 | f"axis; got {len(window_shape)} window_shape " |
| 844 | f"elements and {len(axis)} axes elements." |
| 845 | ) |
| 846 | |
| 847 | depths = [0] * x.ndim |
| 848 | for ax, window in zip(axis, window_shape): |
| 849 | depths[ax] += window - 1 |
| 850 | |
| 851 | # Ensure that each chunk is big enough to leave at least a size-1 chunk |
| 852 | # after windowing (this is only really necessary for the last chunk). |
| 853 | safe_chunks = list( |
| 854 | ensure_minimum_chunksize(d + 1, c) for d, c in zip(depths, x.chunks) |
| 855 | ) |
| 856 | if automatic_rechunk: |
| 857 | safe_chunks = [ |
| 858 | s if d != 0 else c for d, c, s in zip(depths, x.chunks, safe_chunks) |
| 859 | ] |
| 860 | # safe chunks is our output chunks, so add the new dimensions |
| 861 | safe_chunks.extend([(w,) for w in window_shape]) |
| 862 | max_chunk = reduce(mul, map(max, x.chunks)) |
| 863 | new_chunks = _calculate_new_chunksizes( |
| 864 | x.chunks, |
| 865 | safe_chunks.copy(), |
| 866 | {i for i, d in enumerate(depths) if d == 0}, |
| 867 | max_chunk, |
| 868 | ) |
| 869 | x = x.rechunk(tuple(new_chunks)) |
| 870 | else: |
| 871 | x = x.rechunk(tuple(safe_chunks)) |
| 872 | |
| 873 | # result.shape = x_shape_trimmed + window_shape, |
| 874 | # where x_shape_trimmed is x.shape with every entry |
| 875 | # reduced by one less than the corresponding window size. |
| 876 | # trim chunks to match x_shape_trimmed |
| 877 | newchunks = tuple(c[:-1] + (c[-1] - d,) for d, c in zip(depths, x.chunks)) + tuple( |
| 878 | (window,) for window in window_shape |
| 879 | ) |
nothing calls this directly
no test coverage detected