Source Reconstruction (pyavs.source)¶

The source module provides functions for MEG source reconstruction and analysis.

Forward Modeling¶

Forward modeling for pyAVS package.

This module provides functions for creating forward models, BEM models, and handling coregistration for source reconstruction.

pyavs.source.forward.create_bem_model(subject: str, subjects_dir: str, conductivity: Tuple[float, float, float] = (0.3, 0.006, 0.3), ico: int | None = 4, verbose: bool = True) → mne.bem.ConductorModel[source]¶

Create BEM (Boundary Element Method) model for source reconstruction.

Parameters:

subject (str) – Subject name in FreeSurfer subjects directory
subjects_dir (str) – Path to FreeSurfer subjects directory
conductivity (tuple of float, optional) – Conductivity values for (brain, skull, scalp) (default: (0.3, 0.006, 0.3))
ico (int, optional) – Icosahedral subdivision number (default: 4)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

BEM conductor model

Return type:

mne.bem.ConductorModel

pyavs.source.forward.create_bem_solution(bem_model: mne.bem.ConductorModel, verbose: bool = True) → mne.bem.ConductorModel[source]¶

Create BEM solution from BEM model.

Parameters:

bem_model (mne.bem.ConductorModel) – BEM model
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

BEM solution

Return type:

mne.bem.ConductorModel

pyavs.source.forward.create_source_space(subject: str, subjects_dir: str, spacing: str = 'ico4', surface: str = 'white', add_dist: bool = True, verbose: bool = True) → mne.SourceSpaces[source]¶

Create cortical source space.

Parameters:

subject (str) – Subject name in FreeSurfer subjects directory
subjects_dir (str) – Path to FreeSurfer subjects directory
spacing (str, optional) – Spacing between sources (default: ‘ico4’)
surface (str, optional) – Surface to use (default: ‘white’)
add_dist (bool, optional) – Whether to add distance information (default: True)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Source space

Return type:

mne.SourceSpaces

pyavs.source.forward.create_forward_model(raw: mne.io.Raw, trans: str | mne.transforms.Transform, src: mne.SourceSpaces, bem_solution: mne.bem.ConductorModel, meg: bool = True, eeg: bool = False, mindist: float = 5.0, ignore_ref: bool = True, verbose: bool = True) → mne.Forward[source]¶

Create forward model for source reconstruction.

Parameters:

raw (mne.io.Raw) – MEG/EEG raw data (used for sensor information)
trans (str or mne.transforms.Transform) – Transformation from head to MRI coordinates
src (mne.SourceSpaces) – Source space
bem_solution (mne.bem.ConductorModel) – BEM solution
meg (bool, optional) – Whether to include MEG channels (default: True)
eeg (bool, optional) – Whether to include EEG channels (default: False)
mindist (float, optional) – Minimum distance between sources and inner skull (default: 5.0)
ignore_ref (bool, optional) – Whether to ignore reference channels (default: True)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Forward solution

Return type:

mne.Forward

pyavs.source.forward.setup_coregistration(subject: str, subjects_dir: str, raw: mne.io.Raw, fiducials: str = 'auto', verbose: bool = True) → mne.transforms.Transform[source]¶

Set up coregistration between MEG and MRI coordinate systems.

Parameters:

subject (str) – Subject name in FreeSurfer subjects directory
subjects_dir (str) – Path to FreeSurfer subjects directory
raw (mne.io.Raw) – MEG raw data
fiducials (str, optional) – How to handle fiducials (default: ‘auto’)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Head-to-MRI transformation

Return type:

mne.transforms.Transform

pyavs.source.forward.check_forward_model(fwd: mne.Forward, verbose: bool = True) → Dict[str, Any][source]¶

Check forward model for potential issues.

Parameters:

fwd (mne.Forward) – Forward solution
verbose (bool, optional) – Whether to print check results (default: True)

Returns:

Dictionary with check results

Return type:

dict

pyavs.source.forward.save_forward_model(fwd: mne.Forward, subject_id: int, session: int, data_path: str | None = None, overwrite: bool = True) → str[source]¶

Save forward model to derivatives directory.

Parameters:

fwd (mne.Forward) – Forward solution
subject_id (int) – Subject ID
session (int) – Session number
data_path (str, optional) – Path to data directory. If None, uses configured data path
overwrite (bool, optional) – Whether to overwrite existing files (default: True)

Returns:

Path to saved forward model

Return type:

str

pyavs.source.forward.load_forward_model(subject_id: int, session: int | None = None, data_path: str | None = None, verbose: bool = True) → mne.Forward[source]¶

Load a forward model from the dataset.

Two locations are searched, in order:

pyAVS derivatives (only when session is given) — a forward recomputed with save_forward_model(): {derivatives}/sub-{id:02d}/ses-{sess:02d}/source/ sub-{id:02d}_ses-{sess:02d}_task-avs_fwd.fif
The shipped forward — one per subject, session-independent: {root}/derivatives/freesurfer/sub-{id:02d}/bem/sub-{id:02d}-fwd.fif

The shipped forward is what the release provides; the derivatives path only exists if you recomputed one yourself, in which case it takes precedence.

Parameters:

subject_id (int) – Subject ID
session (int, optional) – Session number. When given, a per-session forward in the pyAVS derivatives tree is preferred over the shipped one. The shipped forward does not depend on session.
data_path (str, optional) – Path to the avs-public root. If None, uses the configured data path.
verbose (bool, optional) – Whether to print loading information (default: True)

Returns:

Forward solution

Return type:

mne.Forward

Source Reconstruction¶

Source reconstruction for pyAVS package.

This module provides functions for MEG source reconstruction including beamforming, minimum norm estimation, and population code analysis.

pyavs.source.reconstruction.setup_source_reconstruction(subject_id: int, session: int, method: str = 'beamformer', data_path: str | None = None, **method_kwargs) → Dict[str, Any][source]¶

Set up source reconstruction for a subject/session.

Parameters:

subject_id (int) – Subject ID
session (int) – Session number
method (str, optional) – Source reconstruction method (‘beamformer’, ‘mne’, ‘lcmv’) (default: ‘beamformer’)
data_path (str, optional) – Path to data directory
**method_kwargs – Additional method-specific parameters

Returns:

Source reconstruction setup parameters

Return type:

dict

pyavs.source.reconstruction.compute_beamformer_filters(epochs: mne.Epochs, forward: mne.Forward, noise_cov: mne.Covariance | None = None, data_cov: mne.Covariance | None = None, reg: float = 0.05, weight_norm: str = 'unit-noise-gain', pick_ori: str = 'max-power', reduce_rank: bool = False, verbose: bool = True) → mne.beamformer.Beamformer[source]¶

Compute LCMV beamformer filters.

Parameters:

epochs (mne.Epochs) – Epoched MEG data
forward (mne.Forward) – Forward solution
noise_cov (mne.Covariance, optional) – Noise covariance matrix (default: None, computed from data)
data_cov (mne.Covariance, optional) – Data covariance matrix (default: None, computed from epochs)
reg (float, optional) – Regularization parameter (default: 0.05)
weight_norm (str, optional) – Weight normalization method (default: ‘unit-noise-gain’)
pick_ori (str, optional) – Orientation selection method (default: ‘max-power’)
reduce_rank (bool, optional) – Whether to reduce rank (default: False)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Beamformer filters

Return type:

mne.beamformer.Beamformer

pyavs.source.reconstruction.apply_beamformer(epochs: mne.Epochs, filters: mne.beamformer.Beamformer, verbose: bool = True) → ndarray[source]¶

Apply beamformer filters to epoched data.

Parameters:

epochs (mne.Epochs) – Epoched MEG data
filters (mne.beamformer.Beamformer) – Beamformer filters
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Source space data with shape (n_epochs, n_sources, n_times)

Return type:

np.ndarray

pyavs.source.reconstruction.compute_minimum_norm_estimate(epochs: mne.Epochs, forward: mne.Forward, noise_cov: mne.Covariance | None = None, lambda2: float = 0.1111111111111111, method: str = 'dSPM', pick_ori: str | None = None, verbose: bool = True) → List[mne.SourceEstimate][source]¶

Compute minimum norm estimate for epoched data.

Parameters:

epochs (mne.Epochs) – Epoched MEG data
forward (mne.Forward) – Forward solution
noise_cov (mne.Covariance, optional) – Noise covariance matrix (default: None)
lambda2 (float, optional) – Regularization parameter (default: 1.0/9.0)
method (str, optional) – Inverse method (‘MNE’, ‘dSPM’, ‘sLORETA’) (default: ‘dSPM’)
pick_ori (str, optional) – Orientation selection (default: None)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Source estimates for each epoch

Return type:

list of mne.SourceEstimate

pyavs.source.reconstruction.compute_source_power(source_data: ndarray, method: str = 'mean', time_window: Tuple[float, float] | None = None, baseline: Tuple[float, float] | None = None, times: ndarray | None = None, verbose: bool = True) → ndarray[source]¶

Compute source power from source space data.

Parameters:

source_data (np.ndarray) – Source space data with shape (n_epochs, n_sources, n_times)
method (str, optional) – Power computation method (‘mean’, ‘peak’, ‘rms’) (default: ‘mean’)
time_window (tuple of float, optional) – Time window for power computation (default: None, uses all times)
baseline (tuple of float, optional) – Baseline time window for normalization (default: None)
times (np.ndarray, optional) – Time points in seconds (default: None)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Source power with shape (n_epochs, n_sources)

Return type:

np.ndarray

pyavs.source.reconstruction.extract_roi_data(source_data: ndarray, src: mne.SourceSpaces, roi_labels: List[str], subjects_dir: str | None = None, subject: str = 'fsaverage', method: str = 'mean', verbose: bool = True) → Dict[str, ndarray][source]¶

Extract data from regions of interest.

Parameters:

source_data (np.ndarray) – Source space data with shape (n_epochs, n_sources, n_times)
src (mne.SourceSpaces) – Source space
roi_labels (list of str) – List of ROI label names
subjects_dir (str, optional) – FreeSurfer subjects directory (default: None, uses get_default_subjects_dir())
subject (str, optional) – Subject name (default: ‘fsaverage’)
method (str, optional) – Aggregation method (‘mean’, ‘pca’, ‘max’) (default: ‘mean’)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Dictionary mapping ROI names to extracted data

Return type:

dict

pyavs.source.reconstruction.compute_population_codes(source_data: ndarray, events_metadata: DataFrame, conditions: List[str], time_window: Tuple[float, float], times: ndarray, baseline: Tuple[float, float] | None = None, normalize: bool = False, verbose: bool = True) → Dict[str, ndarray][source]¶

Compute population codes for different experimental conditions.

Parameters:

source_data (np.ndarray) – Source space data with shape (n_epochs, n_sources, n_times)
events_metadata (pd.DataFrame) – Metadata for each epoch
conditions (list of str) – Column names in metadata defining conditions
time_window (tuple of float) – Time window for population code computation
times (np.ndarray) – Time points in seconds
baseline (tuple of float, optional) – Baseline time window (default: None)
normalize (bool, optional) – Whether to normalize population codes (default: True)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Dictionary mapping conditions to population codes

Return type:

dict

pyavs.source.reconstruction.apply_source_reconstruction(epochs: mne.Epochs, forward: mne.Forward, method: str = 'beamformer', **method_kwargs) → ndarray[source]¶

Apply source reconstruction to epoched data.

Parameters:

epochs (mne.Epochs) – Epoched MEG data
forward (mne.Forward) – Forward solution
method (str, optional) – Source reconstruction method (default: ‘beamformer’)
**method_kwargs – Method-specific parameters

Returns:

Source space data

Return type:

np.ndarray

pyavs.source.reconstruction.compute_empty_room_covariance(data_path: str, subject_id: int, sessions: List[int], verbose: bool = True) → Tuple[mne.Covariance, str][source]¶

Compute noise covariance from empty room recordings.

Parameters:

data_path (str) – Path to data directory
subject_id (int) – Subject ID
sessions (list of int) – Session numbers to process
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

mne.Covariance – Computed noise covariance matrix
str – Path to saved covariance file

Source Spaces¶

Source spaces for pyAVS package.

This module provides functions for creating and managing source spaces including cortical and volume source spaces.

pyavs.source.spaces.create_source_space(subject: str, subjects_dir: str, spacing: str = 'ico4', surface: str = 'white', add_dist: bool = True, n_jobs: int = 1, verbose: bool = True) → mne.SourceSpaces[source]¶

Create cortical source space.

Parameters:

subject (str) – Subject name in FreeSurfer subjects directory
subjects_dir (str) – Path to FreeSurfer subjects directory
spacing (str, optional) – Spacing between sources (‘ico4’, ‘ico5’, ‘7’, etc.) (default: ‘ico4’)
surface (str, optional) – Surface to use (‘white’, ‘pial’, ‘inflated’) (default: ‘white’)
add_dist (bool, optional) – Whether to add distance information (default: True)
n_jobs (int, optional) – Number of parallel jobs (default: 1)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Cortical source space

Return type:

mne.SourceSpaces

pyavs.source.spaces.setup_volume_source_space(subject: str, subjects_dir: str, pos: float = 5.0, mri: str | None = None, bem: str | None = None, surface: str | None = None, mindist: float = 5.0, exclude: float = 0.0, verbose: bool = True) → mne.SourceSpaces[source]¶

Create volume source space.

Parameters:

subject (str) – Subject name in FreeSurfer subjects directory
subjects_dir (str) – Path to FreeSurfer subjects directory
pos (float, optional) – Spacing between sources in mm (default: 5.0)
mri (str, optional) – MRI volume to use (default: None, uses T1)
bem (str, optional) – BEM surface file (default: None)
surface (str, optional) – Surface file for exclusion (default: None)
mindist (float, optional) – Minimum distance from surface in mm (default: 5.0)
exclude (float, optional) – Exclusion distance in mm (default: 0.0)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Volume source space

Return type:

mne.SourceSpaces

pyavs.source.spaces.get_roi_labels(roi_names: str | List[str], subjects_dir: str, subject: str = 'fsaverage', parc: str = 'aparc', verbose: bool = True) → List[mne.Label][source]¶

Get ROI labels from FreeSurfer parcellation.

Parameters:

roi_names (str or list of str) – ROI names or ‘all’ for all ROIs
subjects_dir (str) – FreeSurfer subjects directory
subject (str, optional) – Subject name (default: ‘fsaverage’)
parc (str, optional) – Parcellation name (default: ‘aparc’)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

List of ROI labels

Return type:

list of mne.Label

pyavs.source.spaces.get_glasser_roi_labels(area: str = 'all', subjects_dir: str | None = None, subject: str = 'fsaverage', verbose: bool = True) → List[str][source]¶

Get Glasser atlas ROI names.

Parameters:

area (str, optional) – Area category (‘all’, ‘high_visual’, ‘early_visual’, ‘intermediate_visual’) (default: ‘all’)
subjects_dir (str, optional) – FreeSurfer subjects directory (default: None, uses default from config)
subject (str, optional) – Subject name (default: ‘fsaverage’)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

List of ROI names

Return type:

list of str

pyavs.source.spaces.create_mixed_source_space(subject: str, subjects_dir: str, surface_spacing: str = 'ico4', volume_spacing: float = 7.0, add_interpolator: bool = True, verbose: bool = True) → mne.SourceSpaces[source]¶

Create mixed source space with both surface and volume sources.

Parameters:

subject (str) – Subject name in FreeSurfer subjects directory
subjects_dir (str) – Path to FreeSurfer subjects directory
surface_spacing (str, optional) – Spacing for surface sources (default: ‘ico4’)
volume_spacing (float, optional) – Spacing for volume sources in mm (default: 7.0)
add_interpolator (bool, optional) – Whether to add interpolation matrix (default: True)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Mixed source space

Return type:

mne.SourceSpaces

pyavs.source.spaces.morph_source_space(src: mne.SourceSpaces, subject_from: str, subject_to: str, subjects_dir: str, spacing: str | None = None, verbose: bool = True) → mne.SourceSpaces[source]¶

Morph source space between subjects.

Parameters:

src (mne.SourceSpaces) – Source space to morph
subject_from (str) – Source subject name
subject_to (str) – Target subject name
subjects_dir (str) – FreeSurfer subjects directory
spacing (str, optional) – Target spacing (default: None, uses original)
verbose (bool, optional) – Whether to print progress information (default: True)

Returns:

Morphed source space

Return type:

mne.SourceSpaces

pyavs.source.spaces.save_source_space(src: mne.SourceSpaces, subject_id: int, session: int, space_type: str = 'surface', data_path: str | None = None, overwrite: bool = True) → str[source]¶

Save source space to derivatives directory.

Parameters:

src (mne.SourceSpaces) – Source space to save
subject_id (int) – Subject ID
session (int) – Session number
space_type (str, optional) – Type of source space (‘surface’, ‘volume’, ‘mixed’) (default: ‘surface’)
data_path (str, optional) – Path to data directory
overwrite (bool, optional) – Whether to overwrite existing files (default: True)

Returns:

Path to saved source space file

Return type:

str

pyavs.source.spaces.load_source_space(subject_id: int, session: int | None = None, space_type: str = 'surface', data_path: str | None = None, verbose: bool = True) → mne.SourceSpaces[source]¶

Load a source space from the dataset.

Two locations are searched, in order:

pyAVS derivatives (only when session is given) — a source space written by save_source_space(): {derivatives}/sub-{id:02d}/ses-{sess:02d}/source/ sub-{id:02d}_ses-{sess:02d}_task-avs_{space_type}-src.fif
The shipped source space — one per subject, session-independent: {root}/derivatives/freesurfer/sub-{id:02d}/bem/sub-{id:02d}_oct6-src.fif (only for space_type='surface', the only kind the release ships).

Parameters:

subject_id (int) – Subject ID
session (int, optional) – Session number. When given, a per-session source space in the pyAVS derivatives tree is preferred over the shipped one.
space_type (str, optional) – Type of source space (‘surface’, ‘volume’, ‘mixed’) (default: ‘surface’)
data_path (str, optional) – Path to the avs-public root. If None, uses the configured data path.
verbose (bool, optional) – Whether to print loading information (default: True)

Returns:

Loaded source space

Return type:

mne.SourceSpaces

LCMV Beamformer Filters¶

Per-session and cross-session LCMV beamformer filter computation – see Cross-Session Beamformer Filters and Source Reconstruction and Population Codes for how these fit into the full pipeline.

LCMV beamformer filter computation and management for pyAVS.

This module implements the per-session LCMV filter computation strategy with event-type specific storage and cross-session data covariance computation.

pyavs.source.filters.lcmv_filter_filename(session: int) → str[source]¶

Filename of a per-session LCMV filter, lcmv_filters_ses-01-lcmv.h5.

Single definition so writers and readers cannot drift apart — they used to disagree (ses-01 vs sess01) for the same artifact.

pyavs.source.filters.compute_cross_session_data_covariance(data_path: str, subject_id: int, sessions: List[int], event_type: str, n_epochs_per_session: int = 350, tmin: float = -0.5, tmax: float = 0.8, filter_params: Dict | None = None, resample_freq: int | None = 500, rois: List[str] | None = None, blocks: List[int] | None = None, hemi: str = 'both', block_selection: str = 'all', random_seed: int = 42, overwrite: bool = True) → mne.Epochs[source]¶

Compute cross-session data covariance by concatenating subsampled epochs.

This uses 350 random epochs per session from all 10 sessions to compute a robust data covariance matrix.

Parameters:

data_path (str) – Path to the dataset
subject_id (int) – Subject ID
sessions (list of int) – Sessions to include (should be all 10 for optimal performance)
event_type (str) – Event type (‘saccade’, ‘fixation’, etc.)
n_epochs_per_session (int) – Number of epochs to randomly sample per session (default: 350)
tmin (float) – Time window for epochs
tmax (float) – Time window for epochs
filter_params (dict, optional) – Filter parameters {‘l_freq’: 0.2, ‘h_freq’: 200}
resample_freq (int, optional) – Resampling frequency (default: 500)
block_selection (str) – Block selection strategy (‘all’, ‘10_only’)
random_seed (int) – Random seed for reproducibility
overwrite (bool) – Whether to overwrite existing files

Returns:

rd_epochs_all_sess – Concatenated epochs from all sessions

Return type:

mne.Epochs

pyavs.source.filters.compute_per_session_lcmv_filters(data_path: str, subject_id: int, sessions: List[int], event_type: str, tmin: float = -0.5, tmax: float = 0.8, filter_params: Dict | None = None, resample_freq: int | None = 500, rois: List[str] | None = None, blocks: List[int] | None = None, hemi: str = 'both', n_epochs_per_session: int = 350, cross_session_epochs: mne.Epochs | None = None, pick_ori: str = 'max-power', reg: float = 0.05, weight_norm: str | None = None, rank: str = 'info', overwrite: bool = False) → Dict[int, mne.beamformer.Beamformer][source]¶

Compute per-session LCMV beamformer filters.

This implements the strategy where: 1. Noise covariance is computed per-session from empty room recordings 2. Data covariance is computed from cross-session epochs 3. Filters are computed per-session and saved

Parameters:

data_path (str) – Path to the dataset
subject_id (int) – Subject ID
sessions (list of int) – Sessions to process
event_type (str) – Event type for filter computation
cross_session_epochs (mne.Epochs, optional) – Pre-computed cross-session epochs. If None, will be computed
pick_ori (str) – Orientation picking for beamformer (‘normal’, ‘max-power’, etc.)
reg (float) – Regularization parameter
weight_norm (str, optional) – Weight normalization method
rank (str) – Rank specification
overwrite (bool) – Whether to overwrite existing filters

Returns:

filters – Dictionary mapping session -> beamformer filters

Return type:

dict

pyavs.source.filters.load_or_compute_lcmv_filters(data_path: str, subject_id: int, sessions: List[int], event_type: str, tmin: float = -0.5, tmax: float = 0.8, filter_params: Dict | None = None, resample_freq: int | None = 500, rois: List[str] | None = None, blocks: List[int] | None = None, hemi: str = 'both', n_epochs_per_session: int = 350, **filter_kwargs) → Dict[int, mne.beamformer.Beamformer][source]¶

Load existing LCMV filters or compute them if they don’t exist.

Parameters:

data_path (str) – Path to the dataset
subject_id (int) – Subject ID
sessions (list of int) – Sessions to process
event_type (str) – Event type for filters
**filter_kwargs – Additional arguments for filter computation

Returns:

filters – Dictionary mapping session -> beamformer filters

Return type:

dict

pyavs.source.filters.apply_lcmv_to_epochs(epochs: mne.Epochs, filters: Dict[int, mne.beamformer.Beamformer], session: int) → List[mne.SourceEstimate][source]¶

Apply LCMV beamformer filters to epochs.

Parameters:

epochs (mne.Epochs) – Epochs to source reconstruct
filters (dict) – Dictionary of beamformer filters per session
session (int) – Session number for filter selection

Returns:

stcs – Source time courses for each epoch

Return type:

list of mne.SourceEstimate