167 lines
6.6 KiB
Python
167 lines
6.6 KiB
Python
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,
|
|
)
|
|
|
|
|
|
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",
|
|
"ai_verdict": "POSITIVE",
|
|
"ai_event_type": "EOT_RESPONSE",
|
|
"ai_subtype": None,
|
|
}
|
|
)
|
|
+ "\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(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",
|
|
"reviewer": "r1",
|
|
"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"],
|
|
"EOT_RESPONSE",
|
|
)
|
|
self.assertFalse(resumed.reviews[candidates[0].candidate_id]["review"]["relabeled"])
|
|
|
|
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_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": "EOT_WEARER_BACKCHANNEL",
|
|
"exclude_reason": None,
|
|
"boundary_correction_seconds": None,
|
|
"confidence": "medium",
|
|
"reviewer": "r2",
|
|
"notes": "brief acknowledgement",
|
|
})
|
|
self.assertEqual(record["ai_event_type"], "EOT_RESPONSE")
|
|
self.assertEqual(record["review"]["human_subtype"], "EOT_WEARER_BACKCHANNEL")
|
|
self.assertTrue(record["review"]["relabeled"])
|
|
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|