Dask's equivalence to np.utils.rolling_window
348612
def rolling_window(a, axis, window, center, fill_value):
    
    orig_shape = a.shape
    if axis < 0:
        axis = a.ndim + axis
    depth = {d: 0 for d in range(a.ndim)}
    depth[axis] = int(window / 2)
    # For evenly sized window, we need to crop the first point of each block.
    offset = 1 if window % 2 == 0 else 0

    if depth[axis] > min(a.chunks[axis]):
        raise ValueError(
            "For window size %d, every chunk should be larger than %d, "
            "but the smallest chunk size is %d. Rechunk your array\n"
            "with a larger chunk size or a chunk size that\n"
            "more evenly divides the shape of your array." %
            (window, depth[axis], min(a.chunks[axis])))

    # Although dask.overlap pads values to boundaries of the array,
    # the size of the generated array is smaller than what we want
    # if center == False.
    if center:
        start = int(window / 2)  # 10 -> 5,  9 -> 4
        end = window - 1 - start
    else:
        start, end = window - 1, 0
    pad_size = max(start, end) + offset - depth[axis]
    drop_size = 0
    # pad_size becomes more than 0 when the overlapped array is smaller than
    # needed. In this case, we need to enlarge the original array by padding
    # before overlapping.
    if pad_size > 0:
        if pad_size < depth[axis]:
            # overlapping requires each chunk larger than depth. If pad_size is
            # smaller than the depth, we enlarge this and truncate it later.
            drop_size = depth[axis] - pad_size
            pad_size = depth[axis]
        shape = list(a.shape)
        shape[axis] = pad_size
        chunks = list(a.chunks)
        chunks[axis] = (pad_size, )
        fill_array = da.full(shape, fill_value, dtype=a.dtype, chunks=chunks)
        a = da.concatenate([fill_array, a], axis=axis)

    boundary = {d: fill_value for d in range(a.ndim)}

    # create overlap arrays
    ag = overlap(a, depth=depth, boundary=boundary)

    # apply rolling func
    def func(x, window, axis=-1):
        x = np.asarray(x)
        rolling = nputils._rolling_window(x, window, axis)
        return rolling[(slice(None), ) * axis + (slice(offset, None), )]

    chunks = list(a.chunks)
    chunks.append(window)
    out = ag.map_blocks(func, dtype=a.dtype, new_axis=a.ndim, chunks=chunks,
                        window=window, axis=axis)

    # crop boundary.
    index = (slice(None),) * axis + (slice(drop_size,
                                           drop_size + orig_shape[axis]), )
    return out[index]
350070
def rolling(self, window, min_periods=None, center=False, win_type=None, on=None, axis=0, closed=None):
        
        kwds = {
            "window": window,
            "min_periods": min_periods,
            "center": center,
            "win_type": win_type,
            "on": on,
            "axis": axis,
            "closed": closed,
        }
        return Rolling(self._obj, self._npartitions, self._dask_threshold, self._scheduler, self._progress_bar, **kwds)
1167443
def batchsd(trace, batches=5):
    

    if len(np.shape(trace)) > 1:

        dims = np.shape(trace)
        # ttrace = np.transpose(np.reshape(trace, (dims[0], sum(dims[1:]))))
        ttrace = np.transpose([t.ravel() for t in trace])

        return np.reshape([batchsd(t, batches) for t in ttrace], dims[1:])

    else:
        if batches == 1:
            return np.std(trace) / np.sqrt(len(trace))

        try:
            batched_traces = np.resize(trace, (batches, int(len(trace) / batches)))
        except ValueError:
            # If batches do not divide evenly, trim excess samples
            resid = len(trace) % batches
            batched_traces = np.resize(trace[:-resid], 
                (batches, len(trace[:-resid]) / batches))

        means = np.mean(batched_traces, 1)

        return np.std(means) / np.sqrt(batches)
166582
def _rolling_lstsq(x, y):
    """"""
    if x.ndim == 2:
        # Treat everything as 3d and avoid AxisError on .swapaxes(1, 2) below
        # This means an original input of:
        #     array([0., 1., 2., 3., 4., 5., 6.])
        # becomes:
        # array([[[0.],
        #         [1.],
        #         [2.],
        #         [3.]],
        #
        #        [[1.],
        #         [2.],
        #         ...
        x = x[:, :, None]
    elif x.ndim <= 1:
        raise np.AxisError("x should have ndmi >= 2")
    return np.squeeze(
        np.matmul(
            np.linalg.inv(np.matmul(x.swapaxes(1, 2), x)),
            np.matmul(x.swapaxes(1, 2), np.atleast_3d(y)),
        )
    )
498283
def _get_mstd(cls, df, column, windows):
        
        window = cls.get_only_one_positive_int(windows)
        column_name = '{}_{}_mstd'.format(column, window)
        df[column_name] = df[column].rolling(min_periods=1, window=window,
                                             center=False).std()
1021418
def roll_sd(input, window, scale = 1.0, ddof = 0):
    ''''''
    nobs, i, j, sx, sxx = 0,0,0,0.,0.
    N = len(input)
    sqrt = np.sqrt

    if window > N:
        raise ValueError('Out of bound')

    output = np.ndarray(N-window+1,dtype=input.dtype)

    for val in input[:window]:
        if val == val:
            nobs += 1
            sx += val
            sxx += val*val

    nn = nobs - ddof
    output[j] = NaN if nn<=0 else sqrt(scale * (sxx - sx*sx/nobs) / nn)

    for val in input[window:]:
        prev = input[j]
        if prev == prev:
            sx -= prev
            sxx -= prev*prev
            nobs -= 1

        if val == val:
            nobs += 1
            sx += val
            sxx += val*val

        j += 1
        nn = nobs - ddof
        output[j] = NaN if nn<=0 else sqrt(scale * (sxx - sx*sx/nobs) / nn)

    return output
524056
def _full_like_variable(other, fill_value,
                        dtype: Union[str, np.dtype, None] = None):
    
    from .variable import Variable

    if isinstance(other.data, dask_array_type):
        import dask.array
        if dtype is None:
            dtype = other.dtype
        data = dask.array.full(other.shape, fill_value, dtype=dtype,
                               chunks=other.data.chunks)
    else:
        data = np.full_like(other, fill_value, dtype=dtype)

    return Variable(dims=other.dims, data=data, attrs=other.attrs)
1099551
def stoch(df, window=14, d=3, k=3, fast=False):
    

    my_df = pd.DataFrame(index=df.index)

    my_df['rolling_max'] = df['high'].rolling(window).max()
    my_df['rolling_min'] = df['low'].rolling(window).min()

    my_df['fast_k'] = 100 * (df['close'] - my_df['rolling_min'])/(my_df['rolling_max'] - my_df['rolling_min'])
    my_df['fast_d'] = my_df['fast_k'].rolling(d).mean()

    if fast:
        return my_df.loc[:, ['fast_k', 'fast_d']]

    my_df['slow_k'] = my_df['fast_k'].rolling(k).mean()
    my_df['slow_d'] = my_df['slow_k'].rolling(d).mean()

    return my_df.loc[:, ['slow_k', 'slow_d']]
647683
def rolling_window(a, axis, window, center, fill_value):
    
    pads = [(0, 0) for s in a.shape]
    if center:
        start = int(window / 2)  # 10 -> 5,  9 -> 4
        end = window - 1 - start
        pads[axis] = (start, end)
    else:
        pads[axis] = (window - 1, 0)
    a = np.pad(a, pads, mode='constant', constant_values=fill_value)
    return _rolling_window(a, window, axis)
872610
def windows_df(self):
        

        import pandas as pd
        if self.windows is None:
            raise Exception("You need to call the block_windows or windows before.")
        df_wins = []
        for row, col, win in zip(self.windows_row, self.windows_col, self.windows):
            df_wins.append(pd.DataFrame({"row":[row], "col":[col], "Window":[win]}))
        df_wins = pd.concat(df_wins).set_index(["row", "col"])
        df_wins["window_index"] = range(df_wins.shape[0])
        df_wins = df_wins.sort_index()
        return df_wins
93092
def mavg(data, xseq, **params):
    
    window = params['method_args']['window']

    # The first average comes after the full window size
    # has been swept over
    rolling = data['y'].rolling(**params['method_args'])
    y = rolling.mean()[window:]
    n = len(data)
    stderr = rolling.std()[window:]
    x = data['x'][window:]
    data = pd.DataFrame({'x': x, 'y': y})
    data.reset_index(inplace=True, drop=True)

    if params['se']:
        df = n - window  # Original - Used
        data['ymin'], data['ymax'] = tdist_ci(
            y, df, stderr, params['level'])
        data['se'] = stderr

    return data
98986
def apply_variable_ufunc(
    func,
    *args,
    signature,
    exclude_dims=frozenset(),
    dask='forbidden',
    output_dtypes=None,
    output_sizes=None,
    keep_attrs=False
):
    
    from .variable import Variable, as_compatible_data

    dim_sizes = unified_dim_sizes((a for a in args if hasattr(a, 'dims')),
                                  exclude_dims=exclude_dims)
    broadcast_dims = tuple(dim for dim in dim_sizes
                           if dim not in signature.all_core_dims)
    output_dims = [broadcast_dims + out for out in signature.output_core_dims]

    input_data = [broadcast_compat_data(arg, broadcast_dims, core_dims)
                  if isinstance(arg, Variable)
                  else arg
                  for arg, core_dims in zip(args, signature.input_core_dims)]

    if any(isinstance(array, dask_array_type) for array in input_data):
        if dask == 'forbidden':
            raise ValueError('apply_ufunc encountered a dask array on an '
                             'argument, but handling for dask arrays has not '
                             'been enabled. Either set the ``dask`` argument '
                             'or load your data into memory first with '
                             '``.load()`` or ``.compute()``')
        elif dask == 'parallelized':
            input_dims = [broadcast_dims + dims
                          for dims in signature.input_core_dims]
            numpy_func = func

            def func(*arrays):
                return _apply_with_dask_atop(
                    numpy_func, arrays, input_dims, output_dims,
                    signature, output_dtypes, output_sizes)
        elif dask == 'allowed':
            pass
        else:
            raise ValueError('unknown setting for dask array handling in '
                             'apply_ufunc: {}'.format(dask))
    result_data = func(*input_data)

    if signature.num_outputs == 1:
        result_data = (result_data,)
    elif (not isinstance(result_data, tuple) or
            len(result_data) != signature.num_outputs):
        raise ValueError('applied function does not have the number of '
                         'outputs specified in the ufunc signature. '
                         'Result is not a tuple of {} elements: {!r}'
                         .format(signature.num_outputs, result_data))

    output = []
    for dims, data in zip(output_dims, result_data):
        data = as_compatible_data(data)
        if data.ndim != len(dims):
            raise ValueError(
                'applied function returned data with unexpected '
                'number of dimensions: {} vs {}, for dimensions {}'
                .format(data.ndim, len(dims), dims))

        var = Variable(dims, data, fastpath=True)
        for dim, new_size in var.sizes.items():
            if dim in dim_sizes and new_size != dim_sizes[dim]:
                raise ValueError(
                    'size of dimension {!r} on inputs was unexpectedly '
                    'changed by applied function from {} to {}. Only '
                    'dimensions specified in ``exclude_dims`` with '
                    'xarray.apply_ufunc are allowed to change size.'
                    .format(dim, dim_sizes[dim], new_size))

        if keep_attrs and isinstance(args[0], Variable):
            var.attrs.update(args[0].attrs)
        output.append(var)

    if signature.num_outputs == 1:
        return output[0]
    else:
        return tuple(output)
1018653
def four_element_average_dask(d):
        """"""
        try:
            offset = d.attrs['area'].crop_offset
        except (KeyError, AttributeError):
            offset = (0, 0)

        res = d.data.map_blocks(_mean4, offset=offset, dtype=d.dtype)
        return xr.DataArray(res, attrs=d.attrs, dims=d.dims, coords=d.coords)
640821
def _bottleneck_reduce(cls, func):
        

        def wrapped_func(self, **kwargs):
            from .dataarray import DataArray

            # bottleneck doesn't allow min_count to be 0, although it should
            # work the same as if min_count = 1
            if self.min_periods is not None and self.min_periods == 0:
                min_count = 1
            else:
                min_count = self.min_periods

            axis = self.obj.get_axis_num(self.dim)

            padded = self.obj.variable
            if self.center:
                if (LooseVersion(np.__version__) < LooseVersion('1.13') and
                        self.obj.dtype.kind == 'b'):
                    # with numpy < 1.13 bottleneck cannot handle np.nan-Boolean
                    # mixed array correctly. We cast boolean array to float.
                    padded = padded.astype(float)

                if isinstance(padded.data, dask_array_type):
                    # Workaround to make the padded chunk size is larger than
                    # self.window-1
                    shift = - (self.window + 1) // 2
                    offset = (self.window - 1) // 2
                    valid = (slice(None), ) * axis + (
                        slice(offset, offset + self.obj.shape[axis]), )
                else:
                    shift = (-self.window // 2) + 1
                    valid = (slice(None), ) * axis + (slice(-shift, None), )
                padded = padded.pad_with_fill_value({self.dim: (0, -shift)})

            if isinstance(padded.data, dask_array_type):
                values = dask_rolling_wrapper(func, padded,
                                              window=self.window,
                                              min_count=min_count,
                                              axis=axis)
            else:
                values = func(padded.data, window=self.window,
                              min_count=min_count, axis=axis)

            if self.center:
                values = values[valid]
            result = DataArray(values, self.obj.coords)

            return result
        return wrapped_func
1083544
def roll_mean(input, window):
    ''''''
    nobs, i, j, sum_x = 0,0,0,0.
    N = len(input)

    if window > N:
        raise ValueError('Out of bound')

    output = np.ndarray(N-window+1,dtype=input.dtype)

    for val in input[:window]:
        if val == val:
            nobs += 1
            sum_x += val

    output[j] = NaN if not nobs else sum_x / nobs

    for val in input[window:]:
        prev = input[j]
        if prev == prev:
            sum_x -= prev
            nobs -= 1

        if val == val:
            nobs += 1
            sum_x += val

        j += 1
        output[j] = NaN if not nobs else sum_x / nobs

    return output
888904
def loopalt_gtd(time: datetime,
                glat: Union[float, np.ndarray], glon: Union[float, np.ndarray],
                altkm: Union[float, List[float], np.ndarray], *,
                f107a: float = None, f107: float = None, Ap: int = None) -> xarray.Dataset:
    
    glat = np.atleast_2d(glat)
    glon = np.atleast_2d(glon)
    assert glat.ndim == glon.ndim == 2

    times = np.atleast_1d(time)
    assert times.ndim == 1

    atmos = xarray.Dataset()

    for k, t in enumerate(times):
        print('computing', t)
        for i in range(glat.shape[0]):
            for j in range(glat.shape[1]):
                # atmos = xarray.concat((atmos, rungtd1d(t, altkm, glat[i,j], glon[i,j])),
                #                      data_vars='minimal',coords='minimal',dim='lon')
                atm = rungtd1d(t, altkm, glat[i, j], glon[i, j],
                               f107a=f107a, f107=f107, Ap=Ap)
                atmos = xarray.merge((atmos, atm))

    atmos.attrs = atm.attrs

    return atmos
934246
def _persist_inplace(self, **kwargs):
        
        # access .data to coerce everything to numpy or dask arrays
        lazy_data = {k: v._data for k, v in self.variables.items()
                     if isinstance(v._data, dask_array_type)}
        if lazy_data:
            import dask

            # evaluate all the dask arrays simultaneously
            evaluated_data = dask.persist(*lazy_data.values(), **kwargs)

            for k, data in zip(lazy_data, evaluated_data):
                self.variables[k].data = data

        return self