Preprocessing API

Preprocessing processors for preparing EEG data before correction.

Filtering

Filter

class facet.preprocessing.Filter(l_freq: float | None = None, h_freq: float | None = None, picks: str | list[str] | None = None, filter_length: str = 'auto', method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming', verbose: bool = False)[source]

Bases: Processor

Apply MNE’s generic band/high/low-pass filter to EEG data.

Wraps mne.io.Raw.filter() with configurable FIR/IIR parameters. If a noise estimate is present in the context, the same filter is applied to the noise so that downstream evaluation steps remain consistent.

Parameters:

l_freq (float, optional) – Low cutoff frequency in Hz (None for no highpass).
h_freq (float, optional) – High cutoff frequency in Hz (None for no lowpass).
picks (str or list of str, optional) – Channels to filter. None filters all channels.
filter_length (str, optional) – Length of the FIR filter (default: 'auto').
method (str, optional) – Filter method, 'fir' or 'iir' (default: 'fir').
phase (str, optional) – Phase of the filter (default: 'zero').
fir_window (str, optional) – Window function for FIR filter (default: 'hamming').
verbose (bool, optional) – MNE verbosity flag passed to raw.filter() (default: False).

name: str = 'filter'

description: str = 'Apply generic filter to EEG data'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

channel_wise: bool = True

__init__(l_freq: float | None = None, h_freq: float | None = None, picks: str | list[str] | None = None, filter_length: str = 'auto', method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming', verbose: bool = False)[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

HighPassFilter

class facet.preprocessing.HighPassFilter(freq: float, picks: str | list[str] | None = None, filter_length: str = 'auto', method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming')[source]

Bases: Filter

Apply a highpass filter to EEG data.

Convenience subclass of Filter that exposes a single freq parameter instead of separate l_freq/h_freq.

Parameters:

freq (float) – Highpass cutoff frequency in Hz.
picks (str or list of str, optional) – Channels to filter. None filters all channels.
filter_length (str, optional) – Length of the FIR filter (default: 'auto').
method (str, optional) – Filter method, 'fir' or 'iir' (default: 'fir').
phase (str, optional) – Phase of the filter (default: 'zero').
fir_window (str, optional) – Window function for FIR filter (default: 'hamming').

name: str = 'highpass_filter'

description: str = 'Apply highpass filter to EEG data'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

__init__(freq: float, picks: str | list[str] | None = None, filter_length: str = 'auto', method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming')[source]: Initialize processor.

LowPassFilter

class facet.preprocessing.LowPassFilter(freq: float, picks: str | list[str] | None = None, filter_length: str = 'auto', method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming')[source]

Bases: Filter

Apply a lowpass filter to EEG data.

Convenience subclass of Filter that exposes a single freq parameter instead of separate l_freq/h_freq.

Parameters:

freq (float) – Lowpass cutoff frequency in Hz.
picks (str or list of str, optional) – Channels to filter. None filters all channels.
filter_length (str, optional) – Length of the FIR filter (default: 'auto').
method (str, optional) – Filter method, 'fir' or 'iir' (default: 'fir').
phase (str, optional) – Phase of the filter (default: 'zero').
fir_window (str, optional) – Window function for FIR filter (default: 'hamming').

name: str = 'lowpass_filter'

description: str = 'Apply lowpass filter to EEG data'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

__init__(freq: float, picks: str | list[str] | None = None, filter_length: str = 'auto', method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming')[source]: Initialize processor.

BandPassFilter

class facet.preprocessing.BandPassFilter(l_freq: float, h_freq: float, picks: str | list[str] | None = None, filter_length: str = 'auto', method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming')[source]

Bases: Filter

Apply a bandpass filter to EEG data.

Convenience subclass of Filter that requires both l_freq and h_freq to be specified.

Parameters:

l_freq (float) – Low cutoff frequency in Hz.
h_freq (float) – High cutoff frequency in Hz.
picks (str or list of str, optional) – Channels to filter. None filters all channels.
filter_length (str, optional) – Length of the FIR filter (default: 'auto').
method (str, optional) – Filter method, 'fir' or 'iir' (default: 'fir').
phase (str, optional) – Phase of the filter (default: 'zero').
fir_window (str, optional) – Window function for FIR filter (default: 'hamming').

name: str = 'bandpass_filter'

description: str = 'Apply bandpass filter to EEG data'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

__init__(l_freq: float, h_freq: float, picks: str | list[str] | None = None, filter_length: str = 'auto', method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming')[source]: Initialize processor.

NotchFilter

class facet.preprocessing.NotchFilter(freqs: float | list[float], picks: str | list[str] | None = None, filter_length: str = 'auto', notch_widths: float | list[float] | None = None, method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming')[source]

Bases: Processor

Apply a notch filter to remove line noise from EEG data.

Wraps mne.io.Raw.notch_filter() with configurable FIR/IIR parameters. Typical use is to remove 50 Hz or 60 Hz mains interference and their harmonics. If a noise estimate is present in the context, the same filter is applied to keep evaluation results consistent.

Parameters:

freqs (float or list of float) – Frequencies to notch out in Hz (e.g., [50, 100]).
picks (str or list of str, optional) – Channels to filter. None filters all channels.
filter_length (str, optional) – Length of the FIR filter (default: 'auto').
notch_widths (float or list of float, optional) – Width of each notch filter. None uses MNE defaults.
method (str, optional) – Filter method, 'fir' or 'iir' (default: 'fir').
phase (str, optional) – Phase of the filter (default: 'zero').
fir_window (str, optional) – Window function for FIR filter (default: 'hamming').

name: str = 'notch_filter'

description: str = 'Apply notch filter to remove line noise'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

channel_wise: bool = True

__init__(freqs: float | list[float], picks: str | list[str] | None = None, filter_length: str = 'auto', notch_widths: float | list[float] | None = None, method: str = 'fir', phase: str = 'zero', fir_window: str = 'hamming')[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

MATLABPreFilter

Bases: Processor

Apply MATLAB FACET-style FFT prefiltering before artifact correction.

Supports: - Piecewise-linear transfer-function filtering in frequency domain. - Gaussian high-pass filtering (DC-safe) in frequency domain. - Optional padded filtering inside the acquisition window.

Parameters:

lp_frequency (float or collections.abc.Sequence of float, optional) – Low-pass cutoff in Hz (scalar or per-channel values). None means no LP from generated transfer functions.
hp_frequency (float or collections.abc.Sequence of float, optional) – High-pass cutoff in Hz (scalar or per-channel values). None means no HP from generated transfer functions.
transfer_frequencies (numpy.ndarray or collections.abc.Sequence of numpy.ndarray, optional) – Custom normalized frequency points in [0, 1] for transfer-function filtering. If set, overrides LP/HP synthesis.
transfer_amplitudes (numpy.ndarray or collections.abc.Sequence of numpy.ndarray, optional) – Amplitude points corresponding to transfer_frequencies.
gauss_hp_frequency (float or collections.abc.Sequence of float, optional) – Gaussian high-pass cutoff in Hz (scalar or per-channel values).
picks (str or collections.abc.Sequence of str) – Channels to filter (default: "eeg").
pad_acquisition (bool) – Use acquisition-window padding during filtering (default: True).

name: str = 'matlab_prefilter'

description: str = 'Apply MATLAB FACET-style FFT/gaussian prefilter'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

channel_wise: bool = True

__init__(lp_frequency: float | Sequence[float] | None = None, hp_frequency: float | Sequence[float] | None = None, transfer_frequencies: np.ndarray | Sequence[np.ndarray] | None = None, transfer_amplitudes: np.ndarray | Sequence[np.ndarray] | None = None, gauss_hp_frequency: float | Sequence[float] | None = None, picks: str | Sequence[str] = 'eeg', pad_acquisition: bool = True) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

Resampling

Resample

class facet.preprocessing.Resample(sfreq: float, npad: str = 'auto', window: str = 'boxcar', n_jobs: int = 1, verbose: bool = False)[source]

Bases: Processor

Resample EEG data to a fixed target sampling frequency.

Wraps mne.io.Raw.resample(). Trigger positions are scaled proportionally after resampling. If a noise estimate is present in the context, it is resampled with identical parameters so that downstream evaluation steps remain consistent.

Parameters:

sfreq (float) – Target sampling frequency in Hz.
npad (str, optional) – Amount to pad the start and end of the data (default: 'auto').
window (str, optional) – Window to use for resampling (default: 'boxcar').
n_jobs (int, optional) – Number of parallel jobs for resampling (default: 1).
verbose (bool, optional) – MNE verbosity flag passed to raw.resample() (default: False).

name: str = 'resample'

description: str = 'Resample EEG data to a new sampling frequency'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

channel_wise: bool = True

__init__(sfreq: float, npad: str = 'auto', window: str = 'boxcar', n_jobs: int = 1, verbose: bool = False)[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

UpSample

class facet.preprocessing.UpSample(factor: int = 10, npad: str = 'auto', window: str = 'boxcar', n_jobs: int = 1, verbose: bool = False)[source]

Bases: Resample

Upsample EEG data by a fixed integer factor.

Increases the sampling frequency by multiplying it by factor. Trigger positions and any noise estimate are scaled accordingly.

Parameters:

factor (int, optional) – Upsampling factor, e.g. 10 increases the sampling frequency ten-fold (default: 10).
npad (str, optional) – Amount to pad the start and end of the data (default: 'auto').
window (str, optional) – Window to use for resampling (default: 'boxcar').
n_jobs (int, optional) – Number of parallel jobs for resampling (default: 1).
verbose (bool, optional) – MNE verbosity flag passed to raw.resample() (default: False).

name: str = 'upsample'

description: str = 'Upsample EEG data by a factor'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

__init__(factor: int = 10, npad: str = 'auto', window: str = 'boxcar', n_jobs: int = 1, verbose: bool = False)[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

DownSample

class facet.preprocessing.DownSample(factor: int = 10, npad: str = 'auto', window: str = 'boxcar', n_jobs: int = 1, verbose: bool = False)[source]

Bases: Resample

Downsample EEG data by a fixed integer factor.

Decreases the sampling frequency by dividing it by factor. Trigger positions and any noise estimate are scaled accordingly.

Parameters:

factor (int, optional) – Downsampling factor, e.g. 10 reduces the sampling frequency ten-fold (default: 10).
npad (str, optional) – Amount to pad the start and end of the data (default: 'auto').
window (str, optional) – Window to use for resampling (default: 'boxcar').
n_jobs (int, optional) – Number of parallel jobs for resampling (default: 1).
verbose (bool, optional) – MNE verbosity flag passed to raw.resample() (default: False).

name: str = 'downsample'

description: str = 'Downsample EEG data by a factor'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

__init__(factor: int = 10, npad: str = 'auto', window: str = 'boxcar', n_jobs: int = 1, verbose: bool = False)[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]: Check that the resulting sampling frequency is at least 1 Hz.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

Trigger Detection

TriggerDetector

class facet.preprocessing.TriggerDetector(regex: str, save_to_annotations: bool = False)[source]

Bases: Processor

Detect triggers from annotations or stim channels using a regex pattern.

Searches the raw data for events whose description (annotation) or integer value (stim channel) matches the supplied regular expression. Detected trigger sample positions are stored in context.metadata.triggers.

The artifact length is estimated from the median inter-trigger interval; volume-level gaps in slice-triggered acquisitions are detected automatically.

Parameters:

regex (str) – Regular expression pattern to match trigger values.
save_to_annotations (bool, optional) – If True, write detected triggers back to the raw annotations (default: False).

name: str = 'trigger_detector'

description: str = 'Detect triggers using regex pattern'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = False

__init__(regex: str, save_to_annotations: bool = False) → None[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

QRSTriggerDetector

class facet.preprocessing.QRSTriggerDetector(save_to_annotations: bool = False)[source]

Bases: Processor

Detect QRS complexes (heartbeats) for BCG artifact correction.

Uses the FMRIB QRS detector from facet.helpers.bcg_detector, which requires the neurokit2 package (available via pip install facetpy[all]).

The artifact length is set to half the median RR interval, centred on each detected R-peak.

Parameters:: save_to_annotations (bool, optional) – If True, write detected QRS peaks back to the raw annotations (default: False).

name: str = 'qrs_trigger_detector'

description: str = 'Detect QRS complexes for BCG correction'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = False

__init__(save_to_annotations: bool = False) → None[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

MissingTriggerDetector

class facet.preprocessing.MissingTriggerDetector(add_to_context: bool = True, correlation_threshold: float = 0.9, search_window_factor: float = 0.1, ref_channel: int = 0)[source]

Bases: Processor

Detect and insert missing triggers by template matching.

Scans the trigger sequence for gaps larger than 1.9× the artifact length and attempts to locate missing artifact epochs by cross-correlating a reference template against the signal. Optionally extends the search one step before the first trigger and one step after the last.

Parameters:

add_to_context (bool, optional) – If True, insert found triggers into metadata and annotations (default: True).
correlation_threshold (float, optional) – Minimum absolute Pearson correlation to accept a candidate trigger (default: 0.9).
search_window_factor (float, optional) – Search window as a fraction of artifact length (default: 0.1).
ref_channel (int, optional) – Reference channel index for template matching (default: 0).

name: str = 'missing_trigger_detector'

description: str = 'Detect and add missing triggers'

version: str = '1.0.0'

requires_triggers: bool = True

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = False

__init__(add_to_context: bool = True, correlation_threshold: float = 0.9, search_window_factor: float = 0.1, ref_channel: int = 0) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

MissingTriggerCompleter

class facet.preprocessing.MissingTriggerCompleter(volumes: int, slices: int, add_to_context: bool = True, add_annotations: bool = True, strict: bool = True)[source]

Bases: Processor

Deterministically complete missing triggers using volume/slice counts.

MATLAB FACET parity processor for FindMissingTriggers(volumes, slices) based on the CompleteTriggers reconstruction heuristic.

Parameters:

volumes (int) – Total number of fMRI volumes expected.
slices (int) – Number of slices per volume.
add_to_context (bool, optional) – If True, replace metadata.triggers with the completed list (default: True).
add_annotations (bool, optional) – If True and raw is available, add annotations for inserted triggers (default: True).
strict (bool, optional) – If True, mismatches raise ProcessorValidationError. If False, best-effort completion is used (default: True).

name: str = 'missing_trigger_completer'

description: str = 'Deterministically complete missing triggers from volume/slice counts'

version: str = '1.0.0'

requires_triggers: bool = True

requires_raw: bool = False

modifies_raw: bool = False

parallel_safe: bool = False

__init__(volumes: int, slices: int, add_to_context: bool = True, add_annotations: bool = True, strict: bool = True) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

SliceTriggerGenerator

class facet.preprocessing.SliceTriggerGenerator(slices: int, duration_samples: float | None = None, relative_position: float = 0.03, add_annotations: bool = False)[source]

Bases: Processor

Generate slice triggers from volume triggers.

MATLAB FACET parity processor for GenerateSliceTriggers.

Parameters:

slices (int) – Number of slices per volume.
duration_samples (float, optional) – Slice duration in samples. If None, derived from trigger spacing.
relative_position (float) – Relative position of first slice trigger with respect to the volume trigger (default: 0.03).
add_annotations (bool, optional) – If True and raw is available, generated triggers are added as annotations (default: False).

name: str = 'slice_trigger_generator'

description: str = 'Generate slice triggers from volume triggers'

version: str = '1.0.0'

requires_triggers: bool = True

requires_raw: bool = False

modifies_raw: bool = False

parallel_safe: bool = False

__init__(slices: int, duration_samples: float | None = None, relative_position: float = 0.03, add_annotations: bool = False) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

TriggerExplorer

class facet.preprocessing.TriggerExplorer(mode: str = 'gui', auto_select: str | None = None, save_to_annotations: bool = False)[source]

Bases: Processor

Interactively explore and select trigger events from annotations or STIM channels.

Scans the loaded data for all available event sources (MNE annotations and STIM channels) and lets the user pick the correct trigger source. Three interaction modes are supported:

"gui" (default): Opens a matplotlib window with a downsampled preview of the first EEG channel. Radio buttons list every discovered event type; selecting one highlights the corresponding trigger positions on the waveform. A Confirm button finalises the choice. Falls back to "terminal" automatically when no GUI backend is available.
"terminal": Prints a Rich table and prompts for a selection number in the terminal.
"auto": Alias for "gui" — kept for backward compatibility.

When auto_select is provided, all interactive prompts are skipped and the first event matching the regex is chosen automatically — useful for scripted / non-interactive pipelines.

Parameters:

mode (str, optional) – Interaction mode: "gui" (default), "terminal", or "auto".
auto_select (str or None, optional) – If given, automatically select the event whose description matches this regex, bypassing any interactive prompt (default: None).
save_to_annotations (bool, optional) – If True, write detected triggers back to the raw annotations (default: False).

name: str = 'trigger_explorer'

description: str = 'Interactively explore and select trigger events'

version: str = '1.1.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = False

__init__(mode: str = 'gui', auto_select: str | None = None, save_to_annotations: bool = False) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

InteractiveTriggerExplorer

facet.preprocessing.InteractiveTriggerExplorer: alias of TriggerExplorer

Alignment

TriggerAligner

class facet.preprocessing.TriggerAligner(ref_trigger_index: int = 0, ref_channel: int | None = None, search_window: int | None = None, save_to_annotations: bool = False, upsample_for_alignment: bool = True)[source]

Bases: Processor

Align triggers to artifact positions using cross-correlation.

Refines trigger positions by finding the lag that maximises cross- correlation with a reference artifact epoch. The search can optionally operate on temporarily upsampled data for sub-sample precision.

Parameters:

ref_trigger_index (int, optional) – Index of the reference trigger to use as template (default: 0).
ref_channel (int, optional) – Reference channel index. Uses the first EEG channel when None (default: None).
search_window (int, optional) – Search window in samples. Derived from the upsampling factor when None (default: None).
save_to_annotations (bool, optional) – If True, save aligned triggers as raw annotations (default: False).
upsample_for_alignment (bool, optional) – Temporarily upsample data before alignment for sub-sample accuracy (default: True).

name: str = 'trigger_aligner'

description: str = 'Align triggers using cross-correlation'

version: str = '1.0.0'

requires_triggers: bool = True

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = True

channel_wise: bool = True

run_once: bool = True

__init__(ref_trigger_index: int = 0, ref_channel: int | None = None, search_window: int | None = None, save_to_annotations: bool = False, upsample_for_alignment: bool = True) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

SliceAligner

class facet.preprocessing.SliceAligner(ref_trigger_index: int = 0, ref_channel: int | None = None, search_window: int | None = None, save_to_annotations: bool = False)[source]

Bases: TriggerAligner

Align slice triggers on already-upsampled data.

Identical to TriggerAligner but skips the internal temporary upsampling step, assuming the data has already been upsampled upstream.

Parameters:

ref_trigger_index (int, optional) – Index of the reference trigger (default: 0).
ref_channel (int, optional) – Reference channel index (default: None).
search_window (int, optional) – Search window in samples (default: None).
save_to_annotations (bool, optional) – Save aligned triggers as annotations (default: False).

name: str = 'slice_aligner'

description: str = 'Align slice triggers on upsampled data'

version: str = '1.0.0'

requires_triggers: bool = True

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = True

channel_wise: bool = True

run_once: bool = True

__init__(ref_trigger_index: int = 0, ref_channel: int | None = None, search_window: int | None = None, save_to_annotations: bool = False) → None[source]: Initialize processor.

SubsampleAligner

class facet.preprocessing.SubsampleAligner(ref_trigger_index: int = 0, ref_channel: int | None = None, search_window: int | None = None, apply_to_raw: bool = False)[source]

Bases: Processor

Refine trigger positions at subsample precision after upsampling.

Mirrors the intent of the MATLAB RAAlignSubSample step. For each trigger a search segment is extracted, cross-correlated against a reference epoch, and the trigger is shifted by the lag that maximises the correlation.

When apply_to_raw=True the corresponding raw data segments are also rolled by the computed shifts; otherwise only the trigger positions in metadata are updated.

Parameters:

ref_trigger_index (int, optional) – Index of the reference trigger used as the alignment template (default: 0).
ref_channel (int, optional) – Reference channel index. Uses the first EEG channel when None (default: None).
search_window (int, optional) – Search radius in samples. Defaults to twice the upsampling factor.
apply_to_raw (bool, optional) – If True, roll raw data segments by the computed shifts (default: False).

name: str = 'subsample_aligner'

description: str = 'Refine trigger alignment at subsample precision'

version: str = '1.0.0'

requires_triggers: bool = True

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = True

channel_wise: bool = True

run_once: bool = True

__init__(ref_trigger_index: int = 0, ref_channel: int | None = None, search_window: int | None = None, apply_to_raw: bool = False) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

Transforms

CutAcquisitionWindow

class facet.preprocessing.CutAcquisitionWindow(pre_padding_samples: int | None = None, post_padding_samples: int | None = None)[source]

Bases: Processor

Derive acquisition window bounds similarly to MATLAB’s RACut step.

Records the acquisition start/end sample indices and the per-trigger pre/post sample counts in metadata so that downstream processors can operate on consistent artifact windows without accessing raw signal data directly.

Parameters:

pre_padding_samples (int, optional) – Explicit number of samples before each trigger. When None the value is derived from metadata.artifact_to_trigger_offset.
post_padding_samples (int, optional) – Explicit number of samples after each trigger. When None the remaining artifact window is used.

name: str = 'cut_acquisition_window'

description: str = 'Record acquisition window boundaries for artifact processing'

version: str = '1.0.0'

requires_triggers: bool = True

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = False

__init__(pre_padding_samples: int | None = None, post_padding_samples: int | None = None) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

PasteAcquisitionWindow

class facet.preprocessing.PasteAcquisitionWindow[source]

Bases: Processor

Finalize acquisition metadata and clear cached window segments.

FACETpy keeps the full-length raw data throughout the pipeline, so this processor ensures acquisition metadata is present and clears any cached segments set by CutAcquisitionWindow.

name: str = 'paste_acquisition_window'

description: str = 'Finalize acquisition metadata and clear cached window segments'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = False

modifies_raw: bool = False

parallel_safe: bool = False

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

Crop

class facet.preprocessing.Crop(tmin: float | None = None, tmax: float | None = None)[source]

Bases: Processor

Crop the Raw recording to a time interval.

A concise alternative to LambdaProcessor for the common pattern of restricting the recording to a specific window before processing.

Parameters:

tmin (float, optional) – Start time in seconds. None keeps the original start.
tmax (float, optional) –
End time in seconds. None keeps the original end.

If both tmin and tmax are None, an interactive Matplotlib selector is opened to choose the crop window.

Examples

Crop(tmin=0, tmax=162)
Crop()  # open interactive selector

name: str = 'crop'

description: str = 'Crop Raw recording to a time interval'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = False

__init__(tmin: float | None = None, tmax: float | None = None)[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

MagicErasor

class facet.preprocessing.MagicErasor(picks: str | list[str] = 'eeg', channel: str | int | None = None, default_mode: str = 'zero', random_seed: int | None = None)[source]

Bases: Processor

Interactively erase selected signal segments with configurable modes.

Opens an interactive matplotlib editor for one preview channel and lets the user select time segments repeatedly. Each selected segment can be replaced using one of four modes:

zero: set samples to zero.
mean: set samples to the channel mean.
interpolate: linearly interpolate between segment boundaries.
generated_eeg: replace with synthetic EEG generated through EEGGenerator, then adapt channel-wise mean and amplitude to the local surrounding signal.

The editor stays open until the user clicks Done, enabling multiple edits in a single session.

Parameters:

picks (str or list[str], optional) – Channels to edit (default: "eeg").
channel (str or int, optional) – Channel used for preview in the interactive window. When None, the first edited EEG channel is used.
default_mode (str, optional) – Initially selected editing mode (default: "zero").
random_seed (int, optional) – Optional seed used when generated_eeg mode is applied.

name: str = 'magic_erasor'

description: str = 'Interactively erase selected segments with multiple replacement modes'

version: str = '1.1.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = False

channel_wise: bool = False

__init__(picks: str | list[str] = 'eeg', channel: str | int | None = None, default_mode: str = 'zero', random_seed: int | None = None) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

RawTransform

class facet.preprocessing.RawTransform(name: str, func: Callable)[source]

Bases: Processor

Apply an arbitrary callable to the Raw object.

A lighter-weight alternative to LambdaProcessor when only the Raw object needs to be modified. The callable receives the current Raw object and must return a new (or modified copy of a) Raw object.

Parameters:

name (str) – Human-readable label shown in pipeline logs and progress.
func (collections.abc.Callable) – Callable[[mne.io.Raw], mne.io.Raw] — receives the current Raw object, must return a (possibly new) Raw object.

Examples

# Drop bad channels inline
RawTransform("drop_bad", lambda raw: raw.copy().pick_channels(
    [ch for ch in raw.ch_names if ch not in ["EKG", "EMG"]]
))

# Set average reference
RawTransform("set_eeg_ref", lambda raw: raw.copy().set_eeg_reference("average"))

description: str = 'Apply a callable transform to the Raw object'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = False

__init__(name: str, func: Callable)[source]: Initialize processor.

name: str = 'raw_transform'

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

PickChannels

class facet.preprocessing.PickChannels(picks: str | list[str], on_missing: str = 'ignore')[source]

Bases: Processor

Keep only the specified channels or channel types.

A named, reusable alternative to the common lambda ctx: ctx.with_raw( ctx.get_raw().copy().pick(...)) pattern.

Parameters:

picks (str or list of str) – Channel type ("eeg", "stim", …) or list of channel names / types accepted by mne.io.Raw.pick().
on_missing (str, optional) – Passed to MNE. "ignore" (default) silently skips channels that are absent from the recording.

Examples

# Keep only EEG and stimulus channels
PickChannels(picks=["eeg", "stim"])

# Keep specific channels by name
PickChannels(picks=["Fp1", "Fp2", "Fz"])

name: str = 'pick_channels'

description: str = 'Keep only the specified channels or channel types'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

__init__(picks: str | list[str], on_missing: str = 'ignore')[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

DropChannels

class facet.preprocessing.DropChannels(channels: list[str], on_missing: str = 'ignore')[source]

Bases: Processor

Remove named channels from the recording.

A named, reusable alternative to the lambda ctx: ...drop_channels(...) pattern commonly seen in inline pipeline steps.

Parameters:

channels (list of str) – List of channel names to remove.
on_missing (str, optional) – "ignore" (default) skips absent names silently; "raise" raises an error when a channel is not found.

Examples

# Drop typical non-EEG channels that may be present in EDF files
DropChannels(channels=["EKG", "EMG", "EOG", "ECG"])

name: str = 'drop_channels'

description: str = 'Remove named channels from the recording'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

__init__(channels: list[str], on_missing: str = 'ignore')[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

ChannelStandardizer

class facet.preprocessing.ChannelStandardizer(standard: str | list[str], on_missing: str = 'ignore', keep_auxiliary: bool = True, rename_aliases: bool = True)[source]

Bases: Processor

Convert EEG channel layouts to predefined standards or custom subsets.

Keeps EEG channels that match the requested standard in a deterministic order, optionally preserving non-EEG channels (e.g., trigger/stim channels). Legacy aliases such as T3 -> T7 and T5 -> P7 are resolved automatically. When alias channels are used, they can be renamed to the target standard labels.

Parameters:

standard (str or list[str]) – Target standard identifier or explicit ordered channel list. Supported built-in identifiers: - "10-20" / "standard_1020_19" - "32" / "standard_1020_32" For other strings, the processor tries to load an MNE standard montage via mne.channels.make_standard_montage() and uses its channel list.
on_missing (str, optional) – Missing-channel behavior: "ignore" (default) keeps available matches only, "raise" fails when any requested channel is missing.
keep_auxiliary (bool, optional) – If True (default), append non-EEG channels after the selected EEG subset.
rename_aliases (bool, optional) – If True (default), rename selected alias channels to the requested target labels when possible (e.g., T3 becomes T7).

name: str = 'channel_standardizer'

description: str = 'Convert EEG channels to standard subsets with optional alias renaming'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = True

parallel_safe: bool = True

channel_wise: bool = False

__init__(standard: str | list[str], on_missing: str = 'ignore', keep_auxiliary: bool = True, rename_aliases: bool = True) → None[source]: Initialize processor.

validate(context: ProcessingContext) → None[source]

Validate that prerequisites are met.

Override this method to add custom validation logic.

Parameters:: context – Processing context
Raises:: ProcessorValidationError – If validation fails

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

PrintMetric

class facet.preprocessing.PrintMetric(*metric_names: str, label: str | None = None)[source]

Bases: Processor

Print one or more evaluation metric values — useful for debugging pipelines.

Inserts a transparent logging step that reads from the shared metrics dict populated by evaluation processors (e.g. SNRCalculator). The context is returned unchanged.

Parameters:

*metric_names (str) – One or more metric names to print (e.g. 'snr', 'rms_ratio').
label (str, optional) – Optional prefix shown in brackets, e.g. "after PCA".

Examples

pipeline = Pipeline([
    ...,
    SNRCalculator(),
    PrintMetric("snr"),          # → "  snr=12.345"
    PCACorrection(...),
    SNRCalculator(),
    PrintMetric("snr", label="after PCA"),   # → "  [after PCA] snr=14.201"
])

name: str = 'print_metric'

description: str = 'Print evaluation metric values for debugging'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = False

modifies_raw: bool = False

parallel_safe: bool = False

__init__(*metric_names: str, label: str | None = None)[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

Diagnostics

AnalyzeDataReport

class facet.preprocessing.AnalyzeDataReport[source]

Bases: Processor

Generate a FACETpy-style data analysis summary.

This mirrors the purpose of MATLAB FACET AnalyzeData while keeping output in FACETpy’s structured metadata + concise logger format.

The generated report is stored in metadata.custom["analyze_data_report"].

name: str = 'analyze_data_report'

description: str = 'Create structured recording and trigger summary report'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = True

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).

CheckDataReport

class facet.preprocessing.CheckDataReport(require_triggers: bool = True, strict: bool = True)[source]

Bases: Processor

Run FACETpy-style data checks and emit a structured report.

This mirrors MATLAB FACET CheckData intent and stores results in metadata.custom["check_data_report"].

Parameters:

require_triggers (bool, optional) – If True, missing triggers are treated as an error (default: True).
strict (bool, optional) – If True, raise ProcessorValidationError when checks fail (default: True).

name: str = 'check_data_report'

description: str = 'Run trigger/data sanity checks and store report'

version: str = '1.0.0'

requires_triggers: bool = False

requires_raw: bool = True

modifies_raw: bool = False

parallel_safe: bool = True

__init__(require_triggers: bool = True, strict: bool = True) → None[source]: Initialize processor.

process(context: ProcessingContext) → ProcessingContext[source]

Process the context.

This is the main method to implement in subclasses.

Parameters:: context – Input processing context
Returns:: Output processing context. If None is returned, the input context is used (no-op behavior).