diff --git a/PROJECT_CONTEXT.md b/PROJECT_CONTEXT.md index 2c0c064..30bba90 100644 --- a/PROJECT_CONTEXT.md +++ b/PROJECT_CONTEXT.md @@ -308,7 +308,9 @@ Do not reorganize them physically without compatibility wrappers. | `audit_easycom_floor_events.py` | Count EOT/takeover/backchannel candidates | | `prepare_easycom_eot_int_review.py` | Retain every raw wearer onset and offset as an unresolved semantic candidate; attach nearby participant boundaries, activity relations, transcripts, and target-of-speech facts without entry/takeover gating; also scan independent hard-negative regions; semantic labels remain AI/human review outputs | | `adjudicate_easycom_eot_int_candidates.py` | Add optional DeepSeek transcript/timing suggestions; never human ground truth | -| `review_floor_events.py` | Serve the local video/transcript workspace and persist human event reviews | +| `prepare_ego4d_av_eot_int_review.py` | Build the equivalent unresolved wearer onset/offset and hard-negative review queue from official Ego4D AV voice, transcription and annotation-only `is_at_me` fields without changing source boundaries | +| `adjudicate_ego4d_av_eot_int_candidates.py` | Apply the checkpointed strict LLM pre-label client with an Ego4D AV-specific evidence policy; every suggestion still requires human review | +| `review_floor_events.py` / `review_ego4d_av_events.py` | Serve the local EasyCom or Ego4D AV video/transcript workspace and atomically persist human event reviews | | `prepare_easycom_oracle_features.py` | Build explicit past-only oracle person-activity sidecars and analysis-only addressee controls | ### Local features diff --git a/adjudicate_ego4d_av_eot_int_candidates.py b/adjudicate_ego4d_av_eot_int_candidates.py new file mode 100644 index 0000000..9d88cf5 --- /dev/null +++ b/adjudicate_ego4d_av_eot_int_candidates.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +"""Add auditable LLM suggestions to Ego4D AV floor-event candidates. + +This reuses the checkpointing, strict JSON validation, retry/bisection, and +privacy opt-in machinery of the EasyCom annotator, but supplies an Ego4D AV +specific evidence policy. Output remains a suggestion for human review, never +ground truth. +""" + +from __future__ import annotations + +import adjudicate_easycom_eot_int_candidates as engine + + +PROMPT_VERSION = "ego4d-av-wearer-boundary-semantic-prelabel-v1" + +SYSTEM_PROMPT = """You conservatively pre-label Ego4D AV candidate regions for a scientific study of wearer-relative conversational floor control. + +Evidence policy: +1. Use only the supplied clip-local timestamps, participant roles, voice activity, transcriptions, and addressed_to_wearer facts. Never alter timestamps or infer unseen gaze, prosody, intent, identity, or acoustic evidence. +2. Ego4D AV voice segments are sound-activity annotations and may join sounds separated by less than one second. A WEARER_ONSET is not automatically a floor entry and a WEARER_OFFSET is not automatically an interruption yield. Overlap is not automatically interruption. Duration and lexical form alone do not establish conversational function. +3. addressed_to_wearer comes from annotation-only social_segments_talking.is_at_me. Treat true as supporting evidence that the other addresses the wearer, false as evidence against direct address, and null as unknown. It is evidence for annotation only, never a model input. +4. For WEARER_ONSET, label POSITIVE/WEARER_ENTRY only when the wearer genuinely enters or re-enters the floor with a contextually complete response or substantive contribution. A brief complete answer to a question can be positive. Use HARD_NEGATIVE/ENTRY_WEARER_BACKCHANNEL when the wearer merely acknowledges or feeds back while an other retains the floor. Use EXCLUDE/OUT_OF_SCOPE_WITHIN_TURN_CONTINUATION for a within-turn resumption. +5. For WEARER_OFFSET, label POSITIVE/INT_YIELD only when a non-wearer genuinely takes over during wearer speech and the wearer relinquishes the floor. A natural completion, later other onset, or end of a wearer backchannel is not INT_YIELD. Crossing the offset is descriptive timing evidence, not proof. +6. Deterministic SAME_SPEAKER_PAUSE and OTHER_TO_OTHER rows are proposed hard-negative regions. Confirm their conversational meaning from the chronology and transcript; do not blindly accept the automatic label. +7. EXCLUDE is the only non-scoring outcome and is appropriate for clear out-of-scope events, insufficient semantic evidence, missing context, recording boundaries, annotation errors, non-speech artifacts, or timestamp corruption. Every item remains subject to human review. + +Top-level verdicts and closed labels: +- POSITIVE event_type: WEARER_ENTRY or INT_YIELD. +- HARD_NEGATIVE subtype: ENTRY_OTHER_MIDTURN_PAUSE, ENTRY_OTHER_TO_OTHER, ENTRY_WEARER_BACKCHANNEL, INT_OTHER_BACKCHANNEL, INT_NON_TAKEOVER_OVERLAP, INT_WEARER_HOLDS_FLOOR, or OTHER_HARD_NEGATIVE. Use only ENTRY_* for wearer_entry tasks and INT_* for int_yield tasks, except OTHER_HARD_NEGATIVE. +- EXCLUDE reason: OUT_OF_SCOPE_NORMAL_COMPLETION, OUT_OF_SCOPE_WEARER_BACKCHANNEL_OFFSET, OUT_OF_SCOPE_WITHIN_TURN_CONTINUATION, OUT_OF_SCOPE_POST_COMPLETION_ACTIVITY, OUT_OF_SCOPE_NON_EVENT, INSUFFICIENT_EVIDENCE, DATA_ANNOTATION_ERROR, DATA_RECORDING_BOUNDARY, DATA_MISSING_CONTEXT, DATA_NON_SPEECH_ARTIFACT, or DATA_TIMESTAMP_CORRUPTION. + +For a positive WEARER_ENTRY, selected_opportunity_uid may be exactly one uid from temporal_evidence.opportunity_options when that supplied other release clearly prompted the entry; otherwise null. It must be null for all other outcomes. + +Return exactly one JSON object and no markdown: +{"items":[{"uid":"item-0000","verdict":"EXCLUDE","event_type":null,"subtype":null,"exclude_reason":"INSUFFICIENT_EVIDENCE","selected_opportunity_uid":null,"confidence":0.25,"reason":"brief evidence-based sentence"}]} +Return every supplied uid exactly once. confidence must be a number in [0,1].""" + + +def configure_engine() -> None: + engine.PROMPT_VERSION = PROMPT_VERSION + engine.SYSTEM_PROMPT = SYSTEM_PROMPT + engine.__doc__ = __doc__ + + +def main() -> None: + configure_engine() + engine.main() + + +if __name__ == "__main__": + main() diff --git a/docs/ego4d_av_llm_annotation_review.md b/docs/ego4d_av_llm_annotation_review.md new file mode 100644 index 0000000..d3d24ea --- /dev/null +++ b/docs/ego4d_av_llm_annotation_review.md @@ -0,0 +1,136 @@ +# Ego4D AV LLM-assisted floor-event annotation + +This workflow mirrors the EasyCom boundary-review pipeline while using the +official Ego4D AV annotation schema. It creates semantic review data for +wearer floor entry and interruption-driven yield; it does not modify Ego4D's +voice activity annotations and does not turn LLM output into ground truth. + +## Data and claim boundary + +Expected dataset layout: + +```text +DATA_ROOT/ + v1/ + annotations/av_train.json + annotations/av_val.json + clips/.mp4 +``` + +The generator reads clip-local `voice_segments`, `transcriptions`, and +`social_segments_talking.is_at_me`. The social label is annotation-only +evidence. It must never be passed to a forecasting model or used after a +strictly causal prediction cutoff. Ego4D's voice annotation guideline joins +sounds separated by less than one second, so a stored segment boundary is an +activity boundary, not automatically a semantic turn boundary. + +The output split remains the split declared by Ego4D (`train` or `val`). Do not +rename validation data to test or tune annotation rules after inspecting the +final evaluation set. + +## 1. Generate deterministic candidates + +Start with one clip and inspect the JSONL before running the full set: + +```bash +python3 prepare_ego4d_av_eot_int_review.py \ + --annotations "$DATA_ROOT/v1/annotations/av_train.json" \ + --output reports/ego4d_av_event_candidates/ego4d_av_train_candidates.jsonl \ + --limit 1 +``` + +Then generate the selected splits: + +```bash +python3 prepare_ego4d_av_eot_int_review.py \ + --annotations \ + "$DATA_ROOT/v1/annotations/av_train.json" \ + "$DATA_ROOT/v1/annotations/av_val.json" \ + --output reports/ego4d_av_event_candidates/ego4d_av_candidates.jsonl +``` + +The command also writes `_timelines.json` beside the JSONL. Each +raw wearer onset and offset is retained. Same-other pauses and other-to-other +handoffs are added as independently reviewable hard-negative proposals. +Timing rules only select and describe candidates; they never assign semantic +positive labels. + +## 2. Inspect what will be uploaded + +The LLM client refuses to upload transcript excerpts without an explicit +opt-in. First inspect the exact prompt locally: + +```bash +python3 adjudicate_ego4d_av_eot_int_candidates.py \ + --input reports/ego4d_av_event_candidates/ego4d_av_candidates.jsonl \ + --output reports/ego4d_av_event_candidates/ego4d_av_candidates_ai.jsonl \ + --max-events 8 \ + --dry-run +``` + +The prompt uses only timestamps, participant roles, provided transcripts, and +provided `is_at_me` facts. It does not upload video or audio. + +## 3. Run LLM pre-labeling + +After confirming that external transcript upload is authorized: + +```bash +export DEEPSEEK_API_KEY=... +python3 adjudicate_ego4d_av_eot_int_candidates.py \ + --input reports/ego4d_av_event_candidates/ego4d_av_candidates.jsonl \ + --output reports/ego4d_av_event_candidates/ego4d_av_candidates_ai.jsonl \ + --passes 2 \ + --batch-size 8 \ + --allow-external-transcript-upload +``` + +The client checkpoints after every successful batch. Two passes must agree; +disagreement becomes `EXCLUDE/INSUFFICIENT_EVIDENCE`. Malformed singleton +responses are written to a pending-human JSONL rather than silently dropped. +Every successful LLM decision has `needs_human_review=true`. + +## 4. Review in the browser + +```bash +python3 review_ego4d_av_events.py \ + --events reports/ego4d_av_event_candidates/ego4d_av_candidates_ai.jsonl \ + --dataset-root "$DATA_ROOT" \ + --host 127.0.0.1 \ + --port 8765 +``` + +Open `http://127.0.0.1:8765`. The UI shows the local video, clip-level speaker +timeline, participant-aligned transcripts, LLM rationale, and the closed human +label taxonomy. Reviews are atomically saved to +`reports/ego4d_av_event_reviews/reviews.jsonl` with a CSV companion and can be +resumed. The server rejects media paths outside `DATA_ROOT` and supports byte +ranges for browser seeking. + +For an SSH host, keep the service on loopback and forward the port: + +```bash +ssh -L 8765:127.0.0.1:8765 USER@HOST +``` + +## 5. Freeze and audit + +Before using these annotations in a paper or evaluator: + +1. Freeze candidate windows and the closed taxonomy on training data. +2. Double-review a random sample plus every low-confidence/disagreement item. +3. Report raw candidates, positives, hard negatives, exclusions, and exclusion + reasons separately for train and validation. +4. Print and independently inspect example timelines for entry, yield, + backchannel, same-speaker pause, and other-to-other cases. +5. Keep the final label named human-adjudicated semantic annotation. LLM output + alone is never ground truth. + +## Tests + +```bash +python3 -m unittest -v \ + test_prepare_ego4d_av_eot_int_review.py \ + test_review_floor_events.py \ + test_adjudicate_easycom_eot_int_candidates.py +``` diff --git a/floor_event_review_web/app.js b/floor_event_review_web/app.js index 49f9901..dc6836d 100644 --- a/floor_event_review_web/app.js +++ b/floor_event_review_web/app.js @@ -102,8 +102,14 @@ function formatTime(value) { return `${String(minutes).padStart(2, "0")}:${(seconds % 60).toFixed(1).padStart(4, "0")}`; } +function participantCode(value) { + const raw = String(value ?? "?"); + return /^(p|person|wearer)/i.test(raw) ? raw : `P${raw}`; +} + function participantIdentity(row, verbose = false) { - const participant = `P${row.participant_id ?? "?"}`; + const raw = String(row.participant_id ?? "?"); + const participant = participantCode(raw); if (row.participant_role === "wearer") { return verbose ? `W · ${participant} · wearer` : `W · ${participant}`; } @@ -282,7 +288,7 @@ function renderDetail(detail) { for (const option of opportunityOptions) { const control = document.createElement("option"); control.value = option.uid; - control.textContent = `${formatTime(option.anchor_seconds)} · P${option.participant_id ?? "?"} · ${option.transcription || "无转写"}`; + control.textContent = `${formatTime(option.anchor_seconds)} · ${participantCode(option.participant_id)} · ${option.transcription || "无转写"}`; elements.opportunitySelect.append(control); } elements.saveState.textContent = detail.review ? "已保存" : "未保存"; @@ -501,7 +507,7 @@ function renderSpeakerTimeline(detail) { const opportunity = document.createElement("i"); opportunity.className = "opportunity-anchor"; opportunity.style.left = `${100 * (Number(option.anchor_seconds) - left) / span}%`; - opportunity.title = `可选 release · ${formatTime(option.anchor_seconds)} · P${option.participant_id ?? "?"}`; + opportunity.title = `可选 release · ${formatTime(option.anchor_seconds)} · ${participantCode(option.participant_id)}`; track.append(opportunity); } const anchor = document.createElement("i"); diff --git a/floor_event_review_web/index.html b/floor_event_review_web/index.html index 45aad7e..673c5f5 100644 --- a/floor_event_review_web/index.html +++ b/floor_event_review_web/index.html @@ -3,19 +3,19 @@ - EasyCom Floor Event Review + Egocentric Floor Event Review
-

Floor Event Review

AI 粗标注 → 人工确认

+

Floor Event Review

EasyCom / Ego4D AV · AI 粗标注 → 人工确认

载入中—
- + diff --git a/prepare_ego4d_av_eot_int_review.py b/prepare_ego4d_av_eot_int_review.py new file mode 100644 index 0000000..df1cb2b --- /dev/null +++ b/prepare_ego4d_av_eot_int_review.py @@ -0,0 +1,533 @@ +#!/usr/bin/env python3 +"""Build Ego4D AV wearer-entry/INT candidates for LLM and human review. + +The input is one or more official ``av_train.json``/``av_val.json`` files. +Official voice boundaries are never moved or merged. This deterministic stage +only attaches nearby speaker activity, transcriptions, and social ``is_at_me`` +facts. Floor-entry and interruption-yield semantics remain unresolved until +LLM pre-labeling and human review. +""" + +from __future__ import annotations + +import argparse +from collections import Counter +from dataclasses import asdict, dataclass +import json +from pathlib import Path +from typing import Any, Iterable, Sequence + + +SCHEMA_VERSION = 1 +PROTOCOL_VERSION = "ego4d-av-wearer-entry-int-candidates-v1" + + +@dataclass(frozen=True) +class CandidateConfig: + context_seconds: float = 4.0 + relation_window_seconds: float = 3.0 + same_speaker_pause_max_seconds: float = 2.0 + other_handover_max_seconds: float = 1.5 + + +@dataclass(frozen=True) +class SpeechSegment: + uid: str + person_id: str + role: str + start: float + end: float + transcription: str + addressed_to_wearer: bool | None + source: dict[str, Any] + + +def _number(value: Any, *, field: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"{field} must be numeric, got {value!r}") + result = float(value) + if result < 0: + raise ValueError(f"{field} must be non-negative, got {result}") + return result + + +def _segment_times(row: dict[str, Any]) -> tuple[float, float]: + if row.get("start_time") is not None and row.get("end_time") is not None: + start = _number(row["start_time"], field="start_time") + end = _number(row["end_time"], field="end_time") + elif row.get("start_time_sec") is not None and row.get("end_time_sec") is not None: + start = _number(row["start_time_sec"], field="start_time_sec") + end = _number(row["end_time_sec"], field="end_time_sec") + else: + raise ValueError("voice segment lacks clip-local start/end time") + if end <= start: + raise ValueError(f"voice segment must have positive duration: {start}..{end}") + return start, end + + +def _person_key(value: Any) -> str: + if value is None: + return "" + return str(value).strip() + + +def _best_transcription( + rows: Sequence[dict[str, Any]], person_id: str, start: float, end: float +) -> str: + matches: list[tuple[float, float, str]] = [] + for row in rows: + if _person_key(row.get("person_id", row.get("person"))) != person_id: + continue + try: + left = _number(row.get("start_time_sec", row.get("start_time")), field="transcript start") + right = _number(row.get("end_time_sec", row.get("end_time")), field="transcript end") + except ValueError: + continue + overlap = max(0.0, min(end, right) - max(start, left)) + if overlap > 0: + matches.append((left, overlap, str(row.get("transcription", "")).strip())) + matches.sort() + return " ".join(text for _, _, text in matches if text) + + +def _addressed_to_wearer( + rows: Sequence[dict[str, Any]], person_id: str, start: float, end: float +) -> bool | None: + overlaps: list[bool] = [] + for row in rows: + if _person_key(row.get("person", row.get("person_id"))) != person_id: + continue + try: + left, right = _segment_times(row) + except ValueError: + continue + if min(end, right) - max(start, left) > 0: + value = row.get("is_at_me") + if isinstance(value, bool): + overlaps.append(value) + if not overlaps: + return None + return any(overlaps) + + +def load_clip_segments(clip: dict[str, Any]) -> tuple[list[SpeechSegment], str]: + """Return validated clip-local speaker segments and the wearer person ID.""" + clip_uid = str(clip.get("clip_uid", "")).strip() + if not clip_uid: + raise ValueError("clip lacks clip_uid") + transcripts = clip.get("transcriptions") or [] + social = clip.get("social_segments_talking") or [] + if not isinstance(transcripts, list) or not isinstance(social, list): + raise ValueError(f"{clip_uid}: transcriptions/social_segments_talking must be lists") + + people: list[dict[str, Any]] = [] + wearer = clip.get("camera_wearer") + if isinstance(wearer, dict): + people.append(wearer) + raw_people = clip.get("persons") or [] + if not isinstance(raw_people, list): + raise ValueError(f"{clip_uid}: persons must be a list") + people.extend(person for person in raw_people if isinstance(person, dict)) + + wearer_ids = { + _person_key(person.get("person_id")) + for person in people + if person.get("camera_wearer") is True + } + wearer_ids.discard("") + if len(wearer_ids) != 1: + raise ValueError(f"{clip_uid}: expected exactly one camera wearer, found {sorted(wearer_ids)}") + wearer_id = next(iter(wearer_ids)) + + segments: list[SpeechSegment] = [] + seen: set[tuple[str, float, float]] = set() + ordinal = 0 + for person in people: + person_id = _person_key(person.get("person_id")) + if not person_id: + raise ValueError(f"{clip_uid}: person lacks person_id") + voice = person.get("voice_segments") or [] + if not isinstance(voice, list): + raise ValueError(f"{clip_uid}/{person_id}: voice_segments must be a list") + for source in voice: + if not isinstance(source, dict): + raise ValueError(f"{clip_uid}/{person_id}: voice segment must be an object") + start, end = _segment_times(source) + key = (person_id, start, end) + if key in seen: + continue + seen.add(key) + segments.append( + SpeechSegment( + uid=f"{clip_uid}/speech/{ordinal:05d}", + person_id=person_id, + role="wearer" if person_id == wearer_id else "other", + start=start, + end=end, + transcription=_best_transcription(transcripts, person_id, start, end), + addressed_to_wearer=( + None if person_id == wearer_id + else _addressed_to_wearer(social, person_id, start, end) + ), + source=dict(source), + ) + ) + ordinal += 1 + segments.sort(key=lambda row: (row.start, row.end, row.person_id, row.uid)) + for left, right in zip(segments, segments[1:]): + if left.person_id == right.person_id and left.start == right.start and left.end == right.end: + raise ValueError(f"{clip_uid}: duplicate voice segment after normalization") + return segments, wearer_id + + +def _timeline_row(row: SpeechSegment) -> dict[str, Any]: + return { + "uid": row.uid, + "participant_id": row.person_id, + "participant_role": row.role, + "start_seconds": row.start, + "end_seconds": row.end, + "duration_seconds": row.end - row.start, + "transcription": row.transcription, + "addressed_to_wearer": row.addressed_to_wearer, + } + + +def _context(segments: Sequence[SpeechSegment], anchor: float, seconds: float) -> list[dict[str, Any]]: + return [ + _timeline_row(row) for row in segments + if row.end > anchor - seconds and row.start < anchor + seconds + ] + + +def _relation(segments: Sequence[SpeechSegment], anchor: float, wearer_id: str, window: float) -> dict[str, Any]: + others = [row for row in segments if row.person_id != wearer_id] + previous = [row for row in others if row.end <= anchor and anchor - row.end <= window] + following = [row for row in others if row.start >= anchor and row.start - anchor <= window] + active = [row for row in others if row.start < anchor < row.end] + prev = max(previous, key=lambda row: row.end, default=None) + nxt = min(following, key=lambda row: row.start, default=None) + return { + "previous_speaker_id": None if prev is None else prev.person_id, + "previous_speaker_offset_seconds": None if prev is None else prev.end, + "gap_from_previous_speaker_offset_seconds": None if prev is None else anchor - prev.end, + "active_other_ids": sorted({row.person_id for row in active}), + "next_speaker_id": None if nxt is None else nxt.person_id, + "next_other_onset_seconds": None if nxt is None else nxt.start, + "next_other_delay_seconds": None if nxt is None else nxt.start - anchor, + } + + +def _opportunity_options( + segments: Sequence[SpeechSegment], wearer: SpeechSegment, wearer_id: str, window: float +) -> list[dict[str, Any]]: + result = [] + for row in segments: + if row.person_id == wearer_id or row.end > wearer.start: + continue + delay = wearer.start - row.end + if delay > window: + continue + intervening = [ + other for other in segments + if other.person_id != wearer_id and other.uid != row.uid + and other.start >= row.end and other.start < wearer.start + ] + result.append({ + "uid": row.uid, + "participant_id": row.person_id, + "anchor_seconds": row.end, + "response_delay_seconds": delay, + "transcription": row.transcription, + "addressed_to_wearer": row.addressed_to_wearer, + "other_onset_between_release_and_onset": bool(intervening), + }) + return sorted(result, key=lambda row: (-float(row["anchor_seconds"]), str(row["uid"]))) + + +def _base_candidate( + *, clip: dict[str, Any], split: str, source: str, task: str, anchor_type: str, + anchor: float, region_start: float, region_end: float, sequence: int, + segments: Sequence[SpeechSegment], evidence: dict[str, Any], rule_type: str, + automatic_label: str = "candidate", automatic_subtype: str | None = None, +) -> dict[str, Any]: + clip_uid = str(clip["clip_uid"]) + uid = f"{clip_uid}/{source.lower()}/{sequence:05d}" + return { + "schema_version": SCHEMA_VERSION, + "protocol_version": PROTOCOL_VERSION, + "dataset": "ego4d_av", + "uid": uid, + "event_id": uid, + "session": clip_uid, + "clip_uid": clip_uid, + "video_uid": clip.get("video_uid"), + "split": split, + "task": task, + "event_type": rule_type, + "candidate_source": source, + "rule_candidate_type": rule_type, + "automatic_protocol_label": automatic_label, + "automatic_hard_negative_type": automatic_subtype, + "automatic_confidence": "deterministic-boundary-only", + "anchor_type": anchor_type, + "anchor_dense_seconds": anchor, + "event_time": anchor, + "region_start_seconds": region_start, + "region_end_seconds": region_end, + "clip_stem": clip_uid, + "temporal_evidence": evidence, + "speaker_timeline": _context(segments, anchor, max(anchor - region_start, region_end - anchor)), + "context": _context(segments, anchor, max(anchor - region_start, region_end - anchor)), + "ai_verdict": None, + "ai_event_type": None, + "ai_subtype": None, + "ai_confidence": None, + "ai_reason": None, + "human_status": "UNREVIEWED", + "human_verdict": None, + "human_event_type": None, + "human_subtype": None, + "exclude_reason": None, + } + + +def derive_clip_candidates( + clip: dict[str, Any], *, split: str, config: CandidateConfig +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + segments, wearer_id = load_clip_segments(clip) + clip_end = _number(clip.get("clip_end_sec", max((row.end for row in segments), default=0.0)), field="clip_end_sec") + outside = [row for row in segments if row.end > clip_end + 1e-6] + if outside: + first = outside[0] + raise ValueError( + f"{clip.get('clip_uid')}: voice segment {first.uid} ends at {first.end:.6f}s " + f"after clip_end_sec={clip_end:.6f}; refusing to clamp or drop it" + ) + candidates: list[dict[str, Any]] = [] + wearer_segments = [row for row in segments if row.person_id == wearer_id] + + for sequence, row in enumerate(wearer_segments): + previous_wearer = max( + (other for other in wearer_segments if other.end <= row.start), + key=lambda other: other.end, + default=None, + ) + evidence = _relation(segments, row.start, wearer_id, config.relation_window_seconds) + options = _opportunity_options(segments, row, wearer_id, config.relation_window_seconds) + evidence.update({ + "behavior_anchor_seconds": row.start, + "wearer_segment_uid": row.uid, + "wearer_transcription": row.transcription, + "wearer_speech_duration_after_onset_seconds": row.end - row.start, + "previous_wearer_offset_seconds": None if previous_wearer is None else previous_wearer.end, + "previous_wearer_silence_duration_seconds": None if previous_wearer is None else row.start - previous_wearer.end, + "opportunity_options": options, + "preceding_release_options_present": bool(options), + "official_voice_segments_are_unmodified": True, + }) + candidates.append(_base_candidate( + clip=clip, split=split, source="WEARER_ONSET", task="wearer_entry", + anchor_type="wearer_onset", anchor=row.start, + region_start=max(0.0, row.start - config.context_seconds), + region_end=min(clip_end, row.start + config.context_seconds), sequence=sequence, + segments=segments, evidence=evidence, rule_type="WEARER_ENTRY_CANDIDATE", + )) + + active_before = [other for other in segments if other.person_id != wearer_id and other.start < row.end < other.end] + new_during = [other for other in segments if other.person_id != wearer_id and row.start < other.start < row.end] + next_other = min( + (other for other in segments if other.person_id != wearer_id and other.start >= row.end), + key=lambda other: other.start, + default=None, + ) + offset_evidence = _relation(segments, row.end, wearer_id, config.relation_window_seconds) + offset_evidence.update({ + "behavior_anchor_seconds": row.end, + "wearer_activity_run_start_seconds": row.start, + "wearer_segment_uid": row.uid, + "wearer_transcription": row.transcription, + "other_crosses_wearer_offset": bool(active_before), + "crossing_other_ids": sorted({other.person_id for other in active_before}), + "new_other_onsets_during_wearer": [_timeline_row(other) for other in new_during], + "new_other_onset_during_wearer": bool(new_during), + "next_speaker_ids": [] if next_other is None else [next_other.person_id], + "next_speech_onset_seconds": None if next_other is None else next_other.start, + "official_voice_segments_are_unmodified": True, + }) + candidates.append(_base_candidate( + clip=clip, split=split, source="WEARER_OFFSET", task="int_yield", + anchor_type="wearer_offset", anchor=row.end, + region_start=max(0.0, row.end - config.context_seconds), + region_end=min(clip_end, row.end + config.context_seconds), sequence=sequence, + segments=segments, evidence=offset_evidence, rule_type="YIELD_CANDIDATE", + )) + + others = [row for row in segments if row.person_id != wearer_id] + control_sequence = 0 + for previous in others: + following = sorted( + (row for row in others if row.start >= previous.end and row.uid != previous.uid), + key=lambda row: (row.start, row.end), + ) + if not following: + continue + nxt = following[0] + gap = nxt.start - previous.end + wearer_in_gap = any(row.start < nxt.start and row.end > previous.end for row in wearer_segments) + subtype = None + source = None + max_gap = None + if nxt.person_id == previous.person_id: + source, subtype = "SAME_SPEAKER_PAUSE", "ENTRY_OTHER_MIDTURN_PAUSE" + max_gap = config.same_speaker_pause_max_seconds + elif nxt.person_id != previous.person_id: + source, subtype = "OTHER_TO_OTHER", "ENTRY_OTHER_TO_OTHER" + max_gap = config.other_handover_max_seconds + if gap > float(max_gap) or wearer_in_gap: + continue + evidence = _relation(segments, previous.end, wearer_id, config.relation_window_seconds) + evidence.update({ + "previous_speaker_id": previous.person_id, + "incoming_speaker_id": nxt.person_id, + "previous_segment_uid": previous.uid, + "incoming_segment_uid": nxt.uid, + "gap_seconds": gap, + "previous_transcription": previous.transcription, + "incoming_transcription": nxt.transcription, + "wearer_activity_in_gap": False, + "official_voice_segments_are_unmodified": True, + }) + candidates.append(_base_candidate( + clip=clip, split=split, source=str(source), task="wearer_entry", + anchor_type="other_offset", anchor=previous.end, + region_start=max(0.0, previous.end - config.context_seconds), + region_end=min(clip_end, previous.end + config.context_seconds), + sequence=control_sequence, segments=segments, evidence=evidence, + rule_type=str(subtype), automatic_label="hard_negative", + automatic_subtype=str(subtype), + )) + control_sequence += 1 + candidates.sort(key=lambda row: (float(row["event_time"]), str(row["uid"]))) + return candidates, [_timeline_row(row) for row in segments] + + +def iter_clips(paths: Iterable[Path]) -> Iterable[tuple[dict[str, Any], str]]: + seen: set[str] = set() + for path in paths: + payload = json.loads(path.read_text(encoding="utf-8")) + videos = payload.get("videos") if isinstance(payload, dict) else None + if not isinstance(videos, list): + raise ValueError(f"{path}: expected a videos list") + for video in videos: + if not isinstance(video, dict): + continue + split = str(video.get("split") or path.stem.removeprefix("av_") or "unknown") + for clip in video.get("clips") or []: + if not isinstance(clip, dict) or clip.get("valid") is False: + continue + uid = str(clip.get("clip_uid", "")) + if uid in seen: + raise ValueError(f"duplicate clip_uid across inputs: {uid}") + seen.add(uid) + yield clip, split + + +def _write_jsonl(path: Path, rows: Sequence[dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".partial") + with temporary.open("w", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=False) + "\n") + temporary.replace(path) + + +def _write_json(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".partial") + temporary.write_text( + json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) + temporary.replace(path) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--annotations", type=Path, nargs="+", required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--timeline-output", type=Path) + parser.add_argument("--clip-uid", action="append", help="Restrict to one or more clip UIDs") + parser.add_argument("--limit", type=int, help="Process at most N selected clips") + parser.add_argument("--context-seconds", type=float, default=4.0) + parser.add_argument("--relation-window-seconds", type=float, default=3.0) + parser.add_argument("--same-speaker-pause-max-seconds", type=float, default=2.0) + parser.add_argument("--other-handover-max-seconds", type=float, default=1.5) + parser.add_argument("--overwrite", action="store_true") + return parser.parse_args() + + +def main() -> None: + args = parse_args() + numeric = [ + args.context_seconds, args.relation_window_seconds, + args.same_speaker_pause_max_seconds, args.other_handover_max_seconds, + ] + if any(value < 0 for value in numeric): + raise ValueError("all time windows must be non-negative") + if args.limit is not None and args.limit <= 0: + raise ValueError("--limit must be positive") + paths = [path.expanduser().resolve() for path in args.annotations] + for path in paths: + if not path.is_file(): + raise FileNotFoundError(path) + selected = list(iter_clips(paths)) + if args.clip_uid: + requested = set(args.clip_uid) + known = {str(clip["clip_uid"]) for clip, _ in selected} + if requested - known: + raise ValueError(f"unknown clip UID(s): {sorted(requested - known)}") + selected = [(clip, split) for clip, split in selected if clip["clip_uid"] in requested] + if args.limit is not None: + selected = selected[: args.limit] + config = CandidateConfig(*numeric) + rows: list[dict[str, Any]] = [] + timelines: dict[str, list[dict[str, Any]]] = {} + for clip, split in selected: + derived, timeline = derive_clip_candidates(clip, split=split, config=config) + rows.extend(derived) + timelines[str(clip["clip_uid"])] = timeline + if not rows: + raise ValueError("selected annotations produced no candidates") + missing_voice_count_by_clip = { + str(clip["clip_uid"]): len(clip.get("missing_voice_segments") or []) + for clip, _ in selected + } + for row in rows: + row["temporal_evidence"]["clip_missing_voice_segments_count"] = ( + missing_voice_count_by_clip[row["clip_uid"]] + ) + output = args.output.expanduser().resolve() + timeline_output = ( + args.timeline_output.expanduser().resolve() if args.timeline_output + else output.with_name(output.stem + "_timelines.json") + ) + existing = [path for path in (output, timeline_output) if path.exists()] + if existing and not args.overwrite: + raise FileExistsError( + "Refusing to overwrite existing output(s): " + + ", ".join(str(path) for path in existing) + + "; pass --overwrite" + ) + _write_jsonl(output, rows) + _write_json( + timeline_output, + {"schema_version": 1, "dataset": "ego4d_av", "sessions": timelines}, + ) + counts = Counter(row["candidate_source"] for row in rows) + print(f"clips={len(selected)} candidates={len(rows)} output={output}") + print("candidate_sources=" + json.dumps(dict(sorted(counts.items())), ensure_ascii=False)) + print(f"timelines={timeline_output}") + + +if __name__ == "__main__": + main() diff --git a/review_ego4d_av_events.py b/review_ego4d_av_events.py new file mode 100644 index 0000000..dcebf83 --- /dev/null +++ b/review_ego4d_av_events.py @@ -0,0 +1,22 @@ +#!/usr/bin/env python3 +"""Launch the browser-based review tool for Ego4D AV floor candidates.""" + +from __future__ import annotations + +import sys + +import review_floor_events + + +def main() -> None: + if "--dataset" not in sys.argv: + sys.argv.extend(["--dataset", "ego4d"]) + if "--output" not in sys.argv: + sys.argv.extend([ + "--output", "reports/ego4d_av_event_reviews/reviews.jsonl", + ]) + review_floor_events.main() + + +if __name__ == "__main__": + main() diff --git a/review_floor_events.py b/review_floor_events.py index 461b224..46728ad 100644 --- a/review_floor_events.py +++ b/review_floor_events.py @@ -1,11 +1,10 @@ #!/usr/bin/env python3 -"""Serve a local, event-centered EasyCom annotation review workspace. +"""Serve a local, event-centered EasyCom or Ego4D AV review workspace. -The server reads JSONL candidates produced by -``derive_easycom_interaction_events.py`` or ``audit_easycom_floor_events.py``. -It serves only media inside the supplied EasyCom dataset root and stores human -reviews outside the source dataset as an atomically rewritten JSONL file plus -a CSV companion. +The server reads JSONL candidates produced by the EasyCom or Ego4D AV +candidate generators. It serves only media inside the supplied dataset root +and stores human reviews outside the source dataset as an atomically rewritten +JSONL file plus a CSV companion. """ from __future__ import annotations @@ -94,7 +93,7 @@ class Candidate: @property def session(self) -> str: - return str(self.payload.get("session", "unknown")) + return str(self.payload.get("session") or self.payload.get("clip_uid") or "unknown") @property def split(self) -> str: @@ -161,6 +160,7 @@ def expand_candidate_paths(values: Iterable[Path], *, output_path: Path) -> list if path.is_dir(): paths.extend(sorted(path.glob("Session_*.events.jsonl"))) paths.extend(sorted(path.glob("easycom_*_candidates.jsonl"))) + paths.extend(sorted(path.glob("ego4d*_candidates.jsonl"))) elif path.is_file(): paths.append(path) else: @@ -239,17 +239,35 @@ class CandidateRepository: candidates: list[Candidate], dataset_root: Path, *, + dataset: str = "auto", context_before_seconds: float, context_after_seconds: float, ) -> None: self.candidates = candidates self.by_id = {candidate.candidate_id: candidate for candidate in candidates} self.dataset_root = dataset_root.expanduser().resolve() + if dataset not in {"auto", "easycom", "ego4d"}: + raise ValueError(f"Unknown dataset: {dataset}") + declared = {str(candidate.payload.get("dataset", "")) for candidate in candidates} + kinds = {"ego4d" if value.startswith("ego4d") else "easycom" for value in declared} + if dataset == "auto" and len(kinds) != 1: + raise ValueError(f"Candidate inputs mix dataset layouts: {sorted(declared)}") + inferred = next(iter(kinds)) + self.dataset = inferred if dataset == "auto" else dataset self.context_before_seconds = context_before_seconds self.context_after_seconds = context_after_seconds self._transcript_cache: dict[tuple[str, str], list[dict[str, Any]]] = {} self.session_timelines: dict[str, list[dict[str, Any]]] = {} - for path in sorted({candidate.source_path.parent / "easycom_session_timelines.json" for candidate in candidates}): + timeline_names = ("easycom_session_timelines.json", "ego4d_av_clip_timelines.json") + timeline_paths = { + candidate.source_path.parent / name + for candidate in candidates for name in timeline_names + } + timeline_paths.update( + candidate.source_path.with_name(candidate.source_path.stem + "_timelines.json") + for candidate in candidates + ) + for path in sorted(timeline_paths): if not path.is_file(): continue payload = json.loads(path.read_text(encoding="utf-8")) @@ -279,6 +297,14 @@ class CandidateRepository: def _location(self, candidate: Candidate) -> tuple[str | None, float | None, int | None]: row = candidate.payload clip_stem = row.get("clip_stem") + if self.dataset == "ego4d": + clip_uid = row.get("clip_uid") or clip_stem or candidate.session + event_time = row.get("event_time", row.get("anchor_dense_seconds")) + return ( + None if clip_uid is None else str(clip_uid), + None if event_time is None else max(0.0, float(event_time)), + None, + ) transcript_index: int | None = None uid = None if row.get("candidate_source") else ( _uid_parts(row.get("uid")) or _uid_parts(row.get("holder_uid")) @@ -308,9 +334,12 @@ class CandidateRepository: clip_stem, _, _ = self._location(candidate) if clip_stem is None: return None - path = self.dataset_root.joinpath( - *VIDEO_ROOT_PARTS, candidate.session, f"{clip_stem}.mp4" - ).resolve() + if self.dataset == "ego4d": + path = self.dataset_root.joinpath("v1", "clips", f"{clip_stem}.mp4").resolve() + else: + path = self.dataset_root.joinpath( + *VIDEO_ROOT_PARTS, candidate.session, f"{clip_stem}.mp4" + ).resolve() if not _within_root(path, self.dataset_root) or not path.is_file(): return None return path @@ -358,7 +387,37 @@ class CandidateRepository: for row in evidence_rows if isinstance(row, dict) and row.get("participant_role") == "wearer" } - if clip_stem is not None: + if self.dataset == "ego4d": + for index, row in enumerate(evidence_rows): + if not isinstance(row, dict): + continue + try: + start = max(0.0, float(row["start_seconds"])) + end = max(start, float(row["end_seconds"])) + except (KeyError, TypeError, ValueError): + continue + if local_seconds is not None and ( + end < local_seconds - self.context_before_seconds + or start > local_seconds + self.context_after_seconds + ): + continue + transcript_rows.append({ + "uid": row.get("uid", f"{candidate.session}/context/{index:04d}"), + "index": index, + "start_seconds": start, + "end_seconds": end, + "participant_id": row.get("participant_id"), + "participant_role": row.get("participant_role", "other"), + "target_of_speech": ( + "camera_wearer" if row.get("addressed_to_wearer") is True + else "not_camera_wearer" if row.get("addressed_to_wearer") is False + else None + ), + "addressed_to_wearer": row.get("addressed_to_wearer"), + "transcription": str(row.get("transcription", "")), + "is_anchor": start <= (local_seconds or -1) <= end, + }) + elif clip_stem is not None: for index, row in enumerate(self._transcript(candidate.session, clip_stem)): try: start = max(0.0, (float(row.get("Start_Frame", 1)) - 1.0) / SOURCE_FPS) @@ -915,6 +974,10 @@ def parse_args() -> argparse.Namespace: help="Candidate JSONL files or directories containing generated candidates.", ) parser.add_argument("--dataset-root", type=Path, required=True) + parser.add_argument( + "--dataset", choices=("auto", "easycom", "ego4d"), default="auto", + help="Dataset layout; auto reads the candidate dataset field (default: %(default)s).", + ) parser.add_argument( "--output", type=Path, @@ -953,6 +1016,7 @@ def main() -> None: repository = CandidateRepository( candidates, dataset_root, + dataset=args.dataset, context_before_seconds=args.context_before_seconds, context_after_seconds=args.context_after_seconds, ) diff --git a/test_prepare_ego4d_av_eot_int_review.py b/test_prepare_ego4d_av_eot_int_review.py new file mode 100644 index 0000000..d0cb454 --- /dev/null +++ b/test_prepare_ego4d_av_eot_int_review.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import json +from pathlib import Path +import tempfile +import unittest + +from prepare_ego4d_av_eot_int_review import ( + CandidateConfig, derive_clip_candidates, iter_clips, load_clip_segments, +) +from review_floor_events import CandidateRepository, load_candidates + + +def segment(person: str, start: float, end: float) -> dict[str, object]: + return {"person": person, "start_time": start, "end_time": end} + + +def fixture_clip() -> dict[str, object]: + return { + "clip_uid": "clip-001", + "video_uid": "video-001", + "clip_start_sec": 0, + "clip_end_sec": 12.0, + "valid": True, + "camera_wearer": { + "person_id": "wearer", + "camera_wearer": True, + "voice_segments": [segment("wearer", 2.0, 3.0), segment("wearer", 6.0, 8.0)], + }, + "persons": [ + { + "person_id": "p1", "camera_wearer": False, + "voice_segments": [segment("p1", 0.5, 1.8), segment("p1", 3.2, 4.0), segment("p1", 6.5, 8.8)], + }, + { + "person_id": "p2", "camera_wearer": False, + "voice_segments": [segment("p2", 4.3, 5.2)], + }, + ], + "transcriptions": [ + {"person_id": "p1", "start_time_sec": 0.5, "end_time_sec": 1.8, "transcription": "Can you answer?"}, + {"person_id": "wearer", "start_time_sec": 2.0, "end_time_sec": 3.0, "transcription": "Yes."}, + {"person_id": "p1", "start_time_sec": 6.5, "end_time_sec": 8.8, "transcription": "Wait, one thing."}, + ], + "social_segments_talking": [ + {"person": "p1", "start_time": 0.5, "end_time": 1.8, "is_at_me": True}, + ], + } + + +class Ego4DAVCandidateTests(unittest.TestCase): + def test_loads_roles_transcript_and_social_evidence(self) -> None: + rows, wearer_id = load_clip_segments(fixture_clip()) + self.assertEqual(wearer_id, "wearer") + question = next(row for row in rows if row.start == 0.5) + self.assertEqual(question.transcription, "Can you answer?") + self.assertIs(question.addressed_to_wearer, True) + self.assertEqual(next(row.role for row in rows if row.person_id == "wearer"), "wearer") + + def test_derives_raw_wearer_boundaries_and_controls(self) -> None: + rows, timeline = derive_clip_candidates( + fixture_clip(), split="train", config=CandidateConfig() + ) + sources = [row["candidate_source"] for row in rows] + self.assertEqual(sources.count("WEARER_ONSET"), 2) + self.assertEqual(sources.count("WEARER_OFFSET"), 2) + self.assertIn("OTHER_TO_OTHER", sources) + onset = next(row for row in rows if row["candidate_source"] == "WEARER_ONSET") + option = onset["temporal_evidence"]["opportunity_options"][0] + self.assertEqual(option["participant_id"], "p1") + self.assertIs(option["addressed_to_wearer"], True) + offset = next(row for row in rows if row["candidate_source"] == "WEARER_OFFSET" and row["event_time"] == 8.0) + self.assertTrue(offset["temporal_evidence"]["other_crosses_wearer_offset"]) + self.assertTrue(offset["temporal_evidence"]["new_other_onset_during_wearer"]) + self.assertTrue(all(row["start_seconds"] < row["end_seconds"] for row in timeline)) + + def test_duplicate_clip_uid_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + payload = {"videos": [{"split": "train", "clips": [fixture_clip()]}]} + paths = [] + for name in ("av_train.json", "av_val.json"): + path = root / name + path.write_text(json.dumps(payload), encoding="utf-8") + paths.append(path) + with self.assertRaisesRegex(ValueError, "duplicate clip_uid"): + list(iter_clips(paths)) + + def test_review_repository_uses_ego4d_media_and_embedded_transcript(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + media = root / "v1" / "clips" / "clip-001.mp4" + media.parent.mkdir(parents=True) + media.write_bytes(b"not-a-real-video") + rows, _ = derive_clip_candidates( + fixture_clip(), split="val", config=CandidateConfig() + ) + events = root / "ego4d_av_candidates.jsonl" + events.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") + candidates = load_candidates([events]) + repository = CandidateRepository( + candidates, root, dataset="auto", + context_before_seconds=3.0, context_after_seconds=3.0, + ) + self.assertEqual(repository.dataset, "ego4d") + self.assertEqual(repository.media_path(candidates[0]), media.resolve()) + detail = repository.detail(candidates[0]) + self.assertTrue(detail["transcript_available"]) + self.assertTrue(any(row["transcription"] == "Can you answer?" for row in detail["transcript"])) + + def test_rejects_missing_camera_wearer(self) -> None: + clip = fixture_clip() + clip.pop("camera_wearer") + with self.assertRaisesRegex(ValueError, "exactly one camera wearer"): + load_clip_segments(clip) + + +if __name__ == "__main__": + unittest.main()