wx44wx/icassp2027-experiment
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revise wearer entry and yield annotation protocol

Browse Source
This commit is contained in:
Xin Wang
2026-09-15 01:25:39 +08:00
parent 026d89978f
commit 71dfd8808a
12 changed files with 478 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_prepare_easycom_eot_int_review.py
+33 -21
View File
@@ -1,4 +1,4 @@
"""Tests for manuscript-aligned EasyCom EOT/INT review regions."""
"""Tests for manuscript-aligned EasyCom wearer-entry/INT review regions."""
from __future__ import annotations
@@ -46,18 +46,19 @@ def clip(seconds: float = 5.0) -> Clip:
)
class EOTINTReviewCandidateTests(unittest.TestCase):
class EntryINTReviewCandidateTests(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:
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, "unsure")
self.assertEqual(event.rule_candidate_type, "EOT_OPPORTUNITY_CANDIDATE")
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"])
@@ -68,7 +69,7 @@ class EOTINTReviewCandidateTests(unittest.TestCase):
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_eot_opportunity(self) -> None:
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."),
@@ -91,7 +92,7 @@ class EOTINTReviewCandidateTests(unittest.TestCase):
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_competing_other_onset_terminates_eot_opportunity(self) -> None:
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."),
@@ -101,8 +102,11 @@ class EOTINTReviewCandidateTests(unittest.TestCase):
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")
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.assertFalse(onset.temporal_evidence["preceding_release_paired"])
def test_short_wearer_pause_without_other_takeover_becomes_resume_control(self) -> None:
rows = [
@@ -111,9 +115,9 @@ class EOTINTReviewCandidateTests(unittest.TestCase):
]
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.task, "wearer_entry_control")
self.assertEqual(resume.rule_candidate_type, "WITHIN_TURN_RESUME_CONTROL")
self.assertEqual(resume.automatic_protocol_label, "not_applicable")
self.assertEqual(resume.automatic_protocol_label, "control")
self.assertTrue(resume.temporal_evidence["floor_episode_merged"])
self.assertAlmostEqual(resume.temporal_evidence["previous_wearer_silence_duration_seconds"], 0.5)
self.assertFalse(any(
@@ -131,21 +135,29 @@ class EOTINTReviewCandidateTests(unittest.TestCase):
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:
def test_short_gap_with_sustained_other_takeover_preserves_positive_entry_candidate(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("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 self.derive(rows)
if row.candidate_source == "WEARER_ONSET_CONTROL" and row.anchor_dense_seconds == 1.6
row for row in derived
if row.candidate_source == "WEARER_ONSET" 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])
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["int_ordering_satisfied"])
def test_same_other_resume_is_eot_midturn_hard_negative(self) -> None:
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."),
@@ -156,10 +168,10 @@ class EOTINTReviewCandidateTests(unittest.TestCase):
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.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_eot_hard_negative(self) -> None:
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."),
@@ -170,7 +182,7 @@ class EOTINTReviewCandidateTests(unittest.TestCase):
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.assertEqual(event.automatic_hard_negative_type, "ENTRY_OTHER_TO_OTHER")
self.assertNotEqual(
event.temporal_evidence["previous_speaker_id"],
event.temporal_evidence["incoming_speaker_id"],
@@ -182,7 +194,7 @@ class EOTINTReviewCandidateTests(unittest.TestCase):
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.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)
@@ -211,7 +223,7 @@ class EOTINTReviewCandidateTests(unittest.TestCase):
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.assertEqual(event.automatic_protocol_label, "control")
self.assertFalse(event.temporal_evidence["other_crosses_wearer_offset"])
def test_one_frame_boundary_tolerance_is_configurable(self) -> None: