(
self,
process,
store,
n_total,
n_samples,
n_overlap,
sfreq,
window="hann",
tol=1e-10,
*,
name="COLA",
verbose=None,
)
| 267 | |
| 268 | @verbose |
| 269 | def __init__( |
| 270 | self, |
| 271 | process, |
| 272 | store, |
| 273 | n_total, |
| 274 | n_samples, |
| 275 | n_overlap, |
| 276 | sfreq, |
| 277 | window="hann", |
| 278 | tol=1e-10, |
| 279 | *, |
| 280 | name="COLA", |
| 281 | verbose=None, |
| 282 | ): |
| 283 | n_samples = _ensure_int(n_samples, "n_samples") |
| 284 | n_overlap = _ensure_int(n_overlap, "n_overlap") |
| 285 | n_total = _ensure_int(n_total, "n_total") |
| 286 | if n_samples <= 0: |
| 287 | raise ValueError(f"n_samples must be > 0, got {n_samples}") |
| 288 | if n_overlap < 0: |
| 289 | raise ValueError(f"n_overlap must be >= 0, got {n_overlap}") |
| 290 | if n_total < 0: |
| 291 | raise ValueError(f"n_total must be >= 0, got {n_total}") |
| 292 | self._n_samples = int(n_samples) |
| 293 | self._n_overlap = int(n_overlap) |
| 294 | del n_samples, n_overlap |
| 295 | if n_total < self._n_samples: |
| 296 | raise ValueError( |
| 297 | f"Number of samples per window ({self._n_samples}) must be at " |
| 298 | f"most the total number of samples ({n_total})" |
| 299 | ) |
| 300 | if not callable(process): |
| 301 | raise TypeError(f"process must be callable, got type {type(process)}") |
| 302 | self._process = process |
| 303 | self._step = self._n_samples - self._n_overlap |
| 304 | self._store = _check_store(store) |
| 305 | self._idx = 0 |
| 306 | self._in_buffers = self._out_buffers = None |
| 307 | self.name = name |
| 308 | |
| 309 | # Create our window boundaries |
| 310 | window_name = window if isinstance(window, str) else "custom" |
| 311 | self._window = get_window( |
| 312 | window, self._n_samples, fftbins=bool((self._n_samples - 1) % 2) |
| 313 | ) |
| 314 | self._window /= _check_cola( |
| 315 | self._window, self._n_samples, self._step, window_name, tol=tol |
| 316 | ) |
| 317 | self.starts = np.arange(0, n_total - self._n_samples + 1, self._step) |
| 318 | self.stops = self.starts + self._n_samples |
| 319 | delta = n_total - self.stops[-1] |
| 320 | self.stops[-1] = n_total |
| 321 | sfreq = float(sfreq) |
| 322 | pl = "s" if len(self.starts) != 1 else "" |
| 323 | logger.info( |
| 324 | f" Processing {len(self.starts):4d} data chunk{pl} of (at least) " |
| 325 | f"{self._n_samples / sfreq:0.1f} s with " |
| 326 | f"{self._n_overlap / sfreq:0.1f} s overlap and {window_name} windowing" |
nothing calls this directly
no test coverage detected