Preprocessing (pyavs.preprocessing)¶

read_meg_sessions(preload: bool = True, compute_missing_prepro: bool = False) → dict[source]¶

Reads the MEG data of a given subject and session per block.

Parameters:

preload (bool, optional) – Whether to preload the data. Defaults to True.
compute_missing_prepro (bool, optional) – Whether to compute missing preprocessing steps. Defaults to False.

Returns:

A dictionary containing the raw MEG data for each block.

Return type:

load_meg_data(compute_missing_prepro: bool = True, preprocessed: bool = True)[source]¶

Loads the preprocessed data of a given subject and session per block. If the preprocessed data is not available, it will be computed.

Parameters:

compute_missing_prepro (bool, optional) – Flag indicating whether to compute missing preprocessed data. Defaults to True.
preprocessed (bool, optional) – Flag indicating whether to load preprocessed data. Defaults to True.

filter_meg_data(l_freq: float | None = None, h_freq: float | None = None, picks=None, causal: bool | None = None, concatenated: bool | None = False, ignore_existing_filter: bool = False)[source]¶

Applies lowpass and/or highpass filters to the MEG data using pyAVS filter_meg function.

Parameters:

l_freq (float, optional) – The lower frequency cutoff for the filter. If None, uses instance variable.
h_freq (float, optional) – The higher frequency cutoff for the filter. If None, uses instance variable.
picks (list or None, optional) – The indices of the channels to filter. If None, all channels are filtered.
causal (bool, optional) – Whether to use a causal filter. If None, uses instance variable.
concatenated (bool or None, optional) – Whether to use the concatenated data for filtering. If None, it uses the concatenated data if available, otherwise it uses the data per block.
ignore_existing_filter (bool, optional) – If True, ignore existing filters and apply new ones anyway. If False (default), raises an error if data is already filtered with different parameters.

Notes

This method applies filtering to the loaded MEG data. Since preprocess_meg_block no longer applies filtering by default, this method is the primary way to filter data in the workflow.

resample_meg_data(target_sfreq: float | None = None)[source]¶

Resamples the MEG data to the target sampling frequency using pyAVS resample_meg function.

Parameters:: target_sfreq (float, optional) – The target sampling frequency in Hz. If None, uses instance variable.

Notes

This method resamples the loaded MEG data. Note that preprocess_meg_block applies resampling by default, so check the current sampling rate before applying additional resampling.

concatenate_raws_per_session()[source]¶: Concatenates the raws for each block into one raw object per session.

find_events_in_raw()[source]¶: Finds the events in the raw data.

make_trigger_locked_epochs(trigger_name: str = 'mic_on', tmin: float = -0.5, tmax: float = 9.0, baseline: Tuple[float, float] | None = None, preload: bool = True) → mne.Epochs[source]¶

Create epochs locked to a specific MEG trigger (e.g. mic_on, caption_on).

Finds all occurrences of the named trigger in the repaired events array, maps each to its trial via preceding trial-number/block triggers, joins with the experiment log to attach metadata (sceneID, caption_task, etc.), and returns an MNE Epochs object with metadata set.

Parameters:

trigger_name (str, optional) – Name of the trigger as defined in get_meg_trigger_dict(). Valid options: ‘mic_on’, ‘mic_off’, ‘caption_on’, ‘caption_off’, ‘scene_on’, ‘scene_off’, ‘fixcross_on’, ‘fixcross_off’. Default: ‘mic_on’.
tmin (float, optional) – Start of epoch relative to trigger onset in seconds (default: -0.2).
tmax (float, optional) – End of epoch relative to trigger onset in seconds (default: 1.0).
baseline (tuple of (float, float) or None, optional) – Baseline correction window. Default: None (no baseline correction).
preload (bool, optional) – Whether to preload epoch data into memory (default: True).

Returns:

Epochs locked to the specified trigger, with trial metadata attached.

Return type:

mne.Epochs

Notes

Requires that load_meg_data(), concatenate_raws_per_session(), and find_events_in_raw() have been called first. The experiment log is loaded on the fly if not already available.

get_et_annotations(event_type: str = 'fixation', recording: str = 'scene', exclude_last_fixation: bool = True, get_object_labels: bool = False, add_cross_event_info: bool = True, preprocessed: bool = True, save_annotated_raw: bool = False, onset_offset: str = 'onset')[source]¶

Extracts annotations from the eye tracker data.

Parameters:

et_event_type (str, optional) – Event type to extract from the eye tracking data. Defaults to “fixation”. Valid options: “fixation”, “saccade”, “blink”.
recording (str, optional) – Recording context to filter events by. Defaults to “scene”. Valid options: [“scene”, “caption”, “microphone”].
exclude_last_fixation (bool, optional) – Whether to exclude the last fixation event on each scene. Defaults to True.
get_object_labels (bool, optional) – Whether to get object labels for the eye tracking events. Defaults to False.
add_cross_event_info (bool, optional) – Whether to add cross event information to the eye tracking events. Defaults to True.
preprocessed (bool, optional) – Whether the eye tracking data is preprocessed. Defaults to True.
save_annotated_raw (bool, optional) – Whether to save the annotated raws. Defaults to False.
onset_offset (str, optional) – Whether to use event onset or offset timing. Defaults to “onset”. Valid options: [“onset”, “offset”]. When “offset”, time_in_trial is shifted by duration to align epochs to event offsets instead of onsets.

make_et_event_epochs(tmin: float, tmax: float, event_type: str, recording: str = 'scene', save_epochs: bool = True, get_metadata: bool = True, get_object_labels: bool = False, baseline: Tuple[float, float] | None = None)[source]¶

This function will make use of the et event annotations to make event epochs.

Parameters:

tmin (float) – The start of the epoch in seconds (around et event onset)
tmax (float) – The end of the epoch in seconds (around et event onset)
event_type (str) – The event type for which we want to make epochs. E.g. “fixation”, “saccade”, “blink” Valid options: “fixation”, “saccade”, “blink”, “scene”
recording (str, optional) – Recording context to filter events by. Defaults to “scene”. Valid options: [“scene”, “caption”, “microphone”].
save_epochs (bool, optional) – Whether to save the epochs to file. Defaults to True.
get_metadata (bool, optional) – Whether to add metadata to the epochs object. Defaults to True.
get_object_labels (bool, optional) – Whether to get the object labels for the fixations. Defaults to False.
baseline (tuple, optional) – The baseline period for the epochs (for AVS currently not recommended). Defaults to None.

add_et_metadata_to_epochs(metadata_colnames: List[str] | None = None)[source]¶

Adds eye tracking metadata to the epochs object.

Parameters:: metadata_colnames (list or None, optional) – List of column names to be added as metadata. If None, all column names available in the events dataframe will be used.

add_scene_metadata_to_epochs(metadata_colnames: List[str] | None = None)[source]¶

Adds trial-level metadata to scene-onset epochs.

Scene epochs are trial-level (one epoch per scene), not event-level, so this builds one metadata row per trial with scalar columns merged from the experiment log (same block/trial_per_block join used by make_trigger_locked_epochs), rather than treating trial-constant values (sceneID, condition, …) as single-element lists. The underlying fixation/saccade/blink sequence for each trial is still attached, as list-valued fixseq_* columns, for callers that want the full gaze sequence alongside the scene epoch.

Parameters:: metadata_colnames (list, optional) – Columns from self.et_events to include as fixseq_* list columns (one list per trial, in chronological within-trial order). If None, all non-identity columns are included.

get_data_summary() → Dict[str, Any][source]¶

Get summary of loaded and processed data.

Returns:: Summary information
Return type:: dict

apply_ica_to_blocks(use_precomputed: bool | None = None, compute_new_ica: bool = False, find_artifacts: bool = True) → None[source]¶

Apply ICA artifact removal to loaded MEG blocks.

This method applies ICA (either precomputed or newly computed) to remove artifacts from the loaded raw MEG data blocks. It operates on unconcatenated data for optimal artifact removal.

Parameters:

use_precomputed (bool, optional) – Whether to use precomputed ICA solutions. If None, uses instance variable.
compute_new_ica (bool, optional) – Whether to compute new ICA if precomputed not available (default: False)
find_artifacts (bool, optional) – Whether to automatically find artifact components when computing new ICA (default: True)

Notes

This method modifies self.raws_dict in place with ICA-cleaned data. ICA is applied before concatenation for optimal results. The usage of (precomputed) ica implies the interpolation of bad channels.

Eye Tracking Preprocessing¶

Eye tracking preprocessing for pyAVS package.

This module provides functions for preprocessing eye tracking data including event detection, artifact removal, and quality assessment.

pyavs.preprocessing.eye.preprocess_eye_events(events_df: DataFrame, remove_blinks: bool = True, remove_short_fixations: bool = False, min_fixation_duration: float = 0.1, remove_long_saccades: bool = False, max_saccade_duration: float = 0.1, remove_outlier_positions: bool = True, screen_resolution: Tuple[int, int] = (1024, 768), verbose: bool = False) → DataFrame[source]¶

Preprocess eye tracking events by removing artifacts and outliers.

Parameters:

events_df (pd.DataFrame) – Eye tracking events dataframe
remove_blinks (bool, optional) – Whether to remove blink events (default: True)
remove_short_fixations (bool, optional) – Whether to remove short fixations (default: False)
min_fixation_duration (float, optional) – Minimum fixation duration in seconds (default: 0.1)
remove_long_saccades (bool, optional) – Whether to remove long saccades (default: False)
max_saccade_duration (float, optional) – Maximum saccade duration in seconds (default: 0.1)
remove_outlier_positions (bool, optional) – Whether to remove events with outlier positions (default: True)
screen_resolution (tuple of int, optional) – Screen resolution (width, height) for outlier detection (default: (1024, 768))
verbose (bool, optional) – Whether to print preprocessing statistics (default: False)

Returns:

Preprocessed events dataframe

Return type:

pd.DataFrame

pyavs.preprocessing.eye.remove_artifacts(events_df: DataFrame, max_fixation_duration: float = 5.0, max_saccade_amplitude: float = 500.0, min_intersaccadic_interval: float = 0.02, verbose: bool = False) → DataFrame[source]¶

Remove artifacts from eye tracking events.

Parameters:

events_df (pd.DataFrame) – Eye tracking events dataframe
max_fixation_duration (float, optional) – Maximum allowed fixation duration in seconds (default: 5.0)
max_saccade_amplitude (float, optional) – Maximum allowed saccade amplitude in pixels (default: 500.0)
min_intersaccadic_interval (float, optional) – Minimum time between saccades in seconds (default: 0.02)
verbose (bool, optional) – Whether to print removal statistics (default: False)

Returns:

Events dataframe with artifacts removed

Return type:

pd.DataFrame

pyavs.preprocessing.eye.compute_eye_tracking_quality_metrics(events_df: DataFrame, recording_duration: float) → Dict[str, float][source]¶

Compute quality metrics for eye tracking data.

Parameters:

events_df (pd.DataFrame) – Eye tracking events dataframe
recording_duration (float) – Total recording duration in seconds

Returns:

Dictionary of quality metrics

Return type:

MEG Preprocessing¶

MEG preprocessing for pyAVS package.

This module provides functions for MEG data preprocessing including Maxwell filtering, bad channel detection, and signal filtering.

pyavs.preprocessing.meg.get_calibration_files(package_dir: str | None = None) → Dict[str, str][source]¶

Get paths to Maxwell filter calibration files.

Parameters:: package_dir (str, optional) – Path to package directory. If None, uses current package location
Returns:: Dictionary with ‘crosstalk’ and ‘fine_cal’ file paths
Return type:: dict

pyavs.preprocessing.meg.apply_maxwell_filter(raw: mne.io.Raw, crosstalk_file: str | None = None, fine_cal_file: str | None = None, find_bad_channels: bool = True, coord_frame: str = 'head', int_order: int = 8, ext_order: int = 3, origin: str | tuple = 'auto', regularize: str = 'in', ignore_ref: bool = True, bad_condition: str = 'error', head_pos: str | None = None, st_duration: float | None = None, st_correlation: float = 0.98, mag_scale: float = 100.0, skip_by_annotation: str | list = 'edge', extended_proj: list = [], verbose: bool = True) → mne.io.Raw[source]¶

Apply Maxwell filtering (tSSS) to MEG data.

Parameters:

raw (mne.io.Raw) – Raw MEG data
crosstalk_file (str, optional) – Path to crosstalk compensation file
fine_cal_file (str, optional) – Path to fine calibration file
find_bad_channels (bool, optional) – Whether to automatically detect bad channels (default: True)
coord_frame (str, optional) – Coordinate frame for Maxwell filtering (default: ‘head’)
int_order (int, optional) – Internal multipole order (default: 8)
ext_order (int, optional) – External multipole order (default: 3)
origin (str or tuple, optional) – Head origin (default: ‘auto’)
regularize (str, optional) – Regularization method (default: ‘in’)
ignore_ref (bool, optional) – Ignore reference channels (default: True)
bad_condition (str, optional) – How to handle bad condition (default: ‘error’)
head_pos (str, optional) – Path to head position file
st_duration (float, optional) – Signal space separation duration (default: None)
st_correlation (float, optional) – Correlation threshold for tSSS (default: 0.98)
mag_scale (float, optional) – Magnetometer scaling factor (default: 100.0)
skip_by_annotation (str or list, optional) – Annotations to skip (default: ‘edge’)
extended_proj (list, optional) – Extended projections (default: [])
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Maxwell filtered raw data

Return type:

pyavs.preprocessing.meg.filter_meg(raw: mne.io.Raw, l_freq: float | None = 0.2, h_freq: float | None = 200.0, picks: str | list | None = 'meg', filter_length: str = 'auto', l_trans_bandwidth: str = 'auto', h_trans_bandwidth: str = 'auto', n_jobs: int = 1, method: str = 'fir', iir_params: dict | None = None, phase: str = 'zero', fir_window: str = 'hamming', fir_design: str = 'firwin', skip_by_annotation: str | list = 'edge', pad: str = 'reflect_limited', causal: bool = False, verbose: bool = True) → mne.io.Raw[source]¶

Apply bandpass filtering to MEG data.

Parameters:

raw (mne.io.Raw) – Raw MEG data
l_freq (float, optional) – Low-pass frequency in Hz (default: 0.2)
h_freq (float, optional) – High-pass frequency in Hz (default: 100.0)
picks (str or list, optional) – Channels to filter (default: ‘meg’)
filter_length (str, optional) – Length of the FIR filter (default: ‘auto’)
l_trans_bandwidth (str, optional) – Low transition bandwidth (default: ‘auto’)
h_trans_bandwidth (str, optional) – High transition bandwidth (default: ‘auto’)
n_jobs (int, optional) – Number of parallel jobs (default: 1)
method (str, optional) – Filtering method (default: ‘fir’)
iir_params (dict, optional) – IIR filter parameters (default: None)
phase (str, optional) – Phase of the filter (default: ‘zero’, ‘zero-double’, ‘minimum’) For causal filtering, use ‘minimum’
fir_window (str, optional) – FIR window function (default: ‘hamming’)
fir_design (str, optional) – FIR design method (default: ‘firwin’)
skip_by_annotation (str or list, optional) – Annotations to skip (default: ‘edge’)
pad (str, optional) – Padding method (default: ‘reflect_limited’)
causal (bool, optional) – Whether to apply causal filtering (default: False) If True, sets phase=’minimum’ for causal response
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Filtered raw data

Return type:

Notes

Causal filtering introduces a phase delay but preserves temporal order, which can be important for real-time applications or when temporal relationships with other signals are critical. Non-causal (zero-phase) filtering provides better frequency response but is not suitable for real-time processing.

pyavs.preprocessing.meg.resample_meg(raw: mne.io.Raw, sfreq: float, npad: str = 'auto', window: str = 'boxcar', stim_picks: str | list | None = None, n_jobs: int = 1, events: ndarray | None = None, pad: str = 'reflect_limited', verbose: bool = True) → mne.io.Raw[source]¶

Resample MEG data to a new sampling frequency.

Parameters:

raw (mne.io.Raw) – Raw MEG data
sfreq (float) – New sampling frequency in Hz
npad (str, optional) – Padding for resampling (default: ‘auto’)
window (str, optional) – Window function for resampling (default: ‘boxcar’)
stim_picks (str or list, optional) – Stimulus channels to resample (default: None)
n_jobs (int, optional) – Number of parallel jobs (default: 1)
events (np.ndarray, optional) – Events array to resample (default: None)
pad (str, optional) – Padding method (default: ‘reflect_limited’)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Resampled raw data

Return type:

pyavs.preprocessing.meg.load_bad_channels(subject_id: int, session: int, block: int, bad_channels_file: str | None = None, include_heated: bool = False) → List[str][source]¶

Load bad channels from logbook for specific subject/session/block.

Parameters:

subject_id (int) – Subject ID
session (int) – Session number
block (int) – Block number
bad_channels_file (str, optional) – Path to bad channels CSV file. If None, uses package default
include_heated (bool, optional) – Whether to include heated channels as bad (default: False)

Returns:

List of bad channel names (with MEG prefix)

Return type:

list of str

pyavs.preprocessing.meg.interpolate_bad_channels(raw: mne.io.Raw, bad_channels: List[str] | None = None, reset_bads: bool = True, mode: str = 'accurate', origin: str | tuple = 'auto', method: Dict[str, str] = {'meg': 'MNE'}, exclude: list = [], verbose: bool = True) → mne.io.Raw[source]¶

Interpolate bad channels in MEG data.

Parameters:

raw (mne.io.Raw) – Raw MEG data
bad_channels (list of str, optional) – List of bad channel names. If None, uses channels in raw.info[‘bads’]
reset_bads (bool, optional) – Whether to reset bad channels list after interpolation (default: True)
mode (str, optional) – Interpolation mode (default: ‘accurate’)
origin (str or tuple, optional) – Head origin (default: ‘auto’)
method (dict, optional) – Interpolation method per channel type (default: {‘meg’: ‘MNE’})
exclude (list, optional) – Channels to exclude from interpolation (default: [])
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Raw data with interpolated channels

Return type:

pyavs.preprocessing.meg.preprocess_meg_block(raw: mne.io.Raw, subject_id: int, session: int, block: int, apply_maxwell: bool = True, apply_filtering: bool = False, apply_resampling: bool = True, interpolate_bads: bool = True, l_freq: float = 0.2, h_freq: float = 200.0, resample_freq: float = 500.0, causal_filter: bool = False, bad_channels_file: str | None = None, crosstalk_file: str | None = None, fine_cal_file: str | None = None, verbose: bool = True) → mne.io.Raw[source]¶

Complete preprocessing pipeline for a single MEG block.

Parameters:

raw (mne.io.Raw) – Raw MEG data
subject_id (int) – Subject ID
session (int) – Session number
block (int) – Block number
apply_maxwell (bool, optional) – Whether to apply Maxwell filtering (default: True)
apply_filtering (bool, optional) – Whether to apply bandpass filtering (default: False)
apply_resampling (bool, optional) – Whether to resample data (default: True)
interpolate_bads (bool, optional) – Whether to interpolate bad channels (default: True)
l_freq (float, optional) – Low-pass frequency in Hz (default: 0.2)
h_freq (float, optional) – High-pass frequency in Hz (default: 100.0)
resample_freq (float, optional) – Resampling frequency in Hz (default: 500.0)
causal_filter (bool, optional) – Whether to apply causal filtering (default: False) If True, uses minimum-phase filtering which preserves temporal order
bad_channels_file (str, optional) – Path to bad channels file
crosstalk_file (str, optional) – Path to crosstalk file
fine_cal_file (str, optional) – Path to fine calibration file
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Preprocessed raw data

Return type:

Notes

By default, this function applies Maxwell filtering, bad channel interpolation, and resampling, but NOT bandpass filtering. Filtering should be applied later using the AVS composer filter_meg_data() method to allow for flexible analysis-specific filter parameters.

pyavs.preprocessing.meg.prepare_empty_room_recording(raw_empty_room: mne.io.Raw, raw_reference: mne.io.Raw, bads: str = 'union', annotations: str = 'from_raw', meas_date: str = 'keep', verbose: bool = True) → mne.io.Raw[source]¶

Prepare empty room recording for Maxwell filtering.

Parameters:

raw_empty_room (mne.io.Raw) – Empty room recording
raw_reference (mne.io.Raw) – Reference recording from the same session
bads (str, optional) – How to handle bad channels (default: ‘union’)
annotations (str, optional) – How to handle annotations (default: ‘from_raw’)
meas_date (str, optional) – How to handle measurement date (default: ‘keep’)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Prepared empty room recording

Return type:

pyavs.preprocessing.meg.apply_precomputed_ica(raw: mne.io.Raw, subject_id: int, session: int, data_path: str | None = None, ica_solutions_dir: str | None = None, ica_exclusions_file: str | None = None, verbose: bool = True) → mne.io.Raw[source]¶

Apply precomputed ICA solution to MEG data.

This function loads a precomputed ICA solution and applies it to the MEG data, following standard MEG preprocessing methodology.

Parameters:

raw (mne.io.Raw) – MEG raw data
subject_id (int) – Subject ID
session (int) – Session number
data_path (str, optional) – Path to the avs-public data root. If None, uses configured data path. Ignored if ica_solutions_dir and ica_exclusions_file are given.
ica_solutions_dir (str, optional) – Directory of precomputed ICA solutions in the legacy {dir}/as01a/as01a-ica.fif layout, for a solution computed outside the derivatives/pyavs tree. Only used if both this and ica_exclusions_file are given; otherwise the shipped derivatives/pyavs ICA solution is used.
ica_exclusions_file (str, optional) – JSON file of component exclusions matching ica_solutions_dir.
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

MEG data with precomputed ICA applied

Return type:

Raises:

FileNotFoundError – If ICA solution file is not found
ValueError – If ICA solution is incompatible with data

ICA Artifact Removal¶

ICA (Independent Component Analysis) for pyAVS package.

This module provides functions for computing and applying ICA to MEG data, with eye-movement artifact detection via correlation with continuous XY gaze position from eye tracking. The ET data is loaded from CSV samples, wrapped into an MNE RawArray, temporally aligned to MEG using realign_raw, and then each IC source is correlated with the gx/gy channels to flag eye-related components.

pyavs.preprocessing.ica.build_et_raw_from_samples(samples_df: DataFrame, sfreq: float | None = None) → mne.io.RawArray[source]¶

Wrap eye tracking samples from a CSV DataFrame into an MNE RawArray.

Parameters:

samples_df (pd.DataFrame) – Eye tracking samples with at least ‘smpl_time’ [s], ‘gx’ [px], ‘gy’ [px]
sfreq (float, optional) – Sampling frequency. If None, estimated from median(diff(smpl_time)).

Returns:

Raw object with two channels: ‘gx’ and ‘gy’ of type ‘eyegaze’.

Return type:

mne.io.RawArray

pyavs.preprocessing.ica.extract_scene_onset_times_meg(raw: mne.io.Raw, session: int) → ndarray[source]¶

Extract per-trial scene onset times from MEG data using repaired triggers.

Identifies per-trial anchors using repaired block triggers (block+1000) to disambiguate which trial belongs to which block, then returns the trial-number trigger time (1–30, n_trial_per_block=30) directly.

Parameters:

raw (mne.io.Raw) – MEG raw data with STI101 trigger channel (concatenated session).
session (int) – Session number — needed to identify valid block trigger codes after repair.

Returns:

Array of scene onset times in seconds (relative to raw.first_samp), one per trial, in chronological order.

Return type:

np.ndarray

pyavs.preprocessing.ica.extract_scene_onset_times_meg_per_block(raws_dict: dict, session: int, verbose: bool = False) → dict[source]¶

Extract per-trial scene onset times separately for each MEG block.

Parameters:

raws_dict (dict) – {block_id: mne.io.Raw} — individual (not concatenated) MEG block raws.
session (int) – Session number.
verbose (bool) – Log per-block event counts.

Returns:

{block_id: np.ndarray} of event times in seconds relative to each block’s own first_samp. Blocks with no events map to an empty array.

Return type:

pyavs.preprocessing.ica.extract_scene_onset_times_et_per_block(subject_id: int, session: int, data_path: str | None = None) → dict[source]¶

Extract ET scene onset times grouped by block, using BLOCKID messages.

BLOCKID rows in the ET messages file record the start time of each block (in Eyelink ms). Each TYPE=0 SCENEID_time is assigned to the block whose BLOCKID_time is the largest value ≤ that scene time.

Parameters:

subject_id (int) – Subject ID.
session (int) – Session number.
data_path (str, optional) – Path to data directory.

Returns:

{meg_block_id: np.ndarray} of scene onset times in seconds (absolute Eyelink clock), one array per block in MEG block order.

Return type:

pyavs.preprocessing.ica.extract_scene_onset_times_et(subject_id: int, session: int, data_path: str | None = None) → ndarray[source]¶

Extract scene onset times from the eye tracking messages file.

Parameters:

subject_id (int) – Subject ID.
session (int) – Session number.
data_path (str, optional) – Path to data directory. If None, uses configured data path.

Returns:

Sorted array of scene onset times in seconds.

Return type:

np.ndarray

pyavs.preprocessing.ica.align_et_to_meg(meg_raw: mne.io.Raw, et_raw: mne.io.RawArray, meg_event_times: ndarray, et_event_times: ndarray, verbose: bool = True) → mne.io.RawArray[source]¶

Align ET RawArray to MEG Raw using shared scene onset events.

Uses mne.preprocessing.realign_raw to correct for clock offset and drift between the two recording systems. ET raw is modified in-place.

Parameters:

meg_raw (mne.io.Raw) – MEG data (reference, untouched).
et_raw (mne.io.RawArray) – ET data (aligned in-place to MEG timeline).
meg_event_times (np.ndarray) – Scene onset times in MEG reference frame [s].
et_event_times (np.ndarray) – Corresponding scene onset times in ET reference frame [s].
verbose (bool, optional) – Whether to log progress.

Returns:

Aligned ET raw, cropped and resampled to match MEG.

Return type:

mne.io.RawArray

pyavs.preprocessing.ica.align_et_to_meg_per_block(raws_dict: dict, samples_df: DataFrame, meg_events_per_block: dict, et_events_per_block: dict, verbose: bool = True) → mne.io.RawArray[source]¶

Align ET gaze data to MEG per block, then return a concatenated ET RawArray.

Deprecated since version Use: build_et_gaze_epochs_per_scene() instead. This function is kept for backward compatibility with test_et_alignment.py only.

Parameters:

raws_dict (dict) – {block_id: mne.io.Raw} individual MEG block raws (not concatenated).
samples_df (pd.DataFrame) – ET cleaned samples with at least ‘smpl_time’ [s], ‘gx’, ‘gy’ columns.
meg_events_per_block (dict) – {block_id: np.ndarray} — event times in MEG seconds relative to each block’s first_samp.
et_events_per_block (dict) – {block_id: np.ndarray} — absolute Eyelink clock times in seconds.
verbose (bool) – Log alignment details per block.

Returns:

Concatenated ET raw aligned to MEG, with ‘gx’ and ‘gy’ channels.

Return type:

mne.io.RawArray

pyavs.preprocessing.ica.build_et_gaze_epochs_per_scene(meg_raw: mne.io.Raw, samples_df: DataFrame, session: int, tmin: float = -0.1, tmax: float = 4.0, verbose: bool = True) → Tuple[mne.EpochsArray, DataFrame][source]¶

Align ET gaze samples to MEG trial-by-trial and return as an EpochsArray.

Each scene epoch is aligned independently: ET samples for a given trial are looked up by (block, trial_per_block) from samples_df, which must have been loaded with offset_scene_triggers_ms=60 so that the time_in_trial column already expresses time relative to the MEG scene_on trigger (trigger code 100).

Derivation of the offset:

T_et_scene = T_meg_trigger + 0.060  (ET scene fires 60 ms after MEG)
time_in_trial = smpl_time - T_et_scene + 0.060
              = smpl_time - T_meg_trigger

No clock-drift model, no realign_raw, no trigger-count matching.

Parameters:

meg_raw (mne.io.Raw) – Concatenated MEG session (STI101 trigger channel required).
samples_df (pd.DataFrame) – Eye tracking samples from load_samples_with_scenes(offset_scene_triggers_ms=60). Required columns: time_in_trial, gx, gy, block, trial_per_block, recording.
session (int) – Session number — passed to repair_meg_trigger_events and get_avs_blocks.
tmin (float) – Epoch start in seconds relative to MEG scene_on trigger (default -0.1).
tmax (float) – Epoch end in seconds relative to MEG scene_on trigger (default 4.0).
verbose (bool) – Log per-trial alignment statistics.

Returns:

gaze_epochs (mne.EpochsArray) – Shape (n_trials, 2, n_times), channels gx / gy at MEG sfreq. gaze_epochs.metadata contains block and trial_per_block.
trials_meta (pd.DataFrame) – Same metadata as gaze_epochs.metadata.

pyavs.preprocessing.ica.build_meg_scene_epochs_with_et(meg_raw: mne.io.Raw, samples_df: DataFrame, session: int, tmin: float = -0.1, tmax: float = 4.0, picks: str | list | None = 'meg', verbose: bool = True) → Tuple[mne.Epochs, DataFrame][source]¶

Build MEG scene epochs with gaze channels (gx, gy) appended.

First builds ET gaze epochs via build_et_gaze_epochs_per_scene() to obtain the trial events, then creates matching MEG epochs (no amplitude rejection, so epoch counts are guaranteed to stay in sync), and finally appends the two gaze channels to the MEG epoch object.

Parameters:

meg_raw (mne.io.Raw) – Concatenated MEG session (STI101 required).
samples_df (pd.DataFrame) – ET samples from load_samples_with_scenes(offset_scene_triggers_ms=0).
session (int) – Session number.
tmin (float) – Epoch start in seconds relative to MEG scene_on trigger (default -0.1).
tmax (float) – Epoch end in seconds relative to MEG scene_on trigger (default 4.0).
picks (str or list, optional) – MEG channel selection passed to mne.Epochs (default: 'meg').
verbose (bool) – Log progress.

Returns:

epochs (mne.Epochs) – Scene epochs with MEG channels followed by gx / gy. Shape (n_trials, n_meg_picks + 2, n_times).
trials_meta (pd.DataFrame) – Trial metadata with block and trial_per_block columns.

pyavs.preprocessing.ica.find_eye_components_xy_correlation(ica: mne.preprocessing.ICA, meg_raw: mne.io.Raw, et_gaze_epochs: mne.EpochsArray, top_fraction: float = 0.05, reject: dict | None = None, verbose: bool = True) → Tuple[List[int], DataFrame][source]¶

Find ICA components correlated with per-scene XY gaze position.

IC sources are epoched with the same scene_on events as et_gaze_epochs and then both are flattened across epochs before computing Pearson r. The top top_fraction of components ranked by max(abs(r_gx), abs(r_gy)) are flagged as eye components.

Parameters:

ica (mne.preprocessing.ICA) – Fitted ICA object.
meg_raw (mne.io.Raw) – MEG raw data (unfiltered; used to compute ICA source epochs).
et_gaze_epochs (mne.EpochsArray) – Per-scene gaze epochs from build_et_gaze_epochs_per_scene(), with ‘gx’ and ‘gy’ channels. Its .events and .tmin / .tmax drive the matching MEG epoching.
top_fraction (float, optional) – Fraction of components to flag as eye-related, ranked by max_r (default: 0.05 → top 5 %).
reject (dict or None, optional) – Amplitude rejection thresholds applied when creating MEG epochs (e.g. dict(grad=4000e-13, mag=4e-12)). ET epochs are synced to the surviving MEG epochs after dropping. None keeps all epochs.
verbose (bool, optional) – Whether to log results.

Returns:

(eye_component_indices, scores_df) where scores_df has columns ‘component’, ‘r_gx’, ‘r_gy’, ‘max_r’.

Return type:

pyavs.preprocessing.ica.save_et_scores(scores_df: DataFrame, subject_id: int, session: int, data_path: str | None = None, overwrite: bool = True) → str[source]¶

Save ICA–ET correlation scores to a CSV file in the derivatives directory.

Parameters:

scores_df (pd.DataFrame) – DataFrame with columns ‘component’, ‘r_gx’, ‘r_gy’, ‘max_r’.
subject_id (int) – Subject ID.
session (int) – Session number.
data_path (str, optional) – Path to data directory.
overwrite (bool, optional) – Whether to overwrite existing file (default: True).

Returns:

Path to the saved CSV file.

Return type:

str

pyavs.preprocessing.ica.save_ica_exclusions(eye_exclusions: List[int], cardiac_exclusions: List[int], subject_id: int, session: int, data_path: str | None = None, overwrite: bool = True) → str[source]¶

Save ICA component exclusions to a JSON file in the BIDS derivatives directory.

The format mirrors the legacy ex_components.json used by apply_ica_to_raws():

{"as01": {"1": [0, 3, 12, 15], "2": [1, 5, 22]}}

If the file already exists its contents are merged (read-modify-write), so successive sessions accumulate in the same file.

Parameters:

eye_exclusions (list of int) – ICA component indices flagged as eye-movement artefacts.
cardiac_exclusions (list of int) – ICA component indices flagged as cardiac artefacts.
subject_id (int) – Subject ID.
session (int) – Session number.
data_path (str, optional) – Path to data directory.
overwrite (bool, optional) – Whether to overwrite an existing session entry (default: True).

Returns:

Path to the saved JSON file.

Return type:

str

pyavs.preprocessing.ica.run_ica_et_pipeline(subject_id: int, session: int, data_path: str | None = None, top_fraction: float = 0.05, filter_l_freq: float = 1.0, filter_h_freq: float = 40.0, n_components: int | None = None, reject: dict | None = None, save_results: bool = True, verbose: bool = True) → Tuple[mne.preprocessing.ICA, List[int], List[int], DataFrame][source]¶

Full ICA pipeline with eye tracking XY correlation for one subject/session.

Loads preprocessed MEG blocks, aligns ET samples to MEG per scene trial (60 ms offset, no realign_raw), fits ICA on a filtered copy of the concatenated session, then flags ICs correlated with per-scene gaze.

Parameters:

subject_id (int) – Subject ID.
session (int) – Session number.
data_path (str, optional) – Path to data directory. If None, uses configured data path.
top_fraction (float, optional) – Fraction of components to flag as eye-related by max_r rank (default: 0.05).
filter_l_freq (float, optional) – High-pass cutoff for ICA fitting copy (default: 1.0 Hz).
filter_h_freq (float, optional) – Low-pass cutoff for ICA fitting copy (default: 40.0 Hz).
n_components (int, optional) – Number of ICA components (default: None, uses all available).
save_results (bool, optional) – Whether to save ICA solution and ET scores to derivatives (default: True).
verbose (bool, optional) – Whether to log progress (default: True).

Returns:

(ica, eye_exclusions, cardiac_exclusions, scores_df)

Return type:

pyavs.preprocessing.ica.compute_ica(raw: mne.io.Raw, n_components: int | None = None, method: str = 'fastica', fit_params: dict | None = None, max_iter: int = 200, random_state: int = 42, picks: str | list | None = 'meg', decim: int | None = None, reject: dict | None = False, reject_by_annotation: bool = True, verbose: bool = True) → mne.preprocessing.ICA[source]¶

Compute ICA decomposition on MEG data.

Parameters:

raw (mne.io.Raw) – MEG raw data.
n_components (int, optional) – Number of ICA components (default: None, uses min(80, n_meg_channels)).
method (str, optional) – ICA algorithm (default: ‘infomax’).
fit_params (dict, optional) – Additional parameters for ICA fitting.
max_iter (int, optional) – Maximum number of iterations (default: 200).
random_state (int, optional) – Random seed for reproducibility (default: 42).
picks (str or list, optional) – Channels to include (default: ‘meg’).
decim (int, optional) – Decimation factor (default: None).
reject (dict, optional) – Rejection criteria for fitting.
reject_by_annotation (bool, optional) – Whether to reject by annotations (default: True).
verbose (bool, optional) – Whether to log progress.

Returns:

Fitted ICA object.

Return type:

mne.preprocessing.ICA

pyavs.preprocessing.ica.find_cardiac_components(ica: mne.preprocessing.ICA, raw: mne.io.Raw, threshold: float = 0.8, method: str = 'automatic', verbose: bool = True) → List[int][source]¶

Find ICA components related to cardiac artifacts.

Parameters:

ica (mne.preprocessing.ICA) – Fitted ICA object.
raw (mne.io.Raw) – MEG raw data.
threshold (float, optional) – Detection threshold (default: 0.8).
method (str, optional) – Detection method (‘automatic’, ‘frequency’) (default: ‘automatic’).
verbose (bool, optional) – Whether to log results.

Returns:

Indices of cardiac components.

Return type:

list of int

pyavs.preprocessing.ica.apply_ica(raw: mne.io.Raw, ica: mne.preprocessing.ICA, exclude: List[int] | None = None, verbose: bool = True) → mne.io.Raw[source]¶

Apply ICA to remove specified components.

Parameters:

raw (mne.io.Raw) – MEG raw data.
ica (mne.preprocessing.ICA) – Fitted ICA object.
exclude (list of int, optional) – Component indices to exclude (default: None, uses ica.exclude).
verbose (bool, optional) – Whether to log progress.

Returns:

MEG data with ICA applied.

Return type:

pyavs.preprocessing.ica.plot_ica_components(ica: mne.preprocessing.ICA, raw: mne.io.Raw, picks: List[int] | None = None, ch_type: str = 'mag', image_interp: str = 'bilinear', show: bool = True, save_path: str | None = None) → Figure[source]¶

Plot ICA component topographies.

Parameters:

ica (mne.preprocessing.ICA) – Fitted ICA object.
raw (mne.io.Raw) – MEG raw data (for channel info).
picks (list of int, optional) – Components to plot (default: None, plots all).
ch_type (str, optional) – Channel type for topography (default: ‘mag’).
image_interp (str, optional) – Interpolation method (default: ‘bilinear’).
show (bool, optional) – Whether to show the plot.
save_path (str, optional) – Path to save the plot.

Return type:

plt.Figure

pyavs.preprocessing.ica.plot_ica_sources(ica: mne.preprocessing.ICA, raw: mne.io.Raw, picks: List[int] | None = None, start: float = 0.0, stop: float | None = None, show: bool = True, save_path: str | None = None) → Figure[source]¶

Plot ICA source time courses.

Parameters:

ica (mne.preprocessing.ICA) – Fitted ICA object.
raw (mne.io.Raw) – MEG raw data.
picks (list of int, optional) – Components to plot.
start (float, optional) – Start time in seconds.
stop (float, optional) – Stop time in seconds.
show (bool, optional) – Whether to show the plot.
save_path (str, optional) – Path to save the plot.

Return type:

plt.Figure

pyavs.preprocessing.ica.save_ica(ica: mne.preprocessing.ICA, subject_id: int, session: int, data_path: str | None = None, overwrite: bool = True) → str[source]¶

Save ICA object to BIDS derivatives directory.

Parameters:

ica (mne.preprocessing.ICA) – ICA object to save.
subject_id (int) – Subject ID.
session (int) – Session number.
data_path (str, optional) – Path to data directory.
overwrite (bool, optional) – Whether to overwrite existing files (default: True).

Returns:

Path to saved ICA file.

Return type:

str

pyavs.preprocessing.ica.load_ica(subject_id: int, session: int, data_path: str | None = None, verbose: bool = True) → mne.preprocessing.ICA[source]¶

Load ICA object from BIDS derivatives directory.

Parameters:

subject_id (int) – Subject ID.
session (int) – Session number.
data_path (str, optional) – Path to data directory.
verbose (bool, optional) – Whether to log progress.

Returns:

Loaded ICA object.

Return type:

mne.preprocessing.ICA

pyavs.preprocessing.ica.apply_ica_to_raws(raws_dict: Dict[Any, mne.io.Raw], subject_id: int, session: int, use_precomputed: bool = True, ica_solutions_dir: str | None = None, ica_exclusions_file: str | None = None, data_path: str | None = None, compute_new_ica: bool = False, find_artifacts: bool = True, verbose: bool = True) → Dict[Any, mne.io.Raw][source]¶

Apply ICA artifact removal to a dictionary of raw MEG data.

Applies the precomputed ICA solution shipped in derivatives/pyavs or newly computed ICA to unconcatenated raw MEG blocks. Kept for backward compatibility with AVSComposer.apply_ica_to_blocks().

Parameters:

raws_dict (dict) – Dictionary mapping block IDs to raw MEG data.
subject_id (int) – Subject ID.
session (int) – Session number.
use_precomputed (bool, optional) – Whether to use a precomputed ICA solution (default: True).
ica_solutions_dir (str, optional) – Directory of precomputed ICA solutions in the legacy {dir}/as01a/as01a-ica.fif layout, for solutions computed outside the derivatives/pyavs tree. Only used if both this and ica_exclusions_file are given; otherwise the shipped derivatives/pyavs ICA solution is used.
ica_exclusions_file (str, optional) – JSON file of component exclusions matching ica_solutions_dir.
compute_new_ica (bool, optional) – Whether to compute new ICA if precomputed not available (default: False).
find_artifacts (bool, optional) – Whether to automatically find artifacts when computing new ICA.
verbose (bool, optional) – Whether to log progress.

Returns:

Dictionary mapping block IDs to ICA-cleaned raw MEG data.

Return type:

MEG-ET Alignment¶

MEG-ET alignment and fusion for pyAVS package.

This module provides helper functions for temporal alignment and fusion of MEG and eye-tracking data using the AVS composer approach with trigger-based alignment.

For the main MEG-ET pipeline, use AVSComposer from .composer module.

pyavs.preprocessing.alignment.get_meg_trigger_mapping() → Dict[str, int][source]¶

Get mapping of MEG trigger names to codes.

Returns:: Dictionary mapping trigger names to codes
Return type:: dict

pyavs.preprocessing.alignment.repair_meg_trigger_events(events: ndarray, session: int, new_block_trigger_offset: int = 1000, initial_block_trigger_offset: int = 50, verbose: bool = True) → ndarray[source]¶

Repair corrupted MEG trigger events.

This function corrects trigger events where block numbers across the entire experiment were sent instead of block numbers per session.

Parameters:

events (np.ndarray) – Events array with shape (n_events, 3)
session (int) – Session number
new_block_trigger_offset (int, optional) – Offset for new block triggers (default: 1000)
initial_block_trigger_offset (int, optional) – Initial offset for block triggers (default: 50)
verbose (bool, optional) – Whether to print repair information (default: True)

Returns:

Repaired events array

Return type:

np.ndarray

pyavs.preprocessing.alignment.create_et_event_epochs(raw: mne.io.Raw, eye_events_df: DataFrame, event_type: str = 'saccade', recording: str = 'scene', tmin: float = -0.2, tmax: float = 0.8, baseline: Tuple[float, float] | None = None, picks: str | list | None = 'meg', reject: dict | None = None, reject_by_annotation: bool = True, preload: bool = True, offset_scene_triggers_ms: float = 20.0, verbose: bool = True, **kwargs) → Tuple[mne.Epochs, DataFrame][source]¶

Create MEG epochs based on eye tracking events using the AVS composer approach.

This function aligns MEG and ET data by using scene onset triggers (code 100) as temporal anchors and then adding the eye tracking event’s time_in_trial relative timing. This follows the methodology from the original AVS-machine-room codebase.

Parameters:

raw (mne.io.Raw) – MEG raw data
eye_events_df (pd.DataFrame) – Eye tracking events dataframe with ‘time_in_trial’, ‘block’, ‘trial_per_block’ columns
event_type (str, optional) – Type of eye tracking event to use (‘scene’, ‘fixation’, ‘saccade’, ‘blink’) (default: ‘saccade’)
recording (str, optional) – Recording context (‘scene’, ‘caption’, ‘microphone’) (default: ‘scene’)
tmin (float, optional) – Start time before event in seconds (default: -0.2)
tmax (float, optional) – End time after event in seconds (default: 0.8)
baseline (tuple or None, optional) – Baseline time window (default: None)
picks (str or list, optional) – Channels to include (default: ‘meg’)
reject (dict, optional) – Rejection criteria (default: None)
reject_by_annotation (bool, optional) – Whether to reject by annotations (default: True)
preload (bool, optional) – Whether to preload epoch data (default: True)
offset_scene_triggers_ms (float, optional) – Systematic offset correction in milliseconds (default: 20.0) This compensates for hardware delays between MEG and ET systems
verbose (bool, optional) – Whether to print epoch information (default: True)

Returns:

(epochs, events_metadata) - MEG epochs and corresponding event metadata

Return type:

Notes

This implementation follows the AVS composer methodology: 1. Find MEG scene onset triggers (code 100) for each trial 2. Calculate MEG event times as: scene_onset_time + time_in_trial + offset 3. Apply systematic 20ms offset correction for hardware delays 4. Create epochs using the calculated MEG sample times

Required columns in eye_events_df: - ‘time_in_trial’: Relative time from scene onset in seconds - ‘block’: Block number - ‘trial_per_block’: Trial number within block - ‘type’: Event type (‘fixation’, ‘saccade’, ‘blink’) - ‘recording’: Recording context (‘scene’, ‘caption’, etc.)

Sample-Scene Attachment¶

Eye tracking samples scene assignment for pyAVS package.

This module provides functionality to load eye tracking samples data and assign them to stimulus scenes, inspired by the avs_combine_events function but adapted for samples data.

Author: P. Sulewski (psulewski@uos.de)

pyavs.preprocessing.samples.attach_scene_ids_to_samples(samples: DataFrame, subject_id: int, session: int, data_path: str | None = None, offset_scene_triggers_ms: int = 20, verbose: bool = True) → DataFrame[source]¶

Attach scene IDs and trial information to eye tracking samples using pyAVS conventions.

This function assigns stimulus scene information to individual eye tracking samples, enabling sample-level analysis of gaze behavior during scene viewing.

Parameters:

samples (pd.DataFrame) – Eye tracking samples dataframe with ‘smpl_time’ column (in seconds)
subject_id (int) – Subject identifier
session (int) – Session number
data_path (str, optional) – Path to the data directory. If None, uses configured data path.
offset_scene_triggers_ms (int, default 20) – Offset to correct for systematic delay between MEG and eyetracker scene onset
verbose (bool, default True) – Whether to print detailed progress information

Returns:

Samples dataframe with added columns: subject, session, trial, recording, sceneID, time_in_trial, block, trial_per_block, caption_task

Return type:

pd.DataFrame

Raises:

KeyError – If required columns are missing from samples dataframe
FileNotFoundError – If required data files cannot be found

pyavs.preprocessing.samples.validate_samples_scene_assignment(samples: DataFrame, verbose: bool = True) → Dict[str, any][source]¶

Validate the scene ID assignment results for samples.

Parameters:

samples (pd.DataFrame) – Samples dataframe with attached scene information
verbose (bool, default True) – Whether to print validation results

Returns:

Dictionary with validation statistics

Return type:

pyavs.preprocessing.samples.load_samples_with_scenes(subject_id: int, session: int, samples_file: str | None = None, data_path: str | None = None, offset_scene_triggers_ms: int = 20, validate_results: bool = True, verbose: bool = True) → DataFrame[source]¶

Convenience function to load samples file and attach scene information.

This function loads eye tracking samples from a file and attaches scene information in a single step, following pyAVS conventions.

Parameters:

subject_id (int) – Subject identifier
session (int) – Session number
samples_file (str, optional) – Path to samples CSV file. If None, attempts to find standard location.
data_path (str, optional) – Path to data directory. If None, uses configured data path.
offset_scene_triggers_ms (int, default 20) – Offset to correct for systematic delay between MEG and eyetracker
validate_results (bool, default True) – Whether to validate and report assignment results
verbose (bool, default True) – Whether to print progress information

Returns:

Samples dataframe with attached scene information

Return type:

pd.DataFrame

Examples

>>> # Load samples with scene information
>>> samples = load_samples_with_scenes(
...     subject_id=1, session=1,
...     data_path="/path/to/avs/data"
... )
>>>
>>> # Check coverage
>>> coverage = (samples['sceneID'].notna().sum() / len(samples)) * 100
>>> print(f"Scene coverage: {coverage:.1f}%")

MEG Trigger Tools¶

Ported from the original AVS-machine-room trigger repair/analysis code.

Trigger tools for MEG-ET alignment in pyAVS package.

This module provides functions to use the subjects eye-tracking data in order to generate fixation based annotations to the MEG data of the AVS experiment.

Adapted from AVS-machine-room/avs_machine_room/prepro/meg/avs_trigger_tools.py Author(s): P. Sulewski (psulewski@uos.de)

pyavs.preprocessing.trigger.tools.get_meg_trigger_dict() → Dict[str, int][source]¶

Here we define a dictionary to map the event codes to the event names.

Returns:: Dictionary mapping trigger names to codes
Return type:: dict

pyavs.preprocessing.trigger.tools.get_avs_blocks(session_num: int, lower_bound: int | None = None, upper_bound: int | None = None, verbose: bool = False) → ndarray[source]¶

Returns a list of all avs blocks in the requested session.

Parameters:

session_num (int) – The session number
lower_bound (int, optional) – The minimum block number of a session
upper_bound (int, optional) – The maximum block number of a session
verbose (bool, optional) – If True, the function will print some information

Returns:

Array of all avs blocks in the requested session

Return type:

numpy.ndarray

pyavs.preprocessing.trigger.tools.repair_meg_trigger_events(events: ndarray, session: int, new_block_trigger_offset: int = 1000, initial_block_trigger_offset: int = 50, verbose: bool = True) → ndarray[source]¶

This function repairs the trigger events of the MEG data. It is necessary because of a mistake in the experiment code. Instead of sending the block number per session, the block number across the whole experiment was sent. This function corrects this mistake.

Note: Due to a bottleneck with the trigger system not being able to send trigger codes above 127 and resetting triggers above that to 0+, we have to adjust for this.

Parameters:

events (numpy.ndarray) – Events structure of the MEG data
session (int) – Session number
new_block_trigger_offset (int, optional) – Offset for new block triggers (default: 1000)
initial_block_trigger_offset (int, optional) – Initial offset for block triggers (default: 50)
verbose (bool, optional) – Whether to print repair information (default: True)

Returns:

Repaired events array

Return type:

numpy.ndarray

pyavs.preprocessing.trigger.tools.get_meg_timestamp(meg_events: ndarray, trial: int, block: int, block_trigger_offset: int = 1000, verbose: bool = False, optimized_timing: bool = True, use_block_trigger: bool = False) → float | None[source]¶

Returns the timestamp of the scene onset in the MEG data.

Parameters:

meg_events (numpy.ndarray) – Events structure of the MEG data
trial (int) – Trial number
block (int) – Block number (across all sessions)
block_trigger_offset (int, optional) – Offset of the block trigger in the MEG data as used in repair_events function (default: 1000)
verbose (bool, optional) – If True prints some information (default: False)
optimized_timing (bool, optional) – If True the function will return the timestamp of the scene onset not based on the trial number trigger but the interpolation between the block and the trial number trigger timestamp. (This aligns better with the actual scene onset, as measured by the photodiode) (default: True)
use_block_trigger (bool, optional) – If True the function will use the block trigger to find the scene onset. If False it will use the trial number trigger. This does not work with “optimized timing” (default: False)

Returns:

Timestamp of the scene onset in the MEG data

Return type:

float or None

pyavs.preprocessing.trigger.tools.get_trigger_epochs_metadata(trigger_events: ndarray, trigger_id: int, blocks: ndarray, block_trigger_offset: int = 1000) → DataFrame[source]¶

For each occurrence of trigger_id in trigger_events, find the associated trial number and block by looking at preceding triggers.

The trigger sequence in the repaired events array is:: block_trigger (1000+block) -> trial_number (1-30) -> … -> trigger_id

This function walks backward from each trigger_id occurrence to find the most recent trial-number trigger (value 1-30 preceded by a block trigger).

Parameters:

trigger_events (numpy.ndarray) – Repaired events array (N x 3: [timestamp, prev_trigger, trigger_value])
trigger_id (int) – The trigger code to find (e.g. 110 for mic_on)
blocks (numpy.ndarray) – Array of block numbers expected in this session
block_trigger_offset (int, optional) – Offset used for block triggers (default: 1000)

Returns:

DataFrame with columns: [‘event_index’, ‘sample’, ‘block’, ‘trial_per_block’] One row per occurrence of trigger_id found in trigger_events.

Return type:

pandas.DataFrame

pyavs.preprocessing.trigger.tools.add_fix_event_trigger(raw: mne.io.Raw, blocks: List[int], et_events: DataFrame, session: int, block_trigger_offset: int = 1000, stim_channel: str = 'STI101', verbose: bool = True, event_types: List[str] = ['fixation', 'saccade', 'blink'], recording: str = 'scene') → Tuple[mne.io.Raw, List[Tuple[int, int]]][source]¶

Adds eye movement based event triggers (fixation, saccade) to the raw neuro data.

Parameters:

raw (mne.io.Raw) – Raw neuro data
blocks (list of int) – List of blocks
et_events (pandas.DataFrame) – Events structure of the eye tracker data
session (int) – Session number
block_trigger_offset (int, optional) – Offset of the block trigger in the MEG data as used in repair_events function (default: 1000)
stim_channel (str, optional) – Name of the stim channel in the raw data (default: ‘STI101’)
verbose (bool, optional) – Print some output (default: True)
event_types (list of str, optional) – List of event types to add (default: [‘fixation’, ‘saccade’]) Valid options: [‘fixation’, ‘saccade’, ‘blink’]
recording (str, optional) – Recording context to filter events by (default: ‘scene’) Valid options: [‘scene’, ‘caption’, ‘microphone’]

Returns:

(raw_with_annotations, missing_trials) - Raw with added eye movement based event annotations and list of missing trials

Return type: