Utilities (pyavs.utils)¶

The utils module provides configuration, validation, and path management utilities.

Configuration Management¶

Configuration management for pyAVS package.

This module provides functions for managing data paths and package configuration.

Historically this module kept its own separate global config dict, independent of the unified pyavs.config (PyAVSConfig/ConfigManager) system. That meant pyavs.set_data_path() (which writes to the unified system) and get_data_path() used internally throughout the core library (which read from this module’s old separate dict) never agreed with each other unless callers passed data_path= explicitly everywhere. The functions below now proxy to the unified global config so both entry points read/write the same underlying store.

pyavs.utils.config.set_data_path(path: str) → None[source]¶

Set the base data path for the AVS dataset.

Parameters:: path (str) – Path to the AVS BIDS dataset directory

pyavs.utils.config.get_data_path() → str | None[source]¶

Get the current data path.

Returns:: Current data path, or None if not set
Return type:: str or None

pyavs.utils.config.setup_data_directory(path: str | None = None) → str[source]¶

Set up data directory with automatic detection.

Parameters:: path (str, optional) – Data path. If None, uses the unified config’s auto-detect cascade (PYAVS_DATA_PATH env var -> ~/.config/pyavs/config.json).
Returns:: Data path that was set
Return type:: str
Raises:: FileNotFoundError – If no path is given and auto-detection fails.

pyavs.utils.config.get_config() → Dict[str, Any][source]¶

Get current configuration as a plain dict.

Deprecated in favor of pyavs.config.get_config(), which returns the richer ConfigManager/PyAVSConfig object. Kept as a thin shim for backward compatibility.

Returns:: Current configuration, with data_path reflecting the live unified config.
Return type:: dict

pyavs.utils.config.update_config(**kwargs) → None[source]¶

Update auxiliary configuration parameters (server, output_prefix, cache_dir, verbose). Use set_data_path() to update the data path itself.

Parameters:: **kwargs – Configuration parameters to update

pyavs.utils.config.get_derivatives_root() → str | None[source]¶

Get the pyAVS derivatives write root.

Defaults to <data_path>/derivatives/pyavs, overridable via the PYAVS_DERIVATIVES_PATH environment variable or the config’s derivatives_path field — useful when the dataset copy is read-only.

Returns:: Derivatives root, or None if no data path is configured.
Return type:: str or None

pyavs.utils.config.save_config(filepath: str) → None[source]¶

Save current configuration to JSON file.

Deprecated in favor of pyavs.config.save_config().

Parameters:: filepath (str) – Path to save configuration file

pyavs.utils.config.load_config(filepath: str) → None[source]¶

Load configuration from JSON file.

Deprecated in favor of pyavs.config.load_config().

Parameters:: filepath (str) – Path to configuration file

Path Utilities¶

Path utilities for pyAVS package.

Naming helpers (sub-XX/ses-XX labels, session letters, native subject-session IDs) and every dataset path now live in pyavs.layout. The session-naming functions below are thin aliases re-exported from there so existing imports keep working; new code should prefer pyavs.layout.

The legacy path builders that used to live here — get_bids_path and get_legacy_paths — are gone. get_bids_path built sub-01/ses-01/meg/sub-01_ses-01_task-avs_run-01_raw.fif, a filename that exists in neither the public release nor the internal tree (the release preserves native scanner names), so every “try BIDS first” branch built on it always missed. get_legacy_paths addressed the internal results/as01_01/ tree, which pyAVS no longer supports. Use pyavs.layout.Layout instead.

pyavs.utils.paths.convert_session_to_letter(session: int) → str¶

Convert a session number to its MEG-filename letter.

The MEG naming convention at MPI represents sessions as letters (1, 2, 3 → a, b, c).

Parameters:: session (int) – Session number (1-based).
Returns:: Session letter.
Return type:: str
Raises:: ValueError – If session is outside 1-26.

pyavs.utils.paths.convert_letter_to_session(letter: str) → int¶

Convert a session letter back to a session number.

Parameters:: letter (str) – Session letter.
Returns:: Session number (1-based).
Return type:: int
Raises:: ValueError – If letter is not a lowercase ASCII letter.

pyavs.utils.paths.get_subject_session_id(subject_id: int, session: int, prefix: str = 'as') → str¶

Build the native subject-session ID used in raw MEG/ET filenames.

Parameters:

subject_id (int) – Subject ID.
session (int) – Session number (1-based).
prefix (str, optional) – Filename prefix (default: 'as').

Returns:

e.g. 'as01a' for subject 1, session 1; 'as01j' for session 10.

Return type:

str

pyavs.utils.paths.get_derivatives_path(data_path: str, subject_id: int, session: int | None = None, datatype: str | None = None) → str[source]¶

Get the pyAVS derivatives directory for a subject (and optionally session).

Parameters:

data_path (str) – avs-public dataset root.
subject_id (int) – Subject ID.
session (int, optional) – Session number. Omitted from the path when None.
datatype (str, optional) – Datatype subdirectory (‘meg’, ‘epochs’, ‘eyetrack’, …).

Returns:

e.g. <root>/derivatives/pyavs/sub-01/ses-01/meg

Return type:

str

pyavs.utils.paths.get_max_blocks(session: int) → int[source]¶

Get maximum number of blocks for a given session.

Parameters:: session (int) – Session number
Returns:: Maximum number of blocks
Return type:: int

pyavs.utils.paths.get_default_subjects_dir() → str[source]¶

Get the FreeSurfer subjects directory.

Checks in order:

The SUBJECTS_DIR environment variable, if it points somewhere that exists.
<data_path>/derivatives/freesurfer from the configured AVS root (see pyavs.configure()), which the public release ships as a ready-to-use MNE SUBJECTS_DIR.

Returns:: Path to the subjects directory.
Return type:: str
Raises:: ValueError – If SUBJECTS_DIR is unset and no AVS data path is configured.

pyavs.utils.paths.get_glasser_rois(area: str) → list[source]¶

Get list of Glasser ROI names for specified area.

Parameters:: area (str) – Area name (‘all’, ‘high_visual’, ‘early_visual’, ‘intermediate_visual’)
Returns:: List of ROI names
Return type:: list

Data Validation¶

Validation utilities for pyAVS package.

This module provides functions for validating data integrity and input parameters.

pyavs.utils.validation.validate_subject_id(subject_id: int) → int[source]¶

Validate subject ID.

Parameters:: subject_id (int) – Subject ID to validate
Raises:: ValueError – If subject ID is invalid

pyavs.utils.validation.validate_session(session: int) → int[source]¶

Validate session number.

Parameters:: session (int) – Session number to validate
Raises:: ValueError – If session number is invalid

pyavs.utils.validation.validate_blocks(blocks: int | List[int] | None, session: int) → List[int][source]¶

Validate and normalize block specification.

Parameters:

blocks (int, list of int, or None) – Block number(s) to validate
session (int) – Session number (for determining max blocks)

Returns:

Validated list of block numbers

Return type:

list of int

Raises:

ValueError – If blocks are invalid

pyavs.utils.validation.validate_data_integrity(data_path: str, subject_id: int, session: int, blocks: List[int] | None = None) → Dict[str, Any][source]¶

Validate data integrity for a subject/session.

Parameters:

data_path (str) – Path to the avs-public dataset root.
subject_id (int) – Subject ID
session (int) – Session number
blocks (list of int, optional) – Block numbers to check

Returns:

Validation results with availability status

Return type:

dict

pyavs.utils.validation.validate_eye_events_dataframe(events_df: DataFrame) → List[str][source]¶

Validate eye events dataframe structure.

Parameters:: events_df (pd.DataFrame) – Eye events dataframe to validate
Returns:: List of validation warnings/errors
Return type:: list of str

pyavs.utils.validation.validate_experiment_log(explog_df: DataFrame) → List[str][source]¶

Validate experiment log dataframe structure.

Parameters:: explog_df (pd.DataFrame) – Experiment log dataframe to validate
Returns:: List of validation warnings/errors
Return type:: list of str

Derivatives Path Conventions¶

BIDS derivatives path/naming conventions – see Dataset Structure.

Unified derivatives directory utilities for pyAVS package.

This module provides standardized functions for creating BIDS-compliant derivatives directory structures and paths for all pyAVS data products.

class pyavs.utils.derivatives.DerivativesManager(data_path: str | None = None)[source]¶

Bases: object

Unified manager for all derivatives directory operations.

Ensures a consistent structure, matching the public release: derivatives/pyavs/sub-{subject_id:02d}/ses-{session:02d}/{datatype}/.

Products keyed by a parameter signature rather than by session (filters/, population_codes/) stay directly under the derivatives root, since they are not per-session artifacts.

__init__(data_path: str | None = None)[source]¶

Initialize derivatives manager.

Parameters:: data_path (str, optional) – Base data path. If None, uses configured data path.

get_preprocessed_path(subject_id: int, session: int, create: bool = False) → Path[source]¶

Get the path for preprocessed (Maxwell-filtered) MEG data.

Structure: derivatives/pyavs/sub-XX/ses-XX/meg/

Parameters:

subject_id (int) – Subject ID
session (int) – Session number
create (bool, optional) – Create the directory. Default False — resolving a path must not write to the dataset, which may be a read-only release copy.

Returns:

Preprocessed data path

Return type:

Path

get_population_codes_path(parameter_signature: str, subject_id: int, session: int, create: bool = False) → Path[source]¶

Get BIDS-compliant path for population codes.

Structure: derivatives/pyavs/population_codes/{signature}/sub-XX/ses-XX/

Parameters:

parameter_signature (str) – Unique parameter signature
subject_id (int) – Subject ID
session (int) – Session number

Returns:

BIDS-compliant population codes path

Return type:

Path

get_source_reconstruction_path(subject_id: int, session: int, method: str = 'beamformer', atlas: str = 'glasser', orientation: str = 'normal', hemisphere: str = 'both', filter_spec: str = 'filter_0.2_200', create: bool = False) → Path[source]¶

Get the path for source reconstruction data.

Structure: derivatives/pyavs/sub-XX/ses-XX/source/{method}/{atlas}/

Parameters:

subject_id (int) – Subject ID
session (int) – Session number
method (str) – Reconstruction method (default: ‘beamformer’)
atlas (str) – Brain atlas (default: ‘glasser’)
orientation (str) – Source orientation (default: ‘normal’)
hemisphere (str) – Hemisphere (default: ‘both’)
filter_spec (str) – Filter specification (default: ‘filter_0.2_200’)

Returns:

BIDS-compliant source reconstruction path

Return type:

Path

get_noise_covariance_path(subject_id: int, create: bool = False) → Path[source]¶

Get the path for noise covariance matrices.

Structure: derivatives/pyavs/sub-XX/source/noise_covariance/

Noise covariance is estimated from empty-room recordings pooled across sessions, so it lives at the subject level rather than under a session.

Parameters:

subject_id (int) – Subject ID
create (bool) – Whether to create the directory (default: False)

Returns:

Noise covariance directory

Return type:

Path

get_filters_path(parameter_signature: str, create: bool = False) → Path[source]¶

Get BIDS-compliant path for beamformer filters.

Structure: derivatives/pyavs/filters/{signature}/

Parameters:: parameter_signature (str) – Unique parameter signature
Returns:: BIDS-compliant filters path
Return type:: Path

get_epochs_path(subject_id: int, session: int, event_type: str = 'saccade', create: bool = False) → Path[source]¶

Get the path for epoched data.

Structure: derivatives/pyavs/sub-XX/ses-XX/epochs/

Parameters:

subject_id (int) – Subject ID
session (int) – Session number
event_type (str) – Event type (default: ‘saccade’)

Returns:

BIDS-compliant epochs path

Return type:

Path

create_bids_filename(subject_id: int, session: int, task: str = 'avs', datatype: str = 'meg', suffix: str = 'raw-sss', extension: str = '.fif', run: int | None = None, recording: str | None = None, **entities) → str[source]¶

Create BIDS-compliant filename.

Parameters:

subject_id (int) – Subject ID
session (int) – Session number
task (str) – Task name (default: ‘avs’)
datatype (str) – Data type (default: ‘meg’)
suffix (str) – File suffix (default: ‘raw-sss’)
extension (str) – File extension (default: ‘.fif’)
run (int, optional) – Run/block number
recording (str, optional) – Recording type (for empty room)
**entities – Additional BIDS entities

Returns:

BIDS-compliant filename

Return type:

str

generate_parameter_signature(**params) → str[source]¶

Generate unique parameter signature for consistent naming.

Parameters:: **params – Parameter dictionary
Returns:: Unique parameter signature
Return type:: str

cleanup_legacy_structure(dry_run: bool = True) → List[str][source]¶

Identify legacy non-BIDS directory structures for cleanup.

Parameters:: dry_run (bool) – If True, only identify without moving (default: True)
Returns:: List of legacy paths identified
Return type:: List[str]

pyavs.utils.derivatives.get_derivatives_manager(data_path: str | None = None) → DerivativesManager[source]¶: Get derivatives manager instance.

pyavs.utils.derivatives.get_bids_preprocessed_path(subject_id: int, session: int, data_path: str | None = None, create: bool = False) → Path[source]¶: Get the preprocessed MEG data directory. Pass create=True to make it.

pyavs.utils.derivatives.get_bids_population_codes_path(parameter_signature: str, subject_id: int, session: int, data_path: str | None = None, create: bool = False) → Path[source]¶: Get the population codes directory. Pass create=True to make it.

pyavs.utils.derivatives.create_bids_meg_filename(subject_id: int, session: int, run: int | None = None, recording: str | None = None, suffix: str = 'raw-sss', data_path: str | None = None) → str[source]¶: Create BIDS-compliant MEG filename.

pyavs.utils.derivatives.generate_parameter_signature(**params) → str[source]¶: Generate parameter signature for consistent naming.

Logging¶

Centralized logging configuration for pyAVS package.

This module provides a unified logging system for all pyAVS components, allowing for consistent log formatting, levels, and output handling.

class pyavs.utils.logging.ColoredFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)[source]¶

Bases: Formatter

Colored formatter for console output.

COLORS = {'CRITICAL': '\x1b[35m', 'DEBUG': '\x1b[36m', 'ERROR': '\x1b[31m', 'INFO': '\x1b[32m', 'RESET': '\x1b[0m', 'WARNING': '\x1b[33m'}¶

format(record)[source]¶

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

class pyavs.utils.logging.PyAVSLogger[source]¶

Bases: object

Centralized logger for pyAVS package.

Provides consistent logging across all modules with configurable output levels, formats, and destinations.

classmethod configure(level: str | int = 'INFO', console: bool = True, file_path: str | Path | None = None, format_string: str | None = None, date_format: str | None = None, use_colors: bool = True, max_file_size: int = 10485760, backup_count: int = 5) → None[source]¶

Configure the global logging system for pyAVS.

Parameters:

level (str or int, optional) – Logging level (‘DEBUG’, ‘INFO’, ‘WARNING’, ‘ERROR’, ‘CRITICAL’) (default: ‘INFO’)
console (bool, optional) – Whether to output to console (default: True)
file_path (str or Path, optional) – Path to log file. If None, no file logging (default: None)
format_string (str, optional) – Custom format string for log messages
date_format (str, optional) – Custom date format for timestamps
use_colors (bool, optional) – Whether to use colored output in console (default: True)
max_file_size (int, optional) – Maximum log file size in bytes before rotation (default: 10MB)
backup_count (int, optional) – Number of backup files to keep during rotation (default: 5)

classmethod get_logger(name: str) → Logger[source]¶

Get a logger instance for a specific module.

Parameters:: name (str) – Logger name, typically the module name
Returns:: Logger instance
Return type:: logging.Logger

classmethod set_level(level: str | int, logger_name: str | None = None) → None[source]¶

Set logging level for specific logger or all loggers.

Parameters:

level (str or int) – Logging level
logger_name (str, optional) – Specific logger name. If None, applies to root pyavs logger

classmethod add_file_handler(file_path: str | Path, level: str | int = 'INFO') → None[source]¶

Add an additional file handler to the logging system.

Parameters:

file_path (str or Path) – Path to additional log file
level (str or int, optional) – Logging level for this handler (default: ‘INFO’)

pyavs.utils.logging.get_logger(name: str) → Logger[source]¶

Get a logger instance (convenience function).

Parameters:: name (str) – Logger name
Returns:: Logger instance
Return type:: logging.Logger

pyavs.utils.logging.configure_logging(**kwargs) → None[source]¶

Configure logging system (convenience function).

Parameters:: **kwargs – Arguments passed to PyAVSLogger.configure()

pyavs.utils.logging.set_log_level(level: str | int, logger_name: str | None = None) → None[source]¶

Set logging level (convenience function).

Parameters:

level (str or int) – Logging level
logger_name (str, optional) – Specific logger name

class pyavs.utils.logging.temporary_log_level(level: str | int, logger_name: str | None = None)[source]¶

Bases: object

Context manager to temporarily change log level.

Usage:

with temporary_log_level(‘DEBUG’):: # Debug logging enabled logger.debug(“This will be shown”)

__init__(level: str | int, logger_name: str | None = None)[source]¶

pyavs.utils.logging.log_processing_start(logger: Logger, operation: str, details: Dict[str, Any] | None = None) → None[source]¶

Log the start of a processing operation.

Parameters:

logger (logging.Logger) – Logger instance
operation (str) – Description of the operation
details (dict, optional) – Additional details to log

pyavs.utils.logging.log_processing_end(logger: Logger, operation: str, success: bool = True, duration: float | None = None, details: Dict[str, Any] | None = None) → None[source]¶

Log the end of a processing operation.

Parameters:

logger (logging.Logger) – Logger instance
operation (str) – Description of the operation
success (bool, optional) – Whether operation was successful (default: True)
duration (float, optional) – Duration in seconds
details (dict, optional) – Additional details to log

Eye-Tracking Utilities¶

Saccade/fixation matching and related eye-tracking helpers.

Eye tracking utilities for pyAVS package.

This module provides utilities for processing and analyzing eye tracking data, including functions for matching saccades to fixations and extracting temporal relationships between eye movement events.

pyavs.utils.eye_tracking.match_saccades_to_fixations(saccades_meta_df: DataFrame, fixations_meta_df: DataFrame, saccade_type: Literal['pre-saccade', 'post-saccade'] = 'pre-saccade') → DataFrame[source]¶

Match saccades to fixations based on temporal adjacency.

This function identifies saccade-fixation pairs by analyzing the temporal sequence of events within each scene. It matches events that occur consecutively with zero time gap between them.

Parameters:

saccades_meta_df (pd.DataFrame) – Metadata for saccades. Must contain columns: ‘sceneID’, ‘type’, ‘start_time’, ‘end_time’, ‘duration’
fixations_meta_df (pd.DataFrame) – Metadata for fixations. Must contain columns: ‘sceneID’, ‘type’, ‘start_time’, ‘end_time’, ‘duration’, ‘fix_sequence’
saccade_type (Literal["pre-saccade", "post-saccade"], default="pre-saccade") – Type of matching to perform: - “pre-saccade”: Match saccade -> fixation sequences - “post-saccade”: Match fixation -> saccade sequences

Returns:

Matched saccades with associated fixation information. Includes all original saccade columns plus: - ‘associated_fix_sequence’: Sequence number of matched fixation - ‘associated_fix_start_time’: Start time of matched fixation - ‘associated_fixation_duration’: Duration of matched fixation

Return type:

pd.DataFrame

Notes

Only pairs with exactly 0 time difference between consecutive events are included (i.e., saccade.end_time == fixation.start_time for pre-saccade, or fixation.end_time == saccade.start_time for post-saccade).

Examples

>>> # Match saccades to subsequent fixations
>>> matched_df = match_saccades_to_fixations(
...     saccades_df, fixations_df, saccade_type="pre-saccade"
... )
>>>
>>> # Access matched fixation durations
>>> fixation_durations = matched_df['associated_fixation_duration']