class_docstr: Constant overlap-add processing helper.
  
  Parameters
  ----------
  process : callable
      A function that takes a chunk of input data with shape
      ``(n_channels, n_samples)`` and processes it.
  store : callable | ndarray
      A function that takes a completed chunk of output data.
      Can also be an ``ndarray``, in which case it is treated as the
      output data in which to store the results.
  n_total : int
      The total number of samples.
  n_samples : int
      The number of samples per window.
  n_overlap : int
      The overlap between windows.
  window : str
      The window to use. Default is "hann".
  tol : float
      The tolerance for COLA checking.
  
  Notes
  -----
  This will process data using overlapping windows to achieve a constant
  output value. For example, for ``n_total=27``, ``n_samples=10``,
  ``n_overlap=5`` and ``window='triang'``::
  
      1 _____               _______
        |    \   /\   /\   /
        |     \ /  \ /  \ /
        |      x    x    x
        |     / \  / \  / \
        |    /   \/   \/   \
      0 +----|----|----|----|----|-
        0    5   10   15   20   25
  
  This produces four windows: the first three are the requested length
  (10 samples) and the last one is longer (12 samples). The first and last
  window are asymmetric.
docstr: Pass in a chunk of data.
docstr_tok: ["Pass","in","a","chunk","of","data","."]
code: def feed(self, *datas, verbose=None, **kwargs):
      """Pass in a chunk of data."""
      # Append to our input buffer
      if self._in_buffers is None:
          self._in_buffers = [None] * len(datas)
      if len(datas) != len(self._in_buffers):
          raise ValueError(
              "Got %d array(s), needed %d"
              % (len(datas), len(self._in_buffers))
          )
      for di, data in enumerate(datas):
          if not isinstance(data, np.ndarray) or data.ndim < 1:
              raise TypeError(
                  "data entry %d must be an 2D ndarray, got %s"
                  % (
                      di,
                      type(data),
                  )
              )
          if self._in_buffers[di] is None:
              # In practice, users can give large chunks, so we use
              # dynamic allocation of the in buffer. We could save some
              # memory allocation by only ever processing max_len at once,
              # but this would increase code complexity.
              self._in_buffers[di] = np.empty(
                  data.shape[:-1] + (0,), data.dtype
              )
          if (
              data.shape[:-1] != self._in_buffers[di].shape[:-1]
              or self._in_buffers[di].dtype != data.dtype
          ):
              raise TypeError(
                  "data must dtype {} and shape[:-1]=={}, got dtype {} shape[:-1]="
                  "{}".format(
                      self._in_buffers[di].dtype,
                      self._in_buffers[di].shape[:-1],
                      data.dtype,
                      data.shape[:-1],
                  )
              )
          logger.debug(
              "    + Appending %d->%d"
              % (self._in_offset, self._in_offset + data.shape[-1])
          )
          self._in_buffers[di] = np.concatenate(
              [self._in_buffers[di], data], -1
          )
          if self._in_offset > self.stops[-1]:
              raise ValueError(
                  f"data (shape {data.shape}) exceeded expected total buffer size ("
                  f"{self._in_offset} > {self.stops[-1]})"
              )
      # Check to see if we can process the next chunk and dump outputs
      while (
          self._idx < len(self.starts)
          and self._in_offset >= self.stops[self._idx]
      ):
          start, stop = self.starts[self._idx], self.stops[self._idx]
          this_len = stop - start
          this_window = self._window.copy()
          if self._idx == len(self.starts) - 1:
              this_window = np.pad(
                  self._window,
                  (0, this_len - len(this_window)),
                  "constant",
              )
              for offset in range(
                  self._step, len(this_window), self._step
              ):
                  n_use = len(this_window) - offset
                  this_window[offset:] += self._window[:n_use]
          if self._idx == 0:
              for offset in range(
                  self._n_samples - self._step, 0, -self._step
              ):
                  this_window[:offset] += self._window[-offset:]
          logger.debug(f"    * Processing {start}->{stop}")
          this_proc = [
              in_[..., :this_len].copy() for in_ in self._in_buffers
          ]
          if not all(
              proc.shape[-1] == this_len == this_window.size
              for proc in this_proc
          ):
              raise RuntimeError("internal indexing error")
          outs = self._process(*this_proc, **kwargs)
          if self._out_buffers is None:
              max_len = np.max(self.stops - self.starts)
              self._out_buffers = [
                  np.zeros(o.shape[:-1] + (max_len,), o.dtype)
                  for o in outs
              ]
          for oi, out in enumerate(outs):
              out *= this_window
              self._out_buffers[oi][..., : stop - start] += out
          self._idx += 1
          if self._idx < len(self.starts):
              next_start = self.starts[self._idx]
          else:
              next_start = self.stops[-1]
          delta = next_start - self.starts[self._idx - 1]
          for di in range(len(self._in_buffers)):
              self._in_buffers[di] = self._in_buffers[di][..., delta:]
          logger.debug(
              "    - Shifting input/output buffers by %d samples"
              % (delta,)
          )
          self._store(*[o[..., :delta] for o in self._out_buffers])
          for ob in self._out_buffers:
              ob[..., :-delta] = ob[..., delta:]
              ob[..., -delta:] = 0.0
  
file_path: mne-python/mne/_ola.py
imports: import numpy
  from scipy.signal import get_window
  from .utils import _ensure_int, logger, verbose
label_desc: Use image node_id 3 for calling the _COLA obj's underlying member method code with example usage: obj.feed() without return types