Conversion to BIDS

Reading EyeLink, Tobii, GazePoint and webcam recordings, normalizing their behavioral tables, and writing the raw BIDS dataset. The original vendor files are preserved verbatim under the dataset's archival sourcedata/.

bids_formatting

The dataset-level entry points: dataset_to_bids converts a folder of recordings, and compute_derivatives_for_dataset computes derivatives from the resulting raw dataset.

Raw-to-BIDS conversion and Polars-native derivative orchestration.

compute_derivatives_for_dataset(bids_dataset_folder, dataset_format, detection_algorithm='remodnav', num_processes=1, force_best_eye=True, overwrite=False, behavioral_columns=None, **kwargs)

Compute canonical BIDS derivatives for every raw BIDS session.

Processing is serial by default to avoid process-startup and serialization costs for small datasets. Set num_processes above one for large collections of independent sessions.

Parameters:

Name	Type	Description	Default
bids_dataset_folder	str or Path	Root of the raw BIDS dataset.	required
dataset_format	('eyelink', 'gaze', 'tobii', 'webgazer')	Source format represented by the dataset.	"eyelink"
detection_algorithm	('eyelink', 'engbert', 'remodnav')	Event source or detector used for every session.	"eyelink"
num_processes	int	Number of independent session workers.	1
force_best_eye	bool	Whether to retain only the best validated EyeLink eye.	True
overwrite	bool	Whether to replace existing derivative sessions.	False
behavioral_columns	sequence of str	Behavioral fields propagated into trial-level derivative tables.	None
**kwargs	object	Detector, segmentation, quality, and metadata options.	{}

Returns:

Type	Description
Path	Root of the generated BIDS Derivatives dataset.

Raises:

Type	Description
TypeError	If num_processes is not an integer.
ValueError	If num_processes is less than one.

Source code in pyxations/bids_formatting.py

def compute_derivatives_for_dataset(
    bids_dataset_folder,
    dataset_format,
    detection_algorithm="remodnav",
    num_processes: int = 1,
    force_best_eye=True,
    overwrite=False,
    behavioral_columns=None,
    **kwargs,
):
    """Compute canonical BIDS derivatives for every raw BIDS session.

    Processing is serial by default to avoid process-startup and serialization
    costs for small datasets. Set ``num_processes`` above one for large
    collections of independent sessions.

    Parameters
    ----------
    bids_dataset_folder : str or pathlib.Path
        Root of the raw BIDS dataset.
    dataset_format : {"eyelink", "gaze", "tobii", "webgazer"}
        Source format represented by the dataset.
    detection_algorithm : {"eyelink", "engbert", "remodnav"}, default "remodnav"
        Event source or detector used for every session.
    num_processes : int, default 1
        Number of independent session workers.
    force_best_eye : bool, default True
        Whether to retain only the best validated EyeLink eye.
    overwrite : bool, default False
        Whether to replace existing derivative sessions.
    behavioral_columns : sequence of str, optional
        Behavioral fields propagated into trial-level derivative tables.
    **kwargs : object
        Detector, segmentation, quality, and metadata options.

    Returns
    -------
    pathlib.Path
        Root of the generated BIDS Derivatives dataset.

    Raises
    ------
    TypeError
        If ``num_processes`` is not an integer.
    ValueError
        If ``num_processes`` is less than one.
    """

    bids_dataset_folder = Path(bids_dataset_folder)
    if isinstance(num_processes, bool) or not isinstance(num_processes, int):
        raise TypeError("num_processes must be an integer")
    if num_processes < 1:
        raise ValueError("num_processes must be at least 1")
    derivatives_folder = Path(f"{bids_dataset_folder}_derivatives")
    initialize_bids_derivative(bids_dataset_folder, derivatives_folder)

    start_times = kwargs.pop("start_times", None)
    end_times = kwargs.pop("end_times", None)
    if behavioral_columns is not None:
        kwargs["behavioral_columns"] = behavioral_columns

    participants = read_tsv(
        bids_dataset_folder / "participants.tsv", has_header=True
    ).with_columns(
        pl.col("subject_id").cast(pl.String).str.pad_start(4, "0"),
        pl.col("old_subject_id").cast(pl.String),
    )
    subject_lookup = dict(
        participants.select("subject_id", "old_subject_id").iter_rows()
    )

    jobs = []
    for subject in sorted(bids_dataset_folder.glob("sub-*")):
        if not subject.is_dir():
            continue
        subject_name = subject_lookup[subject.name[4:]]
        for session in sorted(subject.glob("ses-*")):
            if not session.is_dir():
                continue
            session_name = session.name[4:]
            session_kwargs = dict(kwargs)
            if (
                start_times
                and subject_name in start_times
                and session_name in start_times[subject_name]
            ):
                session_kwargs["start_times"] = start_times[subject_name][session_name]
            if (
                end_times
                and subject_name in end_times
                and session_name in end_times[subject_name]
            ):
                session_kwargs["end_times"] = end_times[subject_name][session_name]
            jobs.append(
                (
                    session,
                    dataset_format,
                    detection_algorithm,
                    derivatives_folder / subject.name / session.name,
                    force_best_eye,
                    overwrite,
                    session_kwargs,
                )
            )

    if num_processes == 1:
        for (
            source,
            format_name,
            algorithm,
            destination,
            choose_eye,
            replace,
            options,
        ) in jobs:
            process_session(
                source,
                format_name,
                algorithm,
                destination,
                choose_eye,
                replace,
                **options,
            )
    else:
        with ProcessPoolExecutor(max_workers=num_processes) as executor:
            futures = [
                executor.submit(
                    process_session,
                    source,
                    format_name,
                    algorithm,
                    destination,
                    choose_eye,
                    replace,
                    **options,
                )
                for source, format_name, algorithm, destination, choose_eye, replace, options in jobs
            ]
            for future in futures:
                future.result()
    return derivatives_folder

dataset_to_bids(target_folder_path, files_folder_path, dataset_name, session_substrings=1, format_name='eyelink', *, task_name='eyetracking', authors=None, behavioral_column_map=None, overwrite=False)

Convert vendor recordings and associated behavior to raw BIDS.

PsychoPy logs are used when no behavioral CSV or TSV exists for the recording. behavioral_column_map can map fields from any behavioral source onto experiment-level names without changing the archived source.

Parameters:

Name	Type	Description	Default
target_folder_path	str or Path	Parent directory in which to create the dataset.	required
files_folder_path	str or Path	Directory containing vendor recordings and associated behavior.	required
dataset_name	str	Output directory and BIDS dataset name.	required
session_substrings	int	Number of underscore-separated filename tokens used for the session.	1
format_name	('eyelink', 'gaze', 'tobii', 'webgazer')	Vendor reader used for the source recordings.	"eyelink"
task_name	str	Fallback BIDS task label.	"eyetracking"
authors	sequence of str	Dataset authors stored in BIDS metadata.	None
behavioral_column_map	mapping of str to str	Mapping from source behavioral fields to experiment concepts.	None
overwrite	bool	Whether to replace an existing non-empty output dataset.	False

Returns:

Type	Description
Path	Root of the generated raw BIDS dataset.

Source code in pyxations/bids_formatting.py

def dataset_to_bids(
    target_folder_path,
    files_folder_path,
    dataset_name,
    session_substrings=1,
    format_name="eyelink",
    *,
    task_name="eyetracking",
    authors=None,
    behavioral_column_map=None,
    overwrite=False,
):
    """Convert vendor recordings and associated behavior to raw BIDS.

    PsychoPy logs are used when no behavioral CSV or TSV exists for the
    recording. ``behavioral_column_map`` can map fields from any behavioral
    source onto experiment-level names without changing the archived source.

    Parameters
    ----------
    target_folder_path : str or pathlib.Path
        Parent directory in which to create the dataset.
    files_folder_path : str or pathlib.Path
        Directory containing vendor recordings and associated behavior.
    dataset_name : str
        Output directory and BIDS dataset name.
    session_substrings : int, default 1
        Number of underscore-separated filename tokens used for the session.
    format_name : {"eyelink", "gaze", "tobii", "webgazer"}
        Vendor reader used for the source recordings.
    task_name : str, default "eyetracking"
        Fallback BIDS task label.
    authors : sequence of str, optional
        Dataset authors stored in BIDS metadata.
    behavioral_column_map : mapping of str to str, optional
        Mapping from source behavioral fields to experiment concepts.
    overwrite : bool, default False
        Whether to replace an existing non-empty output dataset.

    Returns
    -------
    pathlib.Path
        Root of the generated raw BIDS dataset.
    """

    return write_bids_dataset(
        target_folder_path,
        files_folder_path,
        dataset_name,
        session_substrings=session_substrings,
        format_name=format_name,
        task_name=task_name,
        authors=authors,
        behavioral_column_map=behavioral_column_map,
        overwrite=overwrite,
    )

process_bids_session(raw_session_path, dataset_format, detection_algorithm, session_folder_path, force_best_eye, **kwargs)

Compute and write one derivative session from normalized raw BIDS.

Parameters:

Name	Type	Description	Default
raw_session_path	str or Path	Raw BIDS session directory to process.	required
dataset_format	('eyelink', 'gaze', 'tobii', 'webgazer')	Source format recorded in the raw dataset.	"eyelink"
detection_algorithm	('eyelink', 'engbert', 'remodnav')	Event source or detector used for the derivative.	"eyelink"
session_folder_path	str or Path	Destination derivative session directory.	required
force_best_eye	bool	Whether to retain only the eye with the best EyeLink validation.	required
**kwargs	object	Detector, trial-segmentation, quality, and behavioral options.	{}

Source code in pyxations/bids_formatting.py

def process_bids_session(
    raw_session_path,
    dataset_format,
    detection_algorithm,
    session_folder_path,
    force_best_eye,
    **kwargs,
):
    """Compute and write one derivative session from normalized raw BIDS.

    Parameters
    ----------
    raw_session_path : str or pathlib.Path
        Raw BIDS session directory to process.
    dataset_format : {"eyelink", "gaze", "tobii", "webgazer"}
        Source format recorded in the raw dataset.
    detection_algorithm : {"eyelink", "engbert", "remodnav"}
        Event source or detector used for the derivative.
    session_folder_path : str or pathlib.Path
        Destination derivative session directory.
    force_best_eye : bool
        Whether to retain only the eye with the best EyeLink validation.
    **kwargs : object
        Detector, trial-segmentation, quality, and behavioral options.
    """

    raw = read_raw_bids_session(raw_session_path)
    session_folder_path = Path(session_folder_path)
    session_folder_path.mkdir(parents=True, exist_ok=True)

    messages = raw.messages.clone()
    message_keywords = kwargs.pop("msg_keywords", None)
    if message_keywords and not messages.is_empty() and "message" in messages:
        pattern = "(?i)" + "|".join(re.escape(keyword) for keyword in message_keywords)
        messages = messages.filter(
            pl.col("message").cast(pl.String).str.contains(pattern).fill_null(False)
        )

    samples, fixations, saccades, blinks = _detect_from_bids(
        raw,
        dataset_format=dataset_format,
        detection_algorithm=detection_algorithm,
        session_folder_path=session_folder_path,
        kwargs=kwargs,
    )
    if force_best_eye:
        samples, fixations, blinks, saccades = _choose_best_eye(
            raw, samples, fixations, blinks, saccades
        )

    pre_processing = PreProcessing(
        samples,
        fixations,
        saccades,
        blinks,
        messages,
        session_folder_path,
    )
    pre_processing.set_metadata(
        screen_width=raw.screen_width or kwargs.get("screen_width"),
        screen_height=raw.screen_height or kwargs.get("screen_height"),
    )
    # Quality marking and saccade direction depend only on gaze coordinates,
    # so they apply to every input format rather than to EyeLink alone. Both
    # are skipped when the screen size is unknown, since out-of-screen samples
    # cannot be identified without it.
    bad_parameters = {
        key: kwargs[key]
        for key in (
            "screen_height",
            "screen_width",
            "mark_nan_as_bad",
            "inclusive_bounds",
        )
        if key in kwargs
    }
    screen_is_known = (
        pre_processing.metadata.screen_width is not None
        and pre_processing.metadata.screen_height is not None
    )
    quality_steps: dict = {}
    if screen_is_known:
        quality_steps["bad_samples"] = bad_parameters
    # Saccade direction needs endpoint coordinates, which vendor-reported event
    # tables do not always carry.
    direction_columns = {"xStart", "yStart", "xEnd", "yEnd"}
    direction_step: dict = {}
    if direction_columns.issubset(set(pre_processing.saccades.columns)):
        direction_step["saccades_direction"] = (
            {"tol_deg": kwargs["tol_deg"]} if "tol_deg" in kwargs else {}
        )

    segmentation = _segmentation_recipe(pre_processing, kwargs)
    if segmentation:
        name, parameters = segmentation
        recipe = {**quality_steps, name: parameters, **direction_step}
        pre_processing.process(recipe)
    else:
        # Recordings without synchronisation messages still deserve quality
        # marking and saccade direction; only the segmentation step is skipped.
        _assign_default_trials(pre_processing)
        if quality_steps or direction_step:
            pre_processing.process({**quality_steps, **direction_step})

    behavioral_columns = kwargs.get("behavioral_columns")
    if behavioral_columns and not raw.behavioral_events.is_empty():
        if "trial_number" in raw.behavioral_events:
            metadata = raw.behavioral_events.rename({"trial_number": "trial_index"})
        else:
            metadata = raw.behavioral_events
        if "trial_index" in metadata:
            pre_processing.add_trial_metadata(metadata, behavioral_columns)

    processed = SessionTables(
        samples=pre_processing.samples,
        fixations=pre_processing.fixations,
        saccades=pre_processing.saccades,
        blinks=pre_processing.blinks,
        messages=pre_processing.user_messages,
        calibration=raw.calibration,
        header=raw.header,
        behavioral_events=raw.behavioral_events,
        sampling_frequency=raw.sampling_frequency,
        screen_width=pre_processing.metadata.screen_width,
        screen_height=pre_processing.metadata.screen_height,
    )
    BIDSDerivativeExport().write_session(
        session_folder_path,
        processed,
        detection_algorithm=detection_algorithm,
    )

process_session(raw_session_path, dataset_format, detection_algorithm, session_folder_path, force_best_eye, overwrite, **kwargs)

Process one raw BIDS session unless its requested output already exists.

Parameters:

Name	Type	Description	Default
raw_session_path	str or Path	Raw BIDS session directory.	required
dataset_format	('eyelink', 'gaze', 'tobii', 'webgazer')	Source format recorded in the raw dataset.	"eyelink"
detection_algorithm	('eyelink', 'engbert', 'remodnav')	Event source or detector to use.	"eyelink"
session_folder_path	str or Path	Destination derivative session directory.	required
force_best_eye	bool	Whether to retain only the best validated EyeLink eye.	required
overwrite	bool	Whether to replace an existing derivative session.	required
**kwargs	object	Options forwarded to :func:process_bids_session.	{}

Raises:

Type	Description
ValueError	If dataset_format is unsupported.

Source code in pyxations/bids_formatting.py

def process_session(
    raw_session_path,
    dataset_format,
    detection_algorithm,
    session_folder_path,
    force_best_eye,
    overwrite,
    **kwargs,
):
    """Process one raw BIDS session unless its requested output already exists.

    Parameters
    ----------
    raw_session_path : str or pathlib.Path
        Raw BIDS session directory.
    dataset_format : {"eyelink", "gaze", "tobii", "webgazer"}
        Source format recorded in the raw dataset.
    detection_algorithm : {"eyelink", "engbert", "remodnav"}
        Event source or detector to use.
    session_folder_path : str or pathlib.Path
        Destination derivative session directory.
    force_best_eye : bool
        Whether to retain only the best validated EyeLink eye.
    overwrite : bool
        Whether to replace an existing derivative session.
    **kwargs : object
        Options forwarded to :func:`process_bids_session`.

    Raises
    ------
    ValueError
        If ``dataset_format`` is unsupported.
    """

    session_folder_path = Path(session_folder_path)
    label = bids_label(detection_algorithm.lower(), fallback="pyxations")
    if not overwrite and session_folder_path.exists():
        existing = (session_folder_path / "beh").glob(
            f"*_recording-eye1{label}_physio.tsv.gz"
        )
        if next(existing, None) is not None:
            return
    if dataset_format not in {"eyelink", "webgazer", "tobii", "gaze"}:
        raise ValueError(f"Dataset format {dataset_format} not found.")
    process_bids_session(
        raw_session_path,
        dataset_format,
        detection_algorithm,
        session_folder_path,
        force_best_eye,
        **kwargs,
    )

bids

The raw BIDS reader and writer, plus the wrapper around the official BIDS Validator used to check that generated datasets are valid.

BIDS eye-tracking writing and validation utilities.

The writer targets the eye-tracking additions in BIDS 1.11.1. Vendor files are kept under sourcedata while standardized, per-eye physiological recordings are written to the raw BIDS dataset.

BIDSValidationError

Bases: RuntimeError

Raised when the official BIDS Validator reports an invalid dataset.

Source code in pyxations/bids.py

class BIDSValidationError(RuntimeError):
    """Raised when the official BIDS Validator reports an invalid dataset."""

EyeRecording dataclass

Canonical sample-level recording for one eye.

Source code in pyxations/bids.py

@dataclass
class EyeRecording:
    """Canonical sample-level recording for one eye."""

    samples: pl.DataFrame
    recorded_eye: str
    sampling_frequency: float
    timestamp_unit: str
    coordinate_unit: str
    coordinate_description: str
    pupil_unit: str | None = None
    pupil_description: str | None = None
    manufacturer: str | None = None

    def normalized(self) -> EyeRecording:
        """Validate the recording and return a canonical copy of it.

        Checks that the required sample columns are present, that the recorded
        eye is one of the accepted labels and that the sampling frequency is
        positive, then returns an equivalent recording with a float sampling
        frequency.

        Returns
        -------
        EyeRecording
            A validated copy, ready to be written to BIDS.

        Raises
        ------
        ValueError
            If the ``timestamp``, ``x_coordinate`` or ``y_coordinate`` columns
            are missing, if ``recorded_eye`` is not ``"left"``, ``"right"`` or
            ``"cyclopean"``, or if ``sampling_frequency`` is not positive.
        """
        required = ["timestamp", "x_coordinate", "y_coordinate"]
        missing = [column for column in required if column not in self.samples]
        if missing:
            raise ValueError(f"Eye recording is missing required columns: {missing}")
        if self.recorded_eye not in {"left", "right", "cyclopean"}:
            raise ValueError(f"Unsupported RecordedEye value: {self.recorded_eye}")

        columns = required + (
            ["pupil_size"] if "pupil_size" in self.samples.columns else []
        )
        columns += [column for column in self.samples.columns if column not in columns]
        numeric = required + (
            ["pupil_size"] if "pupil_size" in self.samples.columns else []
        )
        samples = self.samples.select(columns).with_columns(
            pl.col(column)
            .cast(pl.Float64, strict=False)
            .replace([float("inf"), float("-inf")], None)
            .alias(column)
            for column in numeric
        )
        if "pupil_size" in samples:
            samples = samples.with_columns(
                pl.when(pl.col("pupil_size") > 0)
                .then(pl.col("pupil_size"))
                .otherwise(None)
                .alias("pupil_size")
            )
        samples = (
            samples.filter(pl.col("timestamp").is_not_null())
            .sort("timestamp", maintain_order=True)
            .unique(subset=["timestamp"], keep="first", maintain_order=True)
        )
        if samples.is_empty():
            raise ValueError(f"No samples were found for the {self.recorded_eye} eye")

        frequency = float(self.sampling_frequency)
        if not math.isfinite(frequency) or frequency <= 0:
            raise ValueError(
                f"SamplingFrequency must be positive, got {self.sampling_frequency}"
            )
        return EyeRecording(
            samples=samples,
            recorded_eye=self.recorded_eye,
            sampling_frequency=frequency,
            timestamp_unit=self.timestamp_unit,
            coordinate_unit=self.coordinate_unit,
            coordinate_description=self.coordinate_description,
            pupil_unit=self.pupil_unit,
            pupil_description=self.pupil_description,
            manufacturer=self.manufacturer,
        )

normalized()

Validate the recording and return a canonical copy of it.

Checks that the required sample columns are present, that the recorded eye is one of the accepted labels and that the sampling frequency is positive, then returns an equivalent recording with a float sampling frequency.

Returns:

Type	Description
EyeRecording	A validated copy, ready to be written to BIDS.

Raises:

Type	Description
ValueError	If the timestamp, x_coordinate or y_coordinate columns are missing, if recorded_eye is not "left", "right" or "cyclopean", or if sampling_frequency is not positive.

Source code in pyxations/bids.py

def normalized(self) -> EyeRecording:
    """Validate the recording and return a canonical copy of it.

    Checks that the required sample columns are present, that the recorded
    eye is one of the accepted labels and that the sampling frequency is
    positive, then returns an equivalent recording with a float sampling
    frequency.

    Returns
    -------
    EyeRecording
        A validated copy, ready to be written to BIDS.

    Raises
    ------
    ValueError
        If the ``timestamp``, ``x_coordinate`` or ``y_coordinate`` columns
        are missing, if ``recorded_eye`` is not ``"left"``, ``"right"`` or
        ``"cyclopean"``, or if ``sampling_frequency`` is not positive.
    """
    required = ["timestamp", "x_coordinate", "y_coordinate"]
    missing = [column for column in required if column not in self.samples]
    if missing:
        raise ValueError(f"Eye recording is missing required columns: {missing}")
    if self.recorded_eye not in {"left", "right", "cyclopean"}:
        raise ValueError(f"Unsupported RecordedEye value: {self.recorded_eye}")

    columns = required + (
        ["pupil_size"] if "pupil_size" in self.samples.columns else []
    )
    columns += [column for column in self.samples.columns if column not in columns]
    numeric = required + (
        ["pupil_size"] if "pupil_size" in self.samples.columns else []
    )
    samples = self.samples.select(columns).with_columns(
        pl.col(column)
        .cast(pl.Float64, strict=False)
        .replace([float("inf"), float("-inf")], None)
        .alias(column)
        for column in numeric
    )
    if "pupil_size" in samples:
        samples = samples.with_columns(
            pl.when(pl.col("pupil_size") > 0)
            .then(pl.col("pupil_size"))
            .otherwise(None)
            .alias("pupil_size")
        )
    samples = (
        samples.filter(pl.col("timestamp").is_not_null())
        .sort("timestamp", maintain_order=True)
        .unique(subset=["timestamp"], keep="first", maintain_order=True)
    )
    if samples.is_empty():
        raise ValueError(f"No samples were found for the {self.recorded_eye} eye")

    frequency = float(self.sampling_frequency)
    if not math.isfinite(frequency) or frequency <= 0:
        raise ValueError(
            f"SamplingFrequency must be positive, got {self.sampling_frequency}"
        )
    return EyeRecording(
        samples=samples,
        recorded_eye=self.recorded_eye,
        sampling_frequency=frequency,
        timestamp_unit=self.timestamp_unit,
        coordinate_unit=self.coordinate_unit,
        coordinate_description=self.coordinate_description,
        pupil_unit=self.pupil_unit,
        pupil_description=self.pupil_description,
        manufacturer=self.manufacturer,
    )

SourceRecordingBundle dataclass

Normalized raw-BIDS content extracted from one vendor recording.

Source code in pyxations/bids.py

@dataclass
class SourceRecordingBundle:
    """Normalized raw-BIDS content extracted from one vendor recording."""

    recordings: list[EyeRecording]
    events: pl.DataFrame = field(default_factory=empty_frame)
    calibration: pl.DataFrame = field(default_factory=empty_frame)
    header: pl.DataFrame = field(default_factory=empty_frame)
    metadata: dict = field(default_factory=dict)

bids_label(value, *, fallback)

Normalize a value for use as a BIDS entity label.

Parameters:

Name	Type	Description	Default
value	str	Candidate entity value.	required
fallback	str	Value used when normalization removes every character.	required

Returns:

Type	Description
str	Alphanumeric BIDS entity label.

Source code in pyxations/bids.py

def bids_label(value: str, *, fallback: str) -> str:
    """Normalize a value for use as a BIDS entity label.

    Parameters
    ----------
    value : str
        Candidate entity value.
    fallback : str
        Value used when normalization removes every character.

    Returns
    -------
    str
        Alphanumeric BIDS entity label.
    """

    label = re.sub(r"[^A-Za-z0-9]+", "", str(value))
    return label or fallback

read_bids_task_events(session_path)

Read and combine BIDS task-event tables for one raw session.

Parameters:

Name	Type	Description	Default
session_path	str or Path	Raw BIDS session directory containing beh.	required

Returns:

Type	Description
DataFrame	Combined event rows, including their source BIDS filenames.

Source code in pyxations/bids.py

def read_bids_task_events(session_path: str | Path) -> pl.DataFrame:
    """Read and combine BIDS task-event tables for one raw session.

    Parameters
    ----------
    session_path : str or pathlib.Path
        Raw BIDS session directory containing ``beh``.

    Returns
    -------
    polars.DataFrame
        Combined event rows, including their source BIDS filenames.
    """

    behavior = Path(session_path) / "beh"
    tables = []
    for path in sorted(behavior.glob("*_events.tsv")):
        table = read_tsv(path, has_header=True).with_columns(
            pl.lit(path.name).alias("bids_events_file")
        )
        tables.append(table)
    return pl.concat(tables, how="diagonal_relaxed") if tables else empty_frame()

read_raw_bids_session(session_path)

Load normalized samples and source events from a raw BIDS session.

Parameters:

Name	Type	Description	Default
session_path	str or Path	Raw BIDS session directory containing physiological recordings.	required

Returns:

Type	Description
SessionTables	Samples, tracker events, behavior, metadata, and recording properties.

Raises:

Type	Description
FileNotFoundError	If the session contains no raw BIDS physiological recordings.
ValueError	If per-eye streams disagree about their sampling frequency.

Source code in pyxations/bids.py

def read_raw_bids_session(session_path: str | Path) -> SessionTables:
    """Load normalized samples and source events from a raw BIDS session.

    Parameters
    ----------
    session_path : str or pathlib.Path
        Raw BIDS session directory containing physiological recordings.

    Returns
    -------
    SessionTables
        Samples, tracker events, behavior, metadata, and recording properties.

    Raises
    ------
    FileNotFoundError
        If the session contains no raw BIDS physiological recordings.
    ValueError
        If per-eye streams disagree about their sampling frequency.
    """

    session = Path(session_path)
    behavior = session / "beh"
    physio_paths = sorted(
        path
        for path in behavior.glob("*_physio.tsv.gz")
        if "_physioevents." not in path.name
    )
    if not physio_paths:
        raise FileNotFoundError(f"No raw BIDS physio files found in {behavior}")

    sample_streams = []
    first_metadata = None
    frequencies = []
    for path in physio_paths:
        metadata = json.loads(
            path.with_suffix("").with_suffix(".json").read_text(encoding="utf-8")
        )
        first_metadata = first_metadata or metadata
        frequencies.append(float(metadata["SamplingFrequency"]))
        frame = _read_bids_table(path, metadata)
        time_scale = _milliseconds_per_unit(metadata.get("timestamp", {}).get("Units"))
        stream = frame.select(
            pl.col("timestamp")
            .cast(pl.Float64, strict=False)
            .mul(time_scale)
            .alias("tSample")
        )
        eye = metadata.get("RecordedEye", "cyclopean")
        prefix = {"left": "L", "right": "R"}.get(eye)
        if prefix:
            stream = stream.with_columns(
                frame.get_column("x_coordinate")
                .cast(pl.Float64, strict=False)
                .alias(f"{prefix}X"),
                frame.get_column("y_coordinate")
                .cast(pl.Float64, strict=False)
                .alias(f"{prefix}Y"),
            )
            if "pupil_size" in frame:
                stream = stream.with_columns(
                    frame.get_column("pupil_size")
                    .cast(pl.Float64, strict=False)
                    .alias(f"{prefix}Pupil")
                )
        else:
            stream = stream.with_columns(
                frame.get_column("x_coordinate")
                .cast(pl.Float64, strict=False)
                .alias("X"),
                frame.get_column("y_coordinate")
                .cast(pl.Float64, strict=False)
                .alias("Y"),
            )
            if "pupil_size" in frame:
                stream = stream.with_columns(
                    frame.get_column("pupil_size")
                    .cast(pl.Float64, strict=False)
                    .alias("Pupil")
                )
        auxiliary_names = {
            "calibration_index": "Calib_index",
            "line_number": "Line_number",
            "trial_number": "trial_number",
        }
        canonical_source_columns = {
            "timestamp",
            "x_coordinate",
            "y_coordinate",
            "pupil_size",
        }
        auxiliary_columns = [
            column for column in frame.columns if column not in canonical_source_columns
        ]
        for source_column in auxiliary_columns:
            target_column = auxiliary_names.get(source_column, source_column)
            if source_column in frame and target_column not in stream:
                stream = stream.with_columns(
                    frame.get_column(source_column).alias(target_column)
                )
        sample_streams.append(stream)

    samples = sample_streams[0]
    for stream in sample_streams[1:]:
        duplicate_auxiliary = [
            column
            for column in stream.columns
            if column != "tSample" and column in samples.columns
        ]
        stream = stream.drop(duplicate_auxiliary)
        if samples.height == stream.height and samples.get_column("tSample").equals(
            stream.get_column("tSample")
        ):
            samples = samples.hstack(stream.drop("tSample"))
        else:
            samples = samples.join(stream, on="tSample", how="full", coalesce=True)
    samples = samples.sort("tSample", maintain_order=True)
    sampling_frequency = float(np.nanmedian(frequencies))
    additions = [pl.lit(sampling_frequency).alias("Rate_recorded")]
    if "Calib_index" not in samples:
        additions.append(pl.lit(1).alias("Calib_index"))
    if "Line_number" not in samples:
        additions.append(
            pl.int_range(0, samples.height, eager=True).alias("Line_number")
        )
    if {"LX", "RX"}.intersection(samples.columns):
        eyes_recorded = (
            "LR"
            if {"LX", "RX"}.issubset(samples.columns)
            else "L"
            if "LX" in samples
            else "R"
        )
        additions.append(pl.lit(eyes_recorded).alias("Eyes_recorded"))
    samples = samples.with_columns(additions)

    event_frames = []
    for path in sorted(behavior.glob("*_physioevents.tsv.gz")):
        metadata = json.loads(
            path.with_suffix("").with_suffix(".json").read_text(encoding="utf-8")
        )
        event_frames.append(_read_bids_table(path, metadata))
    events = (
        pl.concat(event_frames, how="diagonal_relaxed")
        if event_frames
        else empty_frame()
    )
    if not events.is_empty():
        # Device-wide messages may be associated with every per-eye
        # physiological recording. Reconstruct them only once in memory.
        events = events.unique(maintain_order=True)
    time_scale = _milliseconds_per_unit(
        first_metadata.get("timestamp", {}).get("Units")
    )
    if not events.is_empty():
        events = events.with_columns(
            pl.col("onset")
            .cast(pl.Float64, strict=False)
            .mul(time_scale)
            .alias("tStart")
        )
        if "end_timestamp" in events:
            events = events.with_columns(
                pl.col("end_timestamp")
                .cast(pl.Float64, strict=False)
                .mul(time_scale)
                .alias("tEnd")
            )
        else:
            duration = (
                pl.col("duration").cast(pl.Float64, strict=False)
                if "duration" in events
                else pl.lit(None, dtype=pl.Float64)
            )
            events = events.with_columns(
                (pl.col("tStart") + duration * 1_000.0).alias("tEnd")
            )
        events = events.with_columns(
            (pl.col("tEnd") - pl.col("tStart")).alias("duration_ms")
        )

    def selected(event_type: str, mapping: dict[str, str]) -> pl.DataFrame:
        if events.is_empty() or "trial_type" not in events:
            return pl.DataFrame({column: [] for column in mapping.values()})
        result = events.filter(pl.col("trial_type") == event_type)
        available = {
            source: target for source, target in mapping.items() if source in result
        }
        return result.select(list(available)).rename(available)

    common = {
        "eye": "eye",
        "tStart": "tStart",
        "tEnd": "tEnd",
        "duration_ms": "duration",
        "line_number": "Line_number",
        "calibration_index": "Calib_index",
    }
    fixations = selected(
        "fixation",
        {
            **common,
            "x_avg": "xAvg",
            "y_avg": "yAvg",
            "pupil_avg": "pupilAvg",
        },
    )
    saccades = selected(
        "saccade",
        {
            **common,
            "x_start": "xStart",
            "y_start": "yStart",
            "x_end": "xEnd",
            "y_end": "yEnd",
            "amplitude": "ampDeg",
            "peak_velocity": "vPeak",
        },
    )
    blinks = selected("blink", common)
    messages = selected(
        "message",
        {
            "tStart": "timestamp",
            "message": "message",
            "line_number": "Line_number",
            "calibration_index": "Calib_index",
        },
    )
    if "Eyes_recorded" in samples:
        session_eye = samples.get_column("Eyes_recorded").item(0)
        for table in (fixations, saccades, blinks, messages):
            table = table.with_columns(
                pl.lit(session_eye).alias("Eyes_recorded"),
                pl.lit(sampling_frequency).alias("Rate_recorded"),
            )
            if table is fixations:
                fixations = table
            elif table is saccades:
                saccades = table
            elif table is blinks:
                blinks = table
            else:
                messages = table

    return SessionTables(
        samples=samples,
        fixations=fixations,
        saccades=saccades,
        blinks=blinks,
        messages=messages,
        calibration=payload_frame(first_metadata.get("PyxationsCalibration")),
        header=payload_frame(first_metadata.get("PyxationsHeader")),
        behavioral_events=read_bids_task_events(session),
        sampling_frequency=sampling_frequency,
        screen_width=first_metadata.get("ScreenWidth"),
        screen_height=first_metadata.get("ScreenHeight"),
    )

validate_bids_dataset(dataset_path, *, command=None)

Validate a dataset with the official BIDS Validator.

Parameters:

Name	Type	Description	Default
dataset_path	str or Path	BIDS dataset root to validate.	required
command	sequence of str	Explicit validator command. By default, discover a native validator or use the pinned Deno invocation.	None

Returns:

Type	Description
CompletedProcess	Successful validator process result.

Raises:

Type	Description
RuntimeError	If no validator executable or Deno runtime is available.
BIDSValidationError	If validation reports one or more errors.
FileNotFoundError	If dataset_path is not an existing directory.

Source code in pyxations/bids.py

def validate_bids_dataset(
    dataset_path: str | Path, *, command: Sequence[str] | None = None
) -> subprocess.CompletedProcess[str]:
    """Validate a dataset with the official BIDS Validator.

    Parameters
    ----------
    dataset_path : str or pathlib.Path
        BIDS dataset root to validate.
    command : sequence of str, optional
        Explicit validator command. By default, discover a native validator or
        use the pinned Deno invocation.

    Returns
    -------
    subprocess.CompletedProcess
        Successful validator process result.

    Raises
    ------
    RuntimeError
        If no validator executable or Deno runtime is available.
    BIDSValidationError
        If validation reports one or more errors.
    FileNotFoundError
        If ``dataset_path`` is not an existing directory.
    """

    dataset = Path(dataset_path)
    if not dataset.is_dir():
        raise FileNotFoundError(f"BIDS dataset not found: {dataset}")
    validator = list(command) if command is not None else validator_command()
    if validator is None:
        raise RuntimeError(
            "The official BIDS Validator is unavailable. Install the "
            "`bids-validator` command or the Deno runtime."
        )
    result = subprocess.run(
        [*validator, str(dataset), "--json"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        report = "\n".join(part for part in (result.stdout, result.stderr) if part)
        raise BIDSValidationError(
            f"BIDS validation failed for {dataset} (exit {result.returncode}):\n"
            f"{report}"
        )
    return result

validator_command()

Return the available official BIDS Validator command.

Returns:

Type	Description
list of str or None	Native validator command, Deno invocation, or None when neither is installed.

Source code in pyxations/bids.py

def validator_command() -> list[str] | None:
    """Return the available official BIDS Validator command.

    Returns
    -------
    list of str or None
        Native validator command, Deno invocation, or ``None`` when neither is
        installed.
    """

    executable = shutil.which("bids-validator")
    if executable:
        return [executable]
    deno = shutil.which("deno")
    if deno:
        return [
            deno,
            "run",
            "-ERWN",
            f"jsr:@bids/validator@{BIDS_VALIDATOR_VERSION}",
        ]
    return None

webgazer_trial_numbering(source)

Map jsPsych trial_index values to sequential trial numbers.

A jsPsych export numbers every screen it presented, including instructions and calibration, so the trials that carry gaze are an arbitrary subset such as 29, 30, 31, 33. Every other Pyxations input format numbers trials 0, 1, 2, ... in presentation order, so the raw jsPsych indices are renumbered to match and the originals are kept in source_trial_index.

The same mapping is applied to gaze samples and to the behavioral events table, which are read from the same source file by different code paths; were they to disagree, no trial would find its behavioral row.

Parameters:

Name	Type	Description	Default
source	DataFrame	The WebGazer export, read verbatim.	required

Returns:

Type	Description
dict	Mapping of original trial_index to sequential trial number. Screens that carry no gaze data are absent, and therefore have no trial number.

Source code in pyxations/bids.py

def webgazer_trial_numbering(source: pl.DataFrame) -> dict[int, int]:
    """Map jsPsych ``trial_index`` values to sequential trial numbers.

    A jsPsych export numbers every screen it presented, including instructions
    and calibration, so the trials that carry gaze are an arbitrary subset such
    as ``29, 30, 31, 33``. Every other Pyxations input format numbers trials
    ``0, 1, 2, ...`` in presentation order, so the raw jsPsych indices are
    renumbered to match and the originals are kept in ``source_trial_index``.

    The same mapping is applied to gaze samples and to the behavioral events
    table, which are read from the same source file by different code paths;
    were they to disagree, no trial would find its behavioral row.

    Parameters
    ----------
    source : polars.DataFrame
        The WebGazer export, read verbatim.

    Returns
    -------
    dict
        Mapping of original ``trial_index`` to sequential trial number. Screens
        that carry no gaze data are absent, and therefore have no trial number.
    """

    if "trial_index" not in source or "webgazer_data" not in source:
        return {}
    indices = (
        _webgazer_gaze_rows(source)
        .get_column("trial_index")
        .drop_nulls()
        .unique()
        .sort()
        .to_list()
    )
    return {int(original): number for number, original in enumerate(indices)}

write_bids_dataset(target_folder_path, files_folder_path, dataset_name, *, session_substrings=1, format_name='eyelink', task_name='eyetracking', authors=None, behavioral_column_map=None, overwrite=False)

Convert supported vendor recordings into a validated BIDS layout.

Original vendor files are retained under sourcedata. Standardized sample-level recordings are emitted as per-eye physiological files.

Parameters:

Name	Type	Description	Default
target_folder_path	str or Path	Parent directory in which to create the dataset.	required
files_folder_path	str or Path	Directory containing vendor recordings and associated behavior.	required
dataset_name	str	Name of the output dataset directory and BIDS dataset.	required
session_substrings	int	Number of underscore-separated filename tokens used for the session.	1
format_name	('eyelink', 'gaze', 'tobii', 'webgazer')	Vendor input reader to use.	"eyelink"
task_name	str	Fallback BIDS task label when the filename has no task- entity.	"eyetracking"
authors	sequence of str	Dataset authors stored in dataset_description.json.	None
behavioral_column_map	mapping of str to str	Source-independent mapping from behavioral columns to task concepts.	None
overwrite	bool	Whether to replace an existing non-empty output dataset.	False

Returns:

Type	Description
Path	Root of the generated raw BIDS dataset.

Raises:

Type	Description
ValueError	If the format, session configuration, source recordings, or overwrite relationship is invalid.
FileNotFoundError	If the source directory does not exist or EyeLink conversion is needed but edf2asc is unavailable.
FileExistsError	If the output exists and overwrite is false.

Source code in pyxations/bids.py

def write_bids_dataset(
    target_folder_path: str | Path,
    files_folder_path: str | Path,
    dataset_name: str,
    *,
    session_substrings: int = 1,
    format_name: str = "eyelink",
    task_name: str = "eyetracking",
    authors: Sequence[str] | None = None,
    behavioral_column_map: Mapping[str, str] | None = None,
    overwrite: bool = False,
) -> Path:
    """Convert supported vendor recordings into a validated BIDS layout.

    Original vendor files are retained under ``sourcedata``. Standardized
    sample-level recordings are emitted as per-eye physiological files.

    Parameters
    ----------
    target_folder_path : str or pathlib.Path
        Parent directory in which to create the dataset.
    files_folder_path : str or pathlib.Path
        Directory containing vendor recordings and associated behavior.
    dataset_name : str
        Name of the output dataset directory and BIDS dataset.
    session_substrings : int, default 1
        Number of underscore-separated filename tokens used for the session.
    format_name : {"eyelink", "gaze", "tobii", "webgazer"}
        Vendor input reader to use.
    task_name : str, default "eyetracking"
        Fallback BIDS task label when the filename has no ``task-`` entity.
    authors : sequence of str, optional
        Dataset authors stored in ``dataset_description.json``.
    behavioral_column_map : mapping of str to str, optional
        Source-independent mapping from behavioral columns to task concepts.
    overwrite : bool, default False
        Whether to replace an existing non-empty output dataset.

    Returns
    -------
    pathlib.Path
        Root of the generated raw BIDS dataset.

    Raises
    ------
    ValueError
        If the format, session configuration, source recordings, or overwrite
        relationship is invalid.
    FileNotFoundError
        If the source directory does not exist or EyeLink conversion is needed
        but ``edf2asc`` is unavailable.
    FileExistsError
        If the output exists and ``overwrite`` is false.
    """

    format_name = format_name.lower()
    if format_name not in READERS:
        raise ValueError(
            f"Unknown eye-tracking format {format_name!r}; "
            f"choose one of {sorted(READERS)}"
        )
    if session_substrings < 1:
        raise ValueError("session_substrings must be at least 1")

    source_root = Path(files_folder_path)
    if not source_root.is_dir():
        raise FileNotFoundError(f"Input directory not found: {source_root}")
    dataset_root = Path(target_folder_path) / dataset_name
    if dataset_root.exists() and any(dataset_root.iterdir()):
        if not overwrite:
            raise FileExistsError(
                f"Dataset already exists and is not empty: {dataset_root}. "
                "Pass overwrite=True to replace it."
            )
        source_resolved = source_root.resolve()
        dataset_resolved = dataset_root.resolve()
        if (
            source_resolved == dataset_resolved
            or dataset_resolved in source_resolved.parents
        ):
            raise ValueError(
                "Refusing to overwrite a dataset containing its source files"
            )
        shutil.rmtree(dataset_root)
    dataset_root.mkdir(parents=True, exist_ok=True)
    shutil.copytree(
        source_root,
        dataset_root / "sourcedata",
        copy_function=shutil.copy2,
    )

    primary = [
        path
        for path in sorted(source_root.rglob("*"))
        if path.is_file() and _is_primary_recording(path, format_name)
    ]
    if format_name == "eyelink":
        edf_stems = {
            path.stem.lower() for path in primary if path.suffix.lower() == ".edf"
        }
        primary = [
            path
            for path in primary
            if path.suffix.lower() != ".asc" or path.stem.lower() not in edf_stems
        ]
    if not primary:
        raise ValueError(f"No {format_name} recordings found in {source_root}")

    subject_names = sorted({path.name.split("_")[0] for path in primary})
    subject_map = {
        old: str(index).zfill(4) for index, old in enumerate(subject_names, start=1)
    }
    participant_rows = []
    for old, subject in subject_map.items():
        participant_rows.append(
            {
                "participant_id": f"sub-{subject}",
                "subject_id": subject,
                "old_subject_id": old,
            }
        )

    _write_json(
        dataset_root / "dataset_description.json",
        {
            "Name": dataset_name,
            "BIDSVersion": BIDS_VERSION,
            "DatasetType": "raw",
            "Authors": list(authors or ["NeuroLIAA"]),
            "GeneratedBy": [
                {
                    "Name": "Pyxations",
                    "Description": "Multi-vendor eye-tracking conversion to BIDS.",
                }
            ],
        },
    )
    write_tsv(
        dataset_root / "participants.tsv",
        pl.DataFrame(participant_rows),
        include_header=True,
        compressed=False,
    )
    _write_json(
        dataset_root / "participants.json",
        {
            "subject_id": {
                "Description": "Pyxations' zero-padded internal subject identifier."
            },
            "old_subject_id": {
                "Description": "Subject identifier present in the source filename."
            },
        },
    )
    (dataset_root / "README").write_text(
        "Eye-tracking dataset converted to BIDS by Pyxations.\n",
        encoding="utf-8",
        newline="\n",
    )

    for old_subject, subject in subject_map.items():
        subject_primary = [
            path for path in primary if path.name.split("_")[0] == old_subject
        ]
        by_session: dict[str, list[Path]] = {}
        for path in subject_primary:
            session = _session_from_filename(path, session_substrings)
            by_session.setdefault(session, []).append(path)

        all_subject_files = [
            path
            for path in sorted(source_root.rglob("*"))
            if path.is_file() and path.name.split("_")[0] == old_subject
        ]
        for session, session_primary in by_session.items():
            session_sources = [
                path
                for path in all_subject_files
                if _session_from_filename(path, session_substrings) == session
            ]
            for run_index, source in enumerate(session_primary, start=1):
                bundle = _read_source_bundle(source, format_name)
                task = _task_from_filename(source, task_name)
                base = f"sub-{subject}_ses-{session}_task-{task}"
                if len(session_primary) > 1:
                    base += f"_run-{run_index:02d}"
                destination = dataset_root / f"sub-{subject}" / f"ses-{session}" / "beh"
                available_eyes = [
                    recording.recorded_eye for recording in bundle.recordings
                ]
                for eye_index, recording in enumerate(bundle.recordings, start=1):
                    prefix = f"{base}_recording-eye{eye_index}"
                    _write_recording(
                        recording,
                        destination=destination,
                        prefix=prefix,
                        extra_metadata=(
                            {
                                "PyxationsCalibration": frame_payload(
                                    bundle.calibration
                                ),
                                "PyxationsHeader": frame_payload(bundle.header),
                                **bundle.metadata,
                            }
                            if eye_index == 1
                            else None
                        ),
                    )
                    _write_physio_events(
                        _events_for_recording(
                            bundle.events,
                            recorded_eye=recording.recorded_eye,
                            available_eyes=available_eyes,
                        ),
                        destination=destination,
                        prefix=prefix,
                    )
                task_events = _prepare_task_events(
                    session_sources,
                    source_root=source_root,
                    primary_source=source,
                    format_name=format_name,
                    behavioral_column_map=behavioral_column_map,
                )
                _write_task_events(
                    task_events,
                    destination=destination,
                    prefix=base,
                )
    return dataset_root

behavior

Normalizing behavioral CSV or TSV tables into BIDS events.tsv, with source-independent column mapping.

Source-independent behavioral table ingestion and column normalization.

normalize_behavioral_events(events, *, column_map=None)

Map source columns onto experiment-level behavioral concepts.

Mapping is intentionally independent of the source format. This lets a PsychoPy log, CSV, TSV, or future adapter satisfy the same experiment schema without embedding task semantics in the format parser.

Parameters:

Name	Type	Description	Default
events	DataFrame	Behavioral event table in its source-specific schema.	required
column_map	mapping of str to str	Mapping from source column names to experiment-level names.	None

Returns:

Type	Description
DataFrame	The original table when no mapping is supplied, otherwise a table with the requested column names.

Raises:

Type	Description
ValueError	If a source column is absent or the mapping would create duplicates.

Source code in pyxations/behavior.py

def normalize_behavioral_events(
    events: pl.DataFrame,
    *,
    column_map: Mapping[str, str] | None = None,
) -> pl.DataFrame:
    """Map source columns onto experiment-level behavioral concepts.

    Mapping is intentionally independent of the source format. This lets a
    PsychoPy log, CSV, TSV, or future adapter satisfy the same experiment
    schema without embedding task semantics in the format parser.

    Parameters
    ----------
    events : polars.DataFrame
        Behavioral event table in its source-specific schema.
    column_map : mapping of str to str, optional
        Mapping from source column names to experiment-level names.

    Returns
    -------
    polars.DataFrame
        The original table when no mapping is supplied, otherwise a table with
        the requested column names.

    Raises
    ------
    ValueError
        If a source column is absent or the mapping would create duplicates.
    """

    if not column_map:
        return events

    normalized = {
        str(source): _column_name(str(destination))
        for source, destination in column_map.items()
    }
    missing = sorted(set(normalized) - set(events.columns))
    if missing:
        raise ValueError(f"Behavioral columns not found for renaming: {missing}")

    resulting = [normalized.get(column, column) for column in events.columns]
    if len(resulting) != len(set(resulting)):
        raise ValueError("Behavioral column mapping creates duplicate columns")
    return events.rename(normalized)

read_behavioral_events(path, *, column_map=None)

Read a CSV, TSV, or PsychoPy log as a behavioral event table.

The returned table retains source values and applies the same optional experiment-level column mapping regardless of its input format. BIDS timing normalization is performed later by the dataset writer.

Parameters:

Name	Type	Description	Default
path	str or Path	Behavioral CSV, TSV, or PsychoPy log to read.	required
column_map	mapping of str to str	Mapping from source column names to experiment-level names.	None

Returns:

Type	Description
DataFrame	Parsed behavioral events in the source-independent tabular form.

Raises:

Type	Description
ValueError	If the file extension is unsupported or the column mapping is invalid.

Source code in pyxations/behavior.py

def read_behavioral_events(
    path: str | Path,
    *,
    column_map: Mapping[str, str] | None = None,
) -> pl.DataFrame:
    """Read a CSV, TSV, or PsychoPy log as a behavioral event table.

    The returned table retains source values and applies the same optional
    experiment-level column mapping regardless of its input format. BIDS
    timing normalization is performed later by the dataset writer.

    Parameters
    ----------
    path : str or pathlib.Path
        Behavioral CSV, TSV, or PsychoPy log to read.
    column_map : mapping of str to str, optional
        Mapping from source column names to experiment-level names.

    Returns
    -------
    polars.DataFrame
        Parsed behavioral events in the source-independent tabular form.

    Raises
    ------
    ValueError
        If the file extension is unsupported or the column mapping is invalid.
    """

    path = Path(path)
    suffix = path.suffix.lower()
    if suffix == ".csv":
        events = pl.read_csv(path, infer_schema_length=None)
    elif suffix == ".tsv":
        events = pl.read_csv(path, separator="\t", infer_schema_length=None)
    elif suffix == ".log":
        from .psychopy import psychopy_log_to_events

        events = psychopy_log_to_events(path)
    else:
        raise ValueError(
            f"Unsupported behavioral file format {suffix!r}; "
            "expected .csv, .tsv, or .log"
        )
    return normalize_behavioral_events(events, column_map=column_map)

psychopy

Parsing standard PsychoPy New trial logs. This does not require PsychoPy to be installed, and PsychoPy-local timestamps are retained without assuming they are synchronized to the eye-tracker clock.

PsychoPy text-log parsing without requiring PsychoPy at runtime.

psychopy_log_to_events(log_file_path)

Parse PsychoPy .log trials into a BIDS-ready Polars table.

The parser creates one row for every New trial record. It retains the TrialHandler condition mapping, subsequent component property updates, and keypresses observed before the next trial. Component updates use columns such as trial_image_image; repeated updates retain their last value.

PsychoPy timestamps use a separate clock that is not necessarily synchronized with the eye tracker. They are therefore stored as psychopy_onset and psychopy_trial_interval. Canonical BIDS onset and duration remain missing rather than claiming a false synchronization.

Parameters:

Name	Type	Description	Default
log_file_path	str or Path	PsychoPy text log to parse.	required

Returns:

Type	Description
DataFrame	One row per logged trial, or an empty table when no trial markers are present.

Source code in pyxations/psychopy.py

def psychopy_log_to_events(
    log_file_path: str | Path,
) -> pl.DataFrame:
    """Parse PsychoPy ``.log`` trials into a BIDS-ready Polars table.

    The parser creates one row for every ``New trial`` record. It retains the
    TrialHandler condition mapping, subsequent component property updates, and
    keypresses observed before the next trial. Component updates use columns
    such as ``trial_image_image``; repeated updates retain their last value.

    PsychoPy timestamps use a separate clock that is not necessarily
    synchronized with the eye tracker. They are therefore stored as
    ``psychopy_onset`` and ``psychopy_trial_interval``. Canonical BIDS
    ``onset`` and ``duration`` remain missing rather than claiming a false
    synchronization.

    Parameters
    ----------
    log_file_path : str or pathlib.Path
        PsychoPy text log to parse.

    Returns
    -------
    polars.DataFrame
        One row per logged trial, or an empty table when no trial markers are
        present.
    """

    path = Path(log_file_path)
    rows: list[dict[str, Any]] = []
    current: dict[str, Any] | None = None
    current_keypresses: list[str] = []

    def finish(next_timestamp: float | None = None) -> None:
        nonlocal current, current_keypresses
        if current is None:
            return
        if next_timestamp is not None:
            current["psychopy_trial_interval"] = (
                next_timestamp - current["psychopy_onset"]
            )
        if current_keypresses:
            current["keypresses"] = current_keypresses.copy()
        rows.append(current)
        current = None
        current_keypresses = []

    with path.open("r", encoding="utf-8", errors="replace") as stream:
        for raw_line in stream:
            match = _LOG_LINE.match(raw_line.rstrip("\r\n"))
            if match is None:
                continue
            timestamp = float(match.group("timestamp"))
            message = match.group("message").strip()

            trial_match = _NEW_TRIAL.match(message)
            if trial_match is not None:
                finish(timestamp)
                trial_number = len(rows)
                current = {
                    "onset": None,
                    "duration": None,
                    "trial_type": "psychopy_trial",
                    "trial_number": trial_number,
                    "trial_index": trial_number,
                    "psychopy_onset": timestamp,
                }
                for context in _CONTEXT_VALUE.finditer(trial_match.group("context")):
                    current[
                        f"psychopy_{_column_name(context.group('name'), prefix='psychopy')}"
                    ] = _literal(context.group("value"))
                condition_values = _condition_values(trial_match.group("payload"))
                if condition_values:
                    current.update(condition_values)
                elif trial_match.group("payload"):
                    current["psychopy_condition_payload"] = trial_match.group("payload")
                continue

            if current is None:
                continue

            keypress_match = _KEYPRESS.match(message)
            if keypress_match is not None:
                current_keypresses.append(keypress_match.group("value").strip())
                continue

            component_match = _COMPONENT_VALUE.match(message)
            if component_match is not None:
                if (
                    component_match.group("attribute").strip().lower()
                    not in _TRACKED_COMPONENT_ATTRIBUTES
                ):
                    continue
                column = _column_name(
                    f"{component_match.group('component')}_"
                    f"{component_match.group('attribute')}",
                    prefix="psychopy",
                )
                current[column] = _literal(component_match.group("value"))

    finish()
    if not rows:
        return pl.DataFrame()
    return pl.from_dicts(rows, infer_schema_length=None, strict=False)