From fb01309a690e4c9da4c4c06f91fbc4f6302aa728 Mon Sep 17 00:00:00 2001 From: saphjra <49561526+saphjra@users.noreply.github.com> Date: Thu, 19 Feb 2026 09:16:40 +0100 Subject: [PATCH 1/5] made the hard coded labnum in line 234 of stimulus.py variable to be able to parse data from same county but different labs --- preprocessing/data_collection/stimulus.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/preprocessing/data_collection/stimulus.py b/preprocessing/data_collection/stimulus.py index 11aa7dd2..8d2abe03 100644 --- a/preprocessing/data_collection/stimulus.py +++ b/preprocessing/data_collection/stimulus.py @@ -242,7 +242,7 @@ def load( ) instruction_image_path = ( stimulus_dir - / f"participant_instructions_images_{lang}_{country}_1/{instruction_row['instruction_screen_img_name']}" + / f"participant_instructions_images_{lang}_{country}_{labnum}/{instruction_row['instruction_screen_img_name']}" ) instruction = class_name( From e7d29619a60299c0b1abefc2a8a2779ccd7a066b Mon Sep 17 00:00:00 2001 From: saphjra <49561526+saphjra@users.noreply.github.com> Date: Thu, 19 Feb 2026 16:36:00 +0100 Subject: [PATCH 2/5] create new function _parse_messages to substitute _parse_from_Asc in the future --- preprocessing/constants.py | 32 ++++++ .../multipleye_data_collection.py | 106 +++++++++++++++++- 2 files changed, 136 insertions(+), 2 deletions(-) create mode 100644 preprocessing/constants.py diff --git a/preprocessing/constants.py b/preprocessing/constants.py new file mode 100644 index 00000000..9a23cbbf --- /dev/null +++ b/preprocessing/constants.py @@ -0,0 +1,32 @@ + +MESSAGES = { + "other_screens": [ + "welcome_screen", + "informed_consent_screen", + "start_experiment", + "stimulus_order_version", + "showing_instruction_screen", + "camera_setup_screen", + "practice_text_starting_screen", + "transition_screen", + "final_validation", + "show_final_screen", + "optional_break_screen", + "fixation_trigger:skipped_by_experimenter", + "fixation_trigger:experimenter_calibration_triggered", + "recalibration", + "empty_screen", + "obligatory_break", + "optional_break", + ], + "break_msgs": [ + "optional_break_duration", + "optional_break_end", + "optional_break_", + "obligatory_break_duration", + "obligatory_break_endobligatory_break", + ], +} + +BREAK_REGEX = re.compile("|".join(map(re.escape, MESSAGES["break_msgs"]))) +OTHER_SCREENS_REGEX = re.compile("|".join(map(re.escape, MESSAGES["other_screens"]))) diff --git a/preprocessing/data_collection/multipleye_data_collection.py b/preprocessing/data_collection/multipleye_data_collection.py index 7955dc56..8189d58a 100644 --- a/preprocessing/data_collection/multipleye_data_collection.py +++ b/preprocessing/data_collection/multipleye_data_collection.py @@ -946,6 +946,108 @@ def _load_session_stimulus_order( f"Please add the used stimulus folder from the experiment. Or check the stimulus order versions file for missing IDs or duplicates." ) + def _create_empty_rt_frame(self, session_identifier: str): + """create an empty dataframe for the reading times of the stimuli in the session, the dataframe will be filled + with the information from the messages extracted by pm. The dataframe is created based on the stimulus trial mapping, + so that all completed stimuli are included in the dataframe. It also serves as a reference frame for the number of expected messages + This is important for the sanity checks + Also the data frame schema is here defined and can be adjusted""" + stimuli_trial_mapping = self.sessions[session_identifier].stimuli_trial_mapping + + rt_schema = pl.Schema( + { + "stimulus_name": pl.String, + "start_ts": pl.String, + "stop_ts": pl.String, + "start_msg": pl.String, + "stop_msg": pl.String, + "duration_ms": pl.String, + "duration_str": pl.String, + "trials": pl.String, + "pages": pl.String, + "status": pl.String, + } + ) + + stimulus_names = list(stimuli_trial_mapping.values()) + num_of_trials = len(stimulus_names) + rt_df = pl.DataFrame( + { + "stimulus_name": stimulus_names, + "start_ts": [None] * num_of_trials, + "stop_ts": [None] * num_of_trials, + "start_msg": [None] * num_of_trials, + "stop_msg": [None] * num_of_trials, + "duration_ms": [None] * num_of_trials, + "duration_str": [None] * num_of_trials, + "trials": [None] * num_of_trials, + "pages": [None] * num_of_trials, + "status": [None] * num_of_trials, + }, + schema=rt_schema, + ) + return rt_df + + def _create_empty_break_frame(self, session_identifier: str): + """create an empty dataframe for the break data, the dataframe will be filled + with the information from the messages extracted by pm. The dataframe is created based on the stimulus trial mapping, + so that all completed stimuli are included in the dataframe. It also serves as a reference frame for the number of expected messages + This is important for the sanity checks + Also the data frame schema is here defined and can be adjusted""" + breaks_schema = pl.Schema( + { + "start_ts": pl.String(), + "stop_ts": pl.String(), + "duration_ms": pl.String(), + "type": pl.String(), + } + ) + num_of_breaks = -1 # we start with -1 because there is always one break less than the number of stimuli + for stimulus in self.sessions[session_identifier].stimuli: + num_of_breaks += 1 if stimulus.type == "experiment" else 0 + + breaks_df = pl.DataFrame( + { + "start_ts": [None] * num_of_breaks, + "stop_ts": [None] * num_of_breaks, + "duration_ms": [None] * num_of_breaks, + "type": [None] * num_of_breaks, + }, + schema=breaks_schema, + ) + + return breaks_df + + def _parse_messages(self, session_identifier: str): + """function to parse the messages create by pm, intended to replace _parse_asc""" + + rt_frame = multipleye._create_empty_rt_frame(session_identifier) + + break_msg = [] + messages = self.sessions[session_identifier].messages.copy() + for _ in range(len(self.sessions[session_identifier].messages)): + msg = messages.pop( + 0 + ) # zero has to be added because otherwise the last item in the list gets popped instead of the first one, which is what we want + if match := BREAK_REGEX.match(msg["message"]): + break_msg.append(msg) + + elif match := START_RECORDING_REGEX.match(msg["message"]): + event = match.groupdict() + timestamp = msg["timestamp"] + event["start_ts"] = timestamp + df = pl.DataFrame(event) + rt_frame = rt_frame.update(df, on="stimulus_name", how="left") + + elif match := STOP_RECORDING_REGEX.match(msg["message"]): + event = match.groupdict() + timestamp = msg["timestamp"] + event["stop_ts"] = timestamp + df = pl.DataFrame(event) + rt_frame = rt_frame.update(df, on="stimulus_name", how="left") + + print(rt_frame) + def _parse_asc(self, session_identifier: str): """ qick fix for now, should be replaced by the summary experiment frame later on, however the extraction of the @@ -987,8 +1089,8 @@ def _parse_asc(self, session_identifier: str): "stop_msg": [], "duration_ms": [], "duration_str": [], - "trials": [], - "pages": [], + "trial": [], + "page": [], "status": [], "stimulus_name": [], } From 1915635cdb3483fd49d4be127244f1e075366bad Mon Sep 17 00:00:00 2001 From: saphjra <49561526+saphjra@users.noreply.github.com> Date: Fri, 20 Feb 2026 12:57:55 +0100 Subject: [PATCH 3/5] adapted run_merid_preprocessing.py to new workflow, changed also the _init_.py for the datacollection subfolder, further adapted the _parse_from_message method in multipleye_data_collection.py --- .../multipleye_data_collection.py | 73 ++++++++++++------- 1 file changed, 47 insertions(+), 26 deletions(-) diff --git a/preprocessing/data_collection/multipleye_data_collection.py b/preprocessing/data_collection/multipleye_data_collection.py index 8189d58a..5469132b 100644 --- a/preprocessing/data_collection/multipleye_data_collection.py +++ b/preprocessing/data_collection/multipleye_data_collection.py @@ -951,8 +951,7 @@ def _create_empty_rt_frame(self, session_identifier: str): with the information from the messages extracted by pm. The dataframe is created based on the stimulus trial mapping, so that all completed stimuli are included in the dataframe. It also serves as a reference frame for the number of expected messages This is important for the sanity checks - Also the data frame schema is here defined and can be adjusted""" - stimuli_trial_mapping = self.sessions[session_identifier].stimuli_trial_mapping + Also the data frame schema is here defined and can be adjusted""" # ToDo unsure if it works with interupted sessions yet rt_schema = pl.Schema( { @@ -969,20 +968,33 @@ def _create_empty_rt_frame(self, session_identifier: str): } ) - stimulus_names = list(stimuli_trial_mapping.values()) - num_of_trials = len(stimulus_names) + num_of_pages_per_trial = { + stimulus.name: [page.number for page in stimulus.pages] + for stimulus in self.sessions[session_identifier].stimuli + } + d = {"stimulus": [], "pages": []} + num_of_pages = 0 + for stimulus in self.sessions[ + session_identifier + ].stimuli_trial_mapping.values(): + d["stimulus"].extend([stimulus] * len(num_of_pages_per_trial[stimulus])) + d["pages"].extend( + [f"page_{num}" for num in num_of_pages_per_trial[stimulus]] + ) + num_of_pages += len(num_of_pages_per_trial[stimulus]) + rt_df = pl.DataFrame( { - "stimulus_name": stimulus_names, - "start_ts": [None] * num_of_trials, - "stop_ts": [None] * num_of_trials, - "start_msg": [None] * num_of_trials, - "stop_msg": [None] * num_of_trials, - "duration_ms": [None] * num_of_trials, - "duration_str": [None] * num_of_trials, - "trials": [None] * num_of_trials, - "pages": [None] * num_of_trials, - "status": [None] * num_of_trials, + "stimulus_name": d["stimulus"], + "start_ts": [None] * num_of_pages, + "stop_ts": [None] * num_of_pages, + "start_msg": [None] * num_of_pages, + "stop_msg": [None] * num_of_pages, + "duration_ms": [None] * num_of_pages, + "duration_str": [None] * num_of_pages, + "trials": [None] * num_of_pages, + "pages": d["pages"], + "status": [None] * num_of_pages, }, schema=rt_schema, ) @@ -1004,7 +1016,15 @@ def _create_empty_break_frame(self, session_identifier: str): ) num_of_breaks = -1 # we start with -1 because there is always one break less than the number of stimuli for stimulus in self.sessions[session_identifier].stimuli: - num_of_breaks += 1 if stimulus.type == "experiment" else 0 + num_of_breaks += ( + 1 + if stimulus.type == "experiment" + and stimulus.name + in list( + self.sessions[session_identifier].stimuli_trial_mapping.values() + ) + else 0 + ) breaks_df = pl.DataFrame( { @@ -1032,21 +1052,21 @@ def _parse_messages(self, session_identifier: str): if match := BREAK_REGEX.match(msg["message"]): break_msg.append(msg) - elif match := START_RECORDING_REGEX.match(msg["message"]): + elif match := START_RECORDING_REGEX_NEW.match(msg["message"]): event = match.groupdict() timestamp = msg["timestamp"] event["start_ts"] = timestamp df = pl.DataFrame(event) rt_frame = rt_frame.update(df, on="stimulus_name", how="left") - elif match := STOP_RECORDING_REGEX.match(msg["message"]): + elif match := STOP_RECORDING_REGEX_NEW.match(msg["message"]): event = match.groupdict() timestamp = msg["timestamp"] event["stop_ts"] = timestamp df = pl.DataFrame(event) rt_frame = rt_frame.update(df, on="stimulus_name", how="left") - print(rt_frame) + return rt_frame def _parse_asc(self, session_identifier: str): """ @@ -1089,8 +1109,8 @@ def _parse_asc(self, session_identifier: str): "stop_msg": [], "duration_ms": [], "duration_str": [], - "trial": [], - "page": [], + "trials": [], + "pages": [], "status": [], "stimulus_name": [], } @@ -1178,12 +1198,13 @@ def _parse_asc(self, session_identifier: str): ) if not in_break: - breaks_df = pd.DataFrame(breaks) - breaks_df.to_csv( - result_folder / f"breaks_{session_identifier}.tsv", - sep="\t", - index=False, - ) + pass + # breaks_df = pd.DataFrame(breaks) + # breaks_df.to_csv( + # result_folder / f"breaks_{session_identifier}.tsv", + # sep="\t", + # index=False, + # ) else: self.logger.warning( f"Session {session_identifier} did not finish a break properly, " From d85c087c184c0a20a2b7c7767fc60e8943f9b5c4 Mon Sep 17 00:00:00 2001 From: saphjra <49561526+saphjra@users.noreply.github.com> Date: Fri, 20 Feb 2026 14:49:20 +0100 Subject: [PATCH 4/5] changed _parse_messages to update dataframe by matching stimulus name and page --- .../data_collection/multipleye_data_collection.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/preprocessing/data_collection/multipleye_data_collection.py b/preprocessing/data_collection/multipleye_data_collection.py index 5469132b..4f053d3d 100644 --- a/preprocessing/data_collection/multipleye_data_collection.py +++ b/preprocessing/data_collection/multipleye_data_collection.py @@ -1041,7 +1041,7 @@ def _create_empty_break_frame(self, session_identifier: str): def _parse_messages(self, session_identifier: str): """function to parse the messages create by pm, intended to replace _parse_asc""" - rt_frame = multipleye._create_empty_rt_frame(session_identifier) + rt_frame = self._create_empty_rt_frame(session_identifier) break_msg = [] messages = self.sessions[session_identifier].messages.copy() @@ -1057,14 +1057,18 @@ def _parse_messages(self, session_identifier: str): timestamp = msg["timestamp"] event["start_ts"] = timestamp df = pl.DataFrame(event) - rt_frame = rt_frame.update(df, on="stimulus_name", how="left") + rt_frame = rt_frame.update( + df, on=["stimulus_name", "pages"], how="left" + ) elif match := STOP_RECORDING_REGEX_NEW.match(msg["message"]): event = match.groupdict() timestamp = msg["timestamp"] event["stop_ts"] = timestamp df = pl.DataFrame(event) - rt_frame = rt_frame.update(df, on="stimulus_name", how="left") + rt_frame = rt_frame.update( + df, on=["stimulus_name", "pages"], how="left" + ) return rt_frame From b17b0ba7a260eaca0b19b0bb07f2688a36af9d51 Mon Sep 17 00:00:00 2001 From: theDebbister Date: Fri, 22 May 2026 14:37:13 +0200 Subject: [PATCH 5/5] integreate new reading times fully --- multipleye_settings_preprocessing.yaml | 7 +- preprocessing/config.py | 57 +++- preprocessing/constants.py | 32 -- .../multipleye_data_collection.py | 283 ++++++++---------- preprocessing/data_collection/session.py | 1 + preprocessing/data_collection/stimulus.py | 5 + preprocessing/io/load.py | 34 ++- preprocessing/metrics/calculate.py | 4 +- preprocessing/metrics/reading/words.py | 4 +- .../scripts/run_multipleye_preprocessing.py | 7 +- 10 files changed, 219 insertions(+), 215 deletions(-) delete mode 100644 preprocessing/constants.py diff --git a/multipleye_settings_preprocessing.yaml b/multipleye_settings_preprocessing.yaml index ac729b23..a45a9406 100644 --- a/multipleye_settings_preprocessing.yaml +++ b/multipleye_settings_preprocessing.yaml @@ -1,4 +1,4 @@ -data_collection_name: "MultiplEYE_DA_DK_Aalborg_1_2026" +data_collection_name: "MultiplEYE_ZH_CH_Zurich_1_2025" experiment_type: "MeRID" # can be either MultiplEYE or MeRID overwrite: False # defines whether written files will be overwritten, if they already exist. If False, preprocessed sessions will be skipped and not reprocessed. @@ -8,10 +8,11 @@ overwrite: False # defines whether written files will be overwritten, if they al # SESSIONS ## pick either include or exclude sessions, if none is specified, all sessions are included include_sessions: + - 089_ZH_CH_1_ET1 exclude_sessions: - - 021_DA_DK_1_ET1 + #- 021_DA_DK_1_ET1 -include_pilots: True +include_pilots: False # HARDWARE expected_sampling_rate_hz: 1000 # Hz diff --git a/preprocessing/config.py b/preprocessing/config.py index dae0c437..c04d981b 100644 --- a/preprocessing/config.py +++ b/preprocessing/config.py @@ -41,7 +41,7 @@ def _init_defaults(self) -> None: #: Whether to include sessions from the pilot folder. self.INCLUDE_PILOTS: bool = False - #: + #: Which type of experiment. E.g., supported are MultiplEYE and MeRID self.EXPERIMENT_TYPE: str = "" # Defines whether written files will be overwritten, if they already exist. @@ -205,12 +205,12 @@ def _init_defaults(self) -> None: #: Regex to identify the start of a recording for a trial/page. self.START_RECORDING_REGEX = re.compile( - rf"MSG\s+(?P\d+)\s+(?Pstart_recording)_(?P<{self.TRIAL_COL}>(?:PRACTICE_)?trial_\d\d?)_(?P<{self.PAGE_COL}>.*)" + r"(?Pstart_recording)_(?P(PRACTICE_)?trial_\d\d?)_stimulus__(?P\d+)_(?P\S*)" ) #: Regex to identify the stop of a recording for a trial/page. self.STOP_RECORDING_REGEX = re.compile( - rf"MSG\s+(?P\d+)\s+(?Pstop_recording)_(?P<{self.TRIAL_COL}>(?:PRACTICE_)?trial_\d\d?)_(?P<{self.PAGE_COL}>.*)" + r"(?Pstop_recording)_(?P(PRACTICE_)?trial_\d\d?)_stimulus_(?P\S*?_\S*?)_(?P\d+)_(?P\S*)" ) #: Glob pattern for raw data files. @@ -219,6 +219,9 @@ def _init_defaults(self) -> None: #: Glob pattern for event data files. self.EVENT_DATA_FILE_GLOB = "*_{event_type}.csv" + #: Glob patter for reading measures files + self.READING_MEASURES_GLOB = "*_reading_measures.csv" + #: Regex to extract the stimulus order version from ASC files. self.STIMULUS_ORDER_VERSION_REGEX = re.compile( r"MSG\s+\d+\s+stimulus_order_version:\s+(?P\d\d?\d?)\n" @@ -230,12 +233,52 @@ def _init_defaults(self) -> None: ) #: Regex to extract trial and stimulus info from raw data file names. - self.RAW_DATA_FILENAME_REGEX = rf".+_(?P<{self.TRIAL_COL}>(?:PRACTICE_)?trial_\d+)_(?P<{self.STIMULUS_COL}>[^_]+_[^_]+_\d+(\.0)?)_raw_data" + self.RAW_DATA_FILENAME_REGEX = rf"[^_]+_[^_]+_[^_]+_[^_]+_[^_]+_(?P<{self.TRIAL_COL}>(PRACTICE_)?trial_\d+)_(?P<{self.STIMULUS_COL}>[^_]+_[^_]+_\d+(\.0)?)_raw_data" #: Regex to extract trial and stimulus info from event data file names. - self.EVENT_DATA_FILENAME_REGEX = rf".+_(?P<{self.TRIAL_COL}>(?:PRACTICE_)?trial_\d+)_(?P<{self.STIMULUS_COL}>[^_]+_[^_]+_\d+(\.0)?)_{{event_type}}.csv" + self.EVENT_DATA_FILENAME_REGEX = rf"[^_]+_[^_]+_[^_]+_[^_]+_[^_]+_(?P<{self.TRIAL_COL}>(PRACTICE_)?trial_\d+)_(?P<{self.STIMULUS_COL}>[^_]+_[^_]+_\d+(\.0)?)_{{event_type}}.csv" + + #: Regex for the reading measures file name + self.READING_MEASURES_FILENAME_REGEX = r"[^_]+_[^_]+_[^_]+_[^_]+_[^_]+_(?P(PRACTICE_)?trial_\d+)_(?P[^_]+_[^_]+_\d+(\.0)?)_reading_measures.csv" + + multipleye_messages = { + "other_screens": [ + "welcome_screen", + "informed_consent_screen", + "start_experiment", + "stimulus_order_version", + "showing_instruction_screen", + "camera_setup_screen", + "practice_text_starting_screen", + "transition_screen", + "final_validation", + "show_final_screen", + "optional_break_screen", + "fixation_trigger:skipped_by_experimenter", + "fixation_trigger:experimenter_calibration_triggered", + "recalibration", + "empty_screen", + "obligatory_break", + "optional_break", + ], + "break_msgs": [ + "optional_break_duration", + "optional_break_end", + "optional_break", + "obligatory_break_duration", + "obligatory_break_end", + "obligatory_break", + ], + } - # --- HARDWARE AND STIMULI MAPPINGS --- + self.BREAK_REGEX = re.compile( + "|".join(map(re.escape, multipleye_messages["break_msgs"])) + ) + self.OTHER_SCREENS_REGEX = re.compile( + "|".join(map(re.escape, multipleye_messages["other_screens"])) + ) + + # --- HARDWARE --- #: Mapping of eye tracker brands to known model names. self.EYETRACKER_NAMES = { @@ -247,6 +290,8 @@ def _init_defaults(self) -> None: ], } + # --- STIMULI --- + #: Mapping of stimulus names to internal numeric IDs. self.STIMULUS_NAME_MAPPING = { "PopSci_MultiplEYE": 1, diff --git a/preprocessing/constants.py b/preprocessing/constants.py deleted file mode 100644 index 9a23cbbf..00000000 --- a/preprocessing/constants.py +++ /dev/null @@ -1,32 +0,0 @@ - -MESSAGES = { - "other_screens": [ - "welcome_screen", - "informed_consent_screen", - "start_experiment", - "stimulus_order_version", - "showing_instruction_screen", - "camera_setup_screen", - "practice_text_starting_screen", - "transition_screen", - "final_validation", - "show_final_screen", - "optional_break_screen", - "fixation_trigger:skipped_by_experimenter", - "fixation_trigger:experimenter_calibration_triggered", - "recalibration", - "empty_screen", - "obligatory_break", - "optional_break", - ], - "break_msgs": [ - "optional_break_duration", - "optional_break_end", - "optional_break_", - "obligatory_break_duration", - "obligatory_break_endobligatory_break", - ], -} - -BREAK_REGEX = re.compile("|".join(map(re.escape, MESSAGES["break_msgs"]))) -OTHER_SCREENS_REGEX = re.compile("|".join(map(re.escape, MESSAGES["other_screens"]))) diff --git a/preprocessing/data_collection/multipleye_data_collection.py b/preprocessing/data_collection/multipleye_data_collection.py index 4f053d3d..957da8f5 100644 --- a/preprocessing/data_collection/multipleye_data_collection.py +++ b/preprocessing/data_collection/multipleye_data_collection.py @@ -16,8 +16,8 @@ from polars.exceptions import ComputeError from tqdm import tqdm -from ..utils import pid_from_session from ..config import settings +from ..utils import pid_from_session from ..utils.conversion import convert_to_time_str from ..utils.logging import get_logger from ..checks.et_quality_checks import ( @@ -58,7 +58,6 @@ class MultipleyeDataCollection: crashed_session_ids: list[str] = [] num_sessions = 1 overview = {} - data_collection_name: str year: int country: str @@ -95,10 +94,6 @@ def __init__( self.year = year self.data_collection_name = data_collection_name - self.include_pilots = kwargs.get("include_pilots", False) - self.reports_dir = kwargs.get("output_dir", "") - self.pilot_folder = kwargs.get("pilot_folder", "") - for short_name, long_name in settings.EYETRACKER_NAMES.items(): if eye_tracker in long_name: self.eye_tracker = short_name @@ -118,20 +113,22 @@ def __init__( self.lab_configuration = lab_configuration self.data_root = data_root self.session_folder_regex = session_folder_regex - self.psychometric_tests = kwargs.get("psychometric_tests", []) self.excluded_sessions = excluded_sessions self.included_sessions = included_sessions self.logger = get_logger(__name__) + self.reports_folder = "reports" + + self.include_pilots = kwargs.get("include_pilots", False) + self.output_dir = kwargs.get("output_dir", "") + self.pilot_folder = kwargs.get("pilot_folder", "") + self.psychometric_tests = kwargs.get("psychometric_tests", []) + self.logger.info( f"MultipleyeDataCollection initialized. data_root: {self.data_root}" ) self.logger.info(f"Main config loaded from {self.config_file}") - if not self.reports_dir: - self.reports_dir = self.data_root.parent / "quality_reports" - self.reports_dir.mkdir(exist_ok=True) - self.add_recorded_sessions(self.data_root, self.session_folder_regex) if len(self.sessions) == 0: @@ -362,6 +359,7 @@ def create_from_data_folder( include_pilots: bool = False, excluded_sessions: list[str] | None = None, included_sessions: list[str] | None = None, + output_dir: Path | None = None, ) -> "MultipleyeDataCollection": """ :param data_dir: str path to the data folder @@ -475,6 +473,7 @@ def create_from_data_folder( ps_tests_path=ps_tests_path, included_sessions=included_sessions, excluded_sessions=excluded_sessions, + output_dir=output_dir, ) def create_sanity_check_report( @@ -501,14 +500,18 @@ def create_sanity_check_report( return if not output_dir: - output_dir = self.reports_dir + output_dir = self.output_dir - session_results = output_dir / session_name / "sanity_checks" + session_results = output_dir / session_name / self.reports_folder os.makedirs(session_results, exist_ok=True) report_file_path = session_results / f"{session_name}_{self.city}_report.txt" self.sessions[session_name].sanity_report_path = report_file_path + self.sessions[session_name].uncategorized_messages = self.parse_messages( + session_name + ) + if not report_file_path.exists() or overwrite: open(report_file_path, "w", encoding="utf-8").close() @@ -643,7 +646,6 @@ def prepare_session_level_information(self): self.sessions[session].completed_stimuli_ids, self.sessions[session].stimuli_trial_mapping, ) = self._load_session_completed_stimuli(session) - self.sessions[session].messages = self._parse_asc(session) self.sessions[session].logfile = self._load_session_logfile(session) self.sessions[ session @@ -951,7 +953,8 @@ def _create_empty_rt_frame(self, session_identifier: str): with the information from the messages extracted by pm. The dataframe is created based on the stimulus trial mapping, so that all completed stimuli are included in the dataframe. It also serves as a reference frame for the number of expected messages This is important for the sanity checks - Also the data frame schema is here defined and can be adjusted""" # ToDo unsure if it works with interupted sessions yet + Also the data frame schema is here defined and can be adjusted""" + # ToDo unsure if it works with interupted sessions yet rt_schema = pl.Schema( { @@ -962,8 +965,8 @@ def _create_empty_rt_frame(self, session_identifier: str): "stop_msg": pl.String, "duration_ms": pl.String, "duration_str": pl.String, - "trials": pl.String, - "pages": pl.String, + "trial": pl.String, + "page": pl.String, "status": pl.String, } ) @@ -992,8 +995,8 @@ def _create_empty_rt_frame(self, session_identifier: str): "stop_msg": [None] * num_of_pages, "duration_ms": [None] * num_of_pages, "duration_str": [None] * num_of_pages, - "trials": [None] * num_of_pages, - "pages": d["pages"], + "trial": [None] * num_of_pages, + "page": d["pages"], "status": [None] * num_of_pages, }, schema=rt_schema, @@ -1038,197 +1041,167 @@ def _create_empty_break_frame(self, session_identifier: str): return breaks_df - def _parse_messages(self, session_identifier: str): + def _categorize_asc_messages(self, session_identifier: str): """function to parse the messages create by pm, intended to replace _parse_asc""" - rt_frame = self._create_empty_rt_frame(session_identifier) - + reading_times_df = self._create_empty_rt_frame(session_identifier) break_msg = [] + other_screens = [] + uncategorized_msgs = [] + + initial_ts = 0 + messages = self.sessions[session_identifier].messages.copy() for _ in range(len(self.sessions[session_identifier].messages)): - msg = messages.pop( - 0 - ) # zero has to be added because otherwise the last item in the list gets popped instead of the first one, which is what we want - if match := BREAK_REGEX.match(msg["message"]): + msg = messages.pop(0) + + if not initial_ts: + initial_ts = msg["timestamp"] + + if settings.BREAK_REGEX.match(msg["message"]): break_msg.append(msg) - elif match := START_RECORDING_REGEX_NEW.match(msg["message"]): + elif settings.OTHER_SCREENS_REGEX.match(msg["message"]): + other_screens.append(msg) + + elif match := settings.START_RECORDING_REGEX.match(msg["message"]): event = match.groupdict() timestamp = msg["timestamp"] event["start_ts"] = timestamp df = pl.DataFrame(event) - rt_frame = rt_frame.update( - df, on=["stimulus_name", "pages"], how="left" + reading_times_df = reading_times_df.update( + df, on=["stimulus_name", "page"], how="left" ) - elif match := STOP_RECORDING_REGEX_NEW.match(msg["message"]): + elif match := settings.STOP_RECORDING_REGEX.match(msg["message"]): event = match.groupdict() timestamp = msg["timestamp"] event["stop_ts"] = timestamp df = pl.DataFrame(event) - rt_frame = rt_frame.update( - df, on=["stimulus_name", "pages"], how="left" + reading_times_df = reading_times_df.update( + df, on=["stimulus_name", "page"], how="left" ) - return rt_frame + else: + uncategorized_msgs.append(msg) + + return ( + reading_times_df, + break_msg, + other_screens, + uncategorized_msgs, + initial_ts, + ) - def _parse_asc(self, session_identifier: str): + def parse_messages(self, session_identifier: str): """ qick fix for now, should be replaced by the summary experiment frame later on, however the extraction of the stimulus order version is essential for other code parts, it cannot be removed without further alterations """ - other_screens = [ - "welcome_screen", - "informed_consent_screen", - "start_experiment", - "stimulus_order_version", - "showing_instruction_screen", - "camera_setup_screen", - "practice_text_starting_screen", - "transition_screen", - "final_validation", - "show_final_screen", - "optional_break_screen", - "fixation_trigger:skipped_by_experimenter", - "fixation_trigger:experimenter_calibration_triggered", - "recalibration", - "empty_screen", - "obligatory_break", - "optional_break", - ] + stimulus_times_df, break_msg, other_screens, uncategorized_msgs, initial_ts = ( + self._categorize_asc_messages(session_identifier) + ) - asc_file = self.sessions[session_identifier].asc_path - stimuli_trial_mapping = self.sessions[session_identifier].stimuli_trial_mapping + self._document_other_screens(session_identifier, other_screens) + self._document_breaks(session_identifier, break_msg) + + result_folder = self.output_dir / session_identifier / self.reports_folder + os.makedirs(result_folder, exist_ok=True) + + # TODO: adapt this function!!! + self._document_reading_times( + initial_ts, stimulus_times_df, result_folder, session_identifier + ) + return uncategorized_msgs + + def _document_other_screens(self, session_idf: str, other_screens: list) -> None: other_screen_appearance = { "timestamp": [], "screen": [], } - reading_times = { + result_folder = self.output_dir / session_idf / self.reports_folder + + for message in other_screens: + msg = message["message"] + ts = message["timestamp"] + + other_screen_appearance["timestamp"].append(ts) + other_screen_appearance["screen"].append(msg) + + other_screens_df = pd.DataFrame(other_screen_appearance) + other_screens_df.to_csv( + result_folder / f"other_screens_{session_idf}.tsv", + sep="\t", + index=False, + ) + + def _document_breaks(self, session_idf: str, breaks: list) -> None: + breaks_df = { "start_ts": [], "stop_ts": [], "start_msg": [], "stop_msg": [], - "duration_ms": [], - "duration_str": [], - "trials": [], - "pages": [], - "status": [], - "stimulus_name": [], } + result_folder = self.output_dir / session_idf / self.reports_folder - breaks = { - "start_ts": [], - "stop_ts": [], - "duration_ms": [], - "type": [], - } - - messages = [] - - initial_ts = 0 - - result_folder = self.reports_dir / session_identifier - os.makedirs(result_folder, exist_ok=True) in_break = False - with open(asc_file, "r", encoding="utf-8") as f: - for line in f.readlines(): - if match := settings.MESSAGE_REGEX.match(line): - messages.append(match.groupdict()) - msg = match.groupdict()["message"] - ts = match.groupdict()["timestamp"] - - if not initial_ts: - initial_ts = ts - - for screen in other_screens: - if screen in line: - other_screen_appearance["screen"].append(msg) - other_screen_appearance["timestamp"].append(ts) - - if msg == "optional_break" and not in_break: - in_break = True - breaks["start_ts"].append(ts) - breaks["type"].append("optional") - elif msg == "optional_break_end" and in_break: - in_break = False - breaks["stop_ts"].append(ts) - elif msg.split()[0] == "optional_break_duration:": - breaks["duration_ms"].append(msg.split()[1]) - - elif msg == "obligatory_break" and not in_break: - in_break = True - breaks["start_ts"].append(ts) - breaks["type"].append("obligatory") - elif msg == "obligatory_break_end" and in_break: - in_break = False - breaks["stop_ts"].append(ts) - elif msg.split()[0] == "obligatory_break_duration:": - breaks["duration_ms"].append(msg.split()[1]) - - if match := settings.START_RECORDING_REGEX.match(line): - reading_times["start_ts"].append(match.groupdict()["timestamp"]) - reading_times["start_msg"].append(match.groupdict()["type"]) - - trial = match.groupdict()["trial"] - # trial = trial.replace('trial_', '') - reading_times["trials"].append(trial) - - if trial in stimuli_trial_mapping: - reading_times["stimulus_name"].append( - stimuli_trial_mapping[trial] - ) - else: - reading_times["stimulus_name"].append("unknown") - - reading_times["pages"].append(match.groupdict()["page"]) - reading_times["status"].append("reading time") - elif match := settings.STOP_RECORDING_REGEX.match(line): - reading_times["stop_ts"].append(match.groupdict()["timestamp"]) - reading_times["stop_msg"].append(match.groupdict()["type"]) + for msg in breaks: + ts = msg["timestamp"] + msg = msg["message"] + + if msg == "optional_break" and not in_break: + in_break = True + breaks_df["start_ts"].append(ts) + breaks_df["type"].append("optional") + elif msg == "optional_break_end" and in_break: + in_break = False + breaks_df["stop_ts"].append(ts) + elif msg.split()[0] == "optional_break_duration:": + breaks_df["duration_ms"].append(msg.split()[1]) + + elif msg == "obligatory_break" and not in_break: + in_break = True + breaks_df["start_ts"].append(ts) + breaks_df["type"].append("obligatory") + elif msg == "obligatory_break_end" and in_break: + in_break = False + breaks_df["stop_ts"].append(ts) + elif msg.split()[0] == "obligatory_break_duration:": + breaks_df["duration_ms"].append(msg.split()[1]) + + if in_break: + breaks_df["stop_ts"].append(pd.NA) - self._document_reading_times( - initial_ts, reading_times, result_folder, session_identifier - ) - - other_screens_df = pd.DataFrame(other_screen_appearance) - other_screens_df.to_csv( - result_folder / f"other_screens_{session_identifier}.tsv", - sep="\t", - index=False, + self.logger.warning( + f"Session {session_idf} did not finish a break properly, " + f"missing end message." ) - if not in_break: - pass - # breaks_df = pd.DataFrame(breaks) - # breaks_df.to_csv( - # result_folder / f"breaks_{session_identifier}.tsv", - # sep="\t", - # index=False, - # ) - else: - self.logger.warning( - f"Session {session_identifier} did not finish a break properly, " - f"missing end message." - ) - - return messages + breaks_df = pd.DataFrame(breaks_df) + breaks_df.to_csv( + result_folder / f"breaks_{session_idf}.tsv", + sep="\t", + index=False, + ) def _document_reading_times( self, initial_ts, reading_times, result_folder, session_identifier ): """ TODO: improve this function!! this is terrible and buggy - :param initial_ts: + :param initial_ts: The timestamp of the first message. :param reading_times: :param result_folder: :param session_identifier: :return: """ + print(reading_times) + stimuli_trial_mapping = self.sessions[session_identifier].stimuli_trial_mapping total_reading_duration_ms = 0 diff --git a/preprocessing/data_collection/session.py b/preprocessing/data_collection/session.py index 66a98fa9..a29cd246 100644 --- a/preprocessing/data_collection/session.py +++ b/preprocessing/data_collection/session.py @@ -30,6 +30,7 @@ class Session: question_order: dict[str, list[str]] = field(default="unknown", init=False) stimulus_order_ids: list[int] = field(default="unknown", init=False) messages: list[dict[str, str]] = field(default="unknown", init=False) + uncategorized_messages: list[dict[str, str]] = field(default="unknown", init=False) stimuli_trial_mapping: dict[str, str] = field(default="unknown", init=False) stimulus_start_end_ts: dict[str, list[str]] = field(default="unknown", init=False) diff --git a/preprocessing/data_collection/stimulus.py b/preprocessing/data_collection/stimulus.py index 8d2abe03..3a072e7c 100644 --- a/preprocessing/data_collection/stimulus.py +++ b/preprocessing/data_collection/stimulus.py @@ -86,6 +86,7 @@ class ComprehensionQuestion: class Stimulus: id: int name: str + full_identifier: str type: Literal["experiment", "practice", "test_practice", "test_experiment"] pages: list[StimulusPage] text_stimulus: pm.stimulus.TextStimulus @@ -266,9 +267,13 @@ def load( f"{stimulus_id} has {len(questions)} questions instead of {settings.NUM_QUESTIONS_PRACTICE}" ) + # in the experiment, mostly the name + the id are used. This is predefined here so + full_identifier = stimulus_name + f"_{stimulus_id}" + stim = cls( id=stimulus_id, name=stimulus_name, + full_identifier=full_identifier, type=stimulus_type, pages=pages, text_stimulus=text_stimulus, diff --git a/preprocessing/io/load.py b/preprocessing/io/load.py index 826acacc..c5ee59ed 100644 --- a/preprocessing/io/load.py +++ b/preprocessing/io/load.py @@ -122,6 +122,7 @@ def load_trial_level_raw_data( "page": pl.Utf8, }, ) + match = re.match(regex_name, file.stem) trial_df = trial_df.with_columns( pl.lit(match.group("trial")).alias("trial"), @@ -265,24 +266,31 @@ def load_trial_level_events_data( def load_reading_measures( data_folder: Path, - file_pattern: str = r".+_(?P(?:PRACTICE_)?trial_\d+)_(?P[^_]+_[^_]+_\d+(\.0)?)_reading_measures.csv", + file_pattern: str = "", ) -> pl.DataFrame: - files = list(data_folder.glob(file_pattern)) - - if len(files) == 0: - raise ValueError(f"No files found in {data_folder} with pattern {file_pattern}") + if file_pattern is None: + file_pattern = settings.READING_MEASURES_FILENAME_REGEX all_trials = [] - for file in files: + for file in data_folder.glob(settings.READING_MEASURES_GLOB): df = pl.read_csv(file) - # get trial and stimulus from file name - match = re.match(file_pattern, file.stem) - trial_df = df.with_columns( - pl.lit(match.group("trial")).alias("trial"), - pl.lit(match.group("stimulus")).alias("stimulus"), - ) - all_trials.append(trial_df) + match = re.match(file_pattern, file.name) + # go over groups in the name regex and add them as columns + if match is None: + logging.info(f"Skipping file {file} for reading measures loading") + else: + trial_df = df.with_columns( + pl.lit(match.group("trial")).alias("trial"), + pl.lit(match.group("stimulus")).alias("stimulus"), + ) + + all_trials.append(trial_df) + + if not all_trials: + raise ValueError( + f"No reading measures files found at {data_folder} which match {file_pattern}." + ) return pl.concat(all_trials) diff --git a/preprocessing/metrics/calculate.py b/preprocessing/metrics/calculate.py index aa10bee5..906ccd67 100644 --- a/preprocessing/metrics/calculate.py +++ b/preprocessing/metrics/calculate.py @@ -19,7 +19,9 @@ def calculate_reading_measures(gaze: pm.Gaze, stimuli: list[Stimulus]) -> pl.Dat for stim in stimuli: aois = stim.text_stimulus.aois words_only = all_tokens_from_aois(aois, trial=stim.trial_id) - words_only = words_only.with_columns(pl.lit(stim.name).alias("stimulus")) + words_only = words_only.with_columns( + pl.lit(stim.full_identifier).alias("stimulus") + ) words_only_all_trials.append(words_only) words_df = pl.concat(words_only_all_trials) diff --git a/preprocessing/metrics/reading/words.py b/preprocessing/metrics/reading/words.py index 4c7527d6..79700254 100644 --- a/preprocessing/metrics/reading/words.py +++ b/preprocessing/metrics/reading/words.py @@ -1,4 +1,5 @@ import polars as pl +from ...config import settings def all_tokens_from_aois( @@ -9,7 +10,6 @@ def all_tokens_from_aois( Returns every AOI token on the page: words, spaces, punctuation — everything that has a word_idx. """ - from ...config import settings aois = ( aois.with_columns([pl.lit(trial).cast(pl.Utf8).alias(settings.TRIAL_COL)]) @@ -29,8 +29,6 @@ def all_tokens_from_aois( def mark_skipped_tokens( all_tokens: pl.DataFrame, fixations: pl.DataFrame ) -> pl.DataFrame: - from ...config import settings - fixated_tokens = ( fixations.select([settings.TRIAL_COL, settings.PAGE_COL, settings.WORD_IDX_COL]) .drop_nulls() diff --git a/preprocessing/scripts/run_multipleye_preprocessing.py b/preprocessing/scripts/run_multipleye_preprocessing.py index 499afb6d..bedb341c 100644 --- a/preprocessing/scripts/run_multipleye_preprocessing.py +++ b/preprocessing/scripts/run_multipleye_preprocessing.py @@ -32,9 +32,10 @@ def run_multipleye_preprocessing(config_path: str | None = None): if settings.EXPERIMENT_TYPE == "MultiplEYE": data_collection = preprocessing.data_collection.MultipleyeDataCollection.create_from_data_folder( data_folder_path, - include_pilots=settings.INCLUDE_PILOTS, excluded_sessions=settings.EXCLUDE_SESSIONS, included_sessions=settings.INCLUDE_SESSIONS, + include_pilots=settings.INCLUDE_PILOTS, + output_dir=settings.OUTPUT_DIR, ) elif settings.EXPERIMENT_TYPE == "MeRID": @@ -44,6 +45,7 @@ def run_multipleye_preprocessing(config_path: str | None = None): include_pilots=settings.INCLUDE_PILOTS, excluded_sessions=settings.EXCLUDE_SESSIONS, included_sessions=settings.INCLUDE_SESSIONS, + output_dir=settings.OUTPUT_DIR, ) ) @@ -107,6 +109,7 @@ def run_multipleye_preprocessing(config_path: str | None = None): sess.pm_gaze_metadata = gaze._metadata sess.calibrations = gaze.calibrations sess.validations = gaze.validations + sess.messages = gaze.messages # preprocess gaze data pbar.set_description(f"Preprocessing samples {idf}:") @@ -228,7 +231,7 @@ def run_multipleye_preprocessing(config_path: str | None = None): preprocessing.save_reading_measures( settings.OUTPUT_DIR, session_save_name, reading_measures ) - data_collection[sess].reading_measures = True + data_collection[sess.session_identifier].reading_measures = True pbar.set_description(f"Creating sanity check report {idf}") data_collection.create_sanity_check_report(