I/O (pyavs.io)

HDF5 population-code read/write (the unified format via pyavs.save_population_codes_h5()), plus reproducibility helpers that read configurations saved alongside population codes.

Reading

Data reading utilities for pyAVS package.

This module provides functions for loading various data types from HDF5 files and other sources used in the AVS dataset.

pyavs.io.read.load_data_h5(subject_id: int, session: int, data_type: str = 'population_codes', event_type: str = 'saccade', data_path: str | None = None, **param_filters) → Tuple[Dict[str, ndarray], DataFrame, Dict[str, Any]][source]

Load data from HDF5 files.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • data_type (str, optional) – Type of data to load (default: ‘population_codes’)

  • event_type (str, optional) – Event type to load (default: ‘saccade’)

  • data_path (str, optional) – Path to data directory

  • **param_filters – Additional parameter filters

Returns:

(data_dict, metadata_df, attributes_dict) - Loaded data, metadata, and file attributes

Return type:

tuple

pyavs.io.read.load_population_codes(subject_id: int, session: int, event_type: str = 'saccade', data_path: str | None = None, **param_filters) → Tuple[Dict[str, ndarray], DataFrame, Dict[str, Any]][source]

Load population codes from HDF5 files.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • event_type (str, optional) – Event type to load (default: ‘saccade’)

  • data_path (str, optional) – Path to data directory

  • **param_filters – Additional parameter filters

Returns:

(population_codes, metadata_df, attributes_dict)

Return type:

tuple

pyavs.io.read.load_epochs_h5(subject_id: int, session: int, event_type: str = 'epochs', data_path: str | None = None) → Tuple[Dict[str, ndarray], DataFrame, Dict[str, Any]][source]

Load epochs from HDF5 files.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • event_type (str, optional) – Event type (default: ‘epochs’)

  • data_path (str, optional) – Path to data directory

Returns:

(epochs_data, metadata_df, attributes_dict)

Return type:

tuple

pyavs.io.read.load_metadata_csv(subject_id: int, session: int, event_type: str, data_path: str | None = None) → DataFrame[source]

Load per-epoch metadata.

Despite the historical name, the file read is Parquet (sub-01_ses-01_fixation_metadata.parquet), which is the format the public release ships.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • event_type (str) – Event type (e.g., ‘fixation’, ‘saccade’); a trailing ‘_scene’ is stripped for the filename.

  • data_path (str, optional) – Path to the avs-public root

Returns:

Metadata DataFrame, empty if the file does not exist.

Return type:

pd.DataFrame

pyavs.io.read.load_epochs(subject_id: int, session: int, event_type: str = 'fixation_scene', data_path: str | None = None) → mne.Epochs[source]

Load MNE Epochs object from HDF5 files.

This function loads epochs data and reconstructs a proper MNE Epochs object with metadata attached, suitable for RSA analysis and other MNE operations.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • event_type (str, optional) – Event type (default: ‘fixation_scene’)

  • data_path (str, optional) – Path to data directory

Returns:

Reconstructed epochs object with metadata

Return type:

mne.Epochs

pyavs.io.read.build_epochs_array(data_dict: Dict[str, ndarray], metadata_df: DataFrame, attributes_dict: Dict[str, Any]) → mne.Epochs[source]

Reconstruct an mne.Epochs object from raw per-ROI arrays + metadata.

Shared by load_epochs() (single whole-session h5) and pyavs.remote.query.EpochQuery (assembled from range-read chunks spanning one or more remote h5 files) — both end up with the same data_dict/metadata_df/attributes_dict shape and need identical channel-naming/timing/metadata-attachment logic.

Parameters:

  • data_dict (dict of str to np.ndarray) – Per-ROI epoch arrays ('grad'/'mag' or a single 'epochs' key), each shaped (n_epochs, n_channels, n_times).

  • metadata_df (pd.DataFrame) – Per-epoch metadata, row-aligned with the epoch axis.

  • attributes_dict (dict) – File attributes; 'times' (sample times) and 'hz' (sampling rate) are used when present.

Return type:

mne.Epochs

pyavs.io.read.load_annotated_raw_h5(subject_id: int, session: int, suffix: str = 'annotated', data_path: str | None = None) → Tuple[Dict[str, ndarray], DataFrame, Dict[str, Any]][source]

Load annotated raw data from HDF5 files.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • suffix (str, optional) – File suffix (default: ‘annotated’)

  • data_path (str, optional) – Path to data directory

Returns:

(raw_data, metadata_df, attributes_dict)

Return type:

tuple

pyavs.io.read.load_meg_raw(subject_id: int, session: int, block: int, data_path: str | None = None, preload: bool = False) → mne.io.Raw[source]

Load raw MEG data for a specific subject, session, and block.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • block (int) – Block number

  • data_path (str, optional) – Path to data directory

  • preload (bool, optional) – Whether to preload the data (default: False)

Returns:

Raw MEG data

Return type:

mne.io.Raw

pyavs.io.read.load_meg_preprocessed(subject_id: int, session: int, block: int, data_path: str | None = None, preload: bool = False) → mne.io.Raw[source]

Load preprocessed MEG data.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • block (int) – Block number

  • data_path (str, optional) – Path to data directory

  • preload (bool, optional) – Whether to preload the data (default: False)

Returns:

Preprocessed MEG data

Return type:

mne.io.Raw

pyavs.io.read.load_eye_events(subjects: List[int], sessions: List[int], data_path: str | None = None, event_types: List[str] | None = None, recording: str = 'scene') → Tuple[DataFrame, DataFrame][source]

Load eye tracking events and experiment log for multiple subjects/sessions.

Parameters:

  • subjects (list of int) – Subject IDs to load

  • sessions (list of int) – Session numbers to load

  • data_path (str, optional) – Path to data directory

  • event_types (list of str, optional) – Event types to load (default: [‘fixation’, ‘saccade’, ‘blink’])

  • recording (str, optional) – Recording type (default: ‘scene’)

Returns:

(experiment_log_df, events_df) - Experiment log and events dataframes

Return type:

tuple

pyavs.io.read.load_eye_events_single(subject_id: int, session: int, event_type: str, data_path: str | None = None, recording: str = 'scene') → DataFrame | None[source]

Load eye tracking events for a single subject/session.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • event_type (str) – Event type (‘fixation’, ‘saccade’, ‘blink’)

  • data_path (str, optional) – Path to data directory

  • recording (str, optional) – Recording type (default: ‘scene’)

Returns:

Eye tracking events dataframe

Return type:

pd.DataFrame or None

pyavs.io.read.load_experiment_log(subject_id: int, session: int, data_path: str | None = None) → DataFrame | None[source]

Load experiment log for a subject/session.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • data_path (str, optional) – Path to data directory

Returns:

Experiment log dataframe

Return type:

pd.DataFrame or None

pyavs.io.read.load_anatomical(subject_id: int, data_path: str | None = None) → mne.SourceSpaces | None[source]

Load anatomical source space for a subject.

Parameters:

  • subject_id (int) – Subject ID

  • data_path (str, optional) – Path to data directory

Returns:

Source space

Return type:

mne.SourceSpaces or None

pyavs.io.read.load_scenes(data_path: str | None = None) → Dict[str, Any] | None[source]

Load scene information and images.

Parameters:

data_path (str, optional) – Path to data directory

Returns:

Scene information dictionary

Return type:

dict or None

pyavs.io.read.load_scene_images(data_path: str | None = None) → Dict[int, str][source]

Load scene images and return mapping from scene IDs to file paths.

This function looks for COCO scene images and creates a mapping from scene IDs (extracted from filenames) to full file paths.

Parameters:

data_path (str, optional) – Path to data directory

Returns:

Dictionary mapping scene IDs to image file paths

Return type:

dict

pyavs.io.read.find_population_codes_files(subject_id: int, session: int, data_path: str | None = None, event_type: str | None = None, sampling_rate: int | None = None, **param_filters) → List[Dict[str, Any]][source]

Find population codes files for a subject with optional parameter filtering.

Parameters:

  • subject_id (int) – Subject ID to search for

  • session (int) – Session number to search for

  • data_path (str, optional) – Path to data directory. If None, uses configured data path

  • event_type (str, optional) – Filter by event type (e.g., ‘saccade’, ‘fixation’)

  • sampling_rate (int, optional) – Filter by sampling rate

  • **param_filters – Additional parameter filters

Returns:

List of dictionaries containing file paths and metadata for matching files

Return type:

list of dict

pyavs.io.read.list_available_parameter_sets(data_path: str | None = None) → List[Dict[str, Any]][source]

List all available parameter sets in the population codes storage.

Parameters:

data_path (str, optional) – Path to data directory. If None, uses configured data path

Returns:

List of parameter sets with their metadata

Return type:

list of dict

pyavs.io.read.load_source_data(subject_id: int, session: int, data_type: str = 'population_codes', event_type: str = 'saccade', data_path: str | None = None, **param_filters) → Tuple[Dict[str, ndarray], DataFrame, Dict[str, Any]]

Load data from HDF5 files.

Parameters:

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • data_type (str, optional) – Type of data to load (default: ‘population_codes’)

  • event_type (str, optional) – Event type to load (default: ‘saccade’)

  • data_path (str, optional) – Path to data directory

  • **param_filters – Additional parameter filters

Returns:

(data_dict, metadata_df, attributes_dict) - Loaded data, metadata, and file attributes

Return type:

tuple

Writing

Data writing utilities for pyAVS package.

This module provides functions for saving various data types in HDF5 format, unified through the save_population_codes_h5() function.

pyavs.io.write.save_data_h5(data: ndarray | mne.Epochs | mne.io.Raw | Dict[str, ndarray], subject_id: int, session: int, data_type: str = 'epochs', metadata: DataFrame | None = None, times: ndarray | None = None, event_type: str = 'general', blocks: List[int] | None = None, rois: List[str] | None = None, sampling_rate: int = 500, filter_params: Dict[str, float] | None = None, data_path: str | None = None, hemi: str = 'both', compression: str = 'gzip', **kwargs) → str[source]

Unified function to save any type of data in HDF5 format.

This function handles all data types (epochs, raws, source data, population codes) and saves them in a consistent HDF5 format.

Parameters:

  • data (np.ndarray, mne.Epochs, mne.io.Raw, or dict) – Data to save. Can be: - np.ndarray: Source space data with shape (n_epochs, n_sources, n_times) - mne.Epochs: Epoched MEG/EEG data - mne.io.Raw: Raw MEG/EEG data (for annotated raws) - Dict[str, np.ndarray]: Population codes dictionary

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • data_type (str, optional) – Type of data (‘epochs’, ‘annotated’, ‘source’, ‘population_codes’) (default: ‘epochs’)

  • metadata (pd.DataFrame, optional) – Metadata for each epoch

  • times (np.ndarray, optional) – Time points array in seconds

  • event_type (str, optional) – Event type (‘saccade’, ‘fixation’, etc.) (default: ‘general’)

  • blocks (list of int, optional) – List of blocks processed

  • rois (list of str, optional) – List of ROI names (for population codes)

  • sampling_rate (int, optional) – Sampling rate in Hz (default: 500)

  • filter_params (dict, optional) – Filter parameters with ‘l_freq’ and ‘h_freq’ keys

  • data_path (str, optional) – Path to data directory

  • hemi (str, optional) – Hemisphere processed (‘lh’, ‘rh’, ‘both’) (default: ‘both’)

  • compression (str, optional) – HDF5 compression method (default: ‘gzip’)

  • **kwargs – Additional parameters

Returns:

Path to saved HDF5 file

Return type:

str

pyavs.io.write.save_annotated_raw(raw: mne.io.Raw, subject_id: int, session: int, data_path: str | None = None, suffix: str = 'annotated', recording_type: str | None = None, **kwargs) → str[source]

Save annotated Raw data in HDF5 format.

Parameters:

  • raw (mne.io.Raw) – Annotated Raw object to save

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • data_path (str, optional) – Path to data directory

  • suffix (str, optional) – File suffix (default: ‘annotated’)

  • recording_type (str, optional) – Recording type to include in filename (‘scene’, ‘microphone’, ‘caption’)

  • **kwargs – Additional parameters for save_data_h5

Returns:

Path to saved file

Return type:

str

pyavs.io.write.save_source_data(data: ndarray | mne.Epochs | Dict[str, ndarray], subject_id: int, session: int, data_type: str = 'source', metadata: DataFrame | None = None, data_path: str | None = None, **kwargs) → str[source]

Save source space data in HDF5 format.

Parameters:

  • data (np.ndarray, mne.Epochs, or dict) – Source space data to save

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • data_type (str, optional) – Type of data being saved (default: ‘source’)

  • metadata (pd.DataFrame, optional) – Metadata for each epoch

  • data_path (str, optional) – Path to data directory

  • **kwargs – Additional parameters

Returns:

Path to saved file

Return type:

str

pyavs.io.write.save_epochs(epochs: mne.Epochs, subject_id: int, session: int, event_type: str = 'epochs', data_path: str | None = None, **kwargs) → str[source]

Save MNE Epochs in HDF5 format.

Parameters:

  • epochs (mne.Epochs) – Epoched data to save

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • event_type (str, optional) – Event type (default: ‘epochs’)

  • data_path (str, optional) – Path to data directory

  • **kwargs – Additional parameters

Returns:

Path to saved file

Return type:

str

pyavs.io.write.save_population_codes_h5(population_codes: Dict[str, ndarray], metadata: DataFrame, subject_id: int, session: int, event_type: str = 'saccade', blocks: List[int] | None = None, times: ndarray | None = None, rois: List[str] | None = None, random_epochs: ndarray | None = None, sampling_rate: int = 500, filter_params: Dict[str, float] | None = None, data_path: str | None = None, hemi: str = 'both', compression: str = 'gzip', data_type: str = 'population_codes', chunk_epochs: int | None = 1, **kwargs) → str[source]

Save population codes to HDF5 file in standardized format.

This is the core saving function that all other save functions ultimately use. It maintains compatibility with the original analysis pipelines.

Parameters:

  • population_codes (dict) – Dictionary where keys are ROI names and values are data arrays with shape (n_epochs, n_sources, n_timepoints)

  • metadata (pd.DataFrame) – Metadata for each epoch

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • event_type (str, optional) – Event type (‘saccade’, ‘fixation’, etc.) (default: ‘saccade’)

  • blocks (list of int, optional) – List of blocks processed (default: None)

  • times (np.ndarray, optional) – Time points array in seconds (default: None)

  • rois (list of str, optional) – List of ROI names (default: None, will use population_codes.keys())

  • random_epochs (np.ndarray, optional) – Indices of randomly selected epochs (default: None)

  • sampling_rate (int, optional) – Sampling rate in Hz (default: 500)

  • filter_params (dict, optional) – Filter parameters with ‘l_freq’ and ‘h_freq’ keys (default: None)

  • data_path (str, optional) – Path to data directory (default: None, uses configured path)

  • hemi (str, optional) – Hemisphere processed (‘lh’, ‘rh’, ‘both’) (default: ‘both’)

  • compression (str, optional) – HDF5 compression method (default: ‘gzip’)

  • data_type (str, optional) – Type of data being saved (default: ‘population_codes’)

  • chunk_epochs (int or None, optional) – Number of epochs per HDF5 chunk (default: 1 – one chunk spans a full epoch’s channel x time extent). Without this, h5py auto-chunks and fragments the channel/time axes too (observed on shipped data: a (2668, 204, 651) array chunked at (84, 13, 41)), so selecting an arbitrary subset of epochs – e.g. a content-filtered query across subjects – ends up touching most of the file’s chunks regardless of how few epochs are wanted. chunk_epochs=1 makes an N-epoch selection cost close to N chunk reads. Pass None to fall back to h5py’s auto-chunking (the previous, unchunked-by-epoch behavior). Larger values trade some of that partial-read benefit for a better gzip ratio (compression sees more redundancy per chunk).

  • **kwargs – Additional parameters

Returns:

Path to saved HDF5 file

Return type:

str

Notes

This function creates standardized HDF5 files for neuroscience data analysis.

pyavs.io.write.save_metadata_csv(metadata: DataFrame, subject_id: int, session: int, event_type: str, data_path: str | None = None) → str[source]

Save epochs metadata alongside the epochs HDF5 file.

Despite the historical name, this writes Parquet (sub-01_ses-01_fixation_metadata.parquet), which is the format the public release ships. Read it back with pyavs.io.read.load_metadata_csv().

Parameters:

  • metadata (pd.DataFrame) – Epochs metadata to save

  • subject_id (int) – Subject ID

  • session (int) – Session number

  • event_type (str) – Event type (e.g., ‘fixation’, ‘saccade’); a trailing ‘_scene’ is stripped for the filename.

  • data_path (str, optional) – Path to the avs-public root

Returns:

Path to the saved Parquet file

Return type:

str

Config I/O and Reproducibility

Configuration I/O utilities for loading configs from saved population codes.

pyavs.io.config_io.load_config_from_population_codes(data_path: str, parameter_signature: str) → ConfigManager | None[source]

Load the configuration that was used to generate population codes.

Parameters:

  • data_path (str) – Path to the dataset

  • parameter_signature (str) – Parameter signature of the population codes

Returns:

Configuration manager if found, None otherwise

Return type:

ConfigManager or None

pyavs.io.config_io.find_configs_for_subject(data_path: str, subject_id: int) → dict[source]

Find all configuration files for a given subject.

Parameters:

  • data_path (str) – Path to the dataset

  • subject_id (int) – Subject ID

Returns:

Dictionary mapping parameter signatures to config file paths

Return type:

dict

pyavs.io.config_io.list_available_configs(data_path: str) → dict[source]

List all available configurations in the derivatives directory.

Parameters:

data_path (str) – Path to the dataset

Returns:

Dictionary with parameter signatures as keys and config info as values

Return type:

dict

pyavs.io.config_io.reproduce_analysis_from_config(config_file: str | Path, subject_id: int | None = None, sessions: list | None = None) → ConfigManager[source]

Load a configuration and optionally override subject/session parameters.

This is useful for reproducing an analysis with the same parameters but for different subjects or sessions.

Parameters:

  • config_file (str or Path) – Path to configuration file

  • subject_id (int, optional) – Override subject ID

  • sessions (list, optional) – Override sessions

Returns:

Configuration ready for analysis

Return type:

ConfigManager