306 lines
15 KiB
Python
306 lines
15 KiB
Python
"""Tests for manuscript-aligned EasyCom wearer-entry/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 EntryINTReviewCandidateTests(unittest.TestCase):
|
|
def derive(self, rows: list[Utterance]):
|
|
return derive_protocol_candidates(rows, [clip()], session="Session_4")
|
|
|
|
def test_entry_candidate_retains_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, "candidate")
|
|
self.assertEqual(event.task, "wearer_entry")
|
|
self.assertEqual(event.rule_candidate_type, "WEARER_ENTRY_CANDIDATE")
|
|
self.assertEqual(event.anchor_type, "wearer_onset")
|
|
self.assertAlmostEqual(event.event_time, 1.2)
|
|
self.assertIsNone(event.temporal_evidence["opportunity_anchor_seconds"])
|
|
self.assertAlmostEqual(
|
|
event.temporal_evidence["opportunity_options"][0]["anchor_seconds"], 1.0
|
|
)
|
|
self.assertAlmostEqual(event.temporal_evidence["behavior_anchor_seconds"], 1.2)
|
|
self.assertAlmostEqual(event.temporal_evidence["minimum_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_entry_release_metadata(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["maximum_response_delay_seconds"], 3.0)
|
|
|
|
def test_one_wearer_onset_has_one_row_with_constrained_release_options(self) -> None:
|
|
rows = [
|
|
utterance("other-4", 4, 0.0, 1.0, "Question one."),
|
|
utterance("other-3", 3, 0.2, 1.0, "Question two."),
|
|
utterance("wearer", 2, 1.4, 2.4, "Wearer answers."),
|
|
]
|
|
events = [
|
|
row for row in self.derive(rows)
|
|
if row.candidate_source == "WEARER_ONSET" and row.anchor_dense_seconds == 1.4
|
|
]
|
|
self.assertEqual(len(events), 1)
|
|
options = events[0].temporal_evidence["opportunity_options"]
|
|
self.assertEqual({option["uid"] for option in options}, {"other-4", "other-3"})
|
|
self.assertEqual(events[0].temporal_evidence["behavior_anchor_seconds"], 1.4)
|
|
|
|
def test_entry_while_other_active_remains_a_primary_candidate(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")
|
|
self.assertEqual(onset.task, "wearer_entry")
|
|
self.assertEqual(onset.automatic_protocol_label, "candidate")
|
|
self.assertEqual(onset.temporal_evidence["entry_context_timing"], "OTHER_ACTIVE")
|
|
self.assertTrue(onset.temporal_evidence["preceding_release_options_present"])
|
|
self.assertTrue(
|
|
any(
|
|
option["other_onset_between_release_and_onset"]
|
|
for option in onset.temporal_evidence["opportunity_options"]
|
|
)
|
|
)
|
|
|
|
def test_short_wearer_pause_still_becomes_raw_onset_candidate(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)
|
|
onset = next(
|
|
row for row in derived
|
|
if row.candidate_source == "WEARER_ONSET" and row.anchor_dense_seconds == 1.5
|
|
)
|
|
self.assertEqual(onset.task, "wearer_entry")
|
|
self.assertEqual(onset.rule_candidate_type, "WEARER_ENTRY_CANDIDATE")
|
|
self.assertEqual(onset.automatic_protocol_label, "candidate")
|
|
self.assertAlmostEqual(onset.temporal_evidence["previous_wearer_silence_duration_seconds"], 0.5)
|
|
self.assertFalse(any(row.candidate_source == "WEARER_RESUME" for row in derived))
|
|
|
|
def test_gap_other_activity_is_recorded_without_feedback_semantics(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."),
|
|
]
|
|
onset = next(
|
|
row for row in self.derive(rows)
|
|
if row.candidate_source == "WEARER_ONSET" and row.anchor_dense_seconds == 1.6
|
|
)
|
|
gap = onset.temporal_evidence["gap_other_intervals"]
|
|
self.assertEqual([item["participant_id"] for item in gap], [4])
|
|
self.assertEqual(gap[0]["transcription"], "Yeah.")
|
|
self.assertAlmostEqual(gap[0]["gap_overlap_seconds"], 0.3)
|
|
self.assertNotIn("feedback_like_other_ids_during_gap", onset.temporal_evidence)
|
|
|
|
def test_short_gap_with_other_activity_preserves_unresolved_entry_candidate(self) -> None:
|
|
rows = [
|
|
utterance("wearer-1", 2, 0.2, 1.0, "First wearer turn."),
|
|
utterance("other", 4, 0.8, 2.2, "I have the floor now."),
|
|
utterance("wearer-2", 2, 1.6, 2.5, "Wearer re-enters."),
|
|
]
|
|
derived = self.derive(rows)
|
|
reentry = next(
|
|
row for row in derived
|
|
if row.candidate_source == "WEARER_ONSET" and row.anchor_dense_seconds == 1.6
|
|
)
|
|
self.assertEqual(reentry.temporal_evidence["gap_other_participant_ids"], [4])
|
|
self.assertTrue(
|
|
reentry.temporal_evidence["gap_other_intervals"][0]["continues_across_wearer_onset"]
|
|
)
|
|
self.assertEqual(reentry.temporal_evidence["entry_context_timing"], "OTHER_ACTIVE")
|
|
self.assertEqual(reentry.automatic_protocol_label, "candidate")
|
|
preceding_yield = next(
|
|
row for row in derived
|
|
if row.candidate_source == "WEARER_OFFSET" and row.anchor_dense_seconds == 1.0
|
|
)
|
|
self.assertTrue(preceding_yield.temporal_evidence["new_other_onset_during_wearer"])
|
|
|
|
def test_same_other_resume_is_entry_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, "ENTRY_OTHER_MIDTURN_PAUSE")
|
|
self.assertAlmostEqual(event.temporal_evidence["resume_delay_seconds"], 0.3)
|
|
|
|
def test_other_to_other_transfer_is_entry_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, "ENTRY_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, "candidate")
|
|
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["new_other_onset_ids_during_wearer"], [4])
|
|
self.assertAlmostEqual(
|
|
event.temporal_evidence["new_other_onsets_during_wearer"][0]["onset_seconds"], 1.0
|
|
)
|
|
self.assertTrue(event.temporal_evidence["new_other_onset_during_wearer"])
|
|
self.assertTrue(event.temporal_evidence["new_other_onset_crosses_offset"])
|
|
self.assertEqual(event.temporal_evidence["next_speaker_ids"], [4])
|
|
self.assertEqual(event.temporal_evidence["next_speaker_basis"], "active_across_offset")
|
|
nearby = event.temporal_evidence["nearby_other_intervals"][0]
|
|
self.assertAlmostEqual(nearby["overlap_with_wearer_before_offset_seconds"], 1.0)
|
|
|
|
def test_preexisting_other_crossing_offset_remains_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")
|
|
self.assertEqual(event.task, "int_yield")
|
|
self.assertEqual(
|
|
event.temporal_evidence["preexisting_other_ids_at_wearer_activity_run_start"], [4]
|
|
)
|
|
self.assertFalse(event.temporal_evidence["new_other_onset_during_wearer"])
|
|
self.assertTrue(event.temporal_evidence["other_crosses_wearer_offset"])
|
|
|
|
def test_other_starting_after_wearer_offset_remains_an_int_candidate(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")
|
|
self.assertEqual(event.rule_candidate_type, "YIELD_CANDIDATE")
|
|
self.assertEqual(event.automatic_protocol_label, "candidate")
|
|
self.assertFalse(event.temporal_evidence["other_crosses_wearer_offset"])
|
|
self.assertEqual(event.temporal_evidence["next_speaker_ids"], [4])
|
|
self.assertAlmostEqual(event.temporal_evidence["next_speech_onset_seconds"], 2.2)
|
|
|
|
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")
|
|
self.assertTrue(tolerant_offset.temporal_evidence["other_crosses_wearer_offset"])
|
|
self.assertFalse(strict_offset.temporal_evidence["other_crosses_wearer_offset"])
|
|
|
|
def test_internal_raw_offset_remains_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")
|
|
self.assertEqual(event.automatic_protocol_label, "candidate")
|
|
|
|
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()
|