Skip to content

Commit

Permalink
[pre-commit.ci] auto fixes from pre-commit.com hooks
Browse files Browse the repository at this point in the history 
for more information, see https://pre-commit.ci
  • Loading branch information
@pre-commit-ci
pre-commit-ci[bot] committed Apr 10, 2024
1 parent 9244429 commit d5bef7f
Show file tree
Hide file tree
Showing 2 changed files with 115 additions and 89 deletions.
190 changes: 104 additions & 86 deletions eye2bids/edf2bids.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,7 @@ def _convert_edf_to_asc_samples(input_file: str | Path) -> Path:
subprocess.run(["edf2asc", "-y", "-s", input_file, "-o", samples_asc_file])
return Path(samples_asc_file).with_suffix(".asc")


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":
Expand All @@ -123,9 +124,11 @@ def _2eyesmode(df: pd.DataFrame) -> bool:
two_eyes = False
return two_eyes


def _calibrations(df: pd.DataFrame) -> pd.DataFrame:
return df[df[3] == "CALIBRATION"]


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

Expand Down Expand Up @@ -245,7 +248,7 @@ def _extract_RecordedEye(df: pd.DataFrame) -> str:
return "Right"
elif eye == "LR":
return ["Left", "Right"]
return ""
return ""


def _extract_ScreenResolution(df: pd.DataFrame) -> list[int]:
Expand Down Expand Up @@ -319,6 +322,7 @@ def _load_asc_file_as_reduced_df(events_asc_file: str | Path) -> pd.DataFrame:
df_ms = _load_asc_file_as_df(events_asc_file)
return pd.DataFrame(df_ms.iloc[0:, 2:])


def edf2bids(
input_file: str | Path | None = None,
metadata_file: str | Path | None = None,
Expand Down Expand Up @@ -352,89 +356,94 @@ def edf2bids(
with open(metadata_file) as f:
metadata = yaml.load(f, Loader=SafeLoader)


# eye-physio.json Metadata
base_json = {
"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"),
"SoftwareVersion": 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)
}
"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"),
"SoftwareVersion": 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])
}
"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])
"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))
}
"AverageCalibrationError": (_extract_AverageCalibrationError(df_ms)[0::2]),
"MaximalCalibrationError": (_extract_MaximalCalibrationError(df_ms)[0::2]),
"RecordedEye": (_extract_RecordedEye(df_ms_reduced)),
}

json_eye1 = base_json | metadata_eye1
if _2eyesmode(df_ms_reduced) == True:
json_eye2 = base_json | metadata_eye2

# to json

output_filename_eye1 = generate_output_filename(
output_dir=output_dir, input_file=input_file, suffix="_recording-eye1_physio", extension="json"
)
output_dir=output_dir,
input_file=input_file,
suffix="_recording-eye1_physio",
extension="json",
)
with open(output_filename_eye1, "w") as outfile:
json.dump(json_eye1, outfile, indent=4)
json.dump(json_eye1, 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_physio", extension="json"
)
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
Expand All @@ -443,30 +452,28 @@ def edf2bids(
"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."
},
"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')."
"Description": "Event type as identified by the eye-tracker's model (either 'n/a' if not applicabble, 'fixation', or 'saccade')."
},
"TaskName": _extract_TaskName(events),
"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"
)
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)

Expand All @@ -475,11 +482,14 @@ def edf2bids(
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"
)
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
Expand All @@ -492,16 +502,23 @@ def edf2bids(
)

samples = pd.read_csv(samples_asc_file, sep="\t", header=None)
samples_eye1 = pd.DataFrame(samples.iloc[:, [2, 1, 3, 0]]).map(lambda x: x.strip() if isinstance(x, str) else x).replace(".", np.nan, regex=False)

samples_eye1 = (
pd.DataFrame(samples.iloc[:, [2, 1, 3, 0]])
.map(lambda x: x.strip() if isinstance(x, str) else x)
.replace(".", np.nan, regex=False)
)

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

# Samples to eye_physio.tsv.gz

output_filename_eye1 = generate_output_filename(
output_dir=output_dir, input_file=input_file, suffix="_recording-eye1_physio", extension="tsv.gz"
)
output_dir=output_dir,
input_file=input_file,
suffix="_recording-eye1_physio",
extension="tsv.gz",
)
content = samples_eye1.to_csv(sep="\t", index=False, na_rep="n/a", header=None)
with gzip.open(output_filename_eye1, "wb") as f:
f.write(content.encode())
Expand All @@ -511,19 +528,20 @@ def edf2bids(
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="tsv.gz"
)
output_dir=output_dir,
input_file=input_file,
suffix="_recording-eye2_physio",
extension="tsv.gz",
)
content = samples_eye2.to_csv(sep="\t", index=False, na_rep="n/a", header=None)
with gzip.open(output_filename_eye2, "wb") as f:
f.write(content.encode())

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

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

# Messages and events to physioevents.tsv.gz - tbc



def generate_output_filename(
output_dir: Path, input_file: Path, suffix: str, extension: str
) -> Path:
Expand Down
14 changes: 11 additions & 3 deletions tests/test_edf2bids.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,9 @@ def test_edf_end_to_end(metadata_file, eyelink_test_data_dir):
output_dir=output_dir,
)

expected_events_sidecar = output_dir / f"{input_file.stem}_recording-eye1_physioevents.json"
expected_events_sidecar = (
output_dir / f"{input_file.stem}_recording-eye1_physioevents.json"
)
assert expected_events_sidecar.exists()
with open(expected_events_sidecar) as f:
events = json.load(f)
Expand Down Expand Up @@ -88,6 +90,7 @@ def test_edf_nan_in_tsv(eyelink_test_data_dir):
count = sum(i == "." for i in df[0])
assert count == 0


@pytest.mark.skipif(not _check_edf2asc_present(), reason="edf2asc missing")
def test_2files_eye1(eyelink_test_data_dir):
"""Check that for datafile with 2eyes 2 eye1 file is created and check input.
Expand All @@ -105,13 +108,16 @@ def test_2files_eye1(eyelink_test_data_dir):
output_dir=output_dir,
)

expected_eyetrack_sidecar = output_dir / f"{input_file.stem}_recording-eye1_physio.json"
expected_eyetrack_sidecar = (
output_dir / f"{input_file.stem}_recording-eye1_physio.json"
)
assert expected_eyetrack_sidecar.exists()
with open(expected_eyetrack_sidecar) as f:
eyetrack = json.load(f)
assert eyetrack["AverageCalibrationError"] == [[0.29]]
assert eyetrack["RecordedEye"] == "Left"


@pytest.mark.skipif(not _check_edf2asc_present(), reason="edf2asc missing")
def test_2files_eye2(eyelink_test_data_dir):
"""Check that for datafile with 2eyes 2 eye2 file is created and check input.
Expand All @@ -129,7 +135,9 @@ def test_2files_eye2(eyelink_test_data_dir):
output_dir=output_dir,
)

expected_eyetrack_sidecar = output_dir / f"{input_file.stem}_recording-eye2_physio.json"
expected_eyetrack_sidecar = (
output_dir / f"{input_file.stem}_recording-eye2_physio.json"
)
assert expected_eyetrack_sidecar.exists()
with open(expected_eyetrack_sidecar) as f:
eyetrack = json.load(f)
Expand Down

0 comments on commit d5bef7f

Please sign in to comment.