diff --git a/.gitignore b/.gitignore index 71cc6ac..3e33579 100644 --- a/.gitignore +++ b/.gitignore @@ -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] diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index e1e8fec..396b691 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -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] diff --git a/eye2bids/_base.py b/eye2bids/_base.py new file mode 100644 index 0000000..f32cd3b --- /dev/null +++ b/eye2bids/_base.py @@ -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) diff --git a/eye2bids/edf2bids.py b/eye2bids/edf2bids.py index ac43bd2..e2abecf 100644 --- a/eye2bids/edf2bids.py +++ b/eye2bids/edf2bids.py @@ -3,9 +3,9 @@ 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 @@ -13,6 +13,7 @@ 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, diff --git a/tests/test_edf2bids.py b/tests/test_edf2bids.py index 90d479b..195497b 100644 --- a/tests/test_edf2bids.py +++ b/tests/test_edf2bids.py @@ -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]],