Add manuscript-aligned EasyCom EOT INT review workflow

test_prepare_easycom_eot_int_review.py

@@ -0,0 +1,115 @@
+"""Tests for manuscript-aligned EasyCom EOT/INT review regions."""
+
+from __future__ import annotations
+
+import unittest
+
+from derive_easycom_interaction_events import Clip, Utterance
+from prepare_easycom_eot_int_review import derive_protocol_candidates
+
+
+def utterance(uid: str, participant: int, start: float, end: float, text: str) -> Utterance:
+    return Utterance(
+        uid=uid,
+        session="Session_4",
+        clip_stem="00-00-000",
+        clip_index=0,
+        segment_id=0,
+        participant_id=participant,
+        target_of_speech=("Group",),
+        transcription=text,
+        source_start_frame=round(start * 20) + 1,
+        source_end_frame=round(end * 20),
+        dense_start_frame=round(start * 20),
+        dense_end_frame_exclusive=round(end * 20),
+        dense_start_seconds=start,
+        dense_end_seconds=end,
+        duration_seconds=end - start,
+        starts_at_clip_boundary=start == 0,
+        ends_at_clip_boundary=False,
+        annotation_end_frame_clamped_by=0,
+        is_cross_clip_continuation=False,
+    )
+
+
+def clip(seconds: float = 5.0) -> Clip:
+    return Clip(
+        path="/dataset/Main/Video_Compressed/Session_4/00-00-000.mp4",
+        transcript_path="/dataset/Main/Speech_Transcriptions/Session_4/00-00-000.json",
+        stem="00-00-000",
+        source_start_seconds=0.0,
+        source_end_seconds=seconds,
+        dense_start_frame=0,
+        num_frames=round(seconds * 20),
+        segment_id=0,
+        discontinuity_before_seconds=None,
+    )
+
+
+class EOTINTReviewCandidateTests(unittest.TestCase):
+    def derive(self, rows: list[Utterance]):
+        return derive_protocol_candidates(rows, [clip()], session="Session_4")
+
+    def test_eot_response_is_anchored_by_wearer_onset(self) -> None:
+        rows = [
+            utterance("other", 4, 0.0, 1.0, "Your turn."),
+            utterance("wearer", 2, 1.2, 2.0, "My response."),
+        ]
+        event = next(row for row in self.derive(rows) if row.task == "eot_response")
+        self.assertEqual(event.automatic_protocol_label, "positive")
+        self.assertEqual(event.anchor_type, "other_offset")
+        self.assertEqual(event.behavior_anchor_type, "wearer_onset")
+        self.assertAlmostEqual(event.anchor_dense_seconds, 1.0)
+        self.assertAlmostEqual(event.behavior_seconds or -1, 1.2)
+
+    def test_same_other_resume_is_eot_midturn_hard_negative(self) -> None:
+        rows = [
+            utterance("first", 4, 0.0, 1.0, "First clause."),
+            utterance("resume", 4, 1.3, 2.5, "Same turn continues."),
+        ]
+        event = next(
+            row
+            for row in self.derive(rows)
+            if row.task == "eot_response" and row.anchor_dense_seconds == 1.0
+        )
+        self.assertEqual(event.automatic_protocol_label, "hard_negative")
+        self.assertEqual(event.automatic_hard_negative_type, "eot_midturn_pause")
+        self.assertEqual(event.region_end_reason, "competing_other_onset")
+
+    def test_other_to_other_transfer_is_eot_hard_negative(self) -> None:
+        rows = [
+            utterance("first", 4, 0.0, 1.0, "First speaker."),
+            utterance("next", 3, 1.2, 2.5, "Second speaker."),
+        ]
+        event = next(
+            row
+            for row in self.derive(rows)
+            if row.task == "eot_response" and row.anchor_dense_seconds == 1.0
+        )
+        self.assertEqual(event.automatic_protocol_label, "hard_negative")
+        self.assertEqual(event.automatic_hard_negative_type, "eot_other_to_other")
+
+    def test_int_yield_is_anchored_by_wearer_offset(self) -> None:
+        rows = [
+            utterance("wearer", 2, 0.0, 2.0, "Wearer is speaking."),
+            utterance("other", 4, 1.0, 3.0, "I will take over now."),
+        ]
+        event = next(row for row in self.derive(rows) if row.task == "int_yield")
+        self.assertEqual(event.automatic_protocol_label, "positive")
+        self.assertEqual(event.anchor_type, "other_onset")
+        self.assertEqual(event.behavior_anchor_type, "wearer_offset")
+        self.assertAlmostEqual(event.behavior_seconds or -1, 2.0)
+
+    def test_brief_feedback_is_int_backchannel_hard_negative(self) -> None:
+        rows = [
+            utterance("wearer", 2, 0.0, 3.0, "Wearer keeps the floor."),
+            utterance("other", 4, 1.0, 1.4, "Yeah."),
+        ]
+        event = next(row for row in self.derive(rows) if row.task == "int_yield")
+        self.assertEqual(event.automatic_protocol_label, "hard_negative")
+        self.assertEqual(event.automatic_hard_negative_type, "int_backchannel")
+        self.assertEqual(event.region_end_reason, "competing_other_offset")
+
+
+if __name__ == "__main__":
+    unittest.main()