diff --git a/README.md b/README.md index d02ffb9..57fc042 100644 --- a/README.md +++ b/README.md @@ -1658,6 +1658,123 @@ semantics. Run the following controlled experiments: The frozen-Omni head is the cleanest representation test. LoRA can be reported as a second-stage ceiling, not as the only result. +## Transcript-derived interaction-event proxies + +EasyCom's released human transcriptions contain inclusive 20 FPS +`Start_Frame`/`End_Frame`, `Participant_ID`, `Transcription`, and +`Target_of_Speech`. They do **not** contain a semantic backchannel class. +[`derive_easycom_interaction_events.py`](derive_easycom_interaction_events.py) +therefore separates observed timing relations from weak semantic proxies: + +- `ordinary_switch`: a non-overlapping recent speaker change followed by a + sustained incoming turn; +- `overlap_takeover`: an observed overlapping onset followed by sustained + takeover, without claiming interruption intent; +- `overlap_hold`: overlap without enough evidence for takeover; +- `backchannel_proxy_high`: a conservative short lexical response while the + same other participant dominates immediately before and after it; +- `interruption_proxy_high`: the previous transcript has a non-boundary + trailing cutoff and the incoming participant takes over; +- `nonverbal_feedback`: laughter-only feedback around another participant's + floor; pure `[H]`, `[C]`, `[T]`, and `[U]` records are ambiguous candidates. + +`backchannel_candidate`, `interruption_candidate`, and +`nonverbal_candidate` are retained for auditing but masked from auxiliary +training. `[X]` continuations across adjacent one-minute files are not new +events. Source-time gaps larger than one second create separate segments, so +the dense EgoSpeak concatenation does not create a false switch across a +redacted interval. A one- or two-frame annotation/media endpoint discrepancy +is clamped and reported; larger disagreement is fatal. + +A read-only audit on the raw-data machine is: + +```bash +cd /mnt/58462F4C462F29E8/Code/icassp2027-experiment + +python3 derive_easycom_interaction_events.py \ + --dataset-root /mnt/58462F4C462F29E8/Datasets/easycom \ + --target-root /mnt/58462F4C462F29E8/Code/EgoSpeak/dataset/EasyCom/target_perframe \ + --examples-per-event 10 +``` + +To persist JSONL/CSV reports outside the dataset, add: + +```bash + --output-dir reports/easycom_interaction_events_v1 \ + --overwrite +``` + +In an environment with NumPy, add `--write-frame-arrays` to also write one +`Session_*.aux_targets.npz` per session. It contains 5 Hz `event_onsets`, +`event_roles`, collision flags, and event-name metadata aligned to +`target_perframe`. The raw EasyCom tree is never modified. + +The initial all-session audit found 6,981 utterances, including 530 +high-confidence backchannel proxies, 78 high-confidence interruption proxies, +937 overlap-takeovers, 686 overlap-holds, and 349 laughter-feedback events. +These counts establish usable weak-supervision coverage, not label accuracy. +Before training, inspect a stratified sample from Sessions 4--12 and freeze the +rules; do not tune thresholds on test Sessions 1--3. The primary evaluator +remains the original EgoSpeak three-label AP/mAP protocol. + +### Optional DeepSeek adjudication of ambiguous candidates + +DeepSeek is not required for the deterministic timing labels above. It is an +optional second-stage weak-label adjudicator for only +`backchannel_candidate` and `interruption_candidate`; sending all 6,981 +utterances to an LLM would add cost and noise without adding observable +prosody, gaze, or intent. The current audit contains 614 such candidates over +all sessions and 497 in the training split (Sessions 4--12). + +[`adjudicate_easycom_interaction_events.py`](adjudicate_easycom_interaction_events.py) +uses DeepSeek JSON Output through the standard-library HTTP client, so no +additional Python package is required. It makes two differently instructed +conservative passes. A candidate is promoted only if both passes return the candidate's +matching semantic label with high confidence; disagreements, medium/low +confidence, short answers, cooperative overlaps, and uncertain cases remain +masked. The result is still a weak proxy and must not be presented as human +ground truth. The append-only JSONL records prompt/model metadata and resumes +completed candidates, while rejecting a mixed model or prompt configuration. + +First inspect exactly what would leave the machine. This makes no API call and +does not require a key: + +```bash +python3 adjudicate_easycom_interaction_events.py \ + --input-dir reports/easycom_interaction_events_v1 \ + --output reports/easycom_interaction_llm_v1.jsonl \ + --max-events 16 \ + --dry-run +``` + +If external transcript upload is authorized, enter the key without placing it +in shell history and run a 32-candidate pilot: + +```bash +read -s -p "DeepSeek API key: " DEEPSEEK_API_KEY; echo +export DEEPSEEK_API_KEY + +python3 adjudicate_easycom_interaction_events.py \ + --input-dir reports/easycom_interaction_events_v1 \ + --output reports/easycom_interaction_llm_pilot_v1.jsonl \ + --max-events 32 \ + --selection-seed 2027 \ + --batch-size 8 \ + --passes 2 \ + --allow-external-transcript-upload + +unset DEEPSEEK_API_KEY +``` + +Audit the pilot rationales and precision manually before running the full +training split. For the full pass, use a new output path, remove +`--max-events`, and keep the default `--splits train`. The default model is +`deepseek-v4-flash`; it can be changed with `--model` or `DEEPSEEK_MODEL` as +the provider's available model names evolve. Never tune the prompt or +acceptance rule using Sessions 1--3. LLM-adjudicated candidates are deliberately +kept in a separate JSONL until pilot precision is accepted; they are not yet +silently merged into the training arrays. + ## Candidate improvement: explicit working and long memory Motivated by EgoSpeak's separation of long-term and working memory, test an diff --git a/adjudicate_easycom_interaction_events.py b/adjudicate_easycom_interaction_events.py new file mode 100644 index 0000000..2ce4b8f --- /dev/null +++ b/adjudicate_easycom_interaction_events.py @@ -0,0 +1,461 @@ +#!/usr/bin/env python3 +"""Optionally adjudicate ambiguous EasyCom interaction proxies with DeepSeek. + +This is a second-stage *weak-label adjudicator*, not a replacement for the +timestamp rules in ``derive_easycom_interaction_events.py``. By default it +only reviews ambiguous training candidates and promotes a semantic label when +two passes agree and both report high confidence. + +Transcript excerpts are sent to an external API only when the caller passes +``--allow-external-transcript-upload``. The API key is read exclusively from +``DEEPSEEK_API_KEY`` and is never written to output. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import random +import time +import urllib.error +import urllib.request +from collections import Counter, defaultdict +from pathlib import Path +from typing import Any, Iterable, Sequence + + +PROMPT_VERSION = "easycom-interaction-adjudicator-v1" +OUTPUT_SCHEMA_VERSION = 1 +DEFAULT_BASE_URL = "https://api.deepseek.com" +DEFAULT_MODEL = "deepseek-v4-flash" +DEFAULT_EVENT_TYPES = ("backchannel_candidate", "interruption_candidate") +ALLOWED_LABELS = { + "backchannel", + "short_answer_or_response", + "interruption", + "cooperative_overlap_takeover", + "other", + "uncertain", +} +ALLOWED_CONFIDENCE = {"high", "medium", "low"} + +SYSTEM_PROMPT = """You are adjudicating weak labels for conversational turn-taking research. +Use only the supplied transcript and timing evidence. Do not infer prosody, gaze, or intention that is not observed. + +Definitions: +- backchannel: a brief acknowledgement/feedback that does not take the conversational floor. +- short_answer_or_response: a brief but content-bearing answer/response, not merely feedback. +- interruption: the incoming speaker competitively cuts off the previous floor holder and takes the floor. +- cooperative_overlap_takeover: overlapping floor transfer without enough evidence of competitive interruption. +- other: none of the above. +- uncertain: the available text/timing cannot reliably distinguish the labels. + +Be conservative. Timing overlap alone does not prove interruption. A short word alone does not prove backchannel. +Return one JSON object exactly in this form: +{"items":[{"uid":"...","label":"uncertain","confidence":"low","rationale":"brief evidence-based sentence"}]} +Return every input uid exactly once. The response must be JSON, with no markdown or extra text.""" + + +def _load_jsonl(path: Path) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + with path.open("r", encoding="utf-8") as handle: + for line_number, line in enumerate(handle, start=1): + line = line.strip() + if not line: + continue + try: + row = json.loads(line) + except json.JSONDecodeError as exc: + raise ValueError(f"Invalid JSON at {path}:{line_number}: {exc}") from exc + if not isinstance(row, dict) or not isinstance(row.get("uid"), str): + raise ValueError(f"Expected object with uid at {path}:{line_number}") + rows.append(row) + return rows + + +def load_candidates( + input_dir: Path, + *, + splits: set[str], + event_types: set[str], +) -> list[dict[str, Any]]: + paths = sorted(input_dir.glob("Session_*.events.jsonl")) + if not paths: + raise FileNotFoundError(f"No Session_*.events.jsonl files in {input_dir}") + candidates: list[dict[str, Any]] = [] + seen: set[str] = set() + for path in paths: + for row in _load_jsonl(path): + uid = row["uid"] + if uid in seen: + raise ValueError(f"Duplicate event uid: {uid}") + seen.add(uid) + if row.get("split") in splits and row.get("event_type") in event_types: + candidates.append(row) + return sorted(candidates, key=lambda row: row["uid"]) + + +def stratified_limit( + rows: Sequence[dict[str, Any]], + limit: int | None, + *, + seed: int, +) -> list[dict[str, Any]]: + """Select a deterministic pilot across sessions and candidate types.""" + + if limit is None or limit >= len(rows): + return list(rows) + if limit < 0: + raise ValueError("--max-events must be non-negative") + rng = random.Random(seed) + buckets: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list) + for row in rows: + buckets[(str(row.get("event_type")), str(row.get("session")))].append(row) + keys = sorted(buckets) + for key in keys: + rng.shuffle(buckets[key]) + rng.shuffle(keys) + selected: list[dict[str, Any]] = [] + while len(selected) < limit: + made_progress = False + for key in keys: + if buckets[key]: + selected.append(buckets[key].pop()) + made_progress = True + if len(selected) == limit: + break + if not made_progress: + break + return sorted(selected, key=lambda row: row["uid"]) + + +def _candidate_for_prompt(row: dict[str, Any]) -> dict[str, Any]: + fields = ( + "uid", + "event_type", + "participant_role", + "participant_id", + "previous_floor_participant_id", + "retained_floor_participant_id", + "target_matches_previous_floor", + "duration_seconds", + "onset_overlap_seconds", + "gap_from_previous_seconds", + "speaker_after_previous_end_seconds", + "previous_transcript_has_cutoff", + "previous_transcription", + "transcription", + "target_of_speech", + ) + return {field: row.get(field) for field in fields} + + +def build_user_prompt(rows: Sequence[dict[str, Any]], pass_index: int) -> str: + if pass_index == 0: + review = "Primary pass: classify conservatively from the supplied evidence." + else: + review = ( + "Skeptical review pass: actively look for alternative explanations. " + "Use uncertain unless the semantic label remains well supported." + ) + payload = [_candidate_for_prompt(row) for row in rows] + return ( + review + + "\nReturn JSON for these candidates:\n" + + json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + ) + + +def prompt_hash(rows: Sequence[dict[str, Any]], passes: int) -> str: + payload = { + "version": PROMPT_VERSION, + "passes": passes, + "system": SYSTEM_PROMPT, + "rows": [_candidate_for_prompt(row) for row in rows], + } + encoded = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _validate_decisions( + payload: Any, + expected_uids: Sequence[str], +) -> list[dict[str, str]]: + if not isinstance(payload, dict) or not isinstance(payload.get("items"), list): + raise ValueError("Response must be a JSON object containing an items list") + expected = set(expected_uids) + actual: dict[str, dict[str, str]] = {} + for item in payload["items"]: + if not isinstance(item, dict): + raise ValueError("Every response item must be an object") + uid = item.get("uid") + label = item.get("label") + confidence = item.get("confidence") + rationale = item.get("rationale") + if uid not in expected or uid in actual: + raise ValueError(f"Unexpected or duplicate uid in response: {uid!r}") + if label not in ALLOWED_LABELS: + raise ValueError(f"Invalid label for {uid}: {label!r}") + if confidence not in ALLOWED_CONFIDENCE: + raise ValueError(f"Invalid confidence for {uid}: {confidence!r}") + if not isinstance(rationale, str) or not rationale.strip(): + raise ValueError(f"Missing rationale for {uid}") + actual[uid] = { + "uid": uid, + "label": label, + "confidence": confidence, + "rationale": rationale.strip(), + } + if set(actual) != expected: + raise ValueError(f"Response uid mismatch; missing={sorted(expected-set(actual))}") + return [actual[uid] for uid in expected_uids] + + +def _post_chat_completion( + *, + api_key: str, + base_url: str, + model: str, + user_prompt: str, + timeout_seconds: float, + max_retries: int, +) -> tuple[dict[str, Any], dict[str, Any]]: + endpoint = base_url.rstrip("/") + "/chat/completions" + body = { + "model": model, + "messages": [ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": user_prompt}, + ], + "response_format": {"type": "json_object"}, + "thinking": {"type": "disabled"}, + "temperature": 0, + "max_tokens": 2048, + "stream": False, + } + request = urllib.request.Request( + endpoint, + data=json.dumps(body, ensure_ascii=False).encode("utf-8"), + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + method="POST", + ) + last_error: Exception | None = None + for attempt in range(max_retries + 1): + try: + with urllib.request.urlopen(request, timeout=timeout_seconds) as response: + envelope = json.loads(response.read().decode("utf-8")) + content = envelope["choices"][0]["message"]["content"] + if not isinstance(content, str) or not content.strip(): + raise ValueError("DeepSeek returned empty JSON content") + return json.loads(content), envelope.get("usage", {}) + except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, KeyError, json.JSONDecodeError, ValueError) as exc: + last_error = exc + if attempt >= max_retries: + break + time.sleep(min(30.0, 2.0**attempt)) + raise RuntimeError(f"DeepSeek request failed after {max_retries + 1} attempts: {last_error}") + + +def adjudicate_batch( + rows: Sequence[dict[str, Any]], + *, + api_key: str, + base_url: str, + model: str, + passes: int, + timeout_seconds: float, + max_retries: int, +) -> tuple[list[dict[str, Any]], Counter[str]]: + expected_uids = [row["uid"] for row in rows] + by_uid: dict[str, list[dict[str, str]]] = {uid: [] for uid in expected_uids} + usage: Counter[str] = Counter() + for pass_index in range(passes): + payload, pass_usage = _post_chat_completion( + api_key=api_key, + base_url=base_url, + model=model, + user_prompt=build_user_prompt(rows, pass_index), + timeout_seconds=timeout_seconds, + max_retries=max_retries, + ) + for key, value in pass_usage.items(): + if isinstance(value, int): + usage[key] += value + for decision in _validate_decisions(payload, expected_uids): + by_uid[decision["uid"]].append(decision) + + request_prompt_sha256 = prompt_hash(rows, passes) + results: list[dict[str, Any]] = [] + for source in rows: + decisions = by_uid[source["uid"]] + labels = {decision["label"] for decision in decisions} + unanimous = len(labels) == 1 + final_label = decisions[0]["label"] if unanimous else "uncertain" + all_high = all(decision["confidence"] == "high" for decision in decisions) + expected_label = { + "backchannel_candidate": "backchannel", + "interruption_candidate": "interruption", + }.get(source.get("event_type")) + accepted = unanimous and all_high and final_label == expected_label + promoted = ( + f"{final_label}_proxy_llm_consensus" if accepted else None + ) + results.append( + { + "schema_version": OUTPUT_SCHEMA_VERSION, + "uid": source["uid"], + "session": source.get("session"), + "split": source.get("split"), + "source_event_type": source.get("event_type"), + "model": model, + "prompt_version": PROMPT_VERSION, + "request_prompt_sha256": request_prompt_sha256, + "passes": passes, + "decisions": decisions, + "unanimous": unanimous, + "final_label": final_label, + "all_high_confidence": all_high, + "accepted_for_auxiliary": accepted, + "promoted_event_type": promoted, + } + ) + return results, usage + + +def _validate_resume_rows( + rows: Sequence[dict[str, Any]], + *, + model: str, + passes: int, +) -> None: + """Prevent an append-only output from silently mixing experiments.""" + + seen: set[str] = set() + for row in rows: + uid = row["uid"] + if uid in seen: + raise ValueError(f"Duplicate uid in existing output: {uid}") + seen.add(uid) + expected = { + "schema_version": OUTPUT_SCHEMA_VERSION, + "model": model, + "prompt_version": PROMPT_VERSION, + "passes": passes, + } + mismatches = { + key: (row.get(key), value) + for key, value in expected.items() + if row.get(key) != value + } + if mismatches: + raise ValueError( + f"Existing output {uid} was produced by a different adjudication " + f"configuration: {mismatches}. Use a new --output path." + ) + + +def _chunks(rows: Sequence[dict[str, Any]], size: int) -> Iterable[Sequence[dict[str, Any]]]: + for start in range(0, len(rows), size): + yield rows[start : start + size] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input-dir", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--splits", nargs="+", default=["train"]) + parser.add_argument("--event-types", nargs="+", default=list(DEFAULT_EVENT_TYPES)) + parser.add_argument("--model", default=os.environ.get("DEEPSEEK_MODEL", DEFAULT_MODEL)) + parser.add_argument("--base-url", default=os.environ.get("DEEPSEEK_BASE_URL", DEFAULT_BASE_URL)) + parser.add_argument("--batch-size", type=int, default=8) + parser.add_argument("--passes", type=int, default=2) + parser.add_argument("--max-events", type=int) + parser.add_argument("--selection-seed", type=int, default=2027) + parser.add_argument("--timeout-seconds", type=float, default=120.0) + parser.add_argument("--max-retries", type=int, default=3) + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--allow-external-transcript-upload", action="store_true") + return parser.parse_args() + + +def main() -> None: + args = parse_args() + if args.batch_size <= 0 or args.passes <= 0 or args.max_retries < 0: + raise ValueError("batch-size/passes must be positive and max-retries non-negative") + candidates = load_candidates( + args.input_dir.resolve(), + splits=set(args.splits), + event_types=set(args.event_types), + ) + candidates = stratified_limit( + candidates, + args.max_events, + seed=args.selection_seed, + ) + + completed: set[str] = set() + if args.output.exists(): + resume_rows = _load_jsonl(args.output) + _validate_resume_rows(resume_rows, model=args.model, passes=args.passes) + completed = {row["uid"] for row in resume_rows} + pending = [row for row in candidates if row["uid"] not in completed] + print( + f"Candidates={len(candidates)} completed={len(completed & {row['uid'] for row in candidates})} " + f"pending={len(pending)} model={args.model} passes={args.passes}" + ) + if not pending: + return + + if args.dry_run: + sample = pending[: args.batch_size] + print("DRY RUN: no API request will be made. First request prompt:\n") + print(build_user_prompt(sample, 0)) + print("\nprompt_sha256:", prompt_hash(sample, args.passes)) + return + if not args.allow_external_transcript_upload: + raise PermissionError( + "Refusing to upload transcript excerpts. Review --dry-run, then pass " + "--allow-external-transcript-upload if authorized." + ) + api_key = os.environ.get("DEEPSEEK_API_KEY") + if not api_key: + raise RuntimeError("Set DEEPSEEK_API_KEY in the environment") + + args.output.parent.mkdir(parents=True, exist_ok=True) + usage: Counter[str] = Counter() + accepted: Counter[str] = Counter() + processed = 0 + with args.output.open("a", encoding="utf-8") as handle: + for batch_number, batch in enumerate(_chunks(pending, args.batch_size), start=1): + results, batch_usage = adjudicate_batch( + batch, + api_key=api_key, + base_url=args.base_url, + model=args.model, + passes=args.passes, + timeout_seconds=args.timeout_seconds, + max_retries=args.max_retries, + ) + usage.update(batch_usage) + for result in results: + handle.write(json.dumps(result, ensure_ascii=False) + "\n") + if result["accepted_for_auxiliary"]: + accepted[result["final_label"]] += 1 + handle.flush() + processed += len(results) + print( + f"batch={batch_number} processed={processed}/{len(pending)} " + f"accepted={dict(accepted)} total_tokens={usage.get('total_tokens', 0)}" + ) + print(f"Saved adjudications to {args.output.resolve()}") + print("Usage:", dict(usage)) + print("Accepted high-confidence consensus:", dict(accepted)) + + +if __name__ == "__main__": + main() diff --git a/derive_easycom_interaction_events.py b/derive_easycom_interaction_events.py new file mode 100644 index 0000000..0c2467a --- /dev/null +++ b/derive_easycom_interaction_events.py @@ -0,0 +1,874 @@ +#!/usr/bin/env python3 +"""Derive auditable EasyCom interaction-event proxies from transcripts. + +The released EasyCom transcription JSON files provide participant-resolved +``Start_Frame``/``End_Frame`` intervals, text, and ``Target_of_Speech``. They +do *not* provide a semantic backchannel label. This script therefore keeps +deterministic timing events separate from high-precision weak proxies: + +* ordinary speaker switch; +* backchannel proxy (short feedback while another participant retains floor); +* interruption proxy (overlapping onset followed by floor takeover); +* overlap without takeover; +* non-verbal feedback; and +* ambiguous candidates, which should be masked from auxiliary training. + +The source dataset is never modified. With ``--output-dir`` the script writes +JSONL/CSV reports and optional 5 Hz ``.npz`` onset arrays to a separate path. +Without it, the command is a read-only audit. +""" + +from __future__ import annotations + +import argparse +import csv +import json +import re +import ast +import struct +from collections import Counter, defaultdict +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Sequence + +from easycom_manifest import ALL_SESSIONS, TEST_SESSIONS, parse_timecode, probe_video + + +SOURCE_FPS = 20 +TASK_FPS = 5 +CAMERA_WEARER_ID = 2 +FRAMES_PER_TASK_STEP = SOURCE_FPS // TASK_FPS + +AUXILIARY_EVENT_NAMES = ( + "ordinary_switch", + "backchannel_proxy_high", + "overlap_takeover", + "interruption_proxy_high", + "overlap_hold", + "nonverbal_feedback", +) +AMBIGUOUS_EVENT_NAMES = ( + "backchannel_candidate", + "interruption_candidate", + "nonverbal_candidate", +) + +MARKER_RE = re.compile(r"\[[A-Za-z]\]") +WORD_RE = re.compile(r"[a-z]+(?:'[a-z]+)?") +TRAILING_CUTOFF_RE = re.compile(r"-\s*(?:\[[A-Za-z]\]\s*)*$") +PURE_NONVERBAL_RE = re.compile(r"^(?:\s*\[(?:L|C|T|H|U)\]\s*[.!?]?\s*)+$") + +# Deliberately conservative. A lexical match is not sufficient: timing must +# also show that another participant retained the floor. +FEEDBACK_PHRASES = { + "alright", + "awesome", + "cool", + "definitely", + "exactly", + "fair", + "good", + "got it", + "great", + "huh", + "i know", + "i see", + "interesting", + "makes sense", + "mhm", + "mm", + "mm hmm", + "nice", + "no", + "no way", + "nope", + "oh", + "oh man", + "okay", + "really", + "right", + "sure", + "true", + "uh huh", + "uh-huh", + "wow", + "yeah", + "yep", + "yes", + "yup", +} +FEEDBACK_TOKENS = { + token for phrase in FEEDBACK_PHRASES for token in WORD_RE.findall(phrase) +} + + +@dataclass(frozen=True) +class Clip: + path: str + transcript_path: str + stem: str + source_start_seconds: float + source_end_seconds: float + dense_start_frame: int + num_frames: int + segment_id: int + discontinuity_before_seconds: float | None + + @property + def dense_start_seconds(self) -> float: + return self.dense_start_frame / SOURCE_FPS + + +@dataclass(frozen=True) +class Utterance: + uid: str + session: str + clip_stem: str + clip_index: int + segment_id: int + participant_id: int + target_of_speech: tuple[int | str, ...] + transcription: str + source_start_frame: int + source_end_frame: int + dense_start_frame: int + dense_end_frame_exclusive: int + dense_start_seconds: float + dense_end_seconds: float + duration_seconds: float + starts_at_clip_boundary: bool + ends_at_clip_boundary: bool + annotation_end_frame_clamped_by: int + is_cross_clip_continuation: bool + + +@dataclass(frozen=True) +class Event: + uid: str + session: str + split: str + event_type: str + confidence: str + trainable_auxiliary: bool + participant_id: int + participant_role: str + previous_floor_participant_id: int | None + retained_floor_participant_id: int | None + target_matches_previous_floor: bool + clip_stem: str + source_start_frame: int + dense_start_seconds: float + task_step_index: int + duration_seconds: float + onset_overlap_seconds: float + gap_from_previous_seconds: float | None + speaker_after_previous_end_seconds: float + previous_transcript_has_cutoff: bool + previous_transcription: str | None + lexical_feedback: bool + pure_nonverbal: bool + transcription: str + target_of_speech: tuple[int | str, ...] + reason: str + + +@dataclass(frozen=True) +class Rules: + context_seconds: float = 1.5 + backchannel_max_seconds: float = 1.0 + max_switch_gap_seconds: float = 0.6 + min_overlap_seconds: float = 0.1 + takeover_candidate_seconds: float = 0.4 + takeover_high_seconds: float = 0.8 + max_contiguous_source_gap_seconds: float = 1.0 + + +def _load_json(path: Path) -> Any: + payload = path.read_bytes() + errors: list[str] = [] + for encoding in ("utf-8", "cp1252", "latin1"): + try: + return json.loads(payload.decode(encoding)) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + errors.append(f"{encoding}: {exc}") + raise ValueError(f"Could not decode {path}: {'; '.join(errors)}") + + +def _normalized_words(text: str) -> tuple[str, ...]: + return tuple(WORD_RE.findall(MARKER_RE.sub(" ", text).lower())) + + +def is_lexical_feedback(text: str) -> bool: + if "?" in text: + return False + words = _normalized_words(text) + if not words or len(words) > 5: + return False + phrase = " ".join(words) + return phrase in FEEDBACK_PHRASES or all(word in FEEDBACK_TOKENS for word in words) + + +def is_pure_nonverbal(text: str) -> bool: + return PURE_NONVERBAL_RE.fullmatch(text.strip()) is not None + + +def has_trailing_cutoff(utterance: Utterance) -> bool: + # [X] at a one-minute boundary denotes a continued/truncated transcript, + # not evidence that another participant interrupted the speaker. + return ( + TRAILING_CUTOFF_RE.search(utterance.transcription.strip()) is not None + and "[X]" not in utterance.transcription.upper() + and not utterance.ends_at_clip_boundary + ) + + +def _target_values(value: Any) -> tuple[int | str, ...]: + values = value if isinstance(value, list) else [value] + normalized: list[int | str] = [] + for item in values: + if isinstance(item, bool): + normalized.append(str(item)) + elif isinstance(item, int): + normalized.append(item) + elif isinstance(item, float) and item.is_integer(): + normalized.append(int(item)) + else: + normalized.append(str(item)) + return tuple(normalized) + + +def _target_length(path: Path | None) -> int | None: + if path is None: + return None + # Read only the NPY header so an audit does not require NumPy on the raw + # data host. NumPy is required only when --write-frame-arrays is requested. + with path.open("rb") as handle: + if handle.read(6) != bytes([147]) + b"NUMPY": + raise ValueError(f"Not a NumPy array: {path}") + major, _minor = handle.read(2) + length_size = 2 if major == 1 else 4 + header_length = struct.unpack( + " list[Clip]: + video_dir = dataset_root / "Main" / "Video_Compressed" / session + transcript_dir = dataset_root / "Main" / "Speech_Transcriptions" / session + if not video_dir.is_dir(): + raise FileNotFoundError(f"Missing video directory: {video_dir}") + if not transcript_dir.is_dir(): + raise FileNotFoundError(f"Missing transcript directory: {transcript_dir}") + + video_paths = sorted(video_dir.glob("*.mp4"), key=lambda p: parse_timecode(p.stem)) + transcript_paths = {path.stem: path for path in transcript_dir.glob("*.json")} + video_stems = {path.stem for path in video_paths} + if video_stems != set(transcript_paths): + missing_json = sorted(video_stems - set(transcript_paths)) + missing_video = sorted(set(transcript_paths) - video_stems) + raise ValueError( + f"{session} video/transcript inventory differs; " + f"missing_json={missing_json}, missing_video={missing_video}" + ) + + clips: list[Clip] = [] + dense_start_frame = 0 + previous_source_end: float | None = None + segment_id = 0 + for path in video_paths: + media = probe_video(path) + frames = int(media["frames_at_20hz"]) + source_start = parse_timecode(path.stem) + source_end = source_start + frames / SOURCE_FPS + delta = None if previous_source_end is None else source_start - previous_source_end + if delta is not None and abs(delta) > rules.max_contiguous_source_gap_seconds: + segment_id += 1 + clips.append( + Clip( + path=str(path.resolve()), + transcript_path=str(transcript_paths[path.stem].resolve()), + stem=path.stem, + source_start_seconds=source_start, + source_end_seconds=source_end, + dense_start_frame=dense_start_frame, + num_frames=frames, + segment_id=segment_id, + discontinuity_before_seconds=delta, + ) + ) + dense_start_frame += frames + previous_source_end = source_end + return clips + + +def load_utterances(session: str, clips: Sequence[Clip]) -> list[Utterance]: + utterances: list[Utterance] = [] + per_clip: list[list[Utterance]] = [] + for clip_index, clip in enumerate(clips): + payload = _load_json(Path(clip.transcript_path)) + if not isinstance(payload, list): + raise ValueError(f"Expected a list in {clip.transcript_path}") + clip_rows: list[Utterance] = [] + for row_index, row in enumerate(payload): + required = { + "Start_Frame", + "End_Frame", + "Participant_ID", + "Transcription", + "Target_of_Speech", + } + if not isinstance(row, dict) or not required.issubset(row): + raise ValueError( + f"Invalid transcription row {row_index} in {clip.transcript_path}: {row}" + ) + start = int(row["Start_Frame"]) + annotation_end = int(row["End_Frame"]) + end_overrun = max(0, annotation_end - clip.num_frames) + # A few released MP4/container frame counts differ from the JSON + # endpoint by one frame. Permit only a tightly bounded clamp and + # expose it in the audit report; larger disagreements remain fatal. + if end_overrun > 2: + raise ValueError( + f"Annotation endpoint exceeds media by {end_overrun} frames in " + f"{clip.transcript_path}: End_Frame={annotation_end}, " + f"clip_frames={clip.num_frames}" + ) + end = min(annotation_end, clip.num_frames) + if start < 1 or end < start: + raise ValueError( + f"Invalid inclusive frame interval [{start},{annotation_end}] in " + f"{clip.transcript_path}; clip has {clip.num_frames} frames" + ) + dense_start = clip.dense_start_frame + start - 1 + dense_end = clip.dense_start_frame + end + uid = f"{session}/{clip.stem}/{row_index:04d}" + clip_rows.append( + Utterance( + uid=uid, + session=session, + clip_stem=clip.stem, + clip_index=clip_index, + segment_id=clip.segment_id, + participant_id=int(row["Participant_ID"]), + target_of_speech=_target_values(row["Target_of_Speech"]), + transcription=str(row["Transcription"]), + source_start_frame=start, + source_end_frame=end, + dense_start_frame=dense_start, + dense_end_frame_exclusive=dense_end, + dense_start_seconds=dense_start / SOURCE_FPS, + dense_end_seconds=dense_end / SOURCE_FPS, + duration_seconds=(end - start + 1) / SOURCE_FPS, + starts_at_clip_boundary=start <= 2, + ends_at_clip_boundary=end >= clip.num_frames - 1, + annotation_end_frame_clamped_by=end_overrun, + is_cross_clip_continuation=False, + ) + ) + per_clip.append(clip_rows) + + # Mark, rather than merge, cross-file continuations. Their timing remains + # useful for floor context, but they must not create a new event onset. + for clip_index, rows in enumerate(per_clip): + previous_rows = per_clip[clip_index - 1] if clip_index > 0 else [] + contiguous = ( + clip_index > 0 and clips[clip_index].segment_id == clips[clip_index - 1].segment_id + ) + for row in rows: + continuation = False + if contiguous and row.starts_at_clip_boundary: + candidates = [ + prev + for prev in previous_rows + if prev.participant_id == row.participant_id + and prev.ends_at_clip_boundary + ] + continuation = bool(candidates) and ( + "[X]" in row.transcription.upper() + or any("[X]" in prev.transcription.upper() for prev in candidates) + ) + utterances.append( + Utterance(**{**asdict(row), "is_cross_clip_continuation": continuation}) + ) + return sorted(utterances, key=lambda u: (u.dense_start_frame, u.dense_end_frame_exclusive, u.uid)) + + +def _overlap_seconds(row: Utterance, start: float, end: float) -> float: + return max(0.0, min(row.dense_end_seconds, end) - max(row.dense_start_seconds, start)) + + +def _coverage_by_participant( + utterances: Sequence[Utterance], + *, + segment_id: int, + start: float, + end: float, + exclude_participant: int | None = None, +) -> dict[int, float]: + intervals: dict[int, list[tuple[float, float]]] = defaultdict(list) + for row in utterances: + if row.segment_id != segment_id or row.participant_id == exclude_participant: + continue + left = max(start, row.dense_start_seconds) + right = min(end, row.dense_end_seconds) + if right > left: + intervals[row.participant_id].append((left, right)) + coverage: dict[int, float] = {} + for participant, spans in intervals.items(): + spans.sort() + total = 0.0 + left, right = spans[0] + for next_left, next_right in spans[1:]: + if next_left <= right: + right = max(right, next_right) + else: + total += right - left + left, right = next_left, next_right + coverage[participant] = total + right - left + return coverage + + +def _dominant(coverage: dict[int, float]) -> int | None: + if not coverage: + return None + participant, seconds = max(coverage.items(), key=lambda item: (item[1], -item[0])) + return participant if seconds > 1e-9 else None + + +def _previous_floor( + row: Utterance, + utterances: Sequence[Utterance], + rules: Rules, +) -> tuple[int | None, Utterance | None, float | None, float]: + candidates = [ + other + for other in utterances + if other.segment_id == row.segment_id + and other.participant_id != row.participant_id + and other.dense_start_seconds < row.dense_start_seconds + and other.dense_end_seconds >= row.dense_start_seconds - rules.max_switch_gap_seconds + ] + if not candidates: + return None, None, None, 0.0 + active = [other for other in candidates if other.dense_end_seconds > row.dense_start_seconds] + previous = max(active or candidates, key=lambda other: other.dense_end_seconds) + gap = row.dense_start_seconds - previous.dense_end_seconds + overlap = max(0.0, -gap) + return previous.participant_id, previous, gap, overlap + + +def classify_utterance( + row: Utterance, + utterances: Sequence[Utterance], + *, + camera_wearer_id: int, + rules: Rules, +) -> Event: + split = "test" if row.session in TEST_SESSIONS else "train" + role = "wearer" if row.participant_id == camera_wearer_id else "other" + lexical = is_lexical_feedback(row.transcription) + nonverbal = is_pure_nonverbal(row.transcription) + previous_pid, previous, gap, onset_overlap = _previous_floor(row, utterances, rules) + + pre_coverage = _coverage_by_participant( + utterances, + segment_id=row.segment_id, + start=row.dense_start_seconds - rules.context_seconds, + end=row.dense_start_seconds, + exclude_participant=row.participant_id, + ) + post_coverage = _coverage_by_participant( + utterances, + segment_id=row.segment_id, + start=row.dense_end_seconds, + end=row.dense_end_seconds + rules.context_seconds, + exclude_participant=row.participant_id, + ) + pre_pid = _dominant(pre_coverage) or previous_pid + post_pid = _dominant(post_coverage) + retained_pid = pre_pid if pre_pid is not None and pre_pid == post_pid else None + + old_end = previous.dense_end_seconds if previous is not None else row.dense_start_seconds + speaker_after_old = max(0.0, row.dense_end_seconds - max(row.dense_start_seconds, old_end)) + previous_cutoff = has_trailing_cutoff(previous) if previous is not None else False + target_matches = previous_pid is not None and previous_pid in row.target_of_speech + short = row.duration_seconds <= rules.backchannel_max_seconds + 1e-9 + overlaps = onset_overlap >= rules.min_overlap_seconds - 1e-9 + takeover_candidate = speaker_after_old >= rules.takeover_candidate_seconds - 1e-9 + takeover_high = speaker_after_old >= rules.takeover_high_seconds - 1e-9 + + event_type = "unclassified" + confidence = "none" + trainable = False + reason = "no other-speaker floor relation meeting a proxy rule" + + laughter_feedback = nonverbal and "[L]" in row.transcription.upper() + + if row.is_cross_clip_continuation: + event_type = "cross_clip_continuation" + reason = "[X] continuation at a contiguous one-minute file boundary" + elif laughter_feedback and (retained_pid is not None or overlaps): + event_type = "nonverbal_feedback" + confidence = "high" + trainable = True + reason = "laughter-only marker during/around another participant's retained floor" + elif nonverbal and (retained_pid is not None or overlaps): + event_type = "nonverbal_candidate" + confidence = "ambiguous" + reason = "pure hesitation/cough/throat-clear/unintelligible marker; feedback function is unknown" + elif short and retained_pid is not None: + if lexical: + event_type = "backchannel_proxy_high" + confidence = "high" + trainable = True + reason = "short lexical feedback; the same other participant dominates before and after" + else: + event_type = "backchannel_candidate" + confidence = "ambiguous" + reason = "short contribution while another participant retains floor, but no conservative lexical cue" + elif previous is not None and (overlaps or (previous_cutoff and gap is not None and gap <= 0.15)): + if previous_cutoff and takeover_candidate: + event_type = "interruption_proxy_high" + confidence = "high" + trainable = True + reason = "previous transcript is cut off and the incoming speaker continues after that floor ends" + elif takeover_high: + event_type = "overlap_takeover" + confidence = "timing" + trainable = True + reason = "overlapping onset followed by sustained takeover, without semantic interruption evidence" + elif takeover_candidate: + event_type = "interruption_candidate" + confidence = "ambiguous" + reason = "overlapping onset followed by a possible short takeover; interruption intent is unknown" + else: + event_type = "overlap_hold" + confidence = "timing" + trainable = True + reason = "overlap without enough evidence that the incoming speaker took the floor" + elif ( + previous is not None + and gap is not None + and 0.0 <= gap <= rules.max_switch_gap_seconds + and row.duration_seconds >= rules.takeover_high_seconds + ): + event_type = "ordinary_switch" + confidence = "timing" + trainable = True + reason = "other speaker ended recently and the incoming speaker sustained a new turn" + + return Event( + uid=row.uid, + session=row.session, + split=split, + event_type=event_type, + confidence=confidence, + trainable_auxiliary=trainable, + participant_id=row.participant_id, + participant_role=role, + previous_floor_participant_id=previous_pid, + retained_floor_participant_id=retained_pid, + target_matches_previous_floor=target_matches, + clip_stem=row.clip_stem, + source_start_frame=row.source_start_frame, + dense_start_seconds=row.dense_start_seconds, + task_step_index=row.dense_start_frame // FRAMES_PER_TASK_STEP, + duration_seconds=row.duration_seconds, + onset_overlap_seconds=onset_overlap, + gap_from_previous_seconds=gap, + speaker_after_previous_end_seconds=speaker_after_old, + previous_transcript_has_cutoff=previous_cutoff, + previous_transcription=None if previous is None else previous.transcription, + lexical_feedback=lexical, + pure_nonverbal=nonverbal, + transcription=row.transcription, + target_of_speech=row.target_of_speech, + reason=reason, + ) + + +def _serialize_event(event: Event) -> dict[str, Any]: + payload = asdict(event) + payload["target_of_speech"] = list(event.target_of_speech) + return payload + + +def _write_session( + output_dir: Path, + session: str, + events: Sequence[Event], + *, + target_steps: int | None, + write_frame_arrays: bool, + overwrite: bool, +) -> None: + output_dir.mkdir(parents=True, exist_ok=True) + rows = [_serialize_event(event) for event in events if event.event_type != "unclassified"] + jsonl_path = output_dir / f"{session}.events.jsonl" + csv_path = output_dir / f"{session}.events.csv" + for path in (jsonl_path, csv_path): + if path.exists() and not overwrite: + raise FileExistsError(f"Refusing to overwrite {path}; pass --overwrite") + with jsonl_path.open("w", encoding="utf-8") as handle: + for row in rows: + handle.write(json.dumps(row, ensure_ascii=False) + "\n") + fieldnames = list(rows[0]) if rows else [field.name for field in Event.__dataclass_fields__.values()] + with csv_path.open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=fieldnames) + writer.writeheader() + for row in rows: + row = dict(row) + row["target_of_speech"] = json.dumps(row["target_of_speech"], ensure_ascii=False) + writer.writerow(row) + + if write_frame_arrays: + if target_steps is None: + raise ValueError("--write-frame-arrays requires --target-root") + npz_path = output_dir / f"{session}.aux_targets.npz" + if npz_path.exists() and not overwrite: + raise FileExistsError(f"Refusing to overwrite {npz_path}; pass --overwrite") + try: + import numpy as np + except ImportError as exc: + raise RuntimeError("numpy is required to write aligned auxiliary arrays") from exc + onset = np.zeros((target_steps, len(AUXILIARY_EVENT_NAMES)), dtype=np.uint8) + role = np.zeros_like(onset, dtype=np.int8) + collisions = np.zeros_like(onset, dtype=np.uint8) + event_to_column = {name: index for index, name in enumerate(AUXILIARY_EVENT_NAMES)} + for event in events: + column = event_to_column.get(event.event_type) + step = event.task_step_index + if column is None or not event.trainable_auxiliary or not 0 <= step < target_steps: + continue + encoded_role = 1 if event.participant_role == "wearer" else 2 + if onset[step, column]: + collisions[step, column] = 1 + if role[step, column] != encoded_role: + role[step, column] = -1 + else: + onset[step, column] = 1 + role[step, column] = encoded_role + np.savez_compressed( + npz_path, + event_onsets=onset, + event_roles=role, + collisions=collisions, + event_names=np.asarray(AUXILIARY_EVENT_NAMES), + task_fps=np.asarray(TASK_FPS), + target_steps=np.asarray(target_steps), + ) + + +def _session_summary( + session: str, + clips: Sequence[Clip], + utterances: Sequence[Utterance], + events: Sequence[Event], + target_steps: int | None, +) -> dict[str, Any]: + counts = Counter(event.event_type for event in events) + trainable = Counter( + event.event_type for event in events if event.trainable_auxiliary + ) + dense_frames = sum(clip.num_frames for clip in clips) + expected_steps = dense_frames // FRAMES_PER_TASK_STEP + return { + "session": session, + "split": "test" if session in TEST_SESSIONS else "train", + "clips": len(clips), + "source_discontinuities": max((clip.segment_id for clip in clips), default=0), + "utterances": len(utterances), + "annotation_endpoints_clamped": sum( + row.annotation_end_frame_clamped_by > 0 for row in utterances + ), + "annotation_frames_clamped_total": sum( + row.annotation_end_frame_clamped_by for row in utterances + ), + "dense_frames": dense_frames, + "expected_5hz_steps": expected_steps, + "target_steps": target_steps, + "target_minus_expected_steps": ( + None if target_steps is None else target_steps - expected_steps + ), + "event_counts": dict(sorted(counts.items())), + "trainable_auxiliary_counts": dict(sorted(trainable.items())), + } + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--dataset-root", type=Path, required=True) + parser.add_argument( + "--target-root", + type=Path, + help="Optional EgoSpeak target_perframe directory for exact 5 Hz length/aligned .npz output", + ) + parser.add_argument("--output-dir", type=Path, help="Omit for a read-only audit") + parser.add_argument( + "--write-frame-arrays", + action="store_true", + help="Write aligned 5 Hz .npz arrays; requires --target-root and numpy", + ) + parser.add_argument("--sessions", nargs="+", default=list(ALL_SESSIONS)) + parser.add_argument("--camera-wearer-id", type=int, default=CAMERA_WEARER_ID) + parser.add_argument("--context-seconds", type=float, default=1.5) + parser.add_argument("--backchannel-max-seconds", type=float, default=1.0) + parser.add_argument("--max-switch-gap-seconds", type=float, default=0.6) + parser.add_argument("--min-overlap-seconds", type=float, default=0.1) + parser.add_argument("--takeover-candidate-seconds", type=float, default=0.4) + parser.add_argument("--takeover-high-seconds", type=float, default=0.8) + parser.add_argument("--max-contiguous-source-gap-seconds", type=float, default=1.0) + parser.add_argument("--examples-per-event", type=int, default=3) + parser.add_argument("--overwrite", action="store_true") + return parser.parse_args() + + +def main() -> None: + args = parse_args() + if SOURCE_FPS % TASK_FPS: + raise RuntimeError("source/task FPS ratio must be integral") + rules = Rules( + context_seconds=args.context_seconds, + backchannel_max_seconds=args.backchannel_max_seconds, + max_switch_gap_seconds=args.max_switch_gap_seconds, + min_overlap_seconds=args.min_overlap_seconds, + takeover_candidate_seconds=args.takeover_candidate_seconds, + takeover_high_seconds=args.takeover_high_seconds, + max_contiguous_source_gap_seconds=args.max_contiguous_source_gap_seconds, + ) + for name, value in asdict(rules).items(): + if value < 0: + raise ValueError(f"{name} must be non-negative") + unknown = sorted(set(args.sessions) - set(ALL_SESSIONS)) + if unknown: + raise ValueError(f"Unknown EasyCom sessions: {unknown}") + if args.write_frame_arrays and args.output_dir is None: + raise ValueError("--write-frame-arrays requires --output-dir") + if args.write_frame_arrays and args.target_root is None: + raise ValueError("--write-frame-arrays requires --target-root") + + all_events: list[Event] = [] + summaries: list[dict[str, Any]] = [] + for session in args.sessions: + clips = build_clips(args.dataset_root.resolve(), session, rules) + utterances = load_utterances(session, clips) + events = [ + classify_utterance( + row, + utterances, + camera_wearer_id=args.camera_wearer_id, + rules=rules, + ) + for row in utterances + ] + target_path = ( + None + if args.target_root is None + else args.target_root.resolve() / f"{session}.npy" + ) + if target_path is not None and not target_path.is_file(): + raise FileNotFoundError(f"Missing target: {target_path}") + target_steps = _target_length(target_path) + summary = _session_summary(session, clips, utterances, events, target_steps) + if summary["target_minus_expected_steps"] not in (None, 0): + raise ValueError( + f"{session} target/media mismatch: {summary['target_minus_expected_steps']:+d} steps" + ) + summaries.append(summary) + all_events.extend(events) + if args.output_dir is not None: + _write_session( + args.output_dir.resolve(), + session, + events, + target_steps=target_steps, + write_frame_arrays=args.write_frame_arrays, + overwrite=args.overwrite, + ) + counts = Counter(event.event_type for event in events) + print( + f"{session}: utterances={len(utterances)} " + + " ".join(f"{name}={count}" for name, count in sorted(counts.items())) + ) + + aggregate = Counter(event.event_type for event in all_events) + trainable = Counter( + event.event_type for event in all_events if event.trainable_auxiliary + ) + report = { + "schema_version": 1, + "dataset": "EasyCom", + "dataset_root": str(args.dataset_root.resolve()), + "target_root": None if args.target_root is None else str(args.target_root.resolve()), + "source_fps": SOURCE_FPS, + "task_fps": TASK_FPS, + "camera_wearer_participant_id": args.camera_wearer_id, + "rules": asdict(rules), + "auxiliary_event_names": list(AUXILIARY_EVENT_NAMES), + "ambiguous_event_names": list(AMBIGUOUS_EVENT_NAMES), + "semantic_warning": ( + "Backchannel and interruption are transcript/timing-derived weak proxies, " + "not released EasyCom ground-truth semantic annotations." + ), + "recommended_training_policy": ( + "Use only trainable_auxiliary events for auxiliary loss; mask candidate and " + "unclassified rows. Keep EgoSpeak's original three-label evaluator unchanged." + ), + "event_counts": dict(sorted(aggregate.items())), + "trainable_auxiliary_counts": dict(sorted(trainable.items())), + "sessions": summaries, + "examples": {}, + } + for event_type in sorted(aggregate): + report["examples"][event_type] = [ + _serialize_event(event) + for event in all_events + if event.event_type == event_type + ][: args.examples_per_event] + + print("\nAggregate event counts") + for name, count in sorted(aggregate.items()): + print(f" {name:<28} {count:>6}") + print("Trainable auxiliary counts") + for name, count in sorted(trainable.items()): + print(f" {name:<28} {count:>6}") + if args.examples_per_event: + print("Examples") + for event_type in sorted(aggregate): + examples = [ + event for event in all_events if event.event_type == event_type + ][: args.examples_per_event] + for event in examples: + text = " ".join(event.transcription.split()) + if len(text) > 100: + text = text[:97] + "..." + print( + f" {event_type:<28} {event.session} " + f"t={event.dense_start_seconds:.2f}s p={event.participant_id}: {text}" + ) + print( + "\nWARNING: backchannel/interruption outputs are weak proxies; " + "ambiguous candidates are intentionally masked." + ) + + if args.output_dir is not None: + output_dir = args.output_dir.resolve() + output_dir.mkdir(parents=True, exist_ok=True) + report_path = output_dir / "summary.json" + if report_path.exists() and not args.overwrite: + raise FileExistsError(f"Refusing to overwrite {report_path}; pass --overwrite") + report_path.write_text( + json.dumps(report, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + print(f"Saved reports to {output_dir}") + + +if __name__ == "__main__": + main() diff --git a/test_adjudicate_easycom_interaction_events.py b/test_adjudicate_easycom_interaction_events.py new file mode 100644 index 0000000..c273b94 --- /dev/null +++ b/test_adjudicate_easycom_interaction_events.py @@ -0,0 +1,101 @@ +"""Tests for DeepSeek interaction-event adjudication without network calls.""" + +from __future__ import annotations + +import unittest + +from adjudicate_easycom_interaction_events import ( + OUTPUT_SCHEMA_VERSION, + PROMPT_VERSION, + _validate_decisions, + _validate_resume_rows, + build_user_prompt, + stratified_limit, +) + + +class AdjudicationTests(unittest.TestCase): + def test_prompt_contains_json_and_candidate_context(self) -> None: + prompt = build_user_prompt( + [{"uid": "u1", "event_type": "backchannel_candidate", "transcription": "Yeah."}], + 0, + ) + self.assertIn("JSON", prompt) + self.assertIn("u1", prompt) + self.assertIn("Yeah.", prompt) + + def test_response_requires_exact_uid_set(self) -> None: + with self.assertRaisesRegex(ValueError, "mismatch"): + _validate_decisions({"items": []}, ["u1"]) + + def test_valid_response_is_returned_in_input_order(self) -> None: + payload = { + "items": [ + {"uid": "u2", "label": "uncertain", "confidence": "low", "rationale": "Not enough evidence."}, + {"uid": "u1", "label": "backchannel", "confidence": "high", "rationale": "Floor is retained."}, + ] + } + actual = _validate_decisions(payload, ["u1", "u2"]) + self.assertEqual([row["uid"] for row in actual], ["u1", "u2"]) + + def test_invalid_label_is_rejected(self) -> None: + payload = { + "items": [ + {"uid": "u1", "label": "guess", "confidence": "high", "rationale": "No."} + ] + } + with self.assertRaisesRegex(ValueError, "Invalid label"): + _validate_decisions(payload, ["u1"]) + + def test_resume_rejects_mixed_model(self) -> None: + rows = [ + { + "uid": "u1", + "schema_version": OUTPUT_SCHEMA_VERSION, + "model": "different-model", + "prompt_version": PROMPT_VERSION, + "passes": 2, + } + ] + with self.assertRaisesRegex(ValueError, "different adjudication"): + _validate_resume_rows(rows, model="deepseek-v4-flash", passes=2) + + def test_resume_accepts_matching_configuration(self) -> None: + rows = [ + { + "uid": "u1", + "schema_version": OUTPUT_SCHEMA_VERSION, + "model": "deepseek-v4-flash", + "prompt_version": PROMPT_VERSION, + "passes": 2, + } + ] + _validate_resume_rows(rows, model="deepseek-v4-flash", passes=2) + + def test_pilot_limit_is_stratified_and_deterministic(self) -> None: + rows = [ + { + "uid": f"{event_type}-{session}-{index}", + "event_type": event_type, + "session": session, + } + for event_type in ("backchannel_candidate", "interruption_candidate") + for session in ("Session_4", "Session_5") + for index in range(3) + ] + first = stratified_limit(rows, 4, seed=2027) + second = stratified_limit(rows, 4, seed=2027) + self.assertEqual(first, second) + self.assertEqual( + {(row["event_type"], row["session"]) for row in first}, + { + ("backchannel_candidate", "Session_4"), + ("backchannel_candidate", "Session_5"), + ("interruption_candidate", "Session_4"), + ("interruption_candidate", "Session_5"), + }, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/test_derive_easycom_interaction_events.py b/test_derive_easycom_interaction_events.py new file mode 100644 index 0000000..7351223 --- /dev/null +++ b/test_derive_easycom_interaction_events.py @@ -0,0 +1,108 @@ +"""Focused tests for transcript-derived EasyCom interaction proxies.""" + +from __future__ import annotations + +import unittest + +from derive_easycom_interaction_events import Rules, Utterance, classify_utterance + + +def utterance( + uid: str, + participant: int, + start: float, + end: float, + text: str, + *, + target: tuple[int | str, ...] = ("Group",), +) -> Utterance: + return Utterance( + uid=uid, + session="Session_7", + clip_stem="00-00-000", + clip_index=0, + segment_id=0, + participant_id=participant, + target_of_speech=target, + transcription=text, + source_start_frame=round(start * 20) + 1, + source_end_frame=round(end * 20), + dense_start_frame=round(start * 20), + dense_end_frame_exclusive=round(end * 20), + dense_start_seconds=start, + dense_end_seconds=end, + duration_seconds=end - start, + starts_at_clip_boundary=False, + ends_at_clip_boundary=False, + annotation_end_frame_clamped_by=0, + is_cross_clip_continuation=False, + ) + + +class InteractionProxyTests(unittest.TestCase): + def setUp(self) -> None: + self.rules = Rules() + + def classify(self, target: Utterance, rows: list[Utterance]): + return classify_utterance( + target, + rows, + camera_wearer_id=2, + rules=self.rules, + ) + + def test_short_feedback_inside_retained_floor_is_backchannel_proxy(self) -> None: + floor = utterance("floor", 4, 0.0, 4.0, "Long explanation") + feedback = utterance("feedback", 2, 1.0, 1.4, "Yeah.", target=(4,)) + event = self.classify(feedback, [floor, feedback]) + self.assertEqual(event.event_type, "backchannel_proxy_high") + self.assertTrue(event.trainable_auxiliary) + self.assertEqual(event.retained_floor_participant_id, 4) + + def test_short_answer_without_floor_return_is_not_backchannel(self) -> None: + question = utterance("question", 4, 0.0, 1.0, "Ready?") + answer = utterance("answer", 2, 1.1, 2.3, "Yeah, let us start.", target=(4,)) + event = self.classify(answer, [question, answer]) + self.assertEqual(event.event_type, "ordinary_switch") + + def test_overlap_takeover_after_cutoff_is_interruption_proxy(self) -> None: + old = utterance("old", 4, 0.0, 2.0, "I was going to-") + incoming = utterance("incoming", 2, 1.7, 3.2, "Wait, that is wrong.", target=(4,)) + event = self.classify(incoming, [old, incoming]) + self.assertEqual(event.event_type, "interruption_proxy_high") + self.assertTrue(event.trainable_auxiliary) + + def test_overlap_takeover_without_cutoff_is_not_called_interruption(self) -> None: + old = utterance("old", 4, 0.0, 2.0, "I finished my point.") + incoming = utterance("incoming", 2, 1.7, 3.2, "Here is my response.", target=(4,)) + event = self.classify(incoming, [old, incoming]) + self.assertEqual(event.event_type, "overlap_takeover") + self.assertTrue(event.trainable_auxiliary) + + def test_short_nonlexical_floor_return_is_masked_candidate(self) -> None: + before = utterance("before", 4, 0.0, 1.2, "First part") + brief = utterance("brief", 2, 1.25, 1.8, "Possibly.", target=(4,)) + after = utterance("after", 4, 1.9, 3.0, "Second part") + event = self.classify(brief, [before, brief, after]) + self.assertEqual(event.event_type, "backchannel_candidate") + self.assertFalse(event.trainable_auxiliary) + + def test_hesitation_marker_is_not_nonverbal_feedback(self) -> None: + floor = utterance("floor", 4, 0.0, 4.0, "Long explanation") + hesitation = utterance("hesitation", 2, 1.0, 1.2, "[H]", target=(4,)) + event = self.classify(hesitation, [floor, hesitation]) + self.assertEqual(event.event_type, "nonverbal_candidate") + self.assertFalse(event.trainable_auxiliary) + + def test_cross_clip_continuation_never_creates_switch(self) -> None: + row = utterance("continued", 2, 60.0, 61.0, "[X] continued") + row = Utterance( + **{**row.__dict__, "is_cross_clip_continuation": True} + ) + event = self.classify(row, [row]) + self.assertEqual(event.event_type, "cross_clip_continuation") + self.assertFalse(event.trainable_auxiliary) + + +if __name__ == "__main__": + unittest.main()