875 lines
33 KiB
Python
875 lines
33 KiB
Python
#!/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(
|
|
"<H" if length_size == 2 else "<I",
|
|
handle.read(length_size),
|
|
)[0]
|
|
header = ast.literal_eval(handle.read(header_length).decode("latin1").strip())
|
|
shape = header.get("shape")
|
|
if not isinstance(shape, tuple) or len(shape) != 2 or shape[1] != 3:
|
|
raise ValueError(f"Expected target [T,3], got {shape}: {path}")
|
|
return int(shape[0])
|
|
|
|
|
|
def build_clips(dataset_root: Path, session: str, rules: Rules) -> 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()
|