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

Delegate wearer boundary semantics to review

Browse Source
This commit is contained in:
Xin Wang
2026-09-15 02:12:18 +08:00
parent 71dfd8808a
commit 3a127a3f9f
12 changed files with 596 additions and 404 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_prepare_easycom_eot_int_review.py
+61 -36
View File
@@ -106,36 +106,47 @@ class EntryINTReviewCandidateTests(unittest.TestCase):
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"])
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_without_other_takeover_becomes_resume_control(self) -> None:
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)
resume = next(row for row in derived if row.candidate_source == "WEARER_RESUME")
self.assertEqual(resume.task, "wearer_entry_control")
self.assertEqual(resume.rule_candidate_type, "WITHIN_TURN_RESUME_CONTROL")
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(
row.candidate_source == "WEARER_ONSET" and row.anchor_dense_seconds == 1.5
for row in derived
))
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_short_feedback_does_not_split_wearer_floor_episode(self) -> None:
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."),
]
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"])
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_sustained_other_takeover_preserves_positive_entry_candidate(self) -> None:
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."),
@@ -146,16 +157,17 @@ class EntryINTReviewCandidateTests(unittest.TestCase):
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["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["int_ordering_satisfied"])
self.assertTrue(preceding_yield.temporal_evidence["new_other_onset_during_wearer"])
def test_same_other_resume_is_entry_midturn_hard_negative(self) -> None:
rows = [
@@ -199,32 +211,43 @@ class EntryINTReviewCandidateTests(unittest.TestCase):
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.assertEqual(event.temporal_evidence["new_other_onset_ids_during_wearer"], [4])
self.assertAlmostEqual(
event.temporal_evidence["eligible_new_other_onsets"][0]["onset_seconds"], 1.0
event.temporal_evidence["new_other_onsets_during_wearer"][0]["onset_seconds"], 1.0
)
self.assertTrue(event.temporal_evidence["int_ordering_satisfied"])
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_is_not_an_int_candidate(self) -> None:
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_CONTROL")
self.assertEqual(event.temporal_evidence["preexisting_crossing_other_ids"], [4])
self.assertFalse(event.temporal_evidence["int_ordering_satisfied"])
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_is_a_separate_control(self) -> None:
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_CONTROL")
self.assertEqual(event.rule_candidate_type, "WEARER_OFFSET_CONTROL")
self.assertEqual(event.automatic_protocol_label, "control")
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 = [
@@ -241,9 +264,11 @@ class EntryINTReviewCandidateTests(unittest.TestCase):
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")
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_is_not_an_int_candidate(self) -> None:
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."),
@@ -253,8 +278,8 @@ class EntryINTReviewCandidateTests(unittest.TestCase):
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"])
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 = [