blinklinmult.train.events¶

Event-level scoring: recovering blinks from a per-frame signal.

This is the evaluation protocol the blink literature uses, and following it is what makes numbers here comparable with published ones. The window a model sees is its receptive field and nothing more — no blink is ever assigned to a window. Events are recovered afterwards:

Sweep the recording with 50%-overlapping windows, annotation-blind.
Average the per-frame predictions wherever windows overlap.
Threshold the averaged signal and merge consecutive above-threshold frames into predicted intervals.
Match those intervals against the annotated blinks.

Steps 3 and 4 are exactly MPEblink’s Figure 6 procedure.

Four matching criteria, all reported. Papers differ in which they use, and they are not interchangeable — a detection one frame short of the annotation is a hit under any and a miss under iou75:

`any`	≥1 frame of intersection — Drutarovsky & Fogelton (2015)
`iou20`	temporal IoU > 0.2 — Fogelton & Beneš (2016)
`iou50`	temporal IoU ≥ 0.50 — MPEblink Blink-AP
`iou75`	temporal IoU ≥ 0.75 — MPEblink Blink-AP

Matching is one-to-one, as in MPEblink’s Hungarian assignment, and that cuts both ways:

Two predicted intervals on the same blink score one true positive and one false positive — otherwise a model that fires twice per blink reports twice the recall it earned.
One long prediction spanning two annotated blinks scores one true positive and one false negative. Failing to separate a close pair is a real failure — MPEblink’s “consecutive rapid eyeblink” case — and crediting both would hide it. A model that resolves the pair gets full credit.

No single threshold. The operating point changes every number, so the headline is an FROC curve: recall against false alarms per minute, swept across thresholds. A single-threshold table is still reported for readability, but the curve is what the comparison rests on.

Attributes¶

`logger`	Module-level logger.
`CRITERIA`	Matching criteria, as minimum temporal IoU.
`DEFAULT_THRESHOLDS`	Operating points swept for the FROC curve.
`Interval`	A closed frame interval `(start, end)`, both inclusive.
`BLINK_AP_TIOU`	The ten tIoU thresholds MPEblink's Blink-AP averages over.

Exceptions¶

EventError

Raised when a signal and its annotation cannot be scored together.

Classes¶

MatchResult

Counts from matching predictions to annotations under one criterion.

Functions¶

`average_overlapping`(→ tuple[numpy.ndarray, numpy.ndarray])	Collapse overlapping window predictions into one signal per frame.
`to_intervals`(→ list[Interval])	Merge consecutive above-threshold frames into predicted intervals.
`temporal_iou`(→ float)	Intersection over union of two inclusive frame intervals.
`segment_iou`(→ numpy.ndarray)	Temporal IoU of one interval against many, in the authors' convention.
`interpolated_average_precision`(→ float)	Area under the precision-recall curve, VOC-style.
`to_exclusive`(→ tuple[float, float])	Convert an inclusive frame interval to the authors' half-open one.
`blink_ap`(→ numpy.ndarray)	MPEblink's Blink-AP: average precision over a sweep of tIoU thresholds.
`blink_ap_summary`(→ dict[str, float])	Name the Blink-AP figures the authors report.
`matches`(→ bool)	Whether one prediction counts as detecting one annotated blink.
`match`(→ MatchResult)	Count hits, misses, and false alarms under one criterion.
`froc`(→ dict[str, numpy.ndarray])	Sweep the threshold, tracing recall against false alarms per minute.
`pool_curves`(→ dict[str, numpy.ndarray])	Combine per-recording FROC curves into one corpus curve.
`average_precision`(→ float)	Area under the precision-recall curve of a threshold sweep.
`event_metrics`(→ dict[str, float])	Every event-level metric, under every criterion.
`annotated_intervals`(→ list[Interval])	Recover annotated blink intervals from a per-frame id array.

Module Contents¶

blinklinmult.train.events.logger¶: Module-level logger.

blinklinmult.train.events.CRITERIA: dict[str, float]¶

Matching criteria, as minimum temporal IoU.

any is the degenerate case: a threshold of 0.0 accepts any non-empty intersection, so it needs no separate code path. iou20 is strictly greater than 0.2 in the source; the difference from >= is one frame in edge cases and is documented on matches().

blinklinmult.train.events.DEFAULT_THRESHOLDS¶

Operating points swept for the FROC curve.

0.01 resolution over [0.01, 0.99]. The original 0.05-step sweep over [0.05, 0.95] was too coarse to distinguish a fitted optimum from a boundary hit: the frame-wise model’s operating point landed on 0.05 – the sweep’s floor – for both RN15 and RN30, which is the signature of a search running out of range rather than finding a minimum. Extending below the old floor makes that distinguishable, and the finer step recovers calibration that 0.05 buckets lost.

blinklinmult.train.events.Interval¶: A closed frame interval (start, end), both inclusive.

exception blinklinmult.train.events.EventError[source]¶

Bases: ValueError

Raised when a signal and its annotation cannot be scored together.

Initialize self. See help(type(self)) for accurate signature.

class blinklinmult.train.events.MatchResult[source]¶

Counts from matching predictions to annotations under one criterion.

Parameters:

true_positives (int) – Annotated blinks that were detected.
false_positives (int) – Predicted intervals matching no annotation.
false_negatives (int) – Annotated blinks that were missed.
matched (dict[int, list[int]]) – Annotation index to the prediction indices that hit it, for inspecting how a blink was detected.

property recall: float¶

Fraction of annotated blinks detected.

Returns:: TP / (TP + FN); 0.0 when nothing was annotated.
Return type:: float

property precision: float¶

Fraction of predictions that hit an annotation.

Returns:: TP / (TP + FP); 0.0 when nothing was predicted.
Return type:: float

property f1: float¶

Harmonic mean of precision and recall.

Returns:: 0.0 when either is zero.
Return type:: float

false_alarms_per_minute(minutes: float) → float[source]¶

False positives normalised by recording length.

The natural rate on continuous footage: a count alone is not comparable between a 3-second clip and a 5-minute recording.

Parameters:: minutes (float) – Recording duration.
Returns:: FP / minutes; 0.0 for a zero-length recording.
Return type:: float

blinklinmult.train.events.average_overlapping(probabilities: collections.abc.Sequence[float] | numpy.ndarray, frame_ids: collections.abc.Sequence[int] | numpy.ndarray, n_frames: int) → tuple[numpy.ndarray, numpy.ndarray][source]¶

Collapse overlapping window predictions into one signal per frame.

A frame covered by several windows receives the mean of their predictions. A frame no window covers has no prediction at all and is masked, not zeroed — a zero is a confident “no blink”, which is a different claim from “not evaluated”.

Parameters:

probabilities (Sequence[float] | np.ndarray) – (N,) per-frame predictions, flattened across windows and eyes.
frame_ids (Sequence[int] | np.ndarray) – (N,) frame id each prediction belongs to.
n_frames (int) – Length of the recording.

Returns:

(n_frames,) averaged signal and its (n_frames,) coverage mask.

Return type:

tuple[np.ndarray, np.ndarray]

Raises:

EventError – If the inputs disagree in length, or a frame id is out of range for the declared recording length.

blinklinmult.train.events.to_intervals(signal: numpy.ndarray, mask: numpy.ndarray, threshold: float, low_threshold: float | None = None) → list[Interval][source]¶

Merge consecutive above-threshold frames into predicted intervals.

An uncovered frame breaks a run rather than extending it: with no prediction there, joining across the gap would assert a continuity the model never claimed.

Hysteresis. With low_threshold set, a run must peak above threshold to count as an event but extends while it stays above low_threshold. A single cut is the crudest possible extractor and is what makes the operating point brittle – it has to be low enough to catch the shallow start of a closure and high enough not to fire on noise, and no single value does both. Two thresholds separate those jobs: the high one decides whether this is a blink, the low one decides where it begins and ends.

This matters for the physical shape of a blink. The eye-state signal ramps down through closing, plateaus while closed, and ramps back up through opening; the ramps cross the low threshold well before the high one, so hysteresis recovers the true onset and offset rather than clipping to the part of the closure that happened to be deepest. It also makes an incomplete blink – one that crosses low but never high – explicitly detectable rather than silently truncated.

Parameters:

signal (np.ndarray) – (T,) averaged per-frame predictions.
mask (np.ndarray) – (T,) coverage mask.
threshold (float) – Frames strictly above this are “blinking”. With hysteresis this is the high threshold, which a run must reach.
low_threshold (float | None) – Extend a run while it stays strictly above this. None – the default – keeps the single-threshold behaviour exactly, so every existing caller is unchanged. Values at or above threshold are treated as None, since they could only shrink a run rather than extend it.

Returns:

Inclusive (start, end) frame intervals, in order.

Return type:

list[Interval]

Raises:

EventError – If the signal and mask disagree in length.

blinklinmult.train.events.temporal_iou(a: Interval, b: Interval) → float[source]¶

Intersection over union of two inclusive frame intervals.

Parameters:

a (Interval) – First interval.
b (Interval) – Second.

Returns:

|a ∩ b| / |a ∪ b| in frames; 0.0 when disjoint.

Return type:

float

blinklinmult.train.events.BLINK_AP_TIOU¶

The ten tIoU thresholds MPEblink’s Blink-AP averages over.

np.linspace(0.5, 0.95, 10) verbatim from the authors’ action_ap, so [0] is Blink-AP@0.5, [5] is @0.75 and the mean is the headline @0.5:0.95. Reproducing the grid exactly is what makes the numbers comparable.

blinklinmult.train.events.segment_iou(target: tuple[float, float], candidates: numpy.ndarray) → numpy.ndarray[source]¶

Temporal IoU of one interval against many, in the authors’ convention.

Endpoints are exclusive here, unlike temporal_iou(): the authors compute t_end - t_start with no +1, so a blink annotated [10, 14] has length 4 rather than 5. Keeping their convention matters – on a five-frame blink the two differ by 20%, which is enough to move a detection across a tIoU threshold and change the reported AP.

Both arguments must already be half-open; use to_exclusive() to convert this project’s inclusive intervals.

Parameters:

target (tuple[float, float]) – The predicted interval, half-open.
candidates (np.ndarray) – (N, 2) annotated intervals, half-open.

Returns:

(N,) temporal IoU against each candidate.

Return type:

np.ndarray

blinklinmult.train.events.interpolated_average_precision(precision: numpy.ndarray, recall: numpy.ndarray) → float[source]¶

Area under the precision-recall curve, VOC-style.

Precision is made monotonically decreasing from the right, then the area is summed over the steps where recall changes. This is the authors’ interpolated_prec_rec verbatim.

Parameters:

precision (np.ndarray) – (N,) cumulative precision.
recall (np.ndarray) – (N,) cumulative recall.

Returns:

The average precision.

Return type:

float

blinklinmult.train.events.to_exclusive(interval: Interval) → tuple[float, float][source]¶

Convert an inclusive frame interval to the authors’ half-open one.

Everything else here treats (start, end) as inclusive of both endpoints, because that is what the corpora annotate: a blink over frames 10 to 14 covers five frames. The authors’ segment_iou measures end - start with no +1, so the same pair means a length of four.

Without this conversion a single-frame prediction has length zero, its IoU against itself is 0/0 = 0, and a perfect detection scores as a miss – measured on the real corpus, that alone cost 2.4 points of AP. The annotation never triggers it (its shortest blink spans two frames, and none is degenerate across 4 809 events), but a thresholded signal readily produces one-frame intervals.

Parameters:: interval (Interval) – Inclusive (start, end).
Returns:: Half-open (start, end + 1), so an inclusive span of n frames has exclusive length n.
Return type:: tuple[float, float]

blinklinmult.train.events.blink_ap(annotated: dict[str, list[Interval]], predicted: dict[str, list[tuple[int, int, float]]], tiou_thresholds: numpy.ndarray | None = None) → numpy.ndarray[source]¶

MPEblink’s Blink-AP: average precision over a sweep of tIoU thresholds.

A faithful reimplementation of the authors’ compute_average_precision_detection, so a number produced here can be placed beside their published 8.58 (@0.5:0.95) and 25.13 (@0.5).

The algorithm, and why each step matters:

Rank every prediction across the whole corpus by score, not per recording. AP integrates one global precision-recall curve, so a confident detection in one clip outranks a doubtful one in another.
For each prediction, compute tIoU against the annotations of its own instance, and try them from highest overlap down.
At each tIoU threshold independently: the first annotation clearing the threshold and not already claimed makes this a true positive and locks it. One already claimed is skipped in favour of the next. A prediction that clears the threshold against nothing unclaimed is a false positive.
Integrate precision against recall with interpolated_average_precision(), once per threshold.

Because matching is one-to-one per threshold, a model firing twice on one blink earns one true positive and one false positive – the same discipline match() applies at a single operating point.

This is oracle-instance Blink-AP. The authors score only the instances their detector already matched; here the instances come from the annotation, so instance detection is free and the result is an upper bound on their joint figure. It must be reported as such – see blinklinmult.preprocess.mpeblink.

Parameters:

annotated (dict) – Instance id to its annotated blink intervals.
predicted (dict) – Instance id to (start, end, score) predictions. Ids absent from annotated contribute only false positives.
tiou_thresholds (np.ndarray | None) – The sweep; defaults to BLINK_AP_TIOU.

Returns:

(len(tiou_thresholds),) average precision per threshold. All zeros when nothing is annotated or nothing is predicted.

Return type:

np.ndarray

blinklinmult.train.events.blink_ap_summary(scores: numpy.ndarray) → dict[str, float][source]¶

Name the Blink-AP figures the authors report.

Parameters:: scores (np.ndarray) – The output of blink_ap(), length 10.
Returns:: blink_ap (the @0.5:0.95 mean, their headline), plus blink_ap50, blink_ap75 and blink_ap95.
Return type:: dict[str, float]

blinklinmult.train.events.matches(predicted: Interval, annotated: Interval, criterion: str) → bool[source]¶

Whether one prediction counts as detecting one annotated blink.

Parameters:

predicted (Interval) – Predicted interval.
annotated (Interval) – Annotated blink.
criterion (str) – A key of CRITERIA.

Returns:

Whether the pair matches.

Return type:

bool

Raises:

EventError – If the criterion is unknown.

blinklinmult.train.events.match(predicted: list[Interval], annotated: list[Interval], criterion: str = 'any') → MatchResult[source]¶

Count hits, misses, and false alarms under one criterion.

An annotated blink is detected when any prediction matches it, and counts once however many do. Every prediction matching no annotation is a false alarm — including the surplus ones when several land on the same blink, which is what stops a jittery model from inflating its recall.

Parameters:

predicted (list[Interval]) – Predicted intervals.
annotated (list[Interval]) – Annotated blinks.
criterion (str) – A key of CRITERIA.

Returns:

The counts, with the annotation-to-prediction map.

Return type:

MatchResult

blinklinmult.train.events.froc(signal: numpy.ndarray, mask: numpy.ndarray, annotated: list[Interval], minutes: float, criterion: str = 'any', thresholds: numpy.ndarray | None = None, low_ratio: float | None = None) → dict[str, numpy.ndarray][source]¶

Sweep the threshold, tracing recall against false alarms per minute.

The headline result. A single operating point is a choice that changes every number reported at it; the curve makes the trade-off visible instead of burying it in a constant.

Parameters:

signal (np.ndarray) – (T,) averaged per-frame predictions.
mask (np.ndarray) – (T,) coverage mask.
annotated (list[Interval]) – Annotated blinks.
minutes (float) – Recording duration, for the false-alarm rate.
criterion (str) – A key of CRITERIA.
thresholds (np.ndarray | None) – Operating points; defaults to DEFAULT_THRESHOLDS.
low_ratio (float | None) – Hysteresis, as a fraction of the swept threshold. Expressed as a ratio rather than an absolute value so the pair stays ordered across the whole sweep – a constant low threshold would exceed the high one at the bottom of the sweep and silently disable hysteresis. None keeps the single cut.

Returns:

thresholds, recall, precision, f1, false_alarms_per_minute, and the tp/fp/fn counts they were derived from, aligned by index.

The counts are returned because curves from several recordings pool by summing counts at each threshold, never by averaging rates — a thirty-second clip must not weigh the same as a five-minute recording. Without them a corpus curve could only be assembled the wrong way.

Return type:

dict[str, np.ndarray]

blinklinmult.train.events.pool_curves(curves: collections.abc.Sequence[dict[str, numpy.ndarray]], minutes: float) → dict[str, numpy.ndarray][source]¶

Combine per-recording FROC curves into one corpus curve.

Counts are summed at each threshold and the rates re-derived, which is the same rule _aggregate() applies to the single operating point. Averaging the per-recording rates instead would give a thirty-second clip the same weight as a five-minute recording, and a corpus is not a mean of its recordings.

Parameters:

curves (Sequence[dict[str, np.ndarray]]) – One froc() result per recording, all swept over the same thresholds.
minutes (float) – Total duration across the recordings, for the false-alarm rate.

Returns:

A curve in the same shape froc() returns. Empty when no curves are given.

Return type:

dict[str, np.ndarray]

Raises:

EventError – If the curves were swept over different thresholds, which would make summing them index-by-index meaningless.

blinklinmult.train.events.average_precision(curve: dict[str, numpy.ndarray]) → float[source]¶

Area under the precision-recall curve of a threshold sweep.

The curve is anchored at recall = 0 before integrating. Without that anchor a perfect detector scores zero: it holds recall at 1.0 across every threshold, the recall axis has no extent, and the trapezoid collapses — the best possible result reported as the worst. Anchoring is what the standard definition does, and it makes the integral the fraction of the recall range achieved at a given precision.

Parameters:: curve (dict[str, np.ndarray]) – Output of froc().
Returns:: Average precision over the swept operating points.
Return type:: float

blinklinmult.train.events.event_metrics(signal: numpy.ndarray, mask: numpy.ndarray, annotated: list[Interval], minutes: float, threshold: float = 0.5, thresholds: numpy.ndarray | None = None, curves: dict[str, dict[str, numpy.ndarray]] | None = None, low_threshold: float | None = None, low_ratio: float | None = None) → dict[str, float][source]¶

Every event-level metric, under every criterion.

Parameters:

signal (np.ndarray) – (T,) averaged per-frame predictions.
mask (np.ndarray) – (T,) coverage mask.
annotated (list[Interval]) – Annotated blinks.
minutes (float) – Recording duration.
threshold (float) – Operating point for the single-threshold counts.
thresholds (np.ndarray | None) – Sweep for the curves.
curves (dict[str, dict[str, np.ndarray]] | None) – Filled in with the FROC curve per criterion when given. The curve is what average_precision and best_f1 are derived from, and callers that pool several recordings need the points themselves — returning only the two scalars discards the headline result.
low_threshold (float | None) – Absolute hysteresis threshold for the single-point counts. See to_intervals().
low_ratio (float | None) – Hysteresis as a fraction of the threshold, used for the swept curves and as a fallback for the single point when low_threshold is not given.

Returns:

event/<criterion>/<metric> for each of CRITERIA, plus descriptive counts that keep a rate readable.

Return type:

dict[str, float]

blinklinmult.train.events.annotated_intervals(blink_ids: numpy.ndarray) → list[Interval][source]¶

Recover annotated blink intervals from a per-frame id array.

Runs of consecutive frames sharing an id, so a reused id yields two events rather than one spanning the gap between them — which is what the annotation means, and what the corpora’s own notes describe when a subject “blinks twice very fast”.

Parameters:: blink_ids (np.ndarray) – (T,) per-frame blink id; negative means no blink.
Returns:: Inclusive intervals, in frame order.
Return type:: list[Interval]