from __future__ import annotations import json from pathlib import Path import tempfile import unittest from review_floor_events import ( CandidateRepository, ReviewStore, expand_candidate_paths, load_candidates, parse_byte_range, stream_file_range, ) class FloorEventReviewTests(unittest.TestCase): def _fixture(self, root: Path) -> tuple[Path, Path]: dataset = root / "dataset" video_dir = dataset / "Main" / "Video_Compressed" / "Session_4" transcript_dir = dataset / "Main" / "Speech_Transcriptions" / "Session_4" video_dir.mkdir(parents=True) transcript_dir.mkdir(parents=True) (video_dir / "00-00-000.mp4").write_bytes(b"mock-video") (transcript_dir / "00-00-000.json").write_text( json.dumps( [ { "Start_Frame": 1, "End_Frame": 20, "Participant_ID": 4, "Transcription": "Are you coming?", "Target_of_Speech": [2], }, { "Start_Frame": 25, "End_Frame": 40, "Participant_ID": 2, "Transcription": "Yes.", "Target_of_Speech": [4], }, ] ), encoding="utf-8", ) events = root / "candidates.jsonl" events.write_text( json.dumps( { "uid": "Session_4/00-00-000/0001", "session": "Session_4", "split": "train", "event_type": "ordinary_switch", "confidence": "timing", "participant_role": "wearer", "clip_stem": "00-00-000", "source_start_frame": 25, "dense_start_seconds": 1.2, "event_time": 1.2, "candidate_source": "WEARER_ONSET", "task": "wearer_entry", "ai_verdict": "POSITIVE", "ai_event_type": "WEARER_ENTRY", "ai_subtype": None, "speaker_timeline": [ { "uid": "Session_4/00-00-000/0001", "participant_id": 2, "participant_role": "wearer", "start_seconds": 1.2, "end_seconds": 2.0, "transcription": "Yes.", } ], } ) + "\n", encoding="utf-8", ) (root / "easycom_session_timelines.json").write_text( json.dumps({"schema_version": 1, "sessions": {"Session_4": [ {"participant_id": 4, "participant_role": "other", "start_seconds": 0.0, "end_seconds": 1.0}, {"participant_id": 2, "participant_role": "wearer", "start_seconds": 1.2, "end_seconds": 2.0}, ]}}), encoding="utf-8", ) return dataset, events def test_candidate_detail_resolves_media_and_transcript(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) dataset, events = self._fixture(root) candidates = load_candidates([events]) repository = CandidateRepository( candidates, dataset, context_before_seconds=3.0, context_after_seconds=3.0, ) detail = repository.detail(candidates[0]) self.assertTrue(detail["media_available"]) self.assertEqual(detail["event_local_seconds"], 1.2) self.assertEqual(len(detail["transcript"]), 2) self.assertTrue(detail["transcript"][1]["is_anchor"]) self.assertEqual(detail["transcript"][1]["participant_role"], "wearer") self.assertEqual(detail["transcript"][1]["uid"], "Session_4/00-00-000/0001") self.assertEqual(len(repository.session_timelines["Session_4"]), 2) def test_review_store_writes_resumable_jsonl_and_csv(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, events = self._fixture(root) candidates = load_candidates([events]) output = root / "reviews" / "reviews.jsonl" store = ReviewStore(output, candidates) review = { "human_action": "accept", "human_verdict": None, "human_event_type": None, "human_subtype": None, "exclude_reason": None, "boundary_correction_seconds": None, "confidence": "high", "notes": "clear acknowledgement", } store.save(candidates[0].candidate_id, review) self.assertTrue(output.is_file()) self.assertTrue(output.with_suffix(".csv").is_file()) resumed = ReviewStore(output, candidates) self.assertEqual( resumed.reviews[candidates[0].candidate_id]["review"]["human_event_type"], "WEARER_ENTRY", ) self.assertFalse(resumed.reviews[candidates[0].candidate_id]["review"]["relabeled"]) self.assertEqual(resumed.reviews[candidates[0].candidate_id]["review"]["reviewer"], "") def test_directory_expansion_excludes_output(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) candidate = root / "Session_4.events.jsonl" output = root / "reviews.jsonl" candidate.write_text("{}\n", encoding="utf-8") output.write_text("{}\n", encoding="utf-8") self.assertEqual( expand_candidate_paths([root], output_path=output), [candidate.resolve()], ) def test_primary_queue_keeps_nonoverlap_wearer_offset_for_semantics(self) -> None: with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "legacy.jsonl" rows = [ { "uid": "Session_4/wearer_offset/00000285", "candidate_source": "WEARER_OFFSET", "task": "int_yield", "temporal_evidence": { "active_other_ids_before_offset": [], "active_other_ids_after_offset": [], }, }, { "uid": "Session_4/wearer_onset/00000338", "candidate_source": "WEARER_ONSET", "task": "wearer_entry", }, ] path.write_text( "".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8" ) primary = load_candidates([path]) all_rows = load_candidates([path], include_offset_controls=True) self.assertEqual([row.uid for row in primary], [row["uid"] for row in rows]) self.assertEqual(len(all_rows), 2) def test_relabel_preserves_ai_and_records_human_decision(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, events = self._fixture(root) candidates = load_candidates([events]) store = ReviewStore(root / "reviews.jsonl", candidates) record = store.save(candidates[0].candidate_id, { "human_action": "relabel", "human_verdict": "HARD_NEGATIVE", "human_event_type": None, "human_subtype": "ENTRY_WEARER_BACKCHANNEL", "exclude_reason": None, "boundary_correction_seconds": None, "confidence": "medium", "reviewer": "r2", "notes": "brief acknowledgement", }) self.assertEqual(record["ai_event_type"], "WEARER_ENTRY") self.assertEqual(record["review"]["human_subtype"], "ENTRY_WEARER_BACKCHANNEL") self.assertTrue(record["review"]["relabeled"]) def test_removed_verdicts_cannot_be_accepted_or_reintroduced(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, events = self._fixture(root) candidates = load_candidates([events]) candidate = candidates[0] store = ReviewStore(root / "reviews.jsonl", candidates) base = { "human_event_type": None, "human_subtype": None, "selected_opportunity_uid": None, "exclude_reason": None, "boundary_correction_seconds": None, "confidence": "medium", "notes": "legacy fallback", } for verdict in ("UNSURE", "NOT_APPLICABLE"): with self.subTest(action="relabel", verdict=verdict), self.assertRaisesRegex( ValueError, "Invalid human_verdict" ): store.save(candidate.candidate_id, { **base, "human_action": "relabel", "human_verdict": verdict, }) candidate.payload["ai_verdict"] = verdict with self.subTest(action="accept", verdict=verdict), self.assertRaisesRegex( ValueError, "Accept requires an AI pre-label" ): store.save(candidate.candidate_id, { **base, "human_action": "accept", "human_verdict": None, }) candidate.payload["ai_verdict"] = "POSITIVE" def test_relabel_subtype_must_match_candidate_family(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, events = self._fixture(root) candidates = load_candidates([events]) store = ReviewStore(root / "reviews.jsonl", candidates) with self.assertRaisesRegex(ValueError, "ENTRY candidate requires ENTRY_"): store.save(candidates[0].candidate_id, { "human_action": "relabel", "human_verdict": "HARD_NEGATIVE", "human_event_type": None, "human_subtype": "INT_OTHER_BACKCHANNEL", "selected_opportunity_uid": None, "exclude_reason": None, "boundary_correction_seconds": None, "confidence": "medium", "notes": "wrong family", }) def test_normal_completion_can_be_excluded_as_out_of_scope(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, events = self._fixture(root) candidates = load_candidates([events]) store = ReviewStore(root / "reviews.jsonl", candidates) record = store.save(candidates[0].candidate_id, { "human_action": "exclude", "human_verdict": None, "human_event_type": None, "human_subtype": None, "selected_opportunity_uid": None, "exclude_reason": "OUT_OF_SCOPE_NORMAL_COMPLETION", "boundary_correction_seconds": None, "confidence": "high", "notes": "wearer completed; laughter did not take over", }) self.assertEqual(record["review"]["human_verdict"], "EXCLUDE") self.assertEqual( record["review"]["exclude_reason"], "OUT_OF_SCOPE_NORMAL_COMPLETION" ) self.assertEqual(record["review"]["exclude_reason_group"], "OUT_OF_SCOPE") def test_accept_ai_exclude_preserves_reason_taxonomy(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, events = self._fixture(root) candidates = load_candidates([events]) candidate = candidates[0] candidate.payload.update( ai_verdict="EXCLUDE", ai_event_type=None, ai_subtype=None, ai_exclude_reason="DATA_MISSING_CONTEXT", ai_exclude_reason_group="DATA_QUALITY", llm_suggestion={ "verdict": "EXCLUDE", "event_type": None, "subtype": None, "exclude_reason": "DATA_MISSING_CONTEXT", }, ) store = ReviewStore(root / "reviews.jsonl", candidates) record = store.save(candidate.candidate_id, { "human_action": "accept", "human_verdict": None, "human_event_type": None, "human_subtype": None, "selected_opportunity_uid": None, "exclude_reason": None, "boundary_correction_seconds": None, "confidence": "medium", "notes": "context is truncated", }) self.assertEqual(record["review"]["exclude_reason"], "DATA_MISSING_CONTEXT") self.assertEqual(record["review"]["exclude_reason_group"], "DATA_QUALITY") def test_entry_review_may_select_only_a_supplied_opportunity_anchor(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) _, events = self._fixture(root) candidates = load_candidates([events]) candidate = candidates[0] candidate.payload["temporal_evidence"] = { "opportunity_options": [ {"uid": "release-1", "anchor_seconds": 1.0, "participant_id": 4} ] } store = ReviewStore(root / "reviews.jsonl", candidates) base = { "human_action": "relabel", "human_verdict": "POSITIVE", "human_event_type": "WEARER_ENTRY", "human_subtype": None, "exclude_reason": None, "boundary_correction_seconds": None, "confidence": "high", "notes": "clear response", } record = store.save( candidate.candidate_id, {**base, "selected_opportunity_uid": "release-1"} ) self.assertEqual(record["review"]["selected_opportunity_uid"], "release-1") self.assertEqual( record["review"]["valid_prediction_interval_seconds"], [1.0, 1.2] ) with self.assertRaisesRegex(ValueError, "only a supplied"): store.save( candidate.candidate_id, {**base, "selected_opportunity_uid": "invented"}, ) optional = store.save( candidate.candidate_id, {**base, "selected_opportunity_uid": None} ) self.assertIsNone(optional["review"]["selected_opportunity_uid"]) def test_byte_ranges(self) -> None: self.assertEqual(parse_byte_range("bytes=2-5", 10), (2, 5)) self.assertEqual(parse_byte_range("bytes=7-", 10), (7, 9)) self.assertEqual(parse_byte_range("bytes=-3", 10), (7, 9)) with self.assertRaises(ValueError): parse_byte_range("bytes=20-30", 10) def test_cancelled_media_stream_is_not_an_error(self) -> None: class Reader: def read(self, _: int) -> bytes: return b"video" def disconnected(_: bytes) -> None: raise BrokenPipeError(32, "Broken pipe") self.assertFalse(stream_file_range(Reader(), disconnected, 5)) if __name__ == "__main__": unittest.main()