diff --git a/src/RespFlow/access_files.py b/src/RespFlow/access_files.py index cf7c646..f6b2f68 100644 --- a/src/RespFlow/access_files.py +++ b/src/RespFlow/access_files.py @@ -46,13 +46,15 @@ def make_paths(root: str | None = None, raw: str | None = None) -> dict[str, str # Create dictionary path_names = { 'raw':raw, - 'detrend':os.path.join(root, '2_detrend'), - 'bandpass':os.path.join(root, '3_bandpass'), - 'fwr':os.path.join(root, '4_fwr'), - 'screened':os.path.join(root, '5_screened'), - 'filled':os.path.join(root, '6_filled'), - 'smooth':os.path.join(root, '7_smoothed'), - 'feature':os.path.join(root, '8_feature') + 'hard_fault':os.path.join(root, '2_hard_fault'), + 'detrend':os.path.join(root, '3_detrend'), + 'micro_interp':os.path.join(root, '4_micro_interp'), + 'bandpass':os.path.join(root, '5_bandpass'), + 'fwr':os.path.join(root, '6_fwr'), + 'screened':os.path.join(root, '7_screened'), + 'filled':os.path.join(root, '8_filled'), + 'smooth':os.path.join(root, '9_smoothed'), + 'feature':os.path.join(root, '10_feature') } # Create folders diff --git a/src/RespFlow/plot_signals.py b/src/RespFlow/plot_signals.py index caf2119..21c1913 100644 --- a/src/RespFlow/plot_signals.py +++ b/src/RespFlow/plot_signals.py @@ -21,7 +21,7 @@ def plot_dashboard(mapped_files : dict[str, str], max_points=10000) -> None: # Define all processing stages in order - stages = ['raw', 'detrend', 'bandpass', 'fwr', 'screened', 'filled', 'smooth', 'feature'] + stages = ['raw', 'hard_fault', 'detrend', 'micro_interp', 'bandpass', 'fwr', 'screened', 'filled', 'smooth', 'feature'] app = Dash() diff --git a/src/RespFlow/preprocess_signals.py b/src/RespFlow/preprocess_signals.py index 40fcccb..445a0b3 100644 --- a/src/RespFlow/preprocess_signals.py +++ b/src/RespFlow/preprocess_signals.py @@ -4,6 +4,7 @@ from pathlib import Path from scipy.ndimage import median_filter import numpy as np +from dataclasses import dataclass # # ============================================================================= @@ -13,6 +14,306 @@ A collection of functions for preprocessing signals. """ +# +# HARD FAULT +# ============================================================================= +# + +@dataclass +class HardFaultConfig: + """ + Hard-fault detection parameters. + + Hard faults are unambiguous sensor/data failures that should be masked + before detrend, micro-gap fill, and bandpass filtering. + """ + # Flatline / stuck sensor + flat_min_s: float = 1.0 # minimum duration to qualify as flatline + flat_sensitivity: float = 0.05 # multiplier on MAD(dx) for flatline threshold + + # Clipping / saturation (data-driven rails) + clip_percentile: float = 0.1 # lower percentile for rail detection (upper = 100 - this) + clip_min_run_s: float = 0.25 # minimum clipping run duration (seconds) + + # Step/discontinuity spikes + step_sensitivity: float = 12.0 # multiplier on MAD(dx) for step threshold + step_pad_s: float = 0.05 # pad around steps (seconds) + step_verify_window_s: float = 0.5 # window (seconds) to check sustained level shift + + # Optional: pad around any hard fault + fault_pad_s: float = 0.0 # additional dilation of final hard-fault mask (seconds) + + +def _runs_from_mask(mask: np.ndarray) -> list[tuple[int, int]]: + """Return (start, end) half-open index pairs for contiguous True-runs.""" + mask = np.asarray(mask, dtype=bool) + if mask.size == 0: + return [] + d = np.diff(mask.astype(np.int8)) + starts = np.where(d == 1)[0] + 1 + ends = np.where(d == -1)[0] + 1 + if mask[0]: + starts = np.r_[0, starts] + if mask[-1]: + ends = np.r_[ends, mask.size] + return list(zip(starts.tolist(), ends.tolist())) + + +def _apply_min_run_length(mask: np.ndarray, min_len: int) -> np.ndarray: + """Keep only True-runs of length >= min_len.""" + mask = np.asarray(mask, dtype=bool) + if min_len <= 1: + return mask + out = np.zeros_like(mask, dtype=bool) + for a, b in _runs_from_mask(mask): + if (b - a) >= min_len: + out[a:b] = True + return out + + +def _dilate_mask(mask: np.ndarray, radius: int) -> np.ndarray: + """Dilate a boolean mask by +/- radius samples using convolution.""" + mask = np.asarray(mask, dtype=bool) + if radius <= 0 or mask.size == 0: + return mask + kernel = np.ones(2 * radius + 1, dtype=int) + return np.convolve(mask.astype(int), kernel, mode="same") > 0 + + +def _robust_mad(x: np.ndarray) -> float: + """NaN-safe median absolute deviation (MAD).""" + x = np.asarray(x, dtype=float) + x = x[~np.isnan(x)] + if x.size == 0: + return np.nan + med = np.median(x) + return float(np.median(np.abs(x - med))) + + +def apply_hard_fault( + signal: np.ndarray, + sampling_rate: int, + config: HardFaultConfig | None = None, +) -> tuple[np.ndarray, dict[str, object]]: + """ + Detect hard faults and set them to NaN. + + Returns: + signal_out: signal with hard-fault samples set to NaN + info: dict with masks and thresholds used + """ + x = np.asarray(signal, dtype=float).copy() + if x.ndim != 1: + raise ValueError("apply_hard_fault expects a 1D signal.") + + N = x.size + fs = float(sampling_rate) + if config is None: + config = HardFaultConfig() + + mask_nan = np.isnan(x) + + # Early return for very short or fully-missing signals + if N < 3 or np.all(mask_nan): + mask_hardfault = mask_nan.copy() + x[mask_hardfault] = np.nan + info = { + "mask_nan": mask_nan, + "mask_flatline": np.zeros(N, dtype=bool), + "mask_clip": np.zeros(N, dtype=bool), + "mask_step": np.zeros(N, dtype=bool), + "mask_hardfault": mask_hardfault, + "runs_hardfault": _runs_from_mask(mask_hardfault), + } + return x, info + + # Robust scales (computed on available data) + mad_x = _robust_mad(x) + dx = np.diff(x) # NaNs propagate into dx where adjacent samples include NaN + mad_dx = _robust_mad(dx) + + if not np.isfinite(mad_x) or mad_x <= 0: + mad_x = np.finfo(float).eps + if not np.isfinite(mad_dx) or mad_dx <= 0: + mad_dx = np.finfo(float).eps + + # ------------------------------------------------------------------------- + # Flatline / stuck sensor (near-zero first differences sustained) + # ------------------------------------------------------------------------- + flat_eps = max( + config.flat_sensitivity * mad_dx, + 1e-4 * mad_x, # safety floor: fraction of signal scale + np.finfo(float).eps, + ) + + dx_abs = np.abs(dx) + mask_dx_small = (dx_abs <= flat_eps) & ~np.isnan(dx) + + min_flat_samples = int(np.ceil(config.flat_min_s * fs)) + mask_flatline = np.zeros(N, dtype=bool) + + # A run of small dx from [a, b) implies constant samples [a, b+1) + for run_start, run_end in _runs_from_mask(mask_dx_small): + sample_start, sample_end = run_start, min(N, run_end + 1) + if (sample_end - sample_start) >= min_flat_samples: + mask_flatline[sample_start:sample_end] = True + + mask_flatline &= ~mask_nan + + # ------------------------------------------------------------------------- + # Clipping / saturation (runs near inferred rails) + # ------------------------------------------------------------------------- + valid_signal = x[~mask_nan] + rail_low = np.percentile(valid_signal, config.clip_percentile) + rail_high = np.percentile(valid_signal, 100 - config.clip_percentile) + clip_tol = 0.01 * mad_x + + mask_clip_raw = (~mask_nan) & ((x <= rail_low + clip_tol) | (x >= rail_high - clip_tol)) + min_clip_samples = int(np.ceil(config.clip_min_run_s * fs)) + mask_clip = _apply_min_run_length(mask_clip_raw, min_clip_samples) + + # ------------------------------------------------------------------------- + # Step/discontinuity spikes (robust threshold on dx) + # ------------------------------------------------------------------------- + valid_dx = dx[~np.isnan(dx)] + dx_med = float(np.median(valid_dx)) if valid_dx.size else 0.0 + + # Floor the threshold so it never collapses for smooth signals + step_threshold = max(config.step_sensitivity * mad_dx, 0.5 * mad_x) + + step_candidates = np.where(np.abs(dx - dx_med) > step_threshold)[0] + mask_step = np.zeros(N, dtype=bool) + step_pad = int(np.ceil(config.step_pad_s * fs)) + + verify_win = int(np.ceil(config.step_verify_window_s * fs)) + min_shift = 0.3 * mad_x + # Massive spikes (3x threshold) bypass verification + spike_bypass_thr = 3.0 * step_threshold + + for i in step_candidates: + dx_mag = abs(float(dx[i]) - dx_med) + + # Massive spike — flag unconditionally, no verification needed + if dx_mag >= spike_bypass_thr: + mask_start = max(0, i - step_pad) + mask_end = min(N, i + 2 + step_pad) + mask_step[mask_start:mask_end] = True + continue + + # Moderate spike — verify sustained level shift + before_start = max(0, i - verify_win) + after_end = min(N, i + 2 + verify_win) + seg_before = x[before_start:i] + seg_after = x[i + 1:after_end] + + seg_before = seg_before[~np.isnan(seg_before)] + seg_after = seg_after[~np.isnan(seg_after)] + + if seg_before.size == 0 or seg_after.size == 0: + continue + + shift = abs(float(np.median(seg_after)) - float(np.median(seg_before))) + if shift < min_shift: + continue + + mask_start = max(0, i - step_pad) + mask_end = min(N, i + 2 + step_pad) + mask_step[mask_start:mask_end] = True + + mask_step &= ~mask_nan + + # ------------------------------------------------------------------------- + # Combine and pad + # ------------------------------------------------------------------------- + mask_hardfault = mask_nan | mask_flatline | mask_clip | mask_step + + if config.fault_pad_s and config.fault_pad_s > 0: + fault_pad = int(np.ceil(config.fault_pad_s * fs)) + mask_hardfault = _dilate_mask(mask_hardfault, fault_pad) + + x[mask_hardfault] = np.nan + + info: dict[str, object] = { + "mask_nan": mask_nan, + "mask_flatline": mask_flatline, + "mask_clip": mask_clip, + "mask_step": mask_step, + "mask_hardfault": mask_hardfault, + "runs_hardfault": _runs_from_mask(mask_hardfault), + "mad_x": mad_x, + "mad_dx": mad_dx, + "flat_eps": flat_eps, + "rail_low": rail_low, + "rail_high": rail_high, + "clip_tol": clip_tol, + "step_threshold": step_threshold, + } + return x, info + + +def apply_hard_fault_to_df( + df: pd.DataFrame, + sampling_rate: int, + config: HardFaultConfig | None = None, + time_col: str = "time", +) -> pd.DataFrame: + """ + Apply hard-fault detection to all non-time columns of a dataframe. + Replaces hard-fault samples with NaN in each signal column. + """ + out = df.copy() + + for column in out.columns: + if column.lower() == time_col.lower(): + continue + + x = out[column].to_numpy(dtype=float) + x_hf, _info = apply_hard_fault(x, sampling_rate, config=config) + out[column] = x_hf + + return out + + +def hard_fault_signals( + in_path: str, + out_path: str, + sampling_rate: int, + config: HardFaultConfig | None = None, +) -> None: + """ + Applies hard-fault detection to all columns except 'time' in all CSV files. + Preserves folder structure from in_path to out_path. + + Parameters + ---------- + in_path : str + Input directory path + out_path : str + Output directory path + sampling_rate : int + Sampling rate in Hz + config : HardFaultConfig, optional + Detection configuration. Uses defaults if None. + """ + mapped_files = map_files(in_path, file_ext='csv') + + in_path_obj = Path(in_path) + out_path_obj = Path(out_path) + + for file_path in mapped_files.values(): + df = pd.read_csv(file_path) + + df2 = apply_hard_fault_to_df(df, sampling_rate, config=config) + + file_path_obj = Path(file_path) + relative_path = file_path_obj.relative_to(in_path_obj) + output_file_path = out_path_obj / relative_path + output_file_path.parent.mkdir(parents=True, exist_ok=True) + + df2.to_csv(output_file_path, index=False) + + print(f"Processed {len(mapped_files)} files from {in_path} to {out_path}") + # # DETREND # ============================================================================= @@ -56,6 +357,7 @@ def apply_detrend(signal: list | tuple, sampling_rate: int, window_size_seconds: return detrended_signal, baseline + def detrend_signals(in_path: str, out_path: str, sampling_rate: int, window_size_seconds: int = 60) -> None: """ Applies detrending to all columns except 'time' in all CSV files. @@ -104,69 +406,23 @@ def detrend_signals(in_path: str, out_path: str, sampling_rate: int, window_size print(f"Processed {len(mapped_files)} files from {in_path} to {out_path}") # -# BANDPASS +# MICRO INTERP # ============================================================================= # -def apply_bandpass(data: list | tuple, sampling_rate: int, lowcut: float = 0.05, highcut: float = 2.0, order: int = 2) -> list | tuple: - """ - Applies a zero-phase Butterworth bandpass filter. - Standard: 0.05-2.0 Hz for RIP belt data. - """ - nyquist = 0.5 * sampling_rate - low = lowcut / nyquist - high = highcut / nyquist +# Physiological constant: typical resting respiratory rate +RESTING_RR_HZ = 0.25 # 0.25 Hz ≈ 15 breaths/min (upper bound for resting adults) - # Design filter - sos = butter(order, [low, high], btype='band', output='sos') - # Apply zero-phase filter (filtfilt) with padlen adjusted for short signals - padlen = min(len(data) - 1, 15) - y = sosfiltfilt(sos, data, padlen=padlen) - - return y - - -def min_viable_length_sosfiltfilt( - sampling_rate: int, - lowcut: float = 0.05, - highcut: float = 2.0, - order: int = 2, -) -> dict: +def default_max_gap(sampling_rate: int, resting_rate: float = RESTING_RR_HZ) -> int: """ - Compute the SciPy sosfiltfilt *default* padlen for a Butterworth bandpass and - return the minimum viable segment length N_min such that padlen < N-1. + Compute a default max_gap (in samples) for NaN micro gap interpolation. - Returns: - { - "n_sections": int, - "padlen_default": int, - "min_sequence_length": int - } + Rule: 30% of one respiratory cycle length. + At 2000 Hz, 0.25 Hz: 0.3 * (2000 / 0.25) = 2400 samples. """ - nyquist = 0.5 * sampling_rate - low = lowcut / nyquist - high = highcut / nyquist + return int(round(0.3 * sampling_rate / resting_rate)) - sos = butter(order, [low, high], btype="band", output="sos") - n_sections = sos.shape[0] - - # From SciPy docs for sosfiltfilt default padding length: - # padlen_default = 3 * (2*n_sections + 1 - min(z0, p0)) - # where z0 is the number of zeros at the origin, p0 is the number of poles at the origin. - z0 = int(np.sum(sos[:, 2] == 0.0)) # b2 == 0 indicates a zero at z=0 - p0 = int(np.sum(sos[:, 5] == 0.0)) # a2 == 0 indicates a pole at z=0 - padlen_default = int(3 * (2 * n_sections + 1 - min(z0, p0))) - - # sosfiltfilt requires padlen < N-1 => N >= padlen + 2 - min_sequence_length = padlen_default + 2 - - return { - "n_sections": int(n_sections), - "padlen_default": int(padlen_default), - "min_sequence_length": int(min_sequence_length), - } - def nan_gap_indices(x: np.ndarray) -> list[tuple[int, int, int]]: """ Return (start, end, length) for each contiguous NaN gap in a 1D array. @@ -189,7 +445,6 @@ def nan_gap_indices(x: np.ndarray) -> list[tuple[int, int, int]]: return [(s, e, e - s) for s, e in zip(starts.tolist(), ends.tolist())] - def interpolate_nan_gaps( data: np.ndarray, method: str, @@ -255,6 +510,152 @@ def interpolate_nan_gaps( return result, nan_mask +def apply_micro_interp( + signal: np.ndarray, + sampling_rate: int, + max_gap: int | None = None, + interp_method: str = "pchip" +) -> np.ndarray: + """ + Interpolate small NaN gaps in a 1D signal. + + Parameters + ---------- + signal : np.ndarray + 1D input signal (may contain NaN). + sampling_rate : int + Sampling rate in Hz. + max_gap : int or None + Maximum gap size (samples) to interpolate. If None, uses + default_max_gap(sampling_rate). + interp_method : str + Interpolation method: "pchip" (default) or "cubic_spline". + + Returns + ------- + np.ndarray + Signal with small NaN gaps filled via interpolation. + """ + signal = np.asarray(signal, dtype=float) + + if max_gap is None: + max_gap = default_max_gap(sampling_rate) + + filled, _nan_mask = interpolate_nan_gaps(signal, method=interp_method, max_gap=max_gap) + return filled + + +def micro_interp_signals( + in_path: str, + out_path: str, + sampling_rate: int, + max_gap: int | None = None, + interp_method: str = "pchip" +) -> None: + """ + Interpolate small NaN gaps in all columns except 'time' in all CSV files. + Preserves folder structure from in_path to out_path. + + Parameters + ---------- + in_path : str + Input directory path + out_path : str + Output directory path + sampling_rate : int + Sampling rate in Hz + max_gap : int, optional + Maximum gap size (samples) to interpolate. If None, uses + default_max_gap(sampling_rate). + interp_method : str, optional + Interpolation method: "pchip" (default) or "cubic_spline". + """ + mapped_files = map_files(in_path, file_ext='csv') + + in_path_obj = Path(in_path) + out_path_obj = Path(out_path) + + for file_path in mapped_files.values(): + df = pd.read_csv(file_path) + + for column in df.columns: + if column.lower() != 'time': + df[column] = apply_micro_interp(df[column].values, sampling_rate, max_gap, interp_method) + + file_path_obj = Path(file_path) + relative_path = file_path_obj.relative_to(in_path_obj) + output_file_path = out_path_obj / relative_path + output_file_path.parent.mkdir(parents=True, exist_ok=True) + + df.to_csv(output_file_path, index=False) + + print(f"Processed {len(mapped_files)} files from {in_path} to {out_path}") + +# +# BANDPASS +# ============================================================================= +# + +def apply_bandpass(data: list | tuple, sampling_rate: int, lowcut: float = 0.05, highcut: float = 2.0, order: int = 2) -> list | tuple: + """ + Applies a zero-phase Butterworth bandpass filter. + Standard: 0.05-2.0 Hz for RIP belt data. + """ + nyquist = 0.5 * sampling_rate + low = lowcut / nyquist + high = highcut / nyquist + + # Design filter + sos = butter(order, [low, high], btype='band', output='sos') + + # Apply zero-phase filter (filtfilt) with padlen adjusted for short signals + padlen = min(len(data) - 1, 15) + y = sosfiltfilt(sos, data, padlen=padlen) + + return y + + +def min_viable_length_sosfiltfilt( + sampling_rate: int, + lowcut: float = 0.05, + highcut: float = 2.0, + order: int = 2, +) -> dict: + """ + Compute the SciPy sosfiltfilt *default* padlen for a Butterworth bandpass and + return the minimum viable segment length N_min such that padlen < N-1. + + Returns: + { + "n_sections": int, + "padlen_default": int, + "min_sequence_length": int + } + """ + nyquist = 0.5 * sampling_rate + low = lowcut / nyquist + high = highcut / nyquist + + sos = butter(order, [low, high], btype="band", output="sos") + n_sections = sos.shape[0] + + # From SciPy docs for sosfiltfilt default padding length: + # padlen_default = 3 * (2*n_sections + 1 - min(z0, p0)) + # where z0 is the number of zeros at the origin, p0 is the number of poles at the origin. + z0 = int(np.sum(sos[:, 2] == 0.0)) # b2 == 0 indicates a zero at z=0 + p0 = int(np.sum(sos[:, 5] == 0.0)) # a2 == 0 indicates a pole at z=0 + padlen_default = int(3 * (2 * n_sections + 1 - min(z0, p0))) + + # sosfiltfilt requires padlen < N-1 => N >= padlen + 2 + min_sequence_length = padlen_default + 2 + + return { + "n_sections": int(n_sections), + "padlen_default": int(padlen_default), + "min_sequence_length": int(min_sequence_length), + } + + def nan_islands(x: np.ndarray) -> list[tuple[int, int]]: """ Return (start, end) index pairs for contiguous non-NaN regions ("islands") @@ -299,19 +700,16 @@ def apply_bandpass_nan_safe( lowcut: float, highcut: float, order: int, - max_gap: int | None = None, - interp_method: str = "pchip" ) -> np.ndarray: """ - NaN-safe bandpass filter using interpolation. + NaN-safe bandpass filter. + + If the signal has no NaNs, filters directly. If NaNs remain (e.g. large + unfilled gaps), filters each contiguous non-NaN island separately. Parameters: data: Input signal (may contain NaN) sampling_rate, lowcut, highcut, order: Filter parameters - max_gap: Max gap size (samples) to interpolate. If None, interpolate all gaps. - Gaps larger than max_gap remain as NaN in output. - interp_method: Specified interpolation method. Defaults to pchip. - Alternatively user can specify "cubic_spline". """ data = np.asarray(data, dtype=float) @@ -319,26 +717,13 @@ def apply_bandpass_nan_safe( if not np.any(np.isnan(data)): return apply_bandpass(data, sampling_rate, lowcut, highcut, order) - # Interpolate gaps - interpolated, original_nan_mask = interpolate_nan_gaps(data, method=interp_method, max_gap=max_gap) - - # Determine which NaNs were NOT filled (large gaps when max_gap is set) - still_nan = np.isnan(interpolated) + # NaNs present — filter each non-NaN island separately + min_len = min_viable_length_sosfiltfilt(sampling_rate, lowcut, highcut, order)["min_sequence_length"] + result = np.full_like(data, np.nan) - if np.any(still_nan): - # Some gaps weren't filled - filter valid segments only - min_len = min_viable_length_sosfiltfilt(sampling_rate, lowcut, highcut, order)["min_sequence_length"] - result = np.full_like(data, np.nan) - - for start, end, segment in iter_nan_islands(interpolated): - if len(segment) >= min_len: - result[start:end] = apply_bandpass(segment, sampling_rate, lowcut, highcut, order) - else: - # All gaps filled - filter entire signal - result = apply_bandpass(interpolated, sampling_rate, lowcut, highcut, order) - - # Restore original NaN positions - # result[original_nan_mask] = np.nan + for start, end, segment in iter_nan_islands(data): + if len(segment) >= min_len: + result[start:end] = apply_bandpass(segment, sampling_rate, lowcut, highcut, order) return result @@ -351,8 +736,6 @@ def bandpass_filter_signals( sampling_rate: int, passband: str | tuple = 'default', order: int = 2, - max_gap: int | None = None, - interp_method: str = "pchip" ) -> None: """ Applies a Butterworth bandpass filter to all columns except 'time' in all CSV files. @@ -370,17 +753,10 @@ def bandpass_filter_signals( Preset name or tuple of (lowcut, highcut) in Hz. Presets: 'default' (0.05-2.0 Hz), 'resting_adult' (0.05-1.0 Hz), 'narrow_band' (0.1-0.35 Hz), 'wide_band' (0.05-3.0 Hz). - Default: 'resting_adult' + Default: 'default' order : int, optional Filter order (default: 2) - max_gap : int, optional - Maximum gap size (in samples) to interpolate over. If None, all NaN gaps - are interpolated before filtering. Gaps larger than max_gap remain as NaN - in the output. (default: None) - interp_method : Specified interpolation method. Defaults to pchip. - Alternatively user can specify "cubic_spline". """ - PASSBANDS = { 'default': (0.05, 2.0), 'resting_adult': (0.05, 1), @@ -411,7 +787,7 @@ def bandpass_filter_signals( # Apply bandpass filter to all columns except 'time' for column in df.columns: if column.lower() != 'time': - df[column] = apply_bandpass_nan_safe(df[column].values, sampling_rate, lowcut, highcut, order, max_gap, interp_method=interp_method) + df[column] = apply_bandpass_nan_safe(df[column].values, sampling_rate, lowcut, highcut, order) # Determine the relative path from in_path to preserve folder structure file_path_obj = Path(file_path) diff --git a/tests/test_access_files.py b/tests/test_access_files.py index 17915d5..9d7ad71 100644 --- a/tests/test_access_files.py +++ b/tests/test_access_files.py @@ -41,7 +41,7 @@ def test_make_paths_defaults(mock_filesystem): paths = make_paths() expected_keys = { - 'raw', 'detrend', 'bandpass', 'fwr', + 'raw', 'hard_fault', 'detrend', 'micro_interp', 'bandpass', 'fwr', 'screened', 'filled', 'smooth', 'feature' } @@ -66,8 +66,9 @@ def test_make_paths_custom_root_raw(mock_filesystem): assert paths['raw'] == "/abs/my_raw" # Assert other folders should be based on custom root - assert paths['detrend'] == "/abs/my_root/2_detrend" - assert paths['bandpass'] == "/abs/my_root/3_bandpass" + assert paths['detrend'] == "/abs/my_root/3_detrend" + assert paths['micro_interp'] == "/abs/my_root/4_micro_interp" + assert paths['bandpass'] == "/abs/my_root/5_bandpass" # Assert makedirs was called for every path assert set(mock_filesystem) == set(paths.values())