diff --git a/preprocessing/checks/preflight.py b/preprocessing/checks/preflight.py index 4ebb58bf..2bc313f6 100644 --- a/preprocessing/checks/preflight.py +++ b/preprocessing/checks/preflight.py @@ -77,6 +77,10 @@ def _print_warnings(warnings: dict[str, list[str]]) -> None: for msg in warnings["Psychometric tests"]: lines.append(f"\n {msg}") + if "Session completeness" in warnings: + for msg in warnings["Session completeness"]: + lines.append(f"\n {msg}") + print("\n".join(lines), file=sys.stderr) @@ -102,6 +106,7 @@ def run_preflight_check(data_collection) -> None: _check_skipped_sessions(data_collection, errors) _check_sessions(data_collection, errors) _check_stimulus_order_coverage(data_collection, errors) + _check_session_completeness(data_collection, warnings) pt_warnings: list[str] = [] _check_psychometric_tests(data_collection, pt_warnings) @@ -424,6 +429,56 @@ def _check_stimulus_order_coverage( ) +def _check_session_completeness( + data_collection, + warnings: dict[str, list[str]], +) -> None: + """Check that every participant has all expected sessions. + + Groups sessions by base_id (participant + language + country + lab) + and detects the expected number of sessions from the data pattern. + Participants with fewer sessions than the maximum observed for any + participant are reported as warnings. + + Non-pilot sessions only; sessions with non-parseable SIDs are skipped. + """ + from ..models.sid import Sid + + base_sessions: dict[str, set[int]] = {} + for session in data_collection.sessions.values(): + if getattr(session, "is_pilot", False): + continue + try: + sid = Sid(session.session_identifier) + except (ValueError, TypeError): + continue + base_sessions.setdefault(sid.base_id, set()).add(sid.session_id) + + if not base_sessions: + return + + max_sessions = max(len(v) for v in base_sessions.values()) + if max_sessions <= 1: + return + + total_participants = len(base_sessions) + complete = sum(1 for v in base_sessions.values() if len(v) == max_sessions) + incomplete = [ + f"{base_id} (has session(s): {sorted(session_ids)}, " + f"missing: ET{','.join(str(s) for s in sorted(set(range(1, max_sessions + 1)) - session_ids))})" + for base_id, session_ids in sorted(base_sessions.items()) + if len(session_ids) < max_sessions + ] + + msg = ( + f"Session completeness: {complete}/{total_participants} participants " + f"have all {max_sessions} expected ET sessions." + ) + warnings.setdefault("Session completeness", []).append(msg) + if incomplete: + warnings["Session completeness"].extend(incomplete) + + def _format_message(groups: dict[str, list[str]]) -> str: """Transform the grouped error dict into a human-readable message.""" n_total = sum(len(v) for v in groups.values()) diff --git a/preprocessing/data_collection/multipleye_data_collection.py b/preprocessing/data_collection/multipleye_data_collection.py index c24042f7..156ddf3e 100644 --- a/preprocessing/data_collection/multipleye_data_collection.py +++ b/preprocessing/data_collection/multipleye_data_collection.py @@ -57,6 +57,52 @@ def wrapper(self): return wrapper +def _compute_session_completeness(data_collection) -> dict: + """Compute session completeness stats grouped by participant base ID. + + Returns a dict with keys ``expected_sessions_per_participant``, + ``total_participants``, ``complete_participants``, and + ``incomplete_participants`` (a list of base IDs with missing sessions). + Non-pilot sessions only; sessions with non-parseable SIDs are skipped. + """ + base_sessions: dict[str, set[int]] = {} + for session in data_collection.sessions.values(): + if getattr(session, "is_pilot", False): + continue + try: + sid = Sid(session.session_identifier) + except (ValueError, TypeError): + continue + base_sessions.setdefault(sid.base_id, set()).add(sid.session_id) + + if not base_sessions: + return { + "expected_sessions_per_participant": 0, + "total_participants": 0, + "complete_participants": 0, + } + + max_sessions = max(len(v) for v in base_sessions.values()) + total = len(base_sessions) + complete = sum(1 for v in base_sessions.values() if len(v) == max_sessions) + + result: dict = { + "expected_sessions_per_participant": max_sessions, + "total_participants": total, + "complete_participants": complete, + } + + incomplete = sorted( + base_id + for base_id, session_ids in base_sessions.items() + if len(session_ids) < max_sessions + ) + if incomplete: + result["incomplete_participants"] = incomplete + + return result + + class MultipleyeDataCollection: participant_data_path: Path | str | None crashed_session_ids: list[str] = [] @@ -652,6 +698,8 @@ def create_dataset_overview(self, path: str | Path = "") -> dict: [session for session in self.sessions if self.sessions[session].is_pilot] ) + session_completeness = _compute_session_completeness(self) + # TODO: add more, check metadata scheme, add stats like num read pages, total reading time etc. overview = { "Title": self.data_collection_name, @@ -662,6 +710,7 @@ def create_dataset_overview(self, path: str | Path = "") -> dict: "Country": self.country, "Year": self.year, "Number of eye-tracking (ET) sessions per participant": self.num_sessions, + "Session_completeness": session_completeness, } # Add warnings to overview diff --git a/tests/unit/preprocessing/checks/test_preflight.py b/tests/unit/preprocessing/checks/test_preflight.py index 581ee0b9..ed9ed649 100644 --- a/tests/unit/preprocessing/checks/test_preflight.py +++ b/tests/unit/preprocessing/checks/test_preflight.py @@ -12,6 +12,7 @@ PreflightError, run_preflight_check, _check_psychometric_tests, + _check_session_completeness, ) from preprocessing.config import settings @@ -21,6 +22,7 @@ class FakeSession: session_identifier: str session_file_path: Path session_folder_path: Path + is_pilot: bool = False @dataclass @@ -610,6 +612,119 @@ def test_preflight_warnings_only(preflight_env): run_preflight_check(dc) # should not raise +# --------------------------------------------------------------------------- +# Session completeness check +# --------------------------------------------------------------------------- + + +def _build_completeness_env(sids): + """Build a FakeDataCollection from a list of (sid, is_pilot) tuples.""" + sessions: dict[str, FakeSession] = {} + for entry in sids: + sid, is_pilot = entry if isinstance(entry, tuple) else (entry, False) + sessions[sid] = FakeSession( + session_identifier=sid, + session_file_path=Path("/fake") / sid / "data.edf", + session_folder_path=Path("/fake") / sid, + is_pilot=is_pilot, + ) + return FakeDataCollection( + stimulus_dir=Path("/fake"), + language="EN", + country="UK", + lab_number=1, + sessions=sessions, + ) + + +@pytest.mark.parametrize( + "sids, expected_summary, expected_missing", + [ + ( + [ + ("001_EN_UK_1_ET1", False), + ("001_EN_UK_1_ET2", False), + ("002_EN_UK_1_ET1", False), + ("002_EN_UK_1_ET2", False), + ], + "2/2 participants have all 2 expected ET sessions", + [], + ), + ( + [ + ("001_EN_UK_1_ET1", False), + ("001_EN_UK_1_ET2", False), + ("002_EN_UK_1_ET1", False), + ], + "1/2 participants have all 2 expected ET sessions", + ["002_EN_UK_1", "missing: ET2"], + ), + ( + [ + ("001_EN_UK_1_ET1", False), + ("001_EN_UK_1_ET2", False), + ("002_EN_UK_1_ET2", False), + ], + "1/2 participants have all 2 expected ET sessions", + ["002_EN_UK_1", "missing: ET1"], + ), + ( + [ + ("001_EN_UK_1_ET1", False), + ("001_EN_UK_1_ET2", False), + ("001_EN_UK_1_ET3", False), + ("002_EN_UK_1_ET1", False), + ("002_EN_UK_1_ET2", False), + ], + "1/2 participants have all 3 expected ET sessions", + ["002_EN_UK_1", "missing: ET3"], + ), + ( + [ + ("001_EN_UK_1_ET1", False), + ("001_EN_UK_1_ET2", False), + ("002_EN_UK_1_ET1", True), + ], + "1/1 participants have all 2 expected ET sessions", + [], + ), + ], + ids=[ + "all_present", + "missing_et2", + "missing_et1", + "three_sessions", + "pilots_excluded", + ], +) +def test_session_completeness_warnings(sids, expected_summary, expected_missing): + dc = _build_completeness_env(sids) + warnings: dict[str, list[str]] = {} + _check_session_completeness(dc, warnings) + assert "Session completeness" in warnings + msgs = warnings["Session completeness"] + assert expected_summary in msgs[0] + for substr in expected_missing: + assert any(substr in m for m in msgs), ( + f"Expected '{substr}' in messages: {msgs}" + ) + + +@pytest.mark.parametrize( + "sids", + [ + [], + ["000_EN_UK_1_ET1", "001_EN_UK_1_ET1", "002_EN_UK_1_ET1"], + ], + ids=["empty", "single_session"], +) +def test_session_completeness_no_warning(sids): + dc = _build_completeness_env(sids) + warnings: dict[str, list[str]] = {} + _check_session_completeness(dc, warnings) + assert "Session completeness" not in warnings + + # --------------------------------------------------------------------------- # Helpers # ---------------------------------------------------------------------------