Evaluate wearer floor changes by event

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")
+    event_csv_path = args.csv_output or args.output.with_name(
-    write_csv(csv_path, payload)
+        args.output.stem + "_events.csv"
-    print_results(payload)
+    )
+    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__":

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.