"""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 CandidateConfig, 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_candidate_pairs_release_and_behavior_anchors(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.candidate_source == "WEARER_ONSET") self.assertEqual(event.automatic_protocol_label, "unsure") self.assertEqual(event.rule_candidate_type, "EOT_OPPORTUNITY_CANDIDATE") self.assertEqual(event.anchor_type, "wearer_onset") self.assertAlmostEqual(event.event_time, 1.2) self.assertAlmostEqual(event.temporal_evidence["opportunity_anchor_seconds"], 1.0) self.assertAlmostEqual(event.temporal_evidence["behavior_anchor_seconds"], 1.2) self.assertAlmostEqual(event.temporal_evidence["response_delay_seconds"], 0.2) self.assertAlmostEqual(event.temporal_evidence["wearer_speech_duration_after_onset_seconds"], 0.8) def test_long_silence_does_not_cancel_eot_opportunity(self) -> None: rows = [ utterance("other", 4, 0.0, 0.5, "Your turn."), utterance("wearer", 2, 3.5, 4.5, "Delayed response."), ] event = next(row for row in self.derive(rows) if row.candidate_source == "WEARER_ONSET") self.assertAlmostEqual(event.temporal_evidence["response_delay_seconds"], 3.0) def test_competing_other_onset_terminates_eot_opportunity(self) -> None: rows = [ utterance("release", 4, 0.0, 1.0, "First turn."), utterance("competitor", 3, 1.5, 3.0, "I take the floor."), utterance("wearer", 2, 2.5, 3.5, "Wearer later."), ] onset = next( row for row in self.derive(rows) if row.anchor_dense_seconds == 2.5 and row.anchor_type == "wearer_onset" ) self.assertEqual(onset.candidate_source, "WEARER_ONSET_CONTROL") self.assertEqual(onset.automatic_protocol_label, "not_applicable") def test_short_wearer_pause_without_other_takeover_becomes_resume_control(self) -> None: rows = [ utterance("wearer-1", 2, 0.2, 1.0, "I was saying"), utterance("wearer-2", 2, 1.5, 2.5, "and then this."), ] derived = self.derive(rows) resume = next(row for row in derived if row.candidate_source == "WEARER_RESUME") self.assertEqual(resume.task, "eot_response_control") self.assertEqual(resume.rule_candidate_type, "WITHIN_TURN_RESUME_CONTROL") self.assertEqual(resume.automatic_protocol_label, "not_applicable") self.assertTrue(resume.temporal_evidence["floor_episode_merged"]) self.assertAlmostEqual(resume.temporal_evidence["previous_wearer_silence_duration_seconds"], 0.5) self.assertFalse(any( row.candidate_source == "WEARER_ONSET" and row.anchor_dense_seconds == 1.5 for row in derived )) def test_short_feedback_does_not_split_wearer_floor_episode(self) -> None: rows = [ utterance("wearer-1", 2, 0.2, 1.0, "First clause"), utterance("feedback", 4, 1.1, 1.4, "Yeah."), utterance("wearer-2", 2, 1.6, 2.5, "same turn continues."), ] resume = next(row for row in self.derive(rows) if row.candidate_source == "WEARER_RESUME") self.assertEqual(resume.temporal_evidence["feedback_like_other_ids_during_gap"], [4]) self.assertFalse(resume.temporal_evidence["sustained_other_takeover_during_gap"]) def test_short_gap_with_sustained_other_takeover_preserves_floor_reentry_but_not_eot(self) -> None: rows = [ utterance("wearer-1", 2, 0.2, 1.0, "First wearer turn."), utterance("other", 4, 1.0, 2.2, "I have the floor now."), utterance("wearer-2", 2, 1.6, 2.5, "Wearer re-enters."), ] reentry = next( row for row in self.derive(rows) if row.candidate_source == "WEARER_ONSET_CONTROL" and row.anchor_dense_seconds == 1.6 ) self.assertFalse(reentry.temporal_evidence["floor_episode_merged"]) self.assertTrue(reentry.temporal_evidence["sustained_other_takeover_during_gap"]) self.assertEqual(reentry.temporal_evidence["sustained_competing_other_ids_during_gap"], [4]) 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.candidate_source == "SAME_SPEAKER_PAUSE" 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.assertAlmostEqual(event.temporal_evidence["resume_delay_seconds"], 0.3) 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.candidate_source == "OTHER_TO_OTHER" 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") self.assertNotEqual( event.temporal_evidence["previous_speaker_id"], event.temporal_evidence["incoming_speaker_id"], ) def test_other_crossing_wearer_offset_is_an_unresolved_yield_candidate(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.candidate_source == "WEARER_OFFSET") self.assertEqual(event.automatic_protocol_label, "unsure") self.assertEqual(event.rule_candidate_type, "YIELD_CANDIDATE") self.assertEqual(event.anchor_type, "wearer_offset") self.assertAlmostEqual(event.event_time, 2.0) self.assertEqual(event.temporal_evidence["tolerance_crossing_other_ids"], [4]) self.assertEqual(event.temporal_evidence["eligible_new_other_onset_ids"], [4]) self.assertAlmostEqual( event.temporal_evidence["eligible_new_other_onsets"][0]["onset_seconds"], 1.0 ) self.assertTrue(event.temporal_evidence["int_ordering_satisfied"]) def test_preexisting_other_crossing_offset_is_not_an_int_candidate(self) -> None: rows = [ utterance("other", 4, 0.0, 3.0, "Other already holds the floor."), utterance("wearer", 2, 1.0, 2.0, "Wearer enters and stops."), ] event = next(row for row in self.derive(rows) if row.anchor_type == "wearer_offset") self.assertEqual(event.candidate_source, "WEARER_OFFSET_CONTROL") self.assertEqual(event.temporal_evidence["preexisting_crossing_other_ids"], [4]) self.assertFalse(event.temporal_evidence["int_ordering_satisfied"]) def test_other_starting_after_wearer_offset_is_a_separate_control(self) -> None: rows = [ utterance("wearer", 2, 0.0, 2.0, "Wearer finishes."), utterance("other", 4, 2.2, 3.0, "Later turn."), ] event = next(row for row in self.derive(rows) if row.anchor_type == "wearer_offset") self.assertEqual(event.candidate_source, "WEARER_OFFSET_CONTROL") self.assertEqual(event.rule_candidate_type, "WEARER_OFFSET_CONTROL") self.assertEqual(event.automatic_protocol_label, "not_applicable") self.assertFalse(event.temporal_evidence["other_crosses_wearer_offset"]) def test_one_frame_boundary_tolerance_is_configurable(self) -> None: rows = [ utterance("wearer", 2, 0.0, 2.0, "Wearer finishes."), utterance("other", 4, 1.0, 2.0, "Boundary-aligned takeover."), ] tolerant = self.derive(rows) strict = derive_protocol_candidates( rows, [clip()], session="Session_4", config=CandidateConfig(int_offset_tolerance_seconds=0.0), ) tolerant_offset = next(row for row in tolerant if row.anchor_type == "wearer_offset") strict_offset = next(row for row in strict if row.anchor_type == "wearer_offset") self.assertEqual(tolerant_offset.candidate_source, "WEARER_OFFSET") self.assertEqual(strict_offset.candidate_source, "WEARER_OFFSET_CONTROL") def test_internal_raw_offset_is_not_an_int_candidate(self) -> None: rows = [ utterance("wearer-1", 2, 0.0, 1.0, "First clause."), utterance("other", 4, 0.8, 1.3, "Yeah."), utterance("wearer-2", 2, 1.2, 2.5, "Same floor episode."), ] event = next( row for row in self.derive(rows) if row.anchor_type == "wearer_offset" and row.anchor_dense_seconds == 1.0 ) self.assertEqual(event.candidate_source, "WEARER_OFFSET_CONTROL") self.assertTrue(event.temporal_evidence["raw_offset_is_within_merged_floor_episode"]) 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.candidate_source == "OTHER_BACKCHANNEL") self.assertEqual(event.automatic_protocol_label, "hard_negative") self.assertEqual(event.automatic_hard_negative_type, "INT_OTHER_BACKCHANNEL") self.assertTrue(event.temporal_evidence["wearer_active_through_region"]) def test_non_takeover_overlap_is_scanned_without_wearer_transition(self) -> None: rows = [ utterance("wearer", 2, 0.0, 4.0, "Long wearer turn."), utterance("other", 4, 1.0, 2.0, "Long overlapping contribution."), ] event = next(row for row in self.derive(rows) if row.candidate_source == "NON_TAKEOVER_OVERLAP") self.assertEqual(event.automatic_hard_negative_type, "INT_NON_TAKEOVER_OVERLAP") self.assertAlmostEqual(event.region_start_seconds, 1.0) if __name__ == "__main__": unittest.main()