Add EasyCom floor event proxy audit

test_audit_easycom_floor_events.py

@@ -0,0 +1,125 @@
+"""Tests for conservative EasyCom floor-event auditing."""
+
+from __future__ import annotations
+
+import unittest
+
+from audit_easycom_floor_events import derive_eot_candidates
+from derive_easycom_interaction_events import Rules, Utterance
+
+
+def utterance(
+    uid: str,
+    participant: int,
+    start: float,
+    end: float,
+    text: str,
+    *,
+    cross_clip: bool = False,
+    starts_at_boundary: bool = False,
+    ends_at_boundary: bool = False,
+) -> Utterance:
+    return Utterance(
+        uid=uid,
+        session="Session_7",
+        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=starts_at_boundary,
+        ends_at_clip_boundary=ends_at_boundary,
+        annotation_end_frame_clamped_by=0,
+        is_cross_clip_continuation=cross_clip,
+    )
+
+
+class EasyComFloorAuditTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.rules = Rules()
+
+    def candidates(self, rows: list[Utterance]):
+        return derive_eot_candidates(
+            rows,
+            camera_wearer_id=2,
+            rules=self.rules,
+        )
+
+    def test_sustained_overlapping_incoming_speaker_is_positive_proxy(self) -> None:
+        holder = utterance("holder", 4, 0.0, 2.0, "I finished my point.")
+        incoming = utterance("incoming", 2, 1.7, 3.2, "Here is my response.")
+        result = {row.holder_uid: row for row in self.candidates([holder, incoming])}
+        self.assertEqual(
+            result["holder"].category,
+            "eot_positive_overlap_takeover_proxy",
+        )
+        self.assertEqual(result["holder"].incoming_uid, "incoming")
+
+    def test_same_speaker_resumption_is_mid_turn_pause_negative(self) -> None:
+        first = utterance("first", 4, 0.0, 1.0, "First clause.")
+        resumed = utterance("resumed", 4, 1.3, 2.5, "Second clause.")
+        result = {row.holder_uid: row for row in self.candidates([first, resumed])}
+        self.assertEqual(
+            result["first"].category,
+            "eot_negative_mid_turn_pause_proxy",
+        )
+        self.assertAlmostEqual(result["first"].pause_seconds, 0.3)
+
+    def test_long_same_speaker_gap_is_ambiguous(self) -> None:
+        first = utterance("first", 4, 0.0, 1.0, "First clause.")
+        resumed = utterance("resumed", 4, 4.0, 5.0, "Much later.")
+        result = {row.holder_uid: row for row in self.candidates([first, resumed])}
+        self.assertEqual(
+            result["first"].category,
+            "eot_ambiguous_long_same_speaker_gap",
+        )
+
+    def test_short_unresolved_incoming_speaker_is_kept_for_review(self) -> None:
+        holder = utterance("holder", 4, 0.0, 1.0, "First speaker.")
+        short = utterance("short", 2, 1.2, 1.6, "Maybe.")
+        resumed = utterance("resumed", 4, 3.5, 4.5, "First speaker continues.")
+        result = {
+            row.holder_uid: row
+            for row in self.candidates([holder, short, resumed])
+        }
+        self.assertEqual(
+            result["holder"].category,
+            "eot_ambiguous_short_or_unresolved_incoming",
+        )
+        self.assertEqual(result["holder"].incoming_uid, "short")
+
+    def test_adjacent_cross_file_continuation_is_excluded(self) -> None:
+        first = utterance(
+            "first",
+            4,
+            0.0,
+            60.0,
+            "Continued [X]",
+            ends_at_boundary=True,
+        )
+        resumed = utterance(
+            "resumed",
+            4,
+            60.0,
+            61.0,
+            "[X] continuation",
+            cross_clip=True,
+            starts_at_boundary=True,
+        )
+        result = {row.holder_uid: row for row in self.candidates([first, resumed])}
+        self.assertEqual(
+            result["first"].category,
+            "eot_excluded_file_continuation",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()