diff --git a/PROJECT_CONTEXT.md b/PROJECT_CONTEXT.md index 199072b..1107d0d 100644 --- a/PROJECT_CONTEXT.md +++ b/PROJECT_CONTEXT.md @@ -304,7 +304,7 @@ Do not reorganize them physically without compatibility wrappers. | `derive_easycom_interaction_events.py` | Deterministic transcript/timing proxies | | `adjudicate_easycom_interaction_events.py` | Optional ambiguous-event adjudication | | `audit_easycom_floor_events.py` | Count EOT/takeover/backchannel candidates | -| `prepare_easycom_eot_int_review.py` | Build manuscript-aligned EOT-response/INT-yield positive and hard-negative review regions | +| `prepare_easycom_eot_int_review.py` | Build wearer-transition and independent hard-negative candidates with structured timing evidence; 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_easycom_oracle_features.py` | Build explicit past-only oracle person-activity sidecars and analysis-only addressee controls | diff --git a/adjudicate_easycom_eot_int_candidates.py b/adjudicate_easycom_eot_int_candidates.py index a2f0460..5477d0f 100644 --- a/adjudicate_easycom_eot_int_candidates.py +++ b/adjudicate_easycom_eot_int_candidates.py @@ -1,10 +1,10 @@ #!/usr/bin/env python3 -"""Add conservative DeepSeek suggestions to EOT/INT review candidates. +"""Add conservative DeepSeek pre-labels to wearer-centric EOT/INT candidates. -Only transcript excerpts and timing metadata are sent. Suggestions are never -treated as gold labels: the output remains input to ``review_floor_events.py`` -and requires an identified human reviewer. The API key is read only from the -``DEEPSEEK_API_KEY`` environment variable and is never persisted. +The model receives the candidate source, structured participant timing and the +transcript excerpt. It cannot change any source timestamp. Its output is +stored as an auditable suggestion beside the untouched rule candidate and must +still be accepted, relabeled, excluded, or sent to second review by a human. """ from __future__ import annotations @@ -16,49 +16,65 @@ import json import os from pathlib import Path import random +import re import time from typing import Any, Sequence import urllib.error import urllib.request -PROMPT_VERSION = "easycom-manuscript-eot-int-adjudicator-v1" +PROMPT_VERSION = "easycom-wearer-centric-eot-int-prelabel-v2" DEFAULT_MODEL = "deepseek-v4-flash" DEFAULT_BASE_URL = "https://api.deepseek.com" -DISPOSITIONS = {"positive", "hard_negative", "exclude", "uncertain"} +VERDICTS = {"POSITIVE", "HARD_NEGATIVE", "NOT_APPLICABLE", "EXCLUDE", "UNSURE"} +EVENT_TYPES = {"EOT_RESPONSE", "INT_YIELD"} HARD_NEGATIVE_TYPES = { - "eot_midturn_pause", - "eot_other_to_other", - "eot_no_response", - "int_backchannel", - "int_non_takeover_overlap", - "int_retain_floor", - "other", - "not_applicable", - "uncertain", + "EOT_MIDTURN_PAUSE", + "EOT_OTHER_TO_OTHER", + "EOT_WEARER_BACKCHANNEL", + "INT_OTHER_BACKCHANNEL", + "INT_NON_TAKEOVER_OVERLAP", + "INT_WEARER_HOLDS_FLOOR", + "OTHER_HARD_NEGATIVE", } -CONFIDENCES = {"high", "medium", "low"} -SYSTEM_PROMPT = """You audit candidate regions for a scientific study of wearer-relative floor control. -Use only the supplied participant-resolved transcript and timing evidence. Do not infer prosody, gaze, intention, or acoustic interruption cues that are not supplied. +SYSTEM_PROMPT = """You conservatively pre-label candidate regions for a scientific study of wearer-relative conversational floor control. -Tasks: -- eot_response positive: a non-wearer releases the floor and the wearer makes a genuine floor-taking response. -- int_yield positive: a non-wearer takes over during wearer speech and the wearer genuinely yields the floor. +Evidence rules: +1. Use candidate source, structured participant timing, per-speaker activity, and transcript together. Timing evidence has priority over a speculative transcript interpretation. +2. Never alter or propose an event timestamp. Do not infer prosody, gaze, intention, or acoustic cues that were not supplied. +3. A wearer onset is not automatically a floor take; a wearer offset is not automatically a yield; an other onset is not automatically a takeover. A pause is not automatically EOT, overlap is not automatically interruption, and words such as "yeah" are not automatically backchannels. Check who speaks sustainably afterward and who resumes. +4. Use UNSURE when the evidence does not establish conversational function. Never force a label for coverage. -Hard negatives: -- eot_midturn_pause: the same non-wearer resumes after a within-turn pause. -- eot_other_to_other: a different non-wearer takes the floor while the wearer remains silent. -- eot_no_response: the wearer correctly remains silent for the labeled region. -- int_backchannel: brief feedback during wearer speech that does not take the floor. -- int_non_takeover_overlap: overlapping other speech without takeover. -- int_retain_floor: the wearer correctly retains the floor for the labeled region. +Top-level verdicts: +- POSITIVE: the system should commit. +- HARD_NEGATIVE: a valid, interpretable region where the system should not commit. +- NOT_APPLICABLE: a real speech transition outside the committed EOT/INT task, such as a wearer starting after long silence or naturally ending into silence. +- EXCLUDE: the region cannot be scored reliably because of annotation corruption, a recording boundary, missing context, a non-speech artifact, timestamp corruption, or another unresolvable defect. +- UNSURE: evidence is insufficient and a second human review is needed. -Use exclude for annotation corruption or an unusable boundary. Use uncertain whenever transcript/timing evidence cannot establish the semantic function. Timing alone does not prove EOT, interruption, backchannel, or takeover. +POSITIVE event_type must be exactly EOT_RESPONSE or INT_YIELD. EOT_RESPONSE means a non-wearer genuinely releases the floor and the wearer then takes it with a sustained response. INT_YIELD means a non-wearer genuinely takes over during wearer speech and the wearer relinquishes the floor. + +HARD_NEGATIVE subtype must be exactly one of EOT_MIDTURN_PAUSE, EOT_OTHER_TO_OTHER, EOT_WEARER_BACKCHANNEL, INT_OTHER_BACKCHANNEL, INT_NON_TAKEOVER_OVERLAP, INT_WEARER_HOLDS_FLOOR, or OTHER_HARD_NEGATIVE. Return exactly one JSON object with this shape and no markdown: -{"items":[{"uid":"...","protocol_disposition":"uncertain","hard_negative_type":"uncertain","confidence":"low","rationale":"brief evidence-based sentence"}]} -Return every supplied uid exactly once.""" +{"items":[{"uid":"...","verdict":"UNSURE","event_type":null,"subtype":null,"confidence":0.25,"reason":"brief evidence-based sentence","needs_human_review":true}]} +Return every supplied uid exactly once. Copy uid byte-for-byte and never add a backslash before an underscore. confidence must be a number from 0 to 1.""" + + +def _normalize_returned_uid(value: Any) -> Any: + """Undo DeepSeek's occasional Markdown escaping without weakening identity checks.""" + if not isinstance(value, str): + return value + return re.sub(r"\\+_", "_", value) + + +def _participant_uid_suffix(row: dict[str, Any], ordinal: int) -> str: + previous = row.get("previous_other_participant_ids") or [] + incoming = row.get("incoming_other_participant_ids") or [] + previous_key = "-".join(f"p{x}" for x in previous) or "none" + incoming_key = "-".join(f"p{x}" for x in incoming) or "none" + return f"prev-{previous_key}__in-{incoming_key}__row-{ordinal}" def read_candidates(path: Path) -> list[dict[str, Any]]: @@ -70,33 +86,50 @@ def read_candidates(path: Path) -> list[dict[str, Any]]: row = json.loads(line) if not isinstance(row, dict) or not isinstance(row.get("uid"), str): raise ValueError(f"Expected candidate object with uid at {path}:{line_number}") - if row.get("task") not in {"eot_response", "int_yield"}: - raise ValueError(f"Unsupported task at {path}:{line_number}: {row.get('task')!r}") + if not (row.get("candidate_source") or row.get("task")): + raise ValueError(f"Missing candidate source/task at {path}:{line_number}") rows.append(row) if not rows: raise ValueError(f"No candidates in {path}") + by_uid: dict[str, list[int]] = {} + for index, row in enumerate(rows): + by_uid.setdefault(row["uid"], []).append(index) + for original_uid, indices in by_uid.items(): + if len(indices) == 1: + continue + for ordinal, index in enumerate(indices, 1): + row = dict(rows[index]) + row["uid_disambiguated_from"] = original_uid + row["uid"] = f"{original_uid}/{_participant_uid_suffix(row, ordinal)}" + row["event_id"] = row["uid"] + rows[index] = row + if len({row["uid"] for row in rows}) != len(rows): + raise ValueError(f"Could not deterministically disambiguate duplicate candidate UIDs in {path}") return rows def candidate_for_prompt(row: dict[str, Any]) -> dict[str, Any]: fields = ( "uid", + "candidate_source", + "rule_candidate_type", "task", - "automatic_protocol_label", - "automatic_hard_negative_type", "anchor_type", "anchor_dense_seconds", - "behavior_anchor_type", - "behavior_seconds", + "event_time", "region_start_seconds", "region_end_seconds", - "region_end_reason", - "previous_other_participant_ids", - "incoming_other_participant_ids", - "automatic_reason", + "temporal_evidence", + "speaker_timeline", "context", ) - return {field: row.get(field) for field in fields} + payload = {field: row.get(field) for field in fields if row.get(field) is not None} + # Older v1 files remain reviewable and can still be pre-labeled. + if "candidate_source" not in payload: + payload["candidate_source"] = row.get("anchor_type", row.get("event_type", "UNKNOWN")) + if "event_time" not in payload: + payload["event_time"] = row.get("anchor_dense_seconds") + return payload def build_user_prompt(rows: Sequence[dict[str, Any]], pass_index: int) -> str: @@ -112,35 +145,59 @@ def build_user_prompt(rows: Sequence[dict[str, Any]], pass_index: int) -> str: ) -def validate_decisions(payload: Any, expected_uids: Sequence[str]) -> list[dict[str, str]]: +def _transport_rows(rows: Sequence[dict[str, Any]]) -> tuple[list[dict[str, Any]], dict[str, str]]: + """Use simple batch-local IDs so the model never has to copy path-like UIDs.""" + transported: list[dict[str, Any]] = [] + original_by_transport: dict[str, str] = {} + for index, row in enumerate(rows): + transport_uid = f"item-{index:04d}" + original_by_transport[transport_uid] = row["uid"] + transported.append({**row, "uid": transport_uid}) + return transported, original_by_transport + + +def validate_decisions(payload: Any, expected_uids: Sequence[str]) -> list[dict[str, Any]]: if not isinstance(payload, dict) or not isinstance(payload.get("items"), list): raise ValueError("Response must be an object with an items list") expected = set(expected_uids) - decisions: dict[str, dict[str, str]] = {} + decisions: dict[str, dict[str, Any]] = {} for item in payload["items"]: if not isinstance(item, dict): raise ValueError("Every decision must be an object") - uid = item.get("uid") - disposition = item.get("protocol_disposition") - negative_type = item.get("hard_negative_type") + raw_uid = item.get("uid") + uid = _normalize_returned_uid(raw_uid) + verdict = item.get("verdict") + event_type = item.get("event_type") + subtype = item.get("subtype") confidence = item.get("confidence") - rationale = item.get("rationale") + reason = item.get("reason") + needs_human_review = item.get("needs_human_review") if uid not in expected or uid in decisions: - raise ValueError(f"Unexpected or duplicate uid: {uid!r}") - if disposition not in DISPOSITIONS: - raise ValueError(f"Invalid protocol_disposition for {uid}: {disposition!r}") - if negative_type not in HARD_NEGATIVE_TYPES: - raise ValueError(f"Invalid hard_negative_type for {uid}: {negative_type!r}") - if confidence not in CONFIDENCES: - 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}") + raise ValueError(f"Unexpected or duplicate uid: {raw_uid!r}") + if verdict not in VERDICTS: + raise ValueError(f"Invalid verdict for {uid}: {verdict!r}") + if verdict == "POSITIVE" and event_type not in EVENT_TYPES: + raise ValueError(f"POSITIVE requires EOT_RESPONSE or INT_YIELD for {uid}") + if verdict != "POSITIVE" and event_type is not None: + raise ValueError(f"Only POSITIVE may have event_type for {uid}") + if verdict == "HARD_NEGATIVE" and subtype not in HARD_NEGATIVE_TYPES: + raise ValueError(f"HARD_NEGATIVE requires a valid subtype for {uid}") + if verdict != "HARD_NEGATIVE" and subtype is not None: + raise ValueError(f"Only HARD_NEGATIVE may have subtype for {uid}") + if isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1: + raise ValueError(f"confidence must be in [0, 1] for {uid}") + if not isinstance(reason, str) or not reason.strip(): + raise ValueError(f"Missing reason for {uid}") + if not isinstance(needs_human_review, bool): + raise ValueError(f"needs_human_review must be boolean for {uid}") decisions[str(uid)] = { "uid": str(uid), - "protocol_disposition": str(disposition), - "hard_negative_type": str(negative_type), - "confidence": str(confidence), - "rationale": rationale.strip(), + "verdict": str(verdict), + "event_type": event_type, + "subtype": subtype, + "confidence": float(confidence), + "reason": reason.strip(), + "needs_human_review": needs_human_review, } if set(decisions) != expected: raise ValueError(f"Response uid mismatch; missing={sorted(expected - set(decisions))}") @@ -156,12 +213,13 @@ def _post( base_url: str, timeout_seconds: float, max_retries: int, -) -> tuple[list[dict[str, str]], dict[str, Any]]: +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + transported, original_by_transport = _transport_rows(rows) body = { "model": model, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, - {"role": "user", "content": build_user_prompt(rows, pass_index)}, + {"role": "user", "content": build_user_prompt(transported, pass_index)}, ], "response_format": {"type": "json_object"}, "thinking": {"type": "disabled"}, @@ -182,7 +240,10 @@ def _post( envelope = json.loads(response.read().decode("utf-8")) content = envelope["choices"][0]["message"]["content"] parsed = json.loads(content) - return validate_decisions(parsed, [row["uid"] for row in rows]), envelope.get("usage", {}) + decisions = validate_decisions(parsed, list(original_by_transport)) + for decision in decisions: + decision["uid"] = original_by_transport[decision["uid"]] + return decisions, envelope.get("usage", {}) except ( urllib.error.HTTPError, urllib.error.URLError, @@ -206,67 +267,80 @@ def add_suggestions( model: str, base_url: str, passes: int, + batch_size: int = 8, timeout_seconds: float, max_retries: int, ) -> tuple[list[dict[str, Any]], Counter[str]]: - by_uid: dict[str, list[dict[str, str]]] = {row["uid"]: [] for row in rows} + by_uid: dict[str, list[dict[str, Any]]] = {row["uid"]: [] for row in rows} + prompt_hashes: dict[str, list[str]] = {row["uid"]: [] for row in rows} usage: Counter[str] = Counter() + if batch_size <= 0: + raise ValueError("batch_size must be positive") for pass_index in range(passes): - decisions, pass_usage = _post( - rows=rows, - pass_index=pass_index, - api_key=api_key, - model=model, - base_url=base_url, - timeout_seconds=timeout_seconds, - max_retries=max_retries, - ) - for decision in decisions: - by_uid[decision["uid"]].append(decision) - for key, value in pass_usage.items(): - if isinstance(value, int): - usage[key] += value + for start in range(0, len(rows), batch_size): + batch = rows[start:start + batch_size] + transported, _ = _transport_rows(batch) + prompt_hash = hashlib.sha256( + (SYSTEM_PROMPT + "\n" + build_user_prompt(transported, pass_index)).encode("utf-8") + ).hexdigest() + decisions, pass_usage = _post( + rows=batch, + pass_index=pass_index, + api_key=api_key, + model=model, + base_url=base_url, + timeout_seconds=timeout_seconds, + max_retries=max_retries, + ) + for decision in decisions: + by_uid[decision["uid"]].append(decision) + prompt_hashes[decision["uid"]].append(prompt_hash) + for key, value in pass_usage.items(): + if isinstance(value, int): + usage[key] += value output: list[dict[str, Any]] = [] for source in rows: decisions = by_uid[source["uid"]] - labels = { - (decision["protocol_disposition"], decision["hard_negative_type"]) - for decision in decisions - } + labels = {(d["verdict"], d["event_type"], d["subtype"]) for d in decisions} unanimous = len(labels) == 1 if unanimous: - disposition, negative_type = next(iter(labels)) - confidence = ( - "high" if all(item["confidence"] == "high" for item in decisions) else "medium" - ) - rationale = " | ".join(item["rationale"] for item in decisions) + verdict, event_type, subtype = next(iter(labels)) + confidence = min(float(item["confidence"]) for item in decisions) + reason = " | ".join(item["reason"] for item in decisions) + needs_human_review = any(item["needs_human_review"] for item in decisions) else: - disposition = "uncertain" - negative_type = "uncertain" - confidence = "low" - rationale = "The independent DeepSeek passes disagreed." - prompt_payload = { - "prompt_version": PROMPT_VERSION, - "model": model, - "passes": passes, - "candidate": candidate_for_prompt(source), - } + verdict, event_type, subtype = "UNSURE", None, None + confidence = min(float(item["confidence"]) for item in decisions) + reason = "The independent DeepSeek passes disagreed." + needs_human_review = True enriched = dict(source) enriched["llm_suggestion"] = { - "protocol_disposition": disposition, - "hard_negative_type": negative_type, + "verdict": verdict, + "event_type": event_type, + "subtype": subtype, "confidence": confidence, - "rationale": rationale, + "reason": reason, + "needs_human_review": needs_human_review, "unanimous": unanimous, } + enriched.update( + ai_verdict=verdict, + ai_event_type=event_type, + ai_subtype=subtype, + ai_confidence=confidence, + ai_reason=reason, + human_status="UNREVIEWED", + human_verdict=None, + human_event_type=None, + human_subtype=None, + exclude_reason=None, + ) enriched["llm_adjudication"] = { "model": model, "prompt_version": PROMPT_VERSION, "passes": passes, - "prompt_sha256": hashlib.sha256( - json.dumps(prompt_payload, sort_keys=True, ensure_ascii=False).encode("utf-8") - ).hexdigest(), + "request_prompt_sha256": prompt_hashes[source["uid"]], "decisions": decisions, "human_review_required": True, } @@ -281,6 +355,7 @@ def parse_args() -> argparse.Namespace: 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("--passes", type=int, default=2) + parser.add_argument("--batch-size", type=int, default=8) 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) @@ -293,8 +368,8 @@ def parse_args() -> argparse.Namespace: def main() -> None: args = parse_args() - if args.passes <= 0 or args.max_retries < 0: - raise ValueError("passes must be positive and max-retries non-negative") + if args.passes <= 0 or args.batch_size <= 0 or args.max_retries < 0: + raise ValueError("passes/batch-size must be positive and max-retries non-negative") rows = read_candidates(args.input.expanduser().resolve()) if args.max_events is not None: if args.max_events < 0: @@ -324,6 +399,7 @@ def main() -> None: model=args.model, base_url=args.base_url, passes=args.passes, + batch_size=args.batch_size, timeout_seconds=args.timeout_seconds, max_retries=args.max_retries, ) diff --git a/docs/easycom_event_review_tool.md b/docs/easycom_event_review_tool.md index d56cbe8..49a1bd5 100644 --- a/docs/easycom_event_review_tool.md +++ b/docs/easycom_event_review_tool.md @@ -1,16 +1,19 @@ -# EasyCom EOT/INT review workspace +# EasyCom wearer-centric EOT/INT pre-label and review pipeline -`review_floor_events.py` is a local annotation workspace for reviewing the -manuscript-aligned regions produced by `prepare_easycom_eot_int_review.py`. -It can also open the older timing/transcript proxy reports from -`derive_easycom_interaction_events.py` and `audit_easycom_floor_events.py`, but -those older reports do not have the directionality or hard-negative region -contract required by the current manuscript. Raw EasyCom media stays read-only; -human decisions are stored separately as JSONL and CSV. +The workflow has three explicit stages. Speech activity timestamps are +read-only throughout; reviewed events never modify the dense 5 Hz activity +ground truth. -## Prepare manuscript-aligned candidates +```text +EasyCom participant speech intervals + -> rule candidates + structured timing evidence + -> DeepSeek semantic pre-label (optional but recommended) + -> human Accept / Relabel / Exclude / Second review +``` -Develop the annotation guide on training-side Sessions 4--12: +## 1. Generate candidates + +Develop the rules on training-side Sessions 4--12: ```bash python3 prepare_easycom_eot_int_review.py \ @@ -19,24 +22,39 @@ python3 prepare_easycom_eot_int_review.py \ --output-dir reports/easycom_floor_event_candidates ``` -For a one-event pilot, add `--max-candidates 1`. The generator follows the -manuscript direction exactly: +The generator emits two independent sources: -- EOT-response begins at an aggregate-other offset while the wearer is silent. -- INT-yield begins at an aggregate-other onset while the wearer is active. -- An observed wearer transition before the competing other transition is a - timing-positive candidate. -- A competing other transition, or a fully observed no-transition window, is a - timing hard-negative candidate. -- A source boundary without a complete outcome/window is marked for exclusion. +- every wearer `0->1` (`WEARER_ONSET`, `ENTRY_CANDIDATE`) and `1->0` + (`WEARER_OFFSET`, `YIELD_CANDIDATE`) transition; +- hard-negative regions that need no wearer transition: + `SAME_SPEAKER_PAUSE`, `OTHER_TO_OTHER`, `OTHER_BACKCHANNEL`, and + `NON_TAKEOVER_OVERLAP`. -Automatic labels remain candidates until independent human semantic review. +Wearer transitions are initially `unsure`; timing alone is not a semantic +positive label. Hard-negative scans are rule hints and still require AI and +human review. The JSONL retains candidate source, original event time, +participant-level activity, raw durations/gaps/overlap evidence, transcript, +and `Target_of_Speech` when present. `Target_of_Speech` is annotation assistance +only and cannot be a deployable model input. -## Optional DeepSeek suggestion +All heuristic durations are configurable rather than frozen paper rules: -DeepSeek suggestions are triage metadata and never count as a human label. The -script sends only the candidate transcript excerpt and timing fields. Inspect -the exact request without network access first: +```text +--relation-window-seconds +--same-speaker-pause-max-seconds +--other-handover-max-seconds +--short-vocalization-max-seconds +--continuation-min-seconds +``` + +Use `--max-candidates 1` for a one-event smoke test. `summary.json` records the +exact configuration and candidate counts. `easycom_session_timelines.json` +stores the read-only participant speech lanes used by the full-session web +timeline; it is not sent to DeepSeek. + +## 2. Add DeepSeek pre-labels + +First inspect exactly what would be uploaded: ```bash python3 adjudicate_easycom_eot_int_candidates.py \ @@ -47,51 +65,52 @@ python3 adjudicate_easycom_eot_int_candidates.py \ --dry-run ``` -After setting `DEEPSEEK_API_KEY`, make the authorized request: +After setting `DEEPSEEK_API_KEY` and obtaining authorization to upload the +candidate excerpts: ```bash python3 adjudicate_easycom_eot_int_candidates.py \ --input reports/easycom_floor_event_candidates/easycom_eot_int_candidates.jsonl \ --output reports/easycom_floor_event_candidates/easycom_eot_int_deepseek.jsonl \ --model deepseek-v4-flash \ - --max-events 1 \ --allow-external-transcript-upload ``` -Point the review server at the DeepSeek-enriched JSONL to show its suggestion -beside the video. Keep suggestions hidden from an initial independent human -pass if anchoring bias is a concern. +DeepSeek sees candidate source, structured timing, per-speaker activity, and +transcript together. It returns one of `POSITIVE`, `HARD_NEGATIVE`, +`NOT_APPLICABLE`, `EXCLUDE`, or `UNSURE`; positives additionally receive +`EOT_RESPONSE` or `INT_YIELD`, and hard negatives receive a named subtype. It +cannot modify timestamps. The output preserves both the original rule candidate +and the complete AI decision/provenance. -## Start the local workspace +## 3. Human review + +Run the web workspace on the AI-enriched JSONL: ```bash python3 review_floor_events.py \ --dataset-root /path/to/easycom \ - --events reports/easycom_floor_event_candidates/easycom_eot_int_candidates.jsonl \ + --events reports/easycom_floor_event_candidates/easycom_eot_int_deepseek.jsonl \ --output reports/easycom_event_reviews/reviewer_1.jsonl ``` -Open `http://127.0.0.1:8765`. The page resolves the source MP4 and transcript, -seeks to a six-second event context, and supports session/split/task/status -filtering. +Open `http://127.0.0.1:8765`. The page shows a session candidate overview, +participant activity near the event, video/audio, transcript, structured timing, +and the AI verdict/confidence/reason. The common path is one operation: -Each save atomically rewrites both JSONL and CSV. The review explicitly records -`positive`, `hard_negative`, `exclude`, or `uncertain`; the hard-negative -subtype; boundary validity; onset function; conversational routing; offset -outcome; three 200 ms anchor/region corrections; confidence; and reviewer ID. +- **Accept** keeps the AI verdict/type; +- **Relabel** expands verdict and subtype controls; +- **Exclude** requires a concrete data-quality reason; +- **Unsure / Second review** preserves the event for adjudication. + +Boundary correction is hidden under an optional disclosure and is limited to +one 200 ms bin. The original timestamp is always retained. Each save atomically +rewrites JSONL and CSV while preserving the rule candidate, AI decision, human +decision, relabel flag, exclusion state, reviewer, and provenance. ## Freeze discipline -Develop the annotation guide and resolve category disagreements on training -Sessions 4--12. Freeze the guide before reviewing official test Sessions 1--3. -Do not use test reviews to change candidate windows, label categories, or the -R=.8 threshold calibration rule. - -For paper-quality labels, use two independent reviewers with separate -`--output` files and adjudicate disagreements in a third output. A non-empty -reviewer ID is required on every saved row. DeepSeek is not one of the two -reviewers. - -This workspace produces audited labels. It does not yet convert them into the -final committed-event matching metrics; that integration follows only after -the schema and real-data pilot are frozen. +Develop the guide and resolve disagreements only on Sessions 4--12. Freeze +categories and thresholds before reviewing official test Sessions 1--3. Use two +independent human review files plus a separate adjudication output for paper- +quality labels. DeepSeek is pre-label metadata, not an independent reviewer. diff --git a/floor_event_review_web/app.css b/floor_event_review_web/app.css index 90487cb..bd6046d 100644 --- a/floor_event_review_web/app.css +++ b/floor_event_review_web/app.css @@ -220,6 +220,27 @@ textarea { resize: vertical; line-height: 1.45; } .llm-suggestion strong { display: block; color: var(--accent); margin-bottom: 3px; } .event-position { color: var(--muted); font-size: 0.8rem; } +.timeline-section { padding: 0 18px 16px; display: grid; gap: 10px; } +.timeline-section .section-heading { margin-top: 2px; } +.local-heading { margin-top: 8px !important; } +.session-timeline { min-height: 34px; border: 1px solid var(--line); border-radius: 8px; position: relative; background: #07101d; padding: 6px 7px 28px; display: grid; gap: 3px; overflow: hidden; } +.session-lane { display: grid; grid-template-columns: 28px 1fr; gap: 5px; align-items: center; min-height: 10px; } +.session-lane > strong { color: var(--muted); text-align: right; font-size: .58rem; } +.session-track { height: 8px; position: relative; background: rgba(255,255,255,.025); } +.session-speech { position: absolute; top: 1px; height: 6px; min-width: 1px; background: #60a5fa; opacity: .75; } +.session-speech.wearer { background: var(--accent); } +.session-events { position: absolute; left: 40px; right: 7px; bottom: 4px; height: 20px; border-top: 1px solid var(--line-strong); } +.session-marker { position: absolute; top: 3px; width: 7px; height: 15px; margin-left: -3px; padding: 0; border: 0; border-radius: 4px; background: var(--amber); cursor: pointer; opacity: .68; } +.session-marker.int { background: #a78bfa; } +.session-marker.current { opacity: 1; box-shadow: 0 0 0 2px white; z-index: 2; } +.speaker-timeline { display: grid; gap: 6px; color: var(--muted); font-size: .76rem; } +.speaker-lane { display: grid; grid-template-columns: 34px 1fr; align-items: center; gap: 8px; } +.speaker-lane > strong { color: var(--muted); text-align: right; font-size: .72rem; } +.speaker-track { height: 22px; position: relative; border: 1px solid var(--line); border-radius: 5px; overflow: hidden; background: #07101d; } +.speaker-track > i { position: absolute; top: 0; bottom: 0; width: 1px; background: var(--amber); z-index: 3; } +.speech-segment { position: absolute; top: 5px; height: 10px; min-width: 2px; border-radius: 3px; background: #60a5fa; } +.speech-segment.wearer { background: var(--accent); } + .video-shell { background: #02050a; aspect-ratio: 16 / 9; position: relative; border-block: 1px solid var(--line); } .video-shell video { width: 100%; height: 100%; display: block; object-fit: contain; } .empty-state { position: absolute; inset: 0; display: grid; place-content: center; text-align: center; gap: 7px; color: var(--muted); } @@ -273,6 +294,11 @@ legend { font-size: .84rem; font-weight: 750; margin-bottom: 8px; } .text-field input, .text-field textarea { font-size: .84rem; } .form-error { color: #fecdd3; background: rgba(251,113,133,.1); border: 1px solid rgba(251,113,133,.4); border-radius: 8px; padding: 9px; font-size: .8rem; } .review-actions { display: grid; grid-template-columns: .9fr 1.4fr; gap: 8px; } +.conditional-fields { padding: 11px; border: 1px solid var(--line); border-radius: 9px; background: rgba(94,234,212,.035); display: grid; gap: 9px; } +.conditional-fields label, .boundary-details label { display: grid; gap: 6px; } +.conditional-fields label > span, .boundary-details label > span { color: var(--muted); font-size: .72rem; font-weight: 700; } +.boundary-details { border-top: 1px solid var(--line); padding-top: 10px; } +.boundary-details summary { color: var(--muted); cursor: pointer; font-size: .78rem; margin-bottom: 9px; } kbd { border: 1px solid rgba(3,24,22,.35); border-bottom-width: 2px; border-radius: 4px; padding: 1px 4px; font-size: .68rem; font-family: inherit; } .shortcuts { border-top: 1px solid var(--line); padding: 11px 15px 14px; display: flex; flex-wrap: wrap; gap: 8px 14px; color: var(--muted); font-size: .71rem; } .shortcuts kbd { color: #cbd7e7; border-color: var(--line-strong); margin-right: 2px; } diff --git a/floor_event_review_web/app.js b/floor_event_review_web/app.js index d199902..d747528 100644 --- a/floor_event_review_web/app.js +++ b/floor_event_review_web/app.js @@ -7,6 +7,7 @@ const state = { currentId: null, currentDetail: null, saving: false, + sessionSpeech: {}, }; const elements = Object.fromEntries( @@ -15,10 +16,11 @@ const elements = Object.fromEntries( "familyFilter", "statusFilter", "searchFilter", "clearFilters", "queueTitle", "queueCount", "queueList", "queueLimit", "familyBadge", "splitBadge", "autoConfidenceBadge", "eventTitle", "eventSubtitle", "eventPosition", - "llmSuggestion", "llmSuggestionText", + "llmSuggestion", "llmSuggestionText", "sessionTimeline", "speakerTimeline", "eventVideo", "mediaEmpty", "jumpEvent", "timeReadout", "clipLabel", "transcriptList", "rawCandidate", "reviewForm", "reviewerInput", "formError", - "saveButton", "uncertainButton", "saveState", "toast", + "saveButton", "uncertainButton", "saveState", "toast", "relabelFields", + "excludeFields", "eventTypeField", "subtypeField", ].map((id) => [id, document.getElementById(id)]) ); @@ -86,7 +88,7 @@ function applyFilters({ keepCurrent = true } = {}) { (elements.splitFilter.value === "all" || item.split === elements.splitFilter.value) && (elements.familyFilter.value === "all" || item.family === elements.familyFilter.value) && (status === "all" || (status === "reviewed" ? reviewed : !reviewed)) && - (!search || `${item.uid} ${item.automatic_label} ${item.automatic_protocol_label || ""} ${item.automatic_hard_negative_type || ""} ${item.session}`.toLowerCase().includes(search)) + (!search || `${item.uid} ${item.candidate_source || ""} ${item.rule_candidate_type || ""} ${item.ai_verdict || ""} ${item.ai_event_type || ""} ${item.ai_subtype || ""} ${item.session}`.toLowerCase().includes(search)) ); }); renderQueue(); @@ -121,9 +123,9 @@ function renderQueue() { meta.textContent = `${item.session} · ${familyNames[item.family] || item.family} · ${formatTime(item.event_time_seconds)}`; const label = document.createElement("div"); label.className = "queue-label"; - label.textContent = item.automatic_protocol_label - ? `${item.automatic_protocol_label} · ${item.automatic_hard_negative_type || item.automatic_label}` - : item.automatic_label; + label.textContent = item.ai_verdict + ? `AI ${item.ai_verdict} · ${item.ai_event_type || item.ai_subtype || "—"}` + : `${item.rule_candidate_type || item.automatic_label} · 待 AI 粗标`; button.append(line, meta, label); button.addEventListener("click", () => selectCandidate(item.candidate_id)); elements.queueList.append(button); @@ -171,7 +173,7 @@ function renderDetail(detail) { elements.familyBadge.textContent = familyNames[detail.family] || detail.family; elements.splitBadge.textContent = `${detail.split} · ${detail.session}`; elements.autoConfidenceBadge.textContent = detail.automatic_confidence ? `auto ${detail.automatic_confidence}` : "auto confidence —"; - elements.eventTitle.textContent = detail.automatic_label; + elements.eventTitle.textContent = detail.candidate_source || detail.automatic_label; elements.eventSubtitle.textContent = `${detail.uid} · ${detail.source_file}:${detail.source_row}`; elements.clipLabel.textContent = detail.clip_stem ? `${detail.session} / ${detail.clip_stem}` : detail.session; elements.rawCandidate.textContent = JSON.stringify(detail.candidate, null, 2); @@ -181,13 +183,16 @@ function renderDetail(detail) { const suggestion = detail.llm_suggestion; elements.llmSuggestion.hidden = !suggestion; if (suggestion) { - const disposition = suggestion.protocol_disposition || suggestion.final_protocol_disposition || "uncertain"; - const subtype = suggestion.hard_negative_type || suggestion.final_hard_negative_type || "uncertain"; + const disposition = suggestion.verdict || suggestion.protocol_disposition || "UNSURE"; + const subtype = suggestion.event_type || suggestion.subtype || suggestion.hard_negative_type || "—"; const confidence = suggestion.confidence || suggestion.final_confidence || "—"; - const rationale = suggestion.rationale || suggestion.final_rationale || ""; + const rationale = suggestion.reason || suggestion.rationale || ""; elements.llmSuggestionText.textContent = `${disposition} · ${subtype} · ${confidence}${rationale ? ` — ${rationale}` : ""}`; } + renderSessionTimeline(detail); + renderSpeakerTimeline(detail); + elements.mediaEmpty.hidden = detail.media_available; if (detail.media_available) { elements.eventVideo.src = `/media?id=${encodeURIComponent(detail.candidate_id)}`; @@ -226,16 +231,87 @@ function renderDetail(detail) { updateTimeReadout(); } +async function renderSessionTimeline(detail) { + elements.sessionTimeline.replaceChildren(); + if (!state.sessionSpeech[detail.session]) { + try { + const payload = await request(`/api/session-timeline?session=${encodeURIComponent(detail.session)}`); + state.sessionSpeech[detail.session] = payload.speech || []; + } catch (_) { + state.sessionSpeech[detail.session] = []; + } + } + if (state.currentId !== detail.candidate_id) return; + const events = state.candidates.filter((item) => item.session === detail.session); + const speech = state.sessionSpeech[detail.session]; + const end = Math.max(1, + ...events.map((item) => Number(item.event_time_seconds) || 0), + ...speech.map((row) => Number(row.end_seconds) || 0)); + const speakers = [...new Set(speech.map((row) => row.participant_role === "wearer" ? "W" : `P${row.participant_id}`))]; + for (const speakerId of speakers) { + const lane = document.createElement("div"); lane.className = "session-lane"; + const label = document.createElement("strong"); label.textContent = speakerId; + const track = document.createElement("div"); track.className = "session-track"; + for (const row of speech.filter((row) => (row.participant_role === "wearer" ? "W" : `P${row.participant_id}`) === speakerId)) { + const segment = document.createElement("span"); + segment.className = `session-speech${speakerId === "W" ? " wearer" : ""}`; + segment.style.left = `${100 * Number(row.start_seconds) / end}%`; + segment.style.width = `${Math.max(0.08, 100 * (Number(row.end_seconds) - Number(row.start_seconds)) / end)}%`; + track.append(segment); + } + lane.append(label, track); elements.sessionTimeline.append(lane); + } + const eventTrack = document.createElement("div"); eventTrack.className = "session-events"; + for (const item of events) { + const marker = document.createElement("button"); + marker.type = "button"; + marker.className = `session-marker ${item.family}${item.candidate_id === detail.candidate_id ? " current" : ""}`; + marker.style.left = `${Math.min(100, 100 * (Number(item.event_time_seconds) || 0) / end)}%`; + marker.title = `${formatTime(item.event_time_seconds)} · ${item.candidate_source || item.automatic_label} · ${item.ai_verdict || "未粗标"}`; + marker.addEventListener("click", () => selectCandidate(item.candidate_id)); + eventTrack.append(marker); + } + elements.sessionTimeline.append(eventTrack); +} + +function renderSpeakerTimeline(detail) { + elements.speakerTimeline.replaceChildren(); + const rows = detail.candidate.speaker_timeline || detail.candidate.context || []; + if (!rows.length) { + elements.speakerTimeline.textContent = "没有 participant activity evidence。"; + return; + } + const eventTime = Number(detail.candidate.event_time ?? detail.candidate.anchor_dense_seconds); + const left = Math.min(...rows.map((row) => Number(row.start_seconds)), eventTime - 3); + const right = Math.max(...rows.map((row) => Number(row.end_seconds)), eventTime + 3); + const span = Math.max(0.001, right - left); + const speakers = [...new Set(rows.map((row) => `${row.participant_role === "wearer" ? "W" : `P${row.participant_id}`}`))]; + for (const speakerId of speakers) { + const lane = document.createElement("div"); + lane.className = "speaker-lane"; + const label = document.createElement("strong"); label.textContent = speakerId; + const track = document.createElement("div"); track.className = "speaker-track"; + for (const row of rows.filter((row) => (row.participant_role === "wearer" ? "W" : `P${row.participant_id}`) === speakerId)) { + const segment = document.createElement("span"); + segment.className = `speech-segment${speakerId === "W" ? " wearer" : ""}`; + segment.style.left = `${100 * (Number(row.start_seconds) - left) / span}%`; + segment.style.width = `${Math.max(0.8, 100 * (Number(row.end_seconds) - Number(row.start_seconds)) / span)}%`; + segment.title = `${formatTime(row.start_seconds)}–${formatTime(row.end_seconds)} ${row.transcription || ""}`; + track.append(segment); + } + const anchor = document.createElement("i"); anchor.style.left = `${100 * (eventTime - left) / span}%`; track.append(anchor); + lane.append(label, track); elements.speakerTimeline.append(lane); + } +} + function restoreForm(review) { elements.reviewForm.reset(); elements.formError.hidden = true; const reviewer = review?.reviewer || localStorage.getItem("floor-reviewer") || ""; elements.reviewerInput.value = reviewer; if (!review) { - elements.reviewForm.elements.anchor_adjustment_bins.value = "0"; - elements.reviewForm.elements.behavior_adjustment_bins.value = "0"; - elements.reviewForm.elements.region_end_adjustment_bins.value = "0"; elements.reviewForm.elements.confidence.value = "medium"; + updateConditionalFields(); return; } for (const [name, value] of Object.entries(review)) { @@ -244,24 +320,20 @@ function restoreForm(review) { if (control instanceof RadioNodeList) control.value = String(value); else control.value = value === null ? "" : String(value); } + updateConditionalFields(); } function formPayload() { const data = new FormData(elements.reviewForm); - const adjustment = (name) => { - const value = data.get(name); - return value === "" ? null : Number(value); - }; + const nullable = (name) => data.get(name) || null; + const correction = data.get("boundary_correction_seconds"); return { - protocol_disposition: data.get("protocol_disposition"), - hard_negative_type: data.get("hard_negative_type"), - validity: data.get("validity"), - onset_function: data.get("onset_function"), - context: data.get("context"), - offset_outcome: data.get("offset_outcome"), - anchor_adjustment_bins: adjustment("anchor_adjustment_bins"), - behavior_adjustment_bins: adjustment("behavior_adjustment_bins"), - region_end_adjustment_bins: adjustment("region_end_adjustment_bins"), + human_action: data.get("human_action"), + human_verdict: nullable("human_verdict"), + human_event_type: nullable("human_event_type"), + human_subtype: nullable("human_subtype"), + exclude_reason: nullable("exclude_reason"), + boundary_correction_seconds: correction === "" ? null : Number(correction), confidence: data.get("confidence"), reviewer: String(data.get("reviewer") || "").trim(), notes: String(data.get("notes") || "").trim(), @@ -270,16 +342,25 @@ function formPayload() { function validateReview(review) { const missing = []; - if (!review.protocol_disposition) missing.push("协议结论"); - if (!review.hard_negative_type) missing.push("hard-negative 类型"); - if (!review.validity) missing.push("活动边界"); - if (!review.onset_function) missing.push("onset 功能"); - if (!review.context) missing.push("对话场景"); - if (!review.offset_outcome) missing.push("offset 结果"); + if (!review.human_action) missing.push("主操作"); + if (review.human_action === "relabel" && !review.human_verdict) missing.push("一级 verdict"); + if (review.human_action === "relabel" && review.human_verdict === "POSITIVE" && !review.human_event_type) missing.push("positive event"); + if (review.human_action === "relabel" && review.human_verdict === "HARD_NEGATIVE" && !review.human_subtype) missing.push("hard-negative subtype"); + if (review.human_action === "exclude" && !review.exclude_reason) missing.push("排除原因"); + if (review.human_action === "accept" && !state.currentDetail?.candidate?.ai_verdict && !state.currentDetail?.llm_suggestion?.verdict) missing.push("AI 预标注（尚未运行，需 Relabel）"); if (!review.reviewer) missing.push("审核者"); return missing; } +function updateConditionalFields() { + const action = elements.reviewForm.elements.human_action.value; + const verdict = elements.reviewForm.elements.human_verdict.value; + elements.relabelFields.hidden = action !== "relabel"; + elements.excludeFields.hidden = action !== "exclude"; + elements.eventTypeField.hidden = action !== "relabel" || verdict !== "POSITIVE"; + elements.subtypeField.hidden = action !== "relabel" || verdict !== "HARD_NEGATIVE"; +} + async function saveCurrent({ advance = true } = {}) { if (!state.currentId || state.saving) return; const review = formPayload(); @@ -376,14 +457,12 @@ function bindEvents() { saveCurrent(); }); elements.uncertainButton.addEventListener("click", () => { - elements.reviewForm.elements.protocol_disposition.value = "uncertain"; - elements.reviewForm.elements.hard_negative_type.value = "uncertain"; - elements.reviewForm.elements.validity.value = "uncertain"; - elements.reviewForm.elements.onset_function.value = "uncertain"; - elements.reviewForm.elements.context.value = "unknown"; - elements.reviewForm.elements.offset_outcome.value = "uncertain"; + elements.reviewForm.elements.human_action.value = "second_review"; elements.reviewForm.elements.confidence.value = "low"; + updateConditionalFields(); }); + elements.reviewForm.elements.human_action.forEach((control) => control.addEventListener("change", updateConditionalFields)); + elements.reviewForm.elements.human_verdict.addEventListener("change", updateConditionalFields); elements.jumpEvent.addEventListener("click", jumpToEvent); document.querySelectorAll("[data-seek]").forEach((button) => button.addEventListener("click", () => seek(Number(button.dataset.seek))) diff --git a/floor_event_review_web/index.html b/floor_event_review_web/index.html index f4f0b5c..81b0962 100644 --- a/floor_event_review_web/index.html +++ b/floor_event_review_web/index.html @@ -1,285 +1,82 @@ - - - - - Floor Event Review - - - - -
-
- -
-

Floor Event Review

-

EasyCom 事件审核

-
-
-
-
- 载入中 - — -
-
-
- -
+ + + + EasyCom Floor Event Review + + + +
+

Floor Event Review

AI 粗标注 → 人工确认

+
载入中—
+ +
-
- - - - - - -
+
+ + + + + + +
-
- +
+ -
-
-
-
- — - — - — -
-

请选择一个候选事件

-

—

-
-
0 / 0
-
+
+
———

请选择候选

—

0 / 0
- + -
- - - -
- -
- - - - - - — -
- -
-
-
- TRANSCRIPT -

事件上下文

-
- — -
-
-
选择事件后显示 transcript。
-
-
- -
- 查看自动规则与原始候选字段 - 
{}
-
+
+
SESSION OVERVIEW

本 Session 候选分布

点击 marker 跳转
+
+
LOCAL SPEAKER ACTIVITY

事件附近 participant timeline

原始标注，只读
+
-
+ + -
- -
- - +
+ + +
+ +
←→ 前后事件[] ±200 msSpace 播放/暂停
+ +
+
+ diff --git a/prepare_easycom_eot_int_review.py b/prepare_easycom_eot_int_review.py index 4352b52..3729d9b 100644 --- a/prepare_easycom_eot_int_review.py +++ b/prepare_easycom_eot_int_review.py @@ -1,18 +1,10 @@ #!/usr/bin/env python3 -"""Prepare manuscript-aligned EasyCom EOT/INT regions for human review. +"""Generate wearer-centric EasyCom EOT/INT candidates for AI and human review. -The released EasyCom annotations provide participant-resolved speech intervals, -not semantic EOT/INT ground truth. This script therefore creates auditable -candidate regions from the original 20 Hz timeline. It keeps the two wearer- -relative directions distinct: - -* EOT-response: an aggregate non-wearer offset while the wearer is silent; -* INT-yield: an aggregate non-wearer onset while the wearer is speaking. - -Observed wearer transitions form timing-positive candidates. Competing other -transitions and fully observed no-transition windows form hard-negative -candidates. Source-boundary cases are retained as exclusions. Every output -row remains an automatic candidate until it is independently human reviewed. +This stage never changes EasyCom speech boundaries and never claims semantic +ground truth. It emits every wearer transition plus independent participant- +resolved hard-negative regions. All timing thresholds are CLI parameters, and +raw evidence is retained for DeepSeek pre-labeling and human correction. """ from __future__ import annotations @@ -26,22 +18,24 @@ import random from typing import Any, Iterable, Sequence from derive_easycom_interaction_events import ( - CAMERA_WEARER_ID, - SOURCE_FPS, - Clip, - Rules, - Utterance, - build_clips, - is_lexical_feedback, - is_pure_nonverbal, - load_utterances, + CAMERA_WEARER_ID, SOURCE_FPS, Clip, Rules, Utterance, build_clips, + is_lexical_feedback, is_pure_nonverbal, load_utterances, ) from easycom_manifest import ALL_SESSIONS, TEST_SESSIONS -SCHEMA_VERSION = 1 -PROTOCOL_VERSION = "easycom-manuscript-eot-int-v1" -DEFAULT_WINDOW_SECONDS = 2.0 +SCHEMA_VERSION = 2 +PROTOCOL_VERSION = "easycom-wearer-centric-eot-int-candidates-v2" + + +@dataclass(frozen=True) +class CandidateConfig: + context_seconds: float = 3.0 + relation_window_seconds: float = 3.0 + same_speaker_pause_max_seconds: float = 2.0 + other_handover_max_seconds: float = 1.5 + short_vocalization_max_seconds: float = 1.0 + continuation_min_seconds: float = 0.4 @dataclass(frozen=True) @@ -49,40 +43,46 @@ class ProtocolCandidate: schema_version: int protocol_version: str uid: str + event_id: str session: str split: str task: str event_type: str + candidate_source: str + rule_candidate_type: str automatic_protocol_label: str - automatic_hard_negative_type: str + automatic_hard_negative_type: str | None automatic_confidence: str anchor_type: str anchor_dense_frame: int anchor_dense_seconds: float - behavior_anchor_type: str | None - behavior_dense_frame: int | None - behavior_seconds: float | None + event_time: float region_start_seconds: float region_end_seconds: float - region_end_reason: str - previous_other_participant_ids: tuple[int, ...] - incoming_other_participant_ids: tuple[int, ...] clip_stem: str source_start_frame: int + previous_other_participant_ids: tuple[int, ...] + incoming_other_participant_ids: tuple[int, ...] + temporal_evidence: dict[str, Any] + speaker_timeline: tuple[dict[str, Any], ...] context: tuple[dict[str, Any], ...] automatic_reason: str + ai_verdict: None = None + ai_event_type: None = None + ai_subtype: None = None + ai_confidence: None = None + ai_reason: None = None + human_status: str = "UNREVIEWED" + human_verdict: None = None + human_event_type: None = None + human_subtype: None = None + exclude_reason: None = None -def _fill_state( - rows: Sequence[Utterance], - *, - segment_start: int, - segment_end: int, - wearer_id: int, -) -> tuple[bytearray, bytearray]: +def _fill_states(rows: Sequence[Utterance], *, segment_start: int, segment_end: int, + wearer_id: int) -> tuple[bytearray, bytearray]: length = segment_end - segment_start - wearer = bytearray(length) - others = bytearray(length) + wearer, others = bytearray(length), bytearray(length) for row in rows: left = max(segment_start, row.dense_start_frame) - segment_start right = min(segment_end, row.dense_end_frame_exclusive) - segment_start @@ -93,371 +93,329 @@ def _fill_state( return wearer, others -def _first_transition( - state: bytearray, - *, - start: int, - stop: int, - rising: bool, -) -> int | None: - for index in range(max(1, start), min(stop, len(state))): - before = bool(state[index - 1]) - after = bool(state[index]) - if (not before and after) if rising else (before and not after): - return index - return None +def _run_after(state: bytearray, index: int, value: bool) -> int: + end = index + while end < len(state) and bool(state[end]) == value: + end += 1 + return end - index -def _active_other_ids( - rows: Sequence[Utterance], - dense_frame: int, - *, - wearer_id: int, -) -> tuple[int, ...]: +def _run_before(state: bytearray, index: int, value: bool) -> int: + start = index - 1 + while start >= 0 and bool(state[start]) == value: + start -= 1 + return index - start - 1 + + +def _all_state(state: bytearray, start: int, stop: int, value: bool) -> bool: + left, right = max(0, start), min(len(state), stop) + return right > left and all(bool(state[i]) == value for i in range(left, right)) + + +def _active_other_ids(rows: Sequence[Utterance], frame: int, *, wearer_id: int) -> tuple[int, ...]: return tuple(sorted({ - row.participant_id - for row in rows + row.participant_id for row in rows if row.participant_id != wearer_id - and row.dense_start_frame <= dense_frame < row.dense_end_frame_exclusive + and row.dense_start_frame <= frame < row.dense_end_frame_exclusive })) -def _clip_for_frame(clips: Sequence[Clip], dense_frame: int) -> Clip | None: - return next( - ( - clip - for clip in clips - if clip.dense_start_frame <= dense_frame < clip.dense_start_frame + clip.num_frames +def _clip_for_frame(clips: Sequence[Clip], frame: int) -> Clip | None: + return next((clip for clip in clips + if clip.dense_start_frame <= frame < clip.dense_start_frame + clip.num_frames), None) + + +def _context_rows(rows: Sequence[Utterance], *, frame: int, context_frames: int, + wearer_id: int) -> tuple[dict[str, Any], ...]: + left, right = frame - context_frames, frame + context_frames + return tuple({ + "uid": row.uid, + "participant_id": row.participant_id, + "participant_role": "wearer" if row.participant_id == wearer_id else "other", + "start_seconds": row.dense_start_seconds, + "end_seconds": row.dense_end_seconds, + "duration_seconds": row.duration_seconds, + "transcription": row.transcription, + "target_of_speech": list(row.target_of_speech), + } for row in rows + if row.dense_end_frame_exclusive > left and row.dense_start_frame < right) + + +def _relation_evidence(rows: Sequence[Utterance], *, frame: int, wearer_id: int, + window_frames: int) -> dict[str, Any]: + previous = [row for row in rows if row.participant_id != wearer_id + and row.dense_end_frame_exclusive <= frame + and frame - row.dense_end_frame_exclusive <= window_frames] + following = [row for row in rows if row.participant_id != wearer_id + and row.dense_start_frame >= frame + and row.dense_start_frame - frame <= window_frames] + prev = max(previous, key=lambda row: row.dense_end_frame_exclusive, default=None) + nxt = min(following, key=lambda row: row.dense_start_frame, default=None) + active_rows = [row for row in rows if row.participant_id != wearer_id + and row.dense_start_frame < frame < row.dense_end_frame_exclusive] + overlapping = sorted({row.participant_id for row in active_rows}) + resumes = [] if prev is None else [row for row in following if row.participant_id == prev.participant_id] + resumed = min(resumes, key=lambda row: row.dense_start_frame, default=None) + nearby_onsets = [row for row in rows if row.participant_id != wearer_id + and abs(row.dense_start_frame - frame) <= window_frames] + closest_onset = min(nearby_onsets, key=lambda row: abs(row.dense_start_frame - frame), default=None) + return { + "previous_speaker_id": None if prev is None else prev.participant_id, + "previous_speaker_offset_seconds": None if prev is None else prev.dense_end_seconds, + "gap_from_previous_speaker_offset_seconds": ( + None if prev is None else (frame - prev.dense_end_frame_exclusive) / SOURCE_FPS), + "overlapping_other_ids": overlapping, + "other_overlap_at_event_seconds": max( + [0.0] + [(row.dense_end_frame_exclusive - frame) / SOURCE_FPS for row in active_rows] ), - None, - ) + "next_speaker_id": None if nxt is None else nxt.participant_id, + "next_other_onset_seconds": None if nxt is None else nxt.dense_start_seconds, + "previous_speaker_resumes": resumed is not None, + "previous_speaker_resume_delay_seconds": ( + None if resumed is None or prev is None + else (resumed.dense_start_frame - prev.dense_end_frame_exclusive) / SOURCE_FPS + ), + "closest_other_onset_relative_seconds": ( + None if closest_onset is None else (closest_onset.dense_start_frame - frame) / SOURCE_FPS + ), + } -def _context_rows( - rows: Sequence[Utterance], - *, - anchor_frame: int, - context_frames: int, - wearer_id: int, -) -> tuple[dict[str, Any], ...]: - left = anchor_frame - context_frames - right = anchor_frame + context_frames - selected = [] - for row in rows: - if row.dense_end_frame_exclusive <= left or row.dense_start_frame >= right: - continue - selected.append( - { - "uid": row.uid, - "participant_id": row.participant_id, - "participant_role": "wearer" if row.participant_id == wearer_id else "other", - "start_seconds": row.dense_start_seconds, - "end_seconds": row.dense_end_seconds, - "transcription": row.transcription, - "target_of_speech": list(row.target_of_speech), - } - ) - return tuple(selected) - - -def _incoming_is_backchannel( - rows: Sequence[Utterance], - *, - dense_frame: int, - wearer_id: int, - max_seconds: float, -) -> bool: - incoming = [ - row - for row in rows - if row.participant_id != wearer_id - and row.dense_start_frame <= dense_frame < row.dense_end_frame_exclusive - and row.dense_start_frame >= dense_frame - 1 - ] - return bool(incoming) and all(row.duration_seconds <= max_seconds for row in incoming) and any( - is_lexical_feedback(row.transcription) or is_pure_nonverbal(row.transcription) - for row in incoming - ) - - -def _candidate( - *, - session: str, - task: str, - automatic_protocol_label: str, - automatic_hard_negative_type: str, - anchor_frame: int, - behavior_frame: int | None, - region_end_frame: int, - region_end_reason: str, - previous_ids: tuple[int, ...], - incoming_ids: tuple[int, ...], - clips: Sequence[Clip], - rows: Sequence[Utterance], - wearer_id: int, - context_frames: int, - reason: str, -) -> ProtocolCandidate: - clip = _clip_for_frame(clips, anchor_frame) +def _candidate(*, session: str, task: str, source: str, rule_type: str, + anchor_type: str, frame: int, region_start: int, region_end: int, + label: str, subtype: str | None, previous_ids: tuple[int, ...], + incoming_ids: tuple[int, ...], evidence: dict[str, Any], clips: Sequence[Clip], + rows: Sequence[Utterance], wearer_id: int, context_frames: int, + reason: str) -> ProtocolCandidate | None: + clip = _clip_for_frame(clips, frame) if clip is None: - raise ValueError(f"No source clip for {session} dense frame {anchor_frame}") - event_type = "eot_response_candidate" if task == "eot_response" else "int_yield_candidate" - anchor_type = "other_offset" if task == "eot_response" else "other_onset" - behavior_type = None - if behavior_frame is not None: - behavior_type = "wearer_onset" if task == "eot_response" else "wearer_offset" + return None + participant_key = "prev-" + ("-".join(f"p{x}" for x in previous_ids) or "none") + participant_key += "__in-" + ("-".join(f"p{x}" for x in incoming_ids) or "none") + uid = f"{session}/{source.lower()}/{frame:08d}/{participant_key}" + context = _context_rows(rows, frame=frame, context_frames=context_frames, wearer_id=wearer_id) return ProtocolCandidate( - schema_version=SCHEMA_VERSION, - protocol_version=PROTOCOL_VERSION, - uid=f"{session}/{task}/{anchor_frame:08d}", - session=session, - split="test" if session in TEST_SESSIONS else "train", - task=task, - event_type=event_type, - automatic_protocol_label=automatic_protocol_label, - automatic_hard_negative_type=automatic_hard_negative_type, - automatic_confidence="timing_candidate", - anchor_type=anchor_type, - anchor_dense_frame=anchor_frame, - anchor_dense_seconds=anchor_frame / SOURCE_FPS, - behavior_anchor_type=behavior_type, - behavior_dense_frame=behavior_frame, - behavior_seconds=None if behavior_frame is None else behavior_frame / SOURCE_FPS, - region_start_seconds=anchor_frame / SOURCE_FPS, - region_end_seconds=region_end_frame / SOURCE_FPS, - region_end_reason=region_end_reason, + schema_version=SCHEMA_VERSION, protocol_version=PROTOCOL_VERSION, + uid=uid, event_id=uid, session=session, + split="test" if session in TEST_SESSIONS else "train", task=task, + event_type=source.lower(), candidate_source=source, rule_candidate_type=rule_type, + automatic_protocol_label=label, automatic_hard_negative_type=subtype, + automatic_confidence="rule_candidate", anchor_type=anchor_type, + anchor_dense_frame=frame, anchor_dense_seconds=frame / SOURCE_FPS, + event_time=frame / SOURCE_FPS, region_start_seconds=region_start / SOURCE_FPS, + region_end_seconds=region_end / SOURCE_FPS, clip_stem=clip.stem, + source_start_frame=frame - clip.dense_start_frame + 1, previous_other_participant_ids=previous_ids, - incoming_other_participant_ids=incoming_ids, - clip_stem=clip.stem, - source_start_frame=anchor_frame - clip.dense_start_frame + 1, - context=_context_rows( - rows, - anchor_frame=anchor_frame, - context_frames=context_frames, - wearer_id=wearer_id, - ), - automatic_reason=reason, + incoming_other_participant_ids=incoming_ids, temporal_evidence=evidence, + speaker_timeline=context, context=context, automatic_reason=reason, ) -def derive_protocol_candidates( - utterances: Sequence[Utterance], - clips: Sequence[Clip], - *, - session: str, - wearer_id: int = CAMERA_WEARER_ID, - window_seconds: float = DEFAULT_WINDOW_SECONDS, - context_seconds: float = 3.0, -) -> list[ProtocolCandidate]: - """Construct timing candidates with manuscript-aligned directionality.""" - - if window_seconds <= 0 or context_seconds < 0: - raise ValueError("window_seconds must be positive and context_seconds non-negative") - window_frames = max(1, round(window_seconds * SOURCE_FPS)) - context_frames = max(0, round(context_seconds * SOURCE_FPS)) +def derive_protocol_candidates(utterances: Sequence[Utterance], clips: Sequence[Clip], *, + session: str, wearer_id: int = CAMERA_WEARER_ID, + config: CandidateConfig | None = None, + window_seconds: float | None = None, + context_seconds: float | None = None) -> list[ProtocolCandidate]: + """Return every wearer transition and independent hard-negative regions.""" + cfg = config or CandidateConfig() + values = asdict(cfg) + if window_seconds is not None: + values["relation_window_seconds"] = window_seconds + if context_seconds is not None: + values["context_seconds"] = context_seconds + cfg = CandidateConfig(**values) + if any(value < 0 for value in asdict(cfg).values()) or cfg.relation_window_seconds == 0: + raise ValueError("candidate timing parameters must be non-negative and relation window positive") + context_frames = round(cfg.context_seconds * SOURCE_FPS) + relation_frames = round(cfg.relation_window_seconds * SOURCE_FPS) + pause_frames = round(cfg.same_speaker_pause_max_seconds * SOURCE_FPS) + handover_frames = round(cfg.other_handover_max_seconds * SOURCE_FPS) + short_frames = round(cfg.short_vocalization_max_seconds * SOURCE_FPS) + continuation_frames = round(cfg.continuation_min_seconds * SOURCE_FPS) rows_by_segment: dict[int, list[Utterance]] = defaultdict(list) clips_by_segment: dict[int, list[Clip]] = defaultdict(list) for row in utterances: rows_by_segment[row.segment_id].append(row) for clip in clips: clips_by_segment[clip.segment_id].append(clip) - candidates: list[ProtocolCandidate] = [] + seen: set[tuple[str, int, tuple[int, ...]]] = set() + + def add(item: ProtocolCandidate | None, key: tuple[str, int, tuple[int, ...]]) -> None: + if item is not None and key not in seen: + seen.add(key) + candidates.append(item) + for segment_id, segment_clips in sorted(clips_by_segment.items()): segment_clips.sort(key=lambda clip: clip.dense_start_frame) - segment_rows = sorted( - rows_by_segment.get(segment_id, []), - key=lambda row: (row.dense_start_frame, row.dense_end_frame_exclusive, row.uid), - ) + rows = sorted(rows_by_segment.get(segment_id, []), + key=lambda row: (row.dense_start_frame, row.dense_end_frame_exclusive, row.uid)) segment_start = segment_clips[0].dense_start_frame segment_end = segment_clips[-1].dense_start_frame + segment_clips[-1].num_frames - wearer, others = _fill_state( - segment_rows, - segment_start=segment_start, - segment_end=segment_end, - wearer_id=wearer_id, - ) + wearer, others = _fill_states(rows, segment_start=segment_start, + segment_end=segment_end, wearer_id=wearer_id) - for index in range(1, len(others)): - dense_frame = segment_start + index - if others[index - 1] and not others[index] and not wearer[index - 1]: - competing = _first_transition( - others, - start=index + 1, - stop=min(len(others), index + window_frames + 1), - rising=True, + # A: all wearer 0->1 and 1->0 transitions, without semantic guessing. + for index in range(1, len(wearer)): + frame = segment_start + index + if not wearer[index - 1] and wearer[index]: + evidence = _relation_evidence(rows, frame=frame, wearer_id=wearer_id, + window_frames=relation_frames) + evidence.update( + wearer_speech_duration_after_onset_seconds=_run_after(wearer, index, True) / SOURCE_FPS, + previous_wearer_silence_duration_seconds=_run_before(wearer, index, False) / SOURCE_FPS, + original_timestamp_seconds=frame / SOURCE_FPS, ) - nominal_end = index + window_frames - observable_end = min(nominal_end, len(others)) - effective_end = min(observable_end, competing if competing is not None else observable_end) - outcome = _first_transition( - wearer, - start=index, - stop=effective_end, - rising=True, - ) - previous_ids = _active_other_ids( - segment_rows, dense_frame - 1, wearer_id=wearer_id - ) - incoming_ids = ( - () - if competing is None - else _active_other_ids( - segment_rows, segment_start + competing, wearer_id=wearer_id - ) - ) - if outcome is not None: - label = "positive" - subtype = "not_applicable" - end_index = outcome - end_reason = "wearer_onset" - reason = "wearer onset observed after other offset and before competing other onset" - elif competing is not None: - label = "hard_negative" - subtype = ( - "eot_midturn_pause" - if set(previous_ids) & set(incoming_ids) - else "eot_other_to_other" - ) - end_index = competing - end_reason = "competing_other_onset" - reason = "wearer remains silent until another non-wearer onset" - elif nominal_end <= len(others): - label = "hard_negative" - subtype = "eot_no_response" - end_index = nominal_end - end_reason = "full_window" - reason = "wearer remains silent for the full observable response window" - else: - label = "exclude" - subtype = "other" - end_index = len(others) - end_reason = "right_censored" - reason = "recording segment ends before a wearer outcome or full window" - candidates.append(_candidate( - session=session, - task="eot_response", - automatic_protocol_label=label, - automatic_hard_negative_type=subtype, - anchor_frame=dense_frame, - behavior_frame=None if outcome is None else segment_start + outcome, - region_end_frame=segment_start + end_index, - region_end_reason=end_reason, - previous_ids=previous_ids, - incoming_ids=incoming_ids, - clips=segment_clips, - rows=segment_rows, - wearer_id=wearer_id, + add(_candidate( + session=session, task="eot_response", source="WEARER_ONSET", + rule_type="ENTRY_CANDIDATE", anchor_type="wearer_onset", frame=frame, + region_start=frame, region_end=min(segment_end, frame + relation_frames), + label="unsure", subtype=None, + previous_ids=_active_other_ids(rows, frame - 1, wearer_id=wearer_id), + incoming_ids=_active_other_ids(rows, frame, wearer_id=wearer_id), + evidence=evidence, clips=segment_clips, rows=rows, wearer_id=wearer_id, context_frames=context_frames, - reason=reason, - )) - - if not others[index - 1] and others[index] and wearer[index - 1]: - competing = _first_transition( - others, - start=index + 1, - stop=min(len(others), index + window_frames + 1), - rising=False, + reason="observed wearer 0->1 transition; semantic EOT-response status is unresolved", + ), ("WEARER_ONSET", frame, ())) + elif wearer[index - 1] and not wearer[index]: + evidence = _relation_evidence(rows, frame=frame, wearer_id=wearer_id, + window_frames=relation_frames) + evidence.update( + wearer_speech_duration_before_offset_seconds=_run_before(wearer, index, True) / SOURCE_FPS, + wearer_silence_duration_after_offset_seconds=_run_after(wearer, index, False) / SOURCE_FPS, + active_other_ids_before_offset=list(_active_other_ids(rows, frame - 1, wearer_id=wearer_id)), + active_other_ids_after_offset=list(_active_other_ids(rows, frame, wearer_id=wearer_id)), + original_timestamp_seconds=frame / SOURCE_FPS, ) - nominal_end = index + window_frames - observable_end = min(nominal_end, len(others)) - effective_end = min(observable_end, competing if competing is not None else observable_end) - outcome = _first_transition( - wearer, - start=index, - stop=effective_end, - rising=False, - ) - previous_ids = _active_other_ids( - segment_rows, dense_frame - 1, wearer_id=wearer_id - ) - incoming_ids = _active_other_ids( - segment_rows, dense_frame, wearer_id=wearer_id - ) - if outcome is not None: - label = "positive" - subtype = "not_applicable" - end_index = outcome - end_reason = "wearer_offset" - reason = "wearer offset observed after other onset and before competing other offset" - elif competing is not None: - label = "hard_negative" - subtype = ( - "int_backchannel" - if _incoming_is_backchannel( - segment_rows, - dense_frame=dense_frame, - wearer_id=wearer_id, - max_seconds=1.0, - ) - else "int_non_takeover_overlap" - ) - end_index = competing - end_reason = "competing_other_offset" - reason = "wearer retains the floor until the incoming other activity ends" - elif nominal_end <= len(others): - label = "hard_negative" - subtype = "int_retain_floor" - end_index = nominal_end - end_reason = "full_window" - reason = "wearer remains active through the full observable yield window" - else: - label = "exclude" - subtype = "other" - end_index = len(others) - end_reason = "right_censored" - reason = "recording segment ends before a wearer outcome or full window" - candidates.append(_candidate( - session=session, - task="int_yield", - automatic_protocol_label=label, - automatic_hard_negative_type=subtype, - anchor_frame=dense_frame, - behavior_frame=None if outcome is None else segment_start + outcome, - region_end_frame=segment_start + end_index, - region_end_reason=end_reason, - previous_ids=previous_ids, - incoming_ids=incoming_ids, - clips=segment_clips, - rows=segment_rows, - wearer_id=wearer_id, + add(_candidate( + session=session, task="int_yield", source="WEARER_OFFSET", + rule_type="YIELD_CANDIDATE", anchor_type="wearer_offset", frame=frame, + region_start=max(segment_start, frame - relation_frames), + region_end=min(segment_end, frame + relation_frames), label="unsure", subtype=None, + previous_ids=_active_other_ids(rows, frame - 1, wearer_id=wearer_id), + incoming_ids=_active_other_ids(rows, frame, wearer_id=wearer_id), + evidence=evidence, clips=segment_clips, rows=rows, wearer_id=wearer_id, context_frames=context_frames, - reason=reason, - )) + reason="observed wearer 1->0 transition; semantic INT-yield status is unresolved", + ), ("WEARER_OFFSET", frame, ())) - return sorted(candidates, key=lambda row: (row.anchor_dense_frame, row.task, row.uid)) + other_rows = [row for row in rows if row.participant_id != wearer_id] + + # B1: same speaker pauses and resumes while the wearer remains silent. + by_speaker: dict[int, list[Utterance]] = defaultdict(list) + for row in other_rows: + by_speaker[row.participant_id].append(row) + for participant_id, speaker_rows in by_speaker.items(): + for first, resumed in zip(speaker_rows, speaker_rows[1:]): + gap = resumed.dense_start_frame - first.dense_end_frame_exclusive + left = first.dense_end_frame_exclusive - segment_start + right = resumed.dense_start_frame - segment_start + if (gap <= 0 or gap > pause_frames + or not _all_state(wearer, left, right, False) + or not _all_state(others, left, right, False)): + continue + frame = first.dense_end_frame_exclusive + evidence = { + "speaker_id": participant_id, "speaker_offset_seconds": first.dense_end_seconds, + "speaker_resume_seconds": resumed.dense_start_seconds, + "resume_delay_seconds": gap / SOURCE_FPS, "wearer_active_during_pause": False, + "original_timestamp_seconds": frame / SOURCE_FPS, + } + add(_candidate( + session=session, task="eot_response", source="SAME_SPEAKER_PAUSE", + rule_type="EOT_HARD_NEGATIVE_REGION", anchor_type="other_offset", frame=frame, + region_start=frame, region_end=resumed.dense_start_frame, + label="hard_negative", subtype="EOT_MIDTURN_PAUSE", + previous_ids=(participant_id,), incoming_ids=(participant_id,), evidence=evidence, + clips=segment_clips, rows=rows, wearer_id=wearer_id, context_frames=context_frames, + reason="same participant resumes after a configurable pause while wearer stays silent", + ), ("SAME_SPEAKER_PAUSE", frame, (participant_id,))) + + # B2: participant-resolved non-wearer to non-wearer handover. + for first in other_rows: + options = [row for row in other_rows if row.participant_id != first.participant_id + and row.dense_start_frame >= first.dense_end_frame_exclusive + and row.dense_start_frame - first.dense_end_frame_exclusive <= handover_frames] + if not options: + continue + incoming = min(options, key=lambda row: row.dense_start_frame) + same_resumes = any(row.participant_id == first.participant_id + and first.dense_end_frame_exclusive < row.dense_start_frame <= incoming.dense_start_frame + for row in other_rows) + left = first.dense_end_frame_exclusive - segment_start + right = max(left + 1, incoming.dense_start_frame - segment_start + 1) + if same_resumes or not _all_state(wearer, left, right, False): + continue + frame = first.dense_end_frame_exclusive + evidence = { + "previous_speaker_id": first.participant_id, "incoming_speaker_id": incoming.participant_id, + "previous_offset_seconds": first.dense_end_seconds, + "incoming_onset_seconds": incoming.dense_start_seconds, + "handover_gap_seconds": (incoming.dense_start_frame - frame) / SOURCE_FPS, + "wearer_active_during_handover": False, "original_timestamp_seconds": frame / SOURCE_FPS, + } + add(_candidate( + session=session, task="eot_response", source="OTHER_TO_OTHER", + rule_type="EOT_HARD_NEGATIVE_REGION", anchor_type="other_offset", frame=frame, + region_start=frame, region_end=incoming.dense_start_frame, + label="hard_negative", subtype="EOT_OTHER_TO_OTHER", + previous_ids=(first.participant_id,), incoming_ids=(incoming.participant_id,), + evidence=evidence, clips=segment_clips, rows=rows, wearer_id=wearer_id, + context_frames=context_frames, + reason="participant-level IDs show a non-wearer to non-wearer handover", + ), ("OTHER_TO_OTHER", frame, (first.participant_id, incoming.participant_id))) + + # B3/B4: other activity ends while the wearer continuously holds and continues the floor. + for other in other_rows: + start = other.dense_start_frame - segment_start + end = other.dense_end_frame_exclusive - segment_start + if (start < 0 or end + continuation_frames > len(wearer) + or not _all_state(wearer, start, end + continuation_frames, True)): + continue + feedback = (other.dense_end_frame_exclusive - other.dense_start_frame <= short_frames + and (is_lexical_feedback(other.transcription) or is_pure_nonverbal(other.transcription))) + source = "OTHER_BACKCHANNEL" if feedback else "NON_TAKEOVER_OVERLAP" + subtype = "INT_OTHER_BACKCHANNEL" if feedback else "INT_NON_TAKEOVER_OVERLAP" + evidence = { + "other_speaker_id": other.participant_id, "other_onset_seconds": other.dense_start_seconds, + "other_offset_seconds": other.dense_end_seconds, "other_duration_seconds": other.duration_seconds, + "wearer_active_through_region": True, + "wearer_continuation_after_other_seconds": continuation_frames / SOURCE_FPS, + "lexical_or_nonverbal_feedback_hint": feedback, + "original_timestamp_seconds": other.dense_start_seconds, + } + add(_candidate( + session=session, task="int_yield", source=source, + rule_type="INT_HARD_NEGATIVE_REGION", anchor_type="other_onset", + frame=other.dense_start_frame, region_start=other.dense_start_frame, + region_end=other.dense_end_frame_exclusive, label="hard_negative", subtype=subtype, + previous_ids=_active_other_ids(rows, other.dense_start_frame - 1, wearer_id=wearer_id), + incoming_ids=(other.participant_id,), evidence=evidence, clips=segment_clips, + rows=rows, wearer_id=wearer_id, context_frames=context_frames, + reason=("short feedback-like other activity ends while wearer holds the floor" + if feedback else "other overlaps but wearer retains and continues the floor"), + ), (source, other.dense_start_frame, (other.participant_id,))) + + return sorted(candidates, key=lambda row: (row.anchor_dense_frame, row.candidate_source, row.uid)) -def stratified_limit( - rows: Sequence[ProtocolCandidate], - limit: int | None, - *, - seed: int, -) -> list[ProtocolCandidate]: +def stratified_limit(rows: Sequence[ProtocolCandidate], limit: int | None, *, seed: int) -> list[ProtocolCandidate]: if limit is None or limit >= len(rows): return list(rows) if limit < 0: raise ValueError("--max-candidates must be non-negative") rng = random.Random(seed) - buckets: dict[tuple[str, str, str], list[ProtocolCandidate]] = defaultdict(list) + buckets: dict[tuple[str, str], list[ProtocolCandidate]] = defaultdict(list) for row in rows: - buckets[(row.task, row.automatic_protocol_label, row.automatic_hard_negative_type)].append(row) - keys = sorted(buckets) - for key in keys: - rng.shuffle(buckets[key]) - rng.shuffle(keys) + buckets[(row.session, row.candidate_source)].append(row) + for bucket in buckets.values(): + rng.shuffle(bucket) selected: list[ProtocolCandidate] = [] - while len(selected) < limit: - progressed = False - for key in keys: - if buckets[key]: + while len(selected) < limit and any(buckets.values()): + for key in sorted(buckets): + if buckets[key] and len(selected) < limit: selected.append(buckets[key].pop()) - progressed = True - if len(selected) == limit: - break - if not progressed: - break - return sorted(selected, key=lambda row: (row.session, row.anchor_dense_frame, row.task)) + return sorted(selected, key=lambda row: (row.session, row.anchor_dense_frame, row.candidate_source)) def _write_jsonl(path: Path, rows: Iterable[ProtocolCandidate]) -> None: @@ -473,20 +431,13 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--sessions", nargs="+", default=list(ALL_SESSIONS)) parser.add_argument("--camera-wearer-id", type=int, default=CAMERA_WEARER_ID) - parser.add_argument("--window-seconds", type=float, default=DEFAULT_WINDOW_SECONDS) parser.add_argument("--context-seconds", type=float, default=3.0) - parser.add_argument( - "--tasks", - nargs="+", - choices=("eot_response", "int_yield"), - default=("eot_response", "int_yield"), - ) - parser.add_argument( - "--automatic-labels", - nargs="+", - choices=("positive", "hard_negative", "exclude"), - default=("positive", "hard_negative", "exclude"), - ) + 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("--short-vocalization-max-seconds", type=float, default=1.0) + parser.add_argument("--continuation-min-seconds", type=float, default=0.4) + parser.add_argument("--candidate-sources", nargs="+") parser.add_argument("--max-candidates", type=int) parser.add_argument("--selection-seed", type=int, default=2027) return parser.parse_args() @@ -497,74 +448,59 @@ def main() -> None: unknown = sorted(set(args.sessions) - set(ALL_SESSIONS)) if unknown: raise ValueError(f"Unknown EasyCom sessions: {unknown}") - dataset_root = args.dataset_root.expanduser().resolve() - output_dir = args.output_dir.expanduser().resolve() + config = CandidateConfig( + context_seconds=args.context_seconds, + relation_window_seconds=args.relation_window_seconds, + same_speaker_pause_max_seconds=args.same_speaker_pause_max_seconds, + other_handover_max_seconds=args.other_handover_max_seconds, + short_vocalization_max_seconds=args.short_vocalization_max_seconds, + continuation_min_seconds=args.continuation_min_seconds, + ) + dataset_root, output_dir = args.dataset_root.expanduser().resolve(), args.output_dir.expanduser().resolve() all_rows: list[ProtocolCandidate] = [] + session_timelines: dict[str, list[dict[str, Any]]] = {} rules = Rules() for session in args.sessions: clips = build_clips(dataset_root, session, rules) utterances = load_utterances(session, clips) - rows = derive_protocol_candidates( - utterances, - clips, - session=session, - wearer_id=args.camera_wearer_id, - window_seconds=args.window_seconds, - context_seconds=args.context_seconds, - ) + session_timelines[session] = [{ + "uid": row.uid, + "participant_id": row.participant_id, + "participant_role": "wearer" if row.participant_id == args.camera_wearer_id else "other", + "start_seconds": row.dense_start_seconds, + "end_seconds": row.dense_end_seconds, + } for row in utterances] + rows = derive_protocol_candidates(utterances, clips, session=session, + wearer_id=args.camera_wearer_id, config=config) all_rows.extend(rows) - counts = Counter((row.task, row.automatic_protocol_label) for row in rows) - print(f"{session}: candidates={len(rows)} counts={dict(sorted(counts.items()))}") - - eligible = [ - row - for row in all_rows - if row.task in set(args.tasks) - and row.automatic_protocol_label in set(args.automatic_labels) - ] + print(f"{session}: candidates={len(rows)} sources={dict(sorted(Counter(row.candidate_source for row in rows).items()))}") + sources = None if args.candidate_sources is None else set(args.candidate_sources) + eligible = [row for row in all_rows if sources is None or row.candidate_source in sources] selected = stratified_limit(eligible, args.max_candidates, seed=args.selection_seed) output_path = output_dir / "easycom_eot_int_candidates.jsonl" _write_jsonl(output_path, selected) - counts = Counter( - (row.split, row.task, row.automatic_protocol_label, row.automatic_hard_negative_type) - for row in selected - ) + counts = Counter((row.split, row.candidate_source, row.rule_candidate_type) for row in selected) summary = { - "schema_version": SCHEMA_VERSION, - "protocol_version": PROTOCOL_VERSION, - "dataset": "EasyCom", - "semantic_warning": ( - "Automatic labels are timing candidates for independent human audit; " - "they are not semantic EOT/INT ground truth." - ), - "sessions": list(args.sessions), - "camera_wearer_id": args.camera_wearer_id, - "window_seconds": args.window_seconds, - "context_seconds": args.context_seconds, - "selection_seed": args.selection_seed, + "schema_version": SCHEMA_VERSION, "protocol_version": PROTOCOL_VERSION, "dataset": "EasyCom", + "semantic_warning": "Rules create candidates only; AI labels need human review and never modify dense activity ground truth.", + "sessions": list(args.sessions), "camera_wearer_id": args.camera_wearer_id, + "candidate_config": asdict(config), "selection_seed": args.selection_seed, "max_candidates": args.max_candidates, - "tasks": list(args.tasks), - "automatic_labels": list(args.automatic_labels), - "num_candidates_before_filter": len(all_rows), - "num_candidates_before_limit": len(eligible), + "candidate_sources": None if sources is None else sorted(sources), + "num_candidates_before_filter": len(all_rows), "num_candidates_before_limit": len(eligible), "num_candidates_written": len(selected), - "counts": [ - { - "split": key[0], - "task": key[1], - "automatic_protocol_label": key[2], - "automatic_hard_negative_type": key[3], - "count": value, - } - for key, value in sorted(counts.items()) - ], + "counts": [{"split": key[0], "candidate_source": key[1], + "rule_candidate_type": key[2], "count": value} + for key, value in sorted(counts.items())], } output_dir.mkdir(parents=True, exist_ok=True) (output_dir / "summary.json").write_text( - json.dumps(summary, indent=2, ensure_ascii=False) + "\n", + json.dumps(summary, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + (output_dir / "easycom_session_timelines.json").write_text( + json.dumps({"schema_version": 1, "sessions": session_timelines}, ensure_ascii=False) + "\n", encoding="utf-8", ) - print(f"Saved {len(selected)} manuscript-aligned review candidates to {output_path}") + print(f"Saved {len(selected)} wearer-centric review candidates to {output_path}") if __name__ == "__main__": diff --git a/review_floor_events.py b/review_floor_events.py index 573d4c6..2e67958 100644 --- a/review_floor_events.py +++ b/review_floor_events.py @@ -32,7 +32,7 @@ SOURCE_FPS = 20.0 STATIC_ROOT = Path(__file__).resolve().parent / "floor_event_review_web" VIDEO_ROOT_PARTS = ("Main", "Video_Compressed") TRANSCRIPT_ROOT_PARTS = ("Main", "Speech_Transcriptions") -REVIEW_SCHEMA_VERSION = 2 +REVIEW_SCHEMA_VERSION = 3 MAX_REQUEST_BYTES = 128 * 1024 PROTOCOL_DISPOSITION_VALUES = {"positive", "hard_negative", "exclude", "uncertain"} @@ -67,6 +67,18 @@ OFFSET_OUTCOME_VALUES = { "uncertain", } CONFIDENCE_VALUES = {"high", "medium", "low"} +HUMAN_ACTION_VALUES = {"accept", "relabel", "exclude", "second_review"} +HUMAN_VERDICT_VALUES = {"POSITIVE", "HARD_NEGATIVE", "NOT_APPLICABLE", "EXCLUDE", "UNSURE"} +HUMAN_EVENT_TYPE_VALUES = {"EOT_RESPONSE", "INT_YIELD"} +HUMAN_SUBTYPE_VALUES = { + "EOT_MIDTURN_PAUSE", "EOT_OTHER_TO_OTHER", "EOT_WEARER_BACKCHANNEL", + "INT_OTHER_BACKCHANNEL", "INT_NON_TAKEOVER_OVERLAP", "INT_WEARER_HOLDS_FLOOR", + "OTHER_HARD_NEGATIVE", +} +EXCLUDE_REASON_VALUES = { + "ANNOTATION_ERROR", "RECORDING_BOUNDARY", "MISSING_CONTEXT", "NON_SPEECH_ARTIFACT", + "TIMESTAMP_CORRUPTION", "UNRESOLVABLE", +} @dataclass(frozen=True) @@ -95,7 +107,8 @@ class Candidate: @property def automatic_label(self) -> str: return str( - self.payload.get("event_type") + self.payload.get("candidate_source") + or self.payload.get("event_type") or self.payload.get("category") or self.payload.get("proxy_group") or "unclassified" @@ -108,6 +121,8 @@ class Candidate: @property def family(self) -> str: + if self.payload.get("candidate_source"): + return "eot" if self.task == "eot_response" else "int" if self.task == "eot_response": return "eot" if self.task == "int_yield": @@ -233,6 +248,16 @@ class CandidateRepository: 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}): + if not path.is_file(): + continue + payload = json.loads(path.read_text(encoding="utf-8")) + sessions = payload.get("sessions") if isinstance(payload, dict) else None + if isinstance(sessions, dict): + for session, rows in sessions.items(): + if isinstance(session, str) and isinstance(rows, list): + self.session_timelines[session] = [row for row in rows if isinstance(row, dict)] def _transcript(self, session: str, clip_stem: str) -> list[dict[str, Any]]: key = (session, clip_stem) @@ -255,7 +280,9 @@ class CandidateRepository: row = candidate.payload clip_stem = row.get("clip_stem") transcript_index: int | None = None - uid = _uid_parts(row.get("uid")) or _uid_parts(row.get("holder_uid")) + uid = None if row.get("candidate_source") else ( + _uid_parts(row.get("uid")) or _uid_parts(row.get("holder_uid")) + ) if uid is not None: uid_session, uid_stem, transcript_index = uid if uid_session == candidate.session: @@ -291,8 +318,10 @@ class CandidateRepository: def summary(self, candidate: Candidate, reviewed: bool) -> dict[str, Any]: row = candidate.payload event_time = row.get( - "anchor_dense_seconds", + "event_time", + row.get("anchor_dense_seconds", row.get("dense_start_seconds", row.get("boundary_seconds")), + ), ) return { "candidate_id": candidate.candidate_id, @@ -306,6 +335,11 @@ class CandidateRepository: "automatic_hard_negative_type": row.get("automatic_hard_negative_type"), "automatic_confidence": row.get("automatic_confidence", row.get("confidence")), "llm_suggestion": row.get("llm_suggestion"), + "candidate_source": row.get("candidate_source"), + "rule_candidate_type": row.get("rule_candidate_type"), + "ai_verdict": row.get("ai_verdict"), + "ai_event_type": row.get("ai_event_type"), + "ai_subtype": row.get("ai_subtype"), "participant_role": row.get("participant_role") or row.get("holder_role"), "event_time_seconds": event_time, "reviewed": reviewed, @@ -389,15 +423,11 @@ class ReviewStore: if not isinstance(payload, dict): raise ValueError("review must be an object") - required_choices = { - "protocol_disposition": PROTOCOL_DISPOSITION_VALUES, - "hard_negative_type": HARD_NEGATIVE_TYPE_VALUES, - "validity": VALIDITY_VALUES, - "onset_function": ONSET_FUNCTION_VALUES, - "context": CONTEXT_VALUES, - "offset_outcome": OFFSET_OUTCOME_VALUES, - "confidence": CONFIDENCE_VALUES, - } + # Schema v3 deliberately keeps the common path small. Legacy v2 review + # payloads remain accepted so existing review files are not stranded. + if "human_action" not in payload: + return self._validated_legacy_review(payload) + required_choices = {"human_action": HUMAN_ACTION_VALUES, "confidence": CONFIDENCE_VALUES} normalized: dict[str, Any] = {} for key, allowed in required_choices.items(): value = payload.get(key) @@ -405,15 +435,48 @@ class ReviewStore: raise ValueError(f"Invalid {key}: {value!r}") normalized[key] = value - for key in ( - "anchor_adjustment_bins", - "behavior_adjustment_bins", - "region_end_adjustment_bins", - ): - adjustment = payload.get(key) - if adjustment not in {-1, 0, 1, None}: - raise ValueError(f"{key} must be -1, 0, 1, or null") - normalized[key] = adjustment + action = normalized["human_action"] + candidate = self.candidates[candidate_id].payload + if action == "accept": + verdict = candidate.get("ai_verdict") or (candidate.get("llm_suggestion") or {}).get("verdict") + event_type = candidate.get("ai_event_type") or (candidate.get("llm_suggestion") or {}).get("event_type") + subtype = candidate.get("ai_subtype") or (candidate.get("llm_suggestion") or {}).get("subtype") + if verdict not in HUMAN_VERDICT_VALUES: + raise ValueError("Accept requires an AI pre-label; use Relabel when no AI label is available") + elif action == "exclude": + verdict, event_type, subtype = "EXCLUDE", None, None + elif action == "second_review": + verdict, event_type, subtype = "UNSURE", None, None + else: + verdict, event_type, subtype = payload.get("human_verdict"), payload.get("human_event_type"), payload.get("human_subtype") + if verdict not in HUMAN_VERDICT_VALUES: + raise ValueError(f"Invalid human_verdict: {verdict!r}") + if verdict == "POSITIVE" and event_type not in HUMAN_EVENT_TYPE_VALUES: + raise ValueError("POSITIVE requires EOT_RESPONSE or INT_YIELD") + if verdict != "POSITIVE" and event_type is not None: + raise ValueError("Only POSITIVE may have a human_event_type") + if verdict == "HARD_NEGATIVE" and subtype not in HUMAN_SUBTYPE_VALUES: + raise ValueError("HARD_NEGATIVE requires a valid human_subtype") + if verdict != "HARD_NEGATIVE" and subtype is not None: + raise ValueError("Only HARD_NEGATIVE may have a human_subtype") + exclude_reason = payload.get("exclude_reason") + if action == "exclude" and exclude_reason not in EXCLUDE_REASON_VALUES: + raise ValueError("Exclude requires an exclude_reason") + if action != "exclude" and exclude_reason is not None: + raise ValueError("exclude_reason is only valid for Exclude") + correction = payload.get("boundary_correction_seconds") + if correction not in {-0.2, 0, 0.2, None}: + raise ValueError("boundary_correction_seconds must be -0.2, 0, 0.2, or null") + normalized.update( + human_status="SECOND_REVIEW" if action == "second_review" else "REVIEWED", + human_verdict=verdict, + human_event_type=event_type, + human_subtype=subtype, + exclude_reason=exclude_reason, + relabeled=action == "relabel", + boundary_correction_seconds=correction, + original_event_time=candidate.get("event_time", candidate.get("anchor_dense_seconds")), + ) reviewer = str(payload.get("reviewer", "")).strip() notes = str(payload.get("notes", "")).strip() @@ -427,6 +490,31 @@ class ReviewStore: normalized["notes"] = notes return normalized + def _validated_legacy_review(self, payload: dict[str, Any]) -> dict[str, Any]: + required_choices = { + "protocol_disposition": PROTOCOL_DISPOSITION_VALUES, + "hard_negative_type": HARD_NEGATIVE_TYPE_VALUES, + "validity": VALIDITY_VALUES, "onset_function": ONSET_FUNCTION_VALUES, + "context": CONTEXT_VALUES, "offset_outcome": OFFSET_OUTCOME_VALUES, + "confidence": CONFIDENCE_VALUES, + } + normalized: dict[str, Any] = {} + for key, allowed in required_choices.items(): + value = payload.get(key) + if value not in allowed: + raise ValueError(f"Invalid {key}: {value!r}") + normalized[key] = value + for key in ("anchor_adjustment_bins", "behavior_adjustment_bins", "region_end_adjustment_bins"): + value = payload.get(key) + if value not in {-1, 0, 1, None}: + raise ValueError(f"{key} must be -1, 0, 1, or null") + normalized[key] = value + reviewer, notes = str(payload.get("reviewer", "")).strip(), str(payload.get("notes", "")).strip() + if not reviewer or len(reviewer) > 100 or len(notes) > 4000: + raise ValueError("invalid reviewer or notes") + normalized.update(reviewer=reviewer, notes=notes) + return normalized + def save(self, candidate_id: str, review: Any) -> dict[str, Any]: normalized = self._validated_review(candidate_id, review) candidate = self.candidates[candidate_id] @@ -445,6 +533,9 @@ class ReviewStore: "automatic_hard_negative_type": candidate.payload.get( "automatic_hard_negative_type" ), + "ai_verdict": candidate.payload.get("ai_verdict"), + "ai_event_type": candidate.payload.get("ai_event_type"), + "ai_subtype": candidate.payload.get("ai_subtype"), "review": normalized, "updated_at": datetime.now(timezone.utc).isoformat(), "candidate": candidate.payload, @@ -497,15 +588,9 @@ class ReviewStore: "automatic_label", "automatic_protocol_label", "automatic_hard_negative_type", - "protocol_disposition", - "hard_negative_type", - "validity", - "onset_function", - "context", - "offset_outcome", - "anchor_adjustment_bins", - "behavior_adjustment_bins", - "region_end_adjustment_bins", + "ai_verdict", "ai_event_type", "ai_subtype", + "human_action", "human_status", "human_verdict", "human_event_type", "human_subtype", + "exclude_reason", "relabeled", "boundary_correction_seconds", "original_event_time", "confidence", "reviewer", "notes", @@ -621,6 +706,13 @@ class ReviewRequestHandler(BaseHTTPRequestHandler): ) self._json(detail) return + if parsed.path == "/api/session-timeline": + session = query.get("session", [""])[0] + if session not in self.server.repository.session_timelines: + self._error("Session timeline unavailable", HTTPStatus.NOT_FOUND) + return + self._json({"session": session, "speech": self.server.repository.session_timelines[session]}) + return if parsed.path == "/api/export.jsonl": self._send_download(self.server.store.output_path, "application/x-ndjson") return diff --git a/test_adjudicate_easycom_eot_int_candidates.py b/test_adjudicate_easycom_eot_int_candidates.py index 3109897..890f744 100644 --- a/test_adjudicate_easycom_eot_int_candidates.py +++ b/test_adjudicate_easycom_eot_int_candidates.py @@ -3,17 +3,25 @@ from __future__ import annotations import unittest +from pathlib import Path +import json +import tempfile -from adjudicate_easycom_eot_int_candidates import build_user_prompt, validate_decisions +from adjudicate_easycom_eot_int_candidates import ( + _transport_rows, build_user_prompt, read_candidates, validate_decisions, +) class EOTINTAdjudicationTests(unittest.TestCase): def test_prompt_contains_task_and_context(self) -> None: prompt = build_user_prompt( - [{"uid": "u1", "task": "eot_response", "context": [{"transcription": "Yes."}]}], + [{"uid": "u1", "candidate_source": "WEARER_ONSET", "task": "eot_response", + "temporal_evidence": {"wearer_duration": 2.0}, + "context": [{"transcription": "Yes."}]}], 0, ) - self.assertIn("eot_response", prompt) + self.assertIn("WEARER_ONSET", prompt) + self.assertIn("wearer_duration", prompt) self.assertIn("Yes.", prompt) def test_decisions_require_exact_uid_and_vocabulary(self) -> None: @@ -21,18 +29,56 @@ class EOTINTAdjudicationTests(unittest.TestCase): "items": [ { "uid": "u1", - "protocol_disposition": "hard_negative", - "hard_negative_type": "int_backchannel", - "confidence": "high", - "rationale": "Brief feedback while the wearer continues.", + "verdict": "HARD_NEGATIVE", + "event_type": None, + "subtype": "INT_OTHER_BACKCHANNEL", + "confidence": 0.91, + "reason": "Brief feedback while the wearer continues.", + "needs_human_review": False, } ] } decisions = validate_decisions(payload, ["u1"]) - self.assertEqual(decisions[0]["hard_negative_type"], "int_backchannel") + self.assertEqual(decisions[0]["subtype"], "INT_OTHER_BACKCHANNEL") with self.assertRaisesRegex(ValueError, "mismatch"): validate_decisions({"items": []}, ["u1"]) + def test_markdown_escaped_uid_is_normalized_but_still_unique(self) -> None: + item = { + "uid": "Session\\_4/wearer\\_onset/00000024", + "verdict": "POSITIVE", + "event_type": "EOT_RESPONSE", + "subtype": None, + "confidence": 0.9, + "reason": "Sustained wearer response after the other speaker stops.", + "needs_human_review": False, + } + expected = "Session_4/wearer_onset/00000024" + decision = validate_decisions({"items": [item]}, [expected])[0] + self.assertEqual(decision["uid"], expected) + with self.assertRaisesRegex(ValueError, "duplicate"): + validate_decisions({"items": [item, {**item, "uid": expected}]}, [expected]) + + def test_duplicate_rule_uids_are_disambiguated_with_provenance(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary) / "candidates.jsonl" + base = {"uid": "Session_4/other_to_other/00003600", "task": "eot_response"} + rows = [ + {**base, "previous_other_participant_ids": [6], "incoming_other_participant_ids": [4]}, + {**base, "previous_other_participant_ids": [3], "incoming_other_participant_ids": [6]}, + ] + path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") + loaded = read_candidates(path) + self.assertEqual(len({row["uid"] for row in loaded}), 2) + self.assertTrue(all(row["uid_disambiguated_from"] == base["uid"] for row in loaded)) + + def test_transport_ids_are_short_unique_and_reversible(self) -> None: + rows = [{"uid": "Session_4/other_to_other/00003600/a"}, + {"uid": "Session_4/other_to_other/00003600/b"}] + transported, mapping = _transport_rows(rows) + self.assertEqual([row["uid"] for row in transported], ["item-0000", "item-0001"]) + self.assertEqual(mapping["item-0001"], rows[1]["uid"]) + if __name__ == "__main__": unittest.main() diff --git a/test_prepare_easycom_eot_int_review.py b/test_prepare_easycom_eot_int_review.py index 192be66..41a5373 100644 --- a/test_prepare_easycom_eot_int_review.py +++ b/test_prepare_easycom_eot_int_review.py @@ -50,17 +50,17 @@ class EOTINTReviewCandidateTests(unittest.TestCase): def derive(self, rows: list[Utterance]): return derive_protocol_candidates(rows, [clip()], session="Session_4") - def test_eot_response_is_anchored_by_wearer_onset(self) -> None: + def test_every_wearer_onset_is_an_unresolved_entry_candidate(self) -> None: rows = [ utterance("other", 4, 0.0, 1.0, "Your turn."), utterance("wearer", 2, 1.2, 2.0, "My response."), ] - event = next(row for row in self.derive(rows) if row.task == "eot_response") - self.assertEqual(event.automatic_protocol_label, "positive") - self.assertEqual(event.anchor_type, "other_offset") - self.assertEqual(event.behavior_anchor_type, "wearer_onset") - self.assertAlmostEqual(event.anchor_dense_seconds, 1.0) - self.assertAlmostEqual(event.behavior_seconds or -1, 1.2) + event = next(row for row in self.derive(rows) if row.candidate_source == "WEARER_ONSET") + self.assertEqual(event.automatic_protocol_label, "unsure") + self.assertEqual(event.rule_candidate_type, "ENTRY_CANDIDATE") + self.assertEqual(event.anchor_type, "wearer_onset") + self.assertAlmostEqual(event.event_time, 1.2) + self.assertAlmostEqual(event.temporal_evidence["wearer_speech_duration_after_onset_seconds"], 0.8) def test_same_other_resume_is_eot_midturn_hard_negative(self) -> None: rows = [ @@ -70,11 +70,11 @@ class EOTINTReviewCandidateTests(unittest.TestCase): event = next( row for row in self.derive(rows) - if row.task == "eot_response" and row.anchor_dense_seconds == 1.0 + if row.candidate_source == "SAME_SPEAKER_PAUSE" and row.anchor_dense_seconds == 1.0 ) self.assertEqual(event.automatic_protocol_label, "hard_negative") - self.assertEqual(event.automatic_hard_negative_type, "eot_midturn_pause") - self.assertEqual(event.region_end_reason, "competing_other_onset") + self.assertEqual(event.automatic_hard_negative_type, "EOT_MIDTURN_PAUSE") + self.assertAlmostEqual(event.temporal_evidence["resume_delay_seconds"], 0.3) def test_other_to_other_transfer_is_eot_hard_negative(self) -> None: rows = [ @@ -84,31 +84,44 @@ class EOTINTReviewCandidateTests(unittest.TestCase): event = next( row for row in self.derive(rows) - if row.task == "eot_response" and row.anchor_dense_seconds == 1.0 + if row.candidate_source == "OTHER_TO_OTHER" and row.anchor_dense_seconds == 1.0 ) self.assertEqual(event.automatic_protocol_label, "hard_negative") - self.assertEqual(event.automatic_hard_negative_type, "eot_other_to_other") + self.assertEqual(event.automatic_hard_negative_type, "EOT_OTHER_TO_OTHER") + self.assertNotEqual( + event.temporal_evidence["previous_speaker_id"], + event.temporal_evidence["incoming_speaker_id"], + ) - def test_int_yield_is_anchored_by_wearer_offset(self) -> None: + def test_every_wearer_offset_is_an_unresolved_yield_candidate(self) -> None: rows = [ utterance("wearer", 2, 0.0, 2.0, "Wearer is speaking."), utterance("other", 4, 1.0, 3.0, "I will take over now."), ] - event = next(row for row in self.derive(rows) if row.task == "int_yield") - self.assertEqual(event.automatic_protocol_label, "positive") - self.assertEqual(event.anchor_type, "other_onset") - self.assertEqual(event.behavior_anchor_type, "wearer_offset") - self.assertAlmostEqual(event.behavior_seconds or -1, 2.0) + event = next(row for row in self.derive(rows) if row.candidate_source == "WEARER_OFFSET") + self.assertEqual(event.automatic_protocol_label, "unsure") + self.assertEqual(event.rule_candidate_type, "YIELD_CANDIDATE") + self.assertEqual(event.anchor_type, "wearer_offset") + self.assertAlmostEqual(event.event_time, 2.0) def test_brief_feedback_is_int_backchannel_hard_negative(self) -> None: rows = [ utterance("wearer", 2, 0.0, 3.0, "Wearer keeps the floor."), utterance("other", 4, 1.0, 1.4, "Yeah."), ] - event = next(row for row in self.derive(rows) if row.task == "int_yield") + event = next(row for row in self.derive(rows) if row.candidate_source == "OTHER_BACKCHANNEL") self.assertEqual(event.automatic_protocol_label, "hard_negative") - self.assertEqual(event.automatic_hard_negative_type, "int_backchannel") - self.assertEqual(event.region_end_reason, "competing_other_offset") + self.assertEqual(event.automatic_hard_negative_type, "INT_OTHER_BACKCHANNEL") + self.assertTrue(event.temporal_evidence["wearer_active_through_region"]) + + def test_non_takeover_overlap_is_scanned_without_wearer_transition(self) -> None: + rows = [ + utterance("wearer", 2, 0.0, 4.0, "Long wearer turn."), + utterance("other", 4, 1.0, 2.0, "Long overlapping contribution."), + ] + event = next(row for row in self.derive(rows) if row.candidate_source == "NON_TAKEOVER_OVERLAP") + self.assertEqual(event.automatic_hard_negative_type, "INT_NON_TAKEOVER_OVERLAP") + self.assertAlmostEqual(event.region_start_seconds, 1.0) if __name__ == "__main__": diff --git a/test_review_floor_events.py b/test_review_floor_events.py index eff825a..2e03ff1 100644 --- a/test_review_floor_events.py +++ b/test_review_floor_events.py @@ -56,11 +56,23 @@ class FloorEventReviewTests(unittest.TestCase): "clip_stem": "00-00-000", "source_start_frame": 25, "dense_start_seconds": 1.2, + "event_time": 1.2, + "candidate_source": "WEARER_ONSET", + "ai_verdict": "POSITIVE", + "ai_event_type": "EOT_RESPONSE", + "ai_subtype": None, } ) + "\n", encoding="utf-8", ) + (root / "easycom_session_timelines.json").write_text( + json.dumps({"schema_version": 1, "sessions": {"Session_4": [ + {"participant_id": 4, "participant_role": "other", "start_seconds": 0.0, "end_seconds": 1.0}, + {"participant_id": 2, "participant_role": "wearer", "start_seconds": 1.2, "end_seconds": 2.0}, + ]}}), + encoding="utf-8", + ) return dataset, events def test_candidate_detail_resolves_media_and_transcript(self) -> None: @@ -79,6 +91,7 @@ class FloorEventReviewTests(unittest.TestCase): self.assertEqual(detail["event_local_seconds"], 1.2) self.assertEqual(len(detail["transcript"]), 2) self.assertTrue(detail["transcript"][1]["is_anchor"]) + self.assertEqual(len(repository.session_timelines["Session_4"]), 2) def test_review_store_writes_resumable_jsonl_and_csv(self) -> None: with tempfile.TemporaryDirectory() as temporary: @@ -88,15 +101,12 @@ class FloorEventReviewTests(unittest.TestCase): output = root / "reviews" / "reviews.jsonl" store = ReviewStore(output, candidates) review = { - "protocol_disposition": "positive", - "hard_negative_type": "not_applicable", - "validity": "valid", - "onset_function": "backchannel", - "context": "addressed_wearer", - "offset_outcome": "not_applicable", - "anchor_adjustment_bins": 0, - "behavior_adjustment_bins": 0, - "region_end_adjustment_bins": None, + "human_action": "accept", + "human_verdict": None, + "human_event_type": None, + "human_subtype": None, + "exclude_reason": None, + "boundary_correction_seconds": None, "confidence": "high", "reviewer": "r1", "notes": "clear acknowledgement", @@ -106,9 +116,10 @@ class FloorEventReviewTests(unittest.TestCase): self.assertTrue(output.with_suffix(".csv").is_file()) resumed = ReviewStore(output, candidates) self.assertEqual( - resumed.reviews[candidates[0].candidate_id]["review"]["onset_function"], - "backchannel", + resumed.reviews[candidates[0].candidate_id]["review"]["human_event_type"], + "EOT_RESPONSE", ) + self.assertFalse(resumed.reviews[candidates[0].candidate_id]["review"]["relabeled"]) def test_directory_expansion_excludes_output(self) -> None: with tempfile.TemporaryDirectory() as temporary: @@ -122,6 +133,27 @@ class FloorEventReviewTests(unittest.TestCase): [candidate.resolve()], ) + def test_relabel_preserves_ai_and_records_human_decision(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + _, events = self._fixture(root) + candidates = load_candidates([events]) + store = ReviewStore(root / "reviews.jsonl", candidates) + record = store.save(candidates[0].candidate_id, { + "human_action": "relabel", + "human_verdict": "HARD_NEGATIVE", + "human_event_type": None, + "human_subtype": "EOT_WEARER_BACKCHANNEL", + "exclude_reason": None, + "boundary_correction_seconds": None, + "confidence": "medium", + "reviewer": "r2", + "notes": "brief acknowledgement", + }) + self.assertEqual(record["ai_event_type"], "EOT_RESPONSE") + self.assertEqual(record["review"]["human_subtype"], "EOT_WEARER_BACKCHANNEL") + self.assertTrue(record["review"]["relabeled"]) + def test_byte_ranges(self) -> None: self.assertEqual(parse_byte_range("bytes=2-5", 10), (2, 5)) self.assertEqual(parse_byte_range("bytes=7-", 10), (7, 9))