From 6f5115f2999b128e820f1976838b7e33636680f5 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Sun, 30 Aug 2026 18:26:15 +0800 Subject: [PATCH] Evaluate wearer floor changes by event --- evaluate_floor_state_changes.py | 376 ++++++++++++++++++++++++++- test_evaluate_floor_state_changes.py | 62 +++++ 2 files changed, 430 insertions(+), 8 deletions(-) diff --git a/evaluate_floor_state_changes.py b/evaluate_floor_state_changes.py index 77eb481..afde477 100644 --- a/evaluate_floor_state_changes.py +++ b/evaluate_floor_state_changes.py @@ -80,7 +80,21 @@ def parse_args() -> argparse.Namespace: "--csv-output", type=Path, default=None, - help="Optional flat per-horizon CSV (default: OUTPUT with .csv suffix).", + help="Optional event-level CSV path (default: OUTPUT stem + _events.csv).", + ) + parser.add_argument( + "--midturn-pause-max-seconds", + type=float, + default=2.0, + help=( + "Maximum aggregate-others silent gap used only to identify the " + "short-resume hard-negative subgroup (default: 2.0)." + ), + ) + parser.add_argument( + "--print-framewise", + action="store_true", + help="Also print the older endpoint/framewise diagnostic table.", ) return parser.parse_args() @@ -360,6 +374,199 @@ def evaluate_horizon( } +def _same_source_path_score( + prediction: np.ndarray, + valid: np.ndarray, + session_ids: np.ndarray, + source_row: int, + horizon_steps: int, + *, + offset: bool, +) -> float | None: + """Pool fixed-source future activity scores through one horizon. + + ``prediction[source+k, :, k-1]`` is the score produced from the same + source row for the target ``k`` steps later. Max pooling asks whether the + behavior occurs anywhere in the event response window; it is a ranking + score, not a calibrated event probability. + """ + + target_rows = source_row + np.arange(1, horizon_steps + 1, dtype=np.int64) + if source_row < 0 or target_rows[-1] >= len(prediction): + return None + if np.any(session_ids[target_rows] != session_ids[source_row]): + return None + horizon_indices = np.arange(horizon_steps, dtype=np.int64) + if not valid[target_rows, horizon_indices].all(): + return None + scores = prediction[target_rows, WEARER_INDEX, horizon_indices] + if offset: + scores = 1.0 - scores + return float(np.max(scores)) + + +def _has_transition_within( + values: np.ndarray, + source_row: int, + horizon_steps: int, + *, + rising: bool, +) -> bool: + window = values[source_row : source_row + horizon_steps + 1] + before = window[:-1] + after = window[1:] + transitions = (~before & after) if rising else (before & ~after) + return bool(transitions.any()) + + +def _has_others_resume_without_wearer_entry( + others: np.ndarray, + wearer: np.ndarray, + session_ids: np.ndarray, + source_row: int, + max_pause_steps: int, +) -> bool: + # The offset is at source+1, so allow max_pause_steps after that boundary. + last = min(len(others) - 1, source_row + 1 + max_pause_steps) + for row in range(source_row + 2, last + 1): + if session_ids[row] != session_ids[source_row]: + break + if not wearer[row - 1] and wearer[row]: + return False + if not others[row - 1] and others[row]: + return True + return False + + +def _event_metric( + scores: list[float], + labels: list[bool], + recalls: list[float], + *, + extra_negative_groups: dict[str, list[bool]] | None = None, +) -> dict[str, Any]: + score = np.asarray(scores, dtype=np.float64) + positive = np.asarray(labels, dtype=bool) + negative = ~positive + metric = _ranking_from_masks(score, positive, negative) + masks = { + name: np.asarray(values, dtype=bool) & negative + for name, values in (extra_negative_groups or {}).items() + } + metric["operating_points"] = _operating_points( + score, + positive, + negative, + recalls, + extra_groups=masks, + ) + for name, mask in masks.items(): + metric[f"num_{name}"] = int(mask.sum()) + return metric + + +def evaluate_event_horizon( + target: np.ndarray, + prediction: np.ndarray, + valid: np.ndarray, + session_ids: np.ndarray, + horizon_index: int, + recalls: list[float], + *, + midturn_pause_max_steps: int, +) -> dict[str, Any]: + """Evaluate each aggregate-other boundary once. + + Other offset events ask whether an inactive wearer starts within the + horizon. Other onset events ask whether an active wearer stops within the + horizon. The labels remain observed-behavior proxies: aggregate activity + cannot establish addressee, speaker identity, or semantic EOT/INT. + """ + + horizon_steps = horizon_index + 1 + wearer = target[:, WEARER_INDEX].astype(bool) + others = target[:, OTHERS_INDEX].astype(bool) + + response_scores: list[float] = [] + response_labels: list[bool] = [] + short_resume_groups: list[bool] = [] + offset_sources = np.flatnonzero(others[:-1] & ~others[1:]) + for source_row in offset_sources: + if wearer[source_row]: + continue + score = _same_source_path_score( + prediction, + valid, + session_ids, + int(source_row), + horizon_steps, + offset=False, + ) + if score is None: + continue + response = _has_transition_within( + wearer, + int(source_row), + horizon_steps, + rising=True, + ) + response_scores.append(score) + response_labels.append(response) + short_resume_groups.append( + (not response) + and _has_others_resume_without_wearer_entry( + others, + wearer, + session_ids, + int(source_row), + midturn_pause_max_steps, + ) + ) + + yield_scores: list[float] = [] + yield_labels: list[bool] = [] + onset_sources = np.flatnonzero(~others[:-1] & others[1:]) + for source_row in onset_sources: + if not wearer[source_row]: + continue + score = _same_source_path_score( + prediction, + valid, + session_ids, + int(source_row), + horizon_steps, + offset=True, + ) + if score is None: + continue + yielded = _has_transition_within( + wearer, + int(source_row), + horizon_steps, + rising=False, + ) + yield_scores.append(score) + yield_labels.append(yielded) + + return { + "horizon_index": horizon_steps, + "nominal_seconds": horizon_steps * STEP_SECONDS, + "other_offset_to_wearer_response_event": _event_metric( + response_scores, + response_labels, + recalls, + extra_negative_groups={ + "short_resume_hard_negative": short_resume_groups, + }, + ), + "other_onset_to_wearer_yield_event": _event_metric( + yield_scores, + yield_labels, + recalls, + ), + } + + def _mean_available(items: Iterable[float | None]) -> float | None: values = [float(item) for item in items if item is not None] return float(np.mean(values)) if values else None @@ -386,6 +593,21 @@ def summarize_horizons(horizons: list[dict[str, Any]]) -> dict[str, float | None } +def summarize_event_horizons( + horizons: list[dict[str, Any]], +) -> dict[str, float | None]: + return { + "mean_other_offset_response_event_ap": _mean_available( + item["other_offset_to_wearer_response_event"]["average_precision"] + for item in horizons + ), + "mean_other_onset_yield_event_ap": _mean_available( + item["other_onset_to_wearer_yield_event"]["average_precision"] + for item in horizons + ), + } + + def _percentage(value: float | None) -> str: return "n/a" if value is None else f"{value * 100:.3f}" @@ -399,11 +621,50 @@ def _operating_point( return None -def print_results(payload: dict[str, Any]) -> None: +def print_results(payload: dict[str, Any], *, print_framewise: bool) -> None: report_recall = min( payload["matched_recalls"], key=lambda value: abs(value - 0.8) ) - print("\nWearer floor-state change evaluation") + print("\nWearer floor-event evaluation (primary event-level protocol)") + print( + "Each aggregate-other boundary is counted once. AP/rates are percentages; " + "other offset/onset remain timing proxies, not semantic EOT/INT." + ) + print( + "run | h | other-offset response AP | response/no-response/short-resume | " + f"no-response trigger @ response recall {report_recall:.0%} | " + f"short-resume trigger @ response recall {report_recall:.0%} | " + "other-onset yield AP | yield/continue | " + f"continue false-stop @ yield recall {report_recall:.0%}" + ) + for name, run in payload["runs"].items(): + for item in run["event_horizons"]: + response = item["other_offset_to_wearer_response_event"] + response_point = _operating_point(response, report_recall) + yield_metric = item["other_onset_to_wearer_yield_event"] + yield_point = _operating_point(yield_metric, report_recall) + print( + f"{name} | {item['nominal_seconds']:.1f}s | " + f"{_percentage(response['average_precision'])} | " + f"{response['num_positive']}/{response['num_negative']}/" + f"{response['num_short_resume_hard_negative']} | " + f"{_percentage(None if response_point is None else response_point['negative_false_positive_rate'])} | " + f"{_percentage(None if response_point is None else response_point.get('short_resume_hard_negative'))} | " + f"{_percentage(yield_metric['average_precision'])} | " + f"{yield_metric['num_positive']}/{yield_metric['num_negative']} | " + f"{_percentage(None if yield_point is None else yield_point['negative_false_positive_rate'])}" + ) + summary = run["event_summary"] + print( + f"{name} event mean | response=" + f"{_percentage(summary['mean_other_offset_response_event_ap'])}, " + f"yield={_percentage(summary['mean_other_onset_yield_event_ap'])}" + ) + + if not print_framewise: + return + + print("\nWearer endpoint/framewise evaluation (secondary diagnostic)") print("AP/rates are percentages; EOT/HOLD and yield are temporal proxies.") print( "run | h | OFF->ON AP | ON->OFF AP | EOT/HOLD AP | " @@ -437,7 +698,7 @@ def print_results(payload: dict[str, Any]) -> None: ) -def write_csv(path: Path, payload: dict[str, Any]) -> None: +def write_framewise_csv(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="") as file: fieldnames = [ @@ -489,10 +750,80 @@ def write_csv(path: Path, payload: dict[str, Any]) -> None: writer.writerow(row) +def write_event_csv(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", newline="") as file: + fieldnames = [ + "run", + "horizon_index", + "nominal_seconds", + "response_event_ap", + "response_count", + "no_response_count", + "short_resume_hard_negative_count", + "yield_event_ap", + "yield_count", + "continue_count", + ] + for recall in payload["matched_recalls"]: + fieldnames += [ + f"response_recall_{recall:g}", + f"no_response_trigger_rate_at_recall_{recall:g}", + f"short_resume_trigger_rate_at_recall_{recall:g}", + f"yield_recall_{recall:g}", + f"continue_false_stop_rate_at_recall_{recall:g}", + ] + writer = csv.DictWriter(file, fieldnames=fieldnames) + writer.writeheader() + for name, run in payload["runs"].items(): + for item in run["event_horizons"]: + response = item["other_offset_to_wearer_response_event"] + yield_metric = item["other_onset_to_wearer_yield_event"] + row: dict[str, Any] = { + "run": name, + "horizon_index": item["horizon_index"], + "nominal_seconds": item["nominal_seconds"], + "response_event_ap": response["average_precision"], + "response_count": response["num_positive"], + "no_response_count": response["num_negative"], + "short_resume_hard_negative_count": response[ + "num_short_resume_hard_negative" + ], + "yield_event_ap": yield_metric["average_precision"], + "yield_count": yield_metric["num_positive"], + "continue_count": yield_metric["num_negative"], + } + for recall in payload["matched_recalls"]: + response_point = _operating_point(response, recall) + yield_point = _operating_point(yield_metric, recall) + assert response_point is not None and yield_point is not None + row[f"response_recall_{recall:g}"] = response_point[ + "achieved_positive_recall" + ] + row[f"no_response_trigger_rate_at_recall_{recall:g}"] = ( + response_point["negative_false_positive_rate"] + ) + row[f"short_resume_trigger_rate_at_recall_{recall:g}"] = ( + response_point["short_resume_hard_negative"] + ) + row[f"yield_recall_{recall:g}"] = yield_point[ + "achieved_positive_recall" + ] + row[f"continue_false_stop_rate_at_recall_{recall:g}"] = ( + yield_point["negative_false_positive_rate"] + ) + writer.writerow(row) + + def main() -> None: args = parse_args() runs = parse_runs(args.run) recalls = validate_recalls(args.matched_recalls) + if args.midturn_pause_max_seconds <= 0: + raise ValueError("--midturn-pause-max-seconds must be positive") + midturn_pause_max_steps = max( + 1, int(math.ceil(args.midturn_pause_max_seconds / STEP_SECONDS)) + ) sessions = [normalize_session_name(value) for value in args.sessions] if len(sessions) != len(set(sessions)): raise ValueError("Duplicate sessions") @@ -511,6 +842,14 @@ def main() -> None: "OFF->ON and ON->OFF are endpoint state changes. EOT/HOLD, early-entry, " "and other-entry yield are timing-derived proxies, not semantic judgments." ), + "event_protocol": ( + "Each aggregate-others 1->0 boundary with wearer initially off is one " + "response candidate; each aggregate-others 0->1 boundary with wearer " + "initially on is one yield candidate. Labels are observed wearer " + "transitions within the horizon. Fixed-source future activity scores " + "are max-pooled only as ranking scores." + ), + "midturn_pause_max_seconds": args.midturn_pause_max_seconds, "runs": {}, } for name, run_dir in runs: @@ -528,19 +867,40 @@ def main() -> None: ) for horizon_index in range(NUM_HORIZONS) ] + event_horizons = [ + evaluate_event_horizon( + target, + prediction, + valid, + session_ids, + horizon_index, + recalls, + midturn_pause_max_steps=midturn_pause_max_steps, + ) + for horizon_index in range(NUM_HORIZONS) + ] payload["runs"][name] = { "result_dir": str(run_dir.expanduser().resolve()), "horizons": horizons, "summary": summarize_horizons(horizons), + "event_horizons": event_horizons, + "event_summary": summarize_event_horizons(event_horizons), } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(payload, indent=2) + "\n") - csv_path = args.csv_output or args.output.with_suffix(".csv") - write_csv(csv_path, payload) - print_results(payload) + event_csv_path = args.csv_output or args.output.with_name( + args.output.stem + "_events.csv" + ) + framewise_csv_path = args.output.with_name( + args.output.stem + "_framewise.csv" + ) + write_event_csv(event_csv_path, payload) + write_framewise_csv(framewise_csv_path, payload) + print_results(payload, print_framewise=args.print_framewise) print(f"\nJSON: {args.output}") - print(f"CSV: {csv_path}") + print(f"Event CSV: {event_csv_path}") + print(f"Framewise CSV: {framewise_csv_path}") if __name__ == "__main__": diff --git a/test_evaluate_floor_state_changes.py b/test_evaluate_floor_state_changes.py index ceb3052..f8bd747 100644 --- a/test_evaluate_floor_state_changes.py +++ b/test_evaluate_floor_state_changes.py @@ -9,6 +9,7 @@ import numpy as np from evaluate_floor_state_changes import ( NUM_CLASSES, NUM_HORIZONS, + evaluate_event_horizon, evaluate_horizon, load_arrays, parse_runs, @@ -21,6 +22,67 @@ def activity(wearer: int, others: int) -> list[int]: class FloorStateMetricTests(unittest.TestCase): + def test_event_level_counts_each_boundary_once(self) -> None: + target = np.asarray( + [ + activity(0, 1), # other offset follows + activity(0, 0), + activity(1, 0), # wearer responds within 0.4s + activity(1, 0), # other entry follows + activity(1, 1), + activity(0, 1), # wearer yields within 0.4s + activity(0, 1), + activity(0, 1), + activity(0, 0), # short pause, no wearer response + activity(0, 1), + activity(1, 0), # another other entry follows + activity(1, 1), + activity(1, 1), # wearer continues + ], + dtype=np.float32, + ) + prediction = np.full( + (len(target), NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32 + ) + # Fixed source row 0: response score max(0.2, 0.9). + prediction[1, 1, 0] = 0.2 + prediction[2, 1, 1] = 0.9 + # Fixed source row 7: no-response score max(0.1, 0.2). + prediction[8, 1, 0] = 0.1 + prediction[9, 1, 1] = 0.2 + # Fixed source row 3: yield score max(0.2, 0.9). + prediction[4, 1, 0] = 0.8 + prediction[5, 1, 1] = 0.1 + # Fixed source row 10: continue score max(0.1, 0.2). + prediction[11, 1, 0] = 0.9 + prediction[12, 1, 1] = 0.8 + valid = np.ones((len(target), NUM_HORIZONS), dtype=bool) + session_ids = np.zeros(len(target), dtype=np.int32) + session_ids[10:] = 1 + + result = evaluate_event_horizon( + target, + prediction, + valid, + session_ids, + 1, + [0.8], + midturn_pause_max_steps=10, + ) + response = result["other_offset_to_wearer_response_event"] + self.assertEqual(response["num_positive"], 1) + self.assertEqual(response["num_negative"], 1) + self.assertEqual(response["num_short_resume_hard_negative"], 1) + self.assertAlmostEqual(response["average_precision"], 1.0) + self.assertAlmostEqual( + response["operating_points"][0]["negative_false_positive_rate"], + 0.0, + ) + yield_metric = result["other_onset_to_wearer_yield_event"] + self.assertEqual(yield_metric["num_positive"], 1) + self.assertEqual(yield_metric["num_negative"], 1) + self.assertAlmostEqual(yield_metric["average_precision"], 1.0) + def test_horizon_one_proxies_and_session_boundary(self) -> None: # t0->t1: EOT-response; t3->t4: HOLD; t4->t5: early entry; # t6->t7: other-entry yield; t8->t9: other-entry continue.