Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'main' into messages
Browse files Browse the repository at this point in the history
Remi-Gau authored Jun 26, 2024
2 parents 6a3e661 + 21d17e5 commit 302c84b
Showing 5 changed files with 297 additions and 193 deletions.
10 changes: 8 additions & 2 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -1,16 +1,22 @@
tests/data/osf
tests/data/output
eye2bids/_version.py

*events.json
*eyetrack.json
*eyetrack.tsv

*.ipynb
2eyes.ipynb

tmp

.vscode

# General
.DS_Store

# hatchling
eye2bids/_version.py

# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
4 changes: 2 additions & 2 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
@@ -22,7 +22,7 @@ repos:
args: [--profile, black]

- repo: https://github.com/asottile/pyupgrade
rev: v3.15.2
rev: v3.16.0
hooks:
- id: pyupgrade
args: [--py38-plus]
@@ -60,7 +60,7 @@ repos:
args: [--config-file=pyproject.toml]

- repo: https://github.com/pyCQA/flake8
rev: 7.0.0
rev: 7.1.0
hooks:
- id: flake8
args: [--config, .flake8, --verbose]
157 changes: 157 additions & 0 deletions eye2bids/_base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,157 @@
"""Base classes for sidecar and events."""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from eye2bids.logger import eye2bids_logger

e2b_log = eye2bids_logger()


class BasePhysioEventsJson(dict[str, Any]):
"""Handle content of physioevents sidedar."""

input_file: Path
two_eyes: bool

def __init__(self, metadata: None | dict[str, Any] = None) -> None:

self["Columns"] = ["onset", "duration", "trial_type", "blink", "message"]
self["Description"] = "Messages logged by the measurement device"
self["ForeignIndexColumn"] = "timestamp"

self["blink"] = {
"Description": "One indicates if the eye was closed, zero if open."
}
self["message"] = {"Description": "String messages logged by the eye-tracker."}
self["trial_type"] = {
"Description": (
"Event type as identified by the eye-tracker's model "
"((either 'n/a' if not applicabble, 'fixation', or 'saccade')."
)
}

self.update_from_metadata(metadata)

def update_from_metadata(self, metadata: None | dict[str, Any] = None) -> None:
"""Update content of json side car based on metadata."""
if metadata is None:
return None

self["InstitutionAddress"] = metadata.get("InstitutionAddress")
self["InstitutionName"] = metadata.get("InstitutionName")
self["StimulusPresentation"] = {
"ScreenDistance": metadata.get("ScreenDistance"),
"ScreenRefreshRate": metadata.get("ScreenRefreshRate"),
"ScreenSize": metadata.get("ScreenSize"),
}

def output_filename(self, recording: str | None = None) -> str:
"""Generate output filename."""
filename = self.input_file.stem
if recording is not None:
return f"{filename}_recording-{recording}_physioevents.json"
return f"{filename}_physioevents.json"

def write(
self,
output_dir: Path,
recording: str | None = None,
extra_metadata: dict[str, str | list[str] | list[float]] | None = None,
) -> None:
"""Write to json."""
if extra_metadata is not None:
for key, value in extra_metadata.items():
self[key] = value

content = {key: value for key, value in self.items() if self[key] is not None}
with open(output_dir / self.output_filename(recording=recording), "w") as outfile:
json.dump(content, outfile, indent=4)


class BasePhysioJson(dict[str, Any]):
"""Handle content of physio sidedar."""

input_file: Path
has_validation: bool
two_eyes: bool

def __init__(self, manufacturer: str, metadata: dict[str, Any] | None = None) -> None:

self["Manufacturer"] = manufacturer

self["Columns"] = ["x_coordinate", "y_coordinate", "pupil_size", "timestamp"]
self["timestamp"] = {
"Description": (
"Timestamp issued by the eye-tracker "
"indexing the continuous recordings "
"corresponding to the sampled eye."
)
}
self["x_coordinate"] = {
"Description": (
"Gaze position x-coordinate of the recorded eye, "
"in the coordinate units specified "
"in the corresponding metadata sidecar."
),
"Units": "a.u.",
}
self["y_coordinate"] = {
"Description": (
"Gaze position y-coordinate of the recorded eye, "
"in the coordinate units specified "
"in the corresponding metadata sidecar."
),
"Units": "a.u.",
}
self["pupil_size"] = {
"Description": (
"Pupil area of the recorded eye as calculated "
"by the eye-tracker in arbitrary units "
"(see EyeLink's documentation for conversion)."
),
"Units": "a.u.",
}

self.update_from_metadata(metadata)

def update_from_metadata(self, metadata: None | dict[str, Any] = None) -> None:
"""Update content of json side car based on metadata."""
if metadata is None:
return None

self["EnvironmentCoordinates"] = metadata.get("EnvironmentCoordinates")
self["SoftwareVersion"] = metadata.get("SoftwareVersion")
self["EyeCameraSettings"] = metadata.get("EyeCameraSettings")
self["EyeTrackerDistance"] = metadata.get("EyeTrackerDistance")
self["FeatureDetectionSettings"] = metadata.get("FeatureDetectionSettings")
self["GazeMappingSettings"] = metadata.get("GazeMappingSettings")
self["RawDataFilters"] = metadata.get("RawDataFilters")
self["SampleCoordinateSystem"] = metadata.get("SampleCoordinateSystem")
self["SampleCoordinateUnits"] = metadata.get("SampleCoordinateUnits")
self["ScreenAOIDefinition"] = metadata.get("ScreenAOIDefinition")

def output_filename(self, recording: str | None = None) -> str:
"""Generate output filename."""
filename = self.input_file.stem
if recording is not None:
return f"{filename}_recording-{recording}_physio.json"
return f"{filename}_physio.json"

def write(
self,
output_dir: Path,
recording: str | None = None,
extra_metadata: dict[str, str | list[str] | list[float]] | None = None,
) -> None:
"""Write to json."""
if extra_metadata is not None:
for key, value in extra_metadata.items():
self[key] = value

content = {key: value for key, value in self.items() if self[key] is not None}
with open(output_dir / self.output_filename(recording=recording), "w") as outfile:
json.dump(content, outfile, indent=4)
302 changes: 127 additions & 175 deletions eye2bids/edf2bids.py
Original file line number Diff line number Diff line change
@@ -3,16 +3,17 @@
from __future__ import annotations

import gzip
import json
import subprocess
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd
import yaml
from rich.prompt import Prompt
from yaml.loader import SafeLoader

from eye2bids._base import BasePhysioEventsJson, BasePhysioJson
from eye2bids._parser import global_parser
from eye2bids.logger import eye2bids_logger
import re
@@ -74,7 +75,7 @@ def _check_inputs(
def _check_output_dir(output_dir: str | Path | None = None) -> Path:
"""Check if output directory is valid."""
if output_dir is None:
output_dir = input("Enter the output directory: ")
output_dir = Path()
if isinstance(output_dir, str):
checked_output_dir = Path(output_dir)
elif isinstance(output_dir, Path):
@@ -119,10 +120,7 @@ def _convert_edf_to_asc_samples(input_file: str | Path) -> Path:

def _2eyesmode(df: pd.DataFrame) -> bool:
eye = df[df[2] == "RECCFG"].iloc[0:1, 5:6].to_string(header=False, index=False)
if eye == "LR":
two_eyes = True
else:
two_eyes = False
two_eyes = eye == "LR"
return two_eyes


@@ -134,45 +132,41 @@ def _extract_CalibrationType(df: pd.DataFrame) -> list[int]:
return _calibrations(df).iloc[0:1, 2:3].to_string(header=False, index=False)


def _extract_CalibrationCount(df: pd.DataFrame) -> int:
if _2eyesmode(df) == True:
return len(_calibrations(df)) // 2
return len(_calibrations(df))
def _extract_CalibrationCount(df: pd.DataFrame, two_eyes: bool) -> int:
return len(_calibrations(df)) // 2 if two_eyes else len(_calibrations(df))


def _get_calibration_positions(df: pd.DataFrame) -> list[int]:
if _2eyesmode(df) == True:
return (
np.array(df[df[2] == "VALIDATE"][8].str.split(",", expand=True))
.astype(int)
.tolist()
)[::2]
return (
np.array(df[df[2] == "VALIDATE"][8].str.split(",", expand=True))
.astype(int)
.tolist()
)
def _extract_CalibrationPosition(df: pd.DataFrame) -> list[list[list[int]]]:

calibration_df = df[df[2] == "VALIDATE"]
calibration_df[5] = pd.to_numeric(calibration_df[5], errors="coerce")

def _extract_CalibrationPosition(df: pd.DataFrame) -> list[list[int]]:
cal_pos = _get_calibration_positions(df)
cal_num = len(cal_pos) // _extract_CalibrationCount(df)
if _2eyesmode(df):
# drop duplicated calibration position
# because they will be the same for both eyes
calibration_df = calibration_df[calibration_df[6] == "LEFT"]

CalibrationPosition: list[list[int]] = []
nb_calibration_postions = calibration_df[5].max() + 1

if len(cal_pos) == 0:
return CalibrationPosition
# initialize
CalibrationPosition: Any = [[[]] * nb_calibration_postions]

for i_pos in range(nb_calibration_postions):

results_for_this_position = calibration_df[calibration_df[5] == i_pos]

for i, calibration in enumerate(results_for_this_position.iterrows()):
values = calibration[1][8].split(",")

if len(CalibrationPosition) < i + 1:
CalibrationPosition.append([[]] * nb_calibration_postions)

CalibrationPosition[i][i_pos] = [int(x) for x in values]

CalibrationPosition.extend(
cal_pos[i : i + cal_num] for i in range(0, len(cal_pos), cal_num)
)
return CalibrationPosition


def _extract_CalibrationUnit(df: pd.DataFrame) -> str:
if len(_get_calibration_positions(df)) == 0:
return ""

cal_unit = (
(df[df[2] == "VALIDATE"][[13]])
.iloc[0:1, 0:1]
@@ -206,14 +200,10 @@ def _has_validation(df: pd.DataFrame) -> bool:


def _extract_MaximalCalibrationError(df: pd.DataFrame) -> list[float]:
if not _has_validation(df):
return []
return np.array(_validations(df)[[11]]).astype(float).tolist()


def _extract_AverageCalibrationError(df: pd.DataFrame) -> list[float]:
if not _has_validation(df):
return []
return np.array(_validations(df)[[9]]).astype(float).tolist()


@@ -241,7 +231,7 @@ def _extract_SamplingFrequency(df: pd.DataFrame) -> int:
return int(df[df[2] == "RECCFG"].iloc[0:1, 2:3].to_string(header=False, index=False))


def _extract_RecordedEye(df: pd.DataFrame) -> str:
def _extract_RecordedEye(df: pd.DataFrame) -> str | list[str]:
eye = df[df[2] == "RECCFG"].iloc[0:1, 5:6].to_string(header=False, index=False)
if eye == "L":
return "Left"
@@ -400,6 +390,85 @@ def _physioevents_eye2 (physioevents_reordered: pd.DataFrame) -> pd.DataFrame:
return physioevents_eye2


def generate_physio_json(
input_file: Path,
metadata_file: str | Path | None,
output_dir: Path,
events_asc_file: Path,
) -> None:
"""Generate the _physio.json."""
if metadata_file is None:
metadata = {}
else:
with open(metadata_file) as f:
metadata = yaml.load(f, Loader=SafeLoader)

events = _load_asc_file(events_asc_file)
df_ms = _load_asc_file_as_df(events_asc_file)
df_ms_reduced = _load_asc_file_as_reduced_df(events_asc_file)

base_json = BasePhysioJson(manufacturer="SR-Research", metadata=metadata)

base_json.input_file = input_file
base_json.has_validation = _has_validation(df_ms_reduced)
base_json.two_eyes = _2eyesmode(df_ms_reduced)

base_json["ManufacturersModelName"] = _extract_ManufacturersModelName(events)
base_json["DeviceSerialNumber"] = _extract_DeviceSerialNumber(events)
base_json["EyeTrackingMethod"] = _extract_EyeTrackingMethod(events)
base_json["PupilFitMethod"] = _extract_PupilFitMethod(df_ms_reduced)
base_json["SamplingFrequency"] = _extract_SamplingFrequency(df_ms_reduced)

base_json["StartTime"] = _extract_StartTime(events)
base_json["StopTime"] = _extract_StopTime(events)

if base_json.two_eyes:
metadata_eye1: dict[str, str | list[str] | list[float]] = {
"RecordedEye": (_extract_RecordedEye(df_ms_reduced)[0]),
}
metadata_eye2: dict[str, str | list[str] | list[float]] = {
"RecordedEye": (_extract_RecordedEye(df_ms_reduced)[1]),
}
else:
metadata_eye1 = {
"RecordedEye": (_extract_RecordedEye(df_ms_reduced)),
}

if base_json.has_validation:

if CalibrationPosition := _extract_CalibrationPosition(df_ms_reduced):
base_json["CalibrationCount"] = _extract_CalibrationCount(
df_ms_reduced, two_eyes=base_json.two_eyes
)
base_json["CalibrationUnit"] = _extract_CalibrationUnit(df_ms_reduced)
base_json["CalibrationType"] = _extract_CalibrationType(df_ms_reduced)

base_json["CalibrationPosition"] = CalibrationPosition
if base_json["CalibrationCount"] == 1:
base_json["CalibrationPosition"] = CalibrationPosition[0]

metadata_eye1["AverageCalibrationError"] = _extract_AverageCalibrationError(
df_ms
)[::2]
metadata_eye1["MaximalCalibrationError"] = _extract_MaximalCalibrationError(
df_ms
)[::2]

if base_json.two_eyes:
metadata_eye2["AverageCalibrationError"] = _extract_AverageCalibrationError(
df_ms
)[1::2]
metadata_eye2["MaximalCalibrationError"] = _extract_MaximalCalibrationError(
df_ms
)[1::2]

base_json.write(output_dir=output_dir, recording="eye1", extra_metadata=metadata_eye1)
if base_json.two_eyes:
base_json.write(
output_dir=output_dir, recording="eye2", extra_metadata=metadata_eye2
)


def edf2bids(
input_file: str | Path | None = None,
metadata_file: str | Path | None = None,
@@ -423,8 +492,12 @@ def edf2bids(
f"{input_file}"
)

# %% Sidecar eye-physio.json
generate_physio_json(input_file, metadata_file, output_dir, events_asc_file)

# %% physioevents.json Metadata
events = _load_asc_file(events_asc_file)
df_ms = _load_asc_file_as_df(events_asc_file)

df_ms_reduced = _load_asc_file_as_reduced_df(events_asc_file)

if metadata_file is None:
@@ -433,145 +506,23 @@ def edf2bids(
with open(metadata_file) as f:
metadata = yaml.load(f, Loader=SafeLoader)

# eye-physio.json Metadata
base_json = {
"PhysioType": "eyetrack",
"Columns": ["x_coordinate", "y_coordinate", "pupil_size", "timestamp"],
"x_coordinate": {
"Description": "Gaze position x-coordinate of the recorded eye, in the coordinate units specified in the corresponding metadata sidecar.",
"Units": "a.u.",
},
"y_coordinate": {
"Description": "Gaze position y-coordinate of the recorded eye, in the coordinate units specified in the corresponding metadata sidecar.",
"Units": "a.u.",
},
"pupil_size": {
"Description": "Pupil area of the recorded eye as calculated by the eye-tracker in arbitrary units (see EyeLink's documentation for conversion).",
"Units": "a.u.",
},
"timestamp": {
"Description": "Timestamp issued by the eye-tracker indexing the continuous recordings corresponding to the sampled eye."
},
"Manufacturer": "SR-Research",
"ManufacturersModelName": _extract_ManufacturersModelName(events),
"DeviceSerialNumber": _extract_DeviceSerialNumber(events),
"EnvironmentCoordinates": metadata.get("EnvironmentCoordinates"),
"SoftwareVersions": metadata.get("SoftwareVersion"),
"EyeCameraSettings": metadata.get("EyeCameraSettings"),
"EyeTrackerDistance": metadata.get("EyeTrackerDistance"),
"FeatureDetectionSettings": metadata.get("FeatureDetectionSettings"),
"GazeMappingSettings": metadata.get("GazeMappingSettings"),
"RawDataFilters": metadata.get("RawDataFilters"),
"SampleCoordinateSystem": metadata.get("SampleCoordinateSystem"),
"SampleCoordinateUnits": metadata.get("SampleCoordinateUnits"),
"ScreenAOIDefinition": metadata.get("ScreenAOIDefinition"),
"EyeTrackingMethod": _extract_EyeTrackingMethod(events),
"PupilFitMethod": _extract_PupilFitMethod(df_ms_reduced),
"SamplingFrequency": _extract_SamplingFrequency(df_ms_reduced),
"StartTime": _extract_StartTime(events),
"StopTime": _extract_StopTime(events),
"CalibrationUnit": _extract_CalibrationUnit(df_ms_reduced),
"CalibrationType": _extract_CalibrationType(df_ms_reduced),
"CalibrationCount": _extract_CalibrationCount(df_ms_reduced),
"CalibrationPosition": _extract_CalibrationPosition(df_ms_reduced),
}

if _2eyesmode(df_ms_reduced) == True:
metadata_eye1 = {
"AverageCalibrationError": (_extract_AverageCalibrationError(df_ms)[0::2]),
"MaximalCalibrationError": (_extract_MaximalCalibrationError(df_ms)[0::2]),
"RecordedEye": (_extract_RecordedEye(df_ms_reduced)[0]),
}

metadata_eye2 = {
"AverageCalibrationError": (_extract_AverageCalibrationError(df_ms)[1::2]),
"MaximalCalibrationError": (_extract_MaximalCalibrationError(df_ms)[1::2]),
"RecordedEye": (_extract_RecordedEye(df_ms_reduced)[1]),
}
else:
metadata_eye1 = {
"AverageCalibrationError": (_extract_AverageCalibrationError(df_ms)[0::2]),
"MaximalCalibrationError": (_extract_MaximalCalibrationError(df_ms)[0::2]),
"RecordedEye": (_extract_RecordedEye(df_ms_reduced))
}

json_eye1 = dict(base_json, **metadata_eye1)
if _2eyesmode(df_ms_reduced) == True:
json_eye2 = dict(base_json, **metadata_eye2)
events_json = BasePhysioEventsJson(metadata)

# to json
events_json.input_file = input_file
events_json.two_eyes = _2eyesmode(df_ms_reduced)

output_filename_eye1 = generate_output_filename(
output_dir=output_dir,
input_file=input_file,
suffix="_recording-eye1_physio",
extension="json",
events_json["TaskName"] = _extract_TaskName(events)
events_json["StimulusPresentation"]["ScreenResolution"] = _extract_ScreenResolution(
df_ms_reduced
)
with open(output_filename_eye1, "w") as outfile:
json.dump(json_eye1, outfile, indent=4)

e2b_log.info(f"file generated: {output_filename_eye1}")
events_json.write(output_dir=output_dir, recording="eye1")
if events_json.two_eyes:
events_json.write(output_dir=output_dir, recording="eye2")

if _2eyesmode(df_ms_reduced) == True:
output_filename_eye2 = generate_output_filename(
output_dir=output_dir,
input_file=input_file,
suffix="_recording-eye2_physio",
extension="json",
)
with open(output_filename_eye2, "w") as outfile:
json.dump(json_eye2, outfile, indent=4)

e2b_log.info(f"file generated: {output_filename_eye2}")

# physioevents.json Metadata

events_json = {
"Columns": ["onset", "duration", "trial_type", "blink", "message"],
"Description": "Messages logged by the measurement device",
"ForeignIndexColumn": "timestamp",
"blink": {"Description": "One indicates if the eye was closed, zero if open."},
"message": {"Description": "String messages logged by the eye-tracker."},
"trial_type": {
"Description": "Event type as identified by the eye-tracker's model (either 'n/a' if not applicabble, 'fixation', or 'saccade')."
},
"TaskName": _extract_TaskName(events),
"InstitutionAddress": metadata.get("InstitutionAddress"),
"InstitutionName": metadata.get("InstitutionName"),
"StimulusPresentation": {
"ScreenDistance": metadata.get("ScreenDistance"),
"ScreenRefreshRate": metadata.get("ScreenRefreshRate"),
"ScreenSize": metadata.get("ScreenSize"),
"ScreenResolution": _extract_ScreenResolution(df_ms_reduced),
},
}

output_filename_eye1 = generate_output_filename(
output_dir=output_dir,
input_file=input_file,
suffix="_recording-eye1_physioevents",
extension="json"
)
with open(output_filename_eye1, "w") as outfile:
json.dump(events_json, outfile, indent=4)

e2b_log.info(f"file generated: {output_filename_eye1}")

if _2eyesmode(df_ms_reduced) == True:

output_filename_eye2 = generate_output_filename(
output_dir=output_dir,
input_file=input_file,
suffix="_recording-eye2_physioevents",
extension="json",
)
with open(output_filename_eye2, "w") as outfile:
json.dump(events_json, outfile, indent=4)

e2b_log.info(f"file generated: {output_filename_eye2}")

# %%
# Samples to dataframe

samples_asc_file = _convert_edf_to_asc_samples(input_file)
if not samples_asc_file.exists():
e2b_log.error(
@@ -586,9 +537,10 @@ def edf2bids(
.replace(".", np.nan, regex=False)
)

if _2eyesmode(df_ms_reduced) == True:
samples_eye2 = pd.DataFrame(samples.iloc[:, [4, 5, 6, 0]])
if _2eyesmode(df_ms_reduced):
samples_eye2 = pd.DataFrame(samples.iloc[:, [0, 4, 5, 6]])

# %%
# Samples to eye_physio.tsv.gz

output_filename_eye1 = generate_output_filename(
@@ -603,7 +555,7 @@ def edf2bids(

e2b_log.info(f"file generated: {output_filename_eye1}")

if _2eyesmode(df_ms_reduced) == True:
if _2eyesmode(df_ms_reduced):

output_filename_eye2 = generate_output_filename(
output_dir=output_dir,
17 changes: 3 additions & 14 deletions tests/test_edf2bids.py
Original file line number Diff line number Diff line change
@@ -262,7 +262,8 @@ def test_2files_eye2(eyelink_test_data_dir):

@pytest.mark.skipif(not _check_edf2asc_present(), reason="edf2asc missing")
def test_number_columns_2eyes_tsv(eyelink_test_data_dir):
"""Check that values for only one eye were extracted in eye1-physio.tsv.gz by number of columns.
"""Check that values for only one eye were extracted \
in eye1-physio.tsv.gz by number of columns.
function _samples_to_data_frame
"""
@@ -348,12 +349,8 @@ def test_extract_ScreenResolution(folder, expected, eyelink_test_data_dir):
@pytest.mark.parametrize(
"folder, expected",
[
("emg", ""),
("lt", "pixel"),
("pitracker", ""),
("rest", "pixel"),
("satf", ""),
("vergence", ""),
("2eyes", "pixel"),
],
)
@@ -367,7 +364,6 @@ def test_extract_CalibrationUnit(folder, expected, eyelink_test_data_dir):
@pytest.mark.parametrize(
"folder, expected",
[
("emg", []),
(
"lt",
[
@@ -395,7 +391,6 @@ def test_extract_CalibrationUnit(folder, expected, eyelink_test_data_dir):
],
],
),
("pitracker", []),
(
"rest",
[
@@ -416,8 +411,6 @@ def test_extract_CalibrationUnit(folder, expected, eyelink_test_data_dir):
]
],
),
("satf", []),
("vergence", []),
(
"2eyes",
[
@@ -435,7 +428,7 @@ def test_extract_CalibrationUnit(folder, expected, eyelink_test_data_dir):
[1126, 636],
[794, 444],
[960, 348],
]
],
],
),
],
@@ -566,12 +559,8 @@ def test_extract_ManufacturersModelName(folder, expected, eyelink_test_data_dir)
@pytest.mark.parametrize(
"folder, expected",
[
("emg", []),
("lt", [[0.32], [0.37]]),
("pitracker", []),
("rest", [[0.9]]),
("satf", []),
("vergence", []),
(
"2eyes",
[[0.62], [1.21]],

0 comments on commit 302c84b

Please sign in to comment.