blinklinmult.train.metrics¶

Masked classification metrics for the two blink tasks.

Both tasks are binary detection under heavy class imbalance, so the metrics are chosen accordingly and every one of them reduces over valid positions only — exactly the mask the losses use.

{split}/{target}/f1: Harmonic mean of precision and recall. The primary metric. Accuracy is not reported as a headline number: a model predicting “no blink” everywhere scores above 95% accuracy on a continuously-sampled recording, which says nothing about whether it detects blinks.
{split}/{target}/precision, {split}/{target}/recall: Reported alongside F1 because the two failure modes — missing blinks and hallucinating them — have different costs downstream, and F1 alone hides which one a run has.
{split}/{target}/average_precision: Area under the precision-recall curve: threshold-free, so it separates a model that ranks well but is badly calibrated from one that genuinely cannot discriminate.
{split}/{target}/accuracy, {split}/{target}/balanced_accuracy: Kept for comparability with the published 1.x numbers, which reported accuracy. Balanced accuracy is the one to read.
{split}/mean_f1: Mean F1 across supervised targets. Drives model selection.

Why not torchmetrics’ own classification metrics. They have no notion of a per-position validity mask, and passing them a flattened, pre-filtered tensor would silently change what “an epoch” means under DDP, where each rank filters a different number of positions. Accumulating raw predictions with their masks and reducing once at epoch end keeps the gathered state exact.

Attributes¶

`logger`	Module-level logger.
`EPSILON`	Guards division when a class is absent from an epoch.
`DEFAULT_THRESHOLD`	Probability above which a position is predicted positive.
`PRIMARY_METRIC`	The metric that selects checkpoints and drives early stopping.
`NO_GROUP`	Sentinel frame-group id, meaning "this position cannot be grouped".
`FRAME_REDUCERS`	Rules for combining the two eyes' predictions into one per frame.
`PRIMARY_FRAME_REDUCER`	The frame reducer whose F1 drives model selection.
`ESR_METRIC`	mean frame-level F1 across targets.
`BPD_METRIC`	window-level F1.
`METRIC_NAMES`	Per-target metrics computed every epoch, in report order.

Classes¶

`BlinkMetrics`	Accumulates an epoch of masked predictions for one target.
`MultiTargetMetrics`	Holds one `BlinkMetrics` per supervised target.

Functions¶

`average_precision`(→ torch.Tensor)	Area under the precision-recall curve over valid positions.
`binary_metrics`(→ dict[str, torch.Tensor])	Every per-target metric, over valid positions.
`frame_group_ids`(→ torch.Tensor \| None)	Map each sample to an integer id shared by the eyes of the same frames.
`aggregate_by_group`(→ tuple[torch.Tensor, torch.Tensor])	Reduce per-position values to one value per frame group.
`frame_level_metrics`(→ dict[str, torch.Tensor])	Score one target at frame level, aggregating the eyes of each frame.
`window_level_metrics`(→ dict[str, torch.Tensor])	Score one target at window level: did this window contain a blink?

Module Contents¶

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

blinklinmult.train.metrics.EPSILON = 1e-08¶: Guards division when a class is absent from an epoch.

blinklinmult.train.metrics.DEFAULT_THRESHOLD = 0.5¶: Probability above which a position is predicted positive.

blinklinmult.train.metrics.PRIMARY_METRIC = 'mean_f1'¶

The metric that selects checkpoints and drives early stopping.

Computed over the frame-level scores when frame-group ids are available, so model selection optimises the number the benchmark reports. See FrameAggregator.

blinklinmult.train.metrics.NO_GROUP = -1¶

Sentinel frame-group id, meaning “this position cannot be grouped”.

Accumulated when a caller supplies no group ids, which leaves the per-eye metrics intact and simply disables frame-level aggregation for that epoch.

blinklinmult.train.metrics.FRAME_REDUCERS: tuple[str, ...] = ('max', 'mean')¶

Rules for combining the two eyes’ predictions into one per frame.

max is the primary rule: a blink is a frame-level event that occurred, and the annotation marks a frame as a blink when a closure is visible. If one eye is occluded by a head turn while the other is plainly closing, max recovers the event where mean halves the score and can push it under the threshold. mean is logged alongside so the choice is measured rather than assumed.

blinklinmult.train.metrics.PRIMARY_FRAME_REDUCER = 'max'¶: The frame reducer whose F1 drives model selection.

blinklinmult.train.metrics.ESR_METRIC = 'mean_f1'¶

mean frame-level F1 across targets.

Type:: Headline eye-state-recognition score

blinklinmult.train.metrics.BPD_METRIC = 'bpd_f1'¶

window-level F1.

Reported only when the run supervises blink presence, since a still-image corpus cannot witness a blink.

Type:: Headline blink-presence-detection score

blinklinmult.train.metrics.METRIC_NAMES: tuple[str, ...] = ('f1', 'precision', 'recall', 'average_precision', 'accuracy', 'balanced_accuracy')¶: Per-target metrics computed every epoch, in report order.

blinklinmult.train.metrics.average_precision(probability: torch.Tensor, target: torch.Tensor, mask: torch.Tensor) → torch.Tensor¶

Area under the precision-recall curve over valid positions.

Computed by the step-wise sum sum_k (R_k - R_{k-1}) * P_k over positions ranked by predicted probability — the standard estimator, which unlike trapezoidal interpolation does not reward a model for the region between two observed operating points.

Parameters:

probability (torch.Tensor) – Predicted probabilities, any shape.
target (torch.Tensor) – Binary targets, same shape.
mask (torch.Tensor) – True where the target is real supervision.

Returns:

Scalar in [0, 1]; 0 when no positive is present, for which average precision is undefined.

Return type:

torch.Tensor

blinklinmult.train.metrics.binary_metrics(probability: torch.Tensor, target: torch.Tensor, mask: torch.Tensor, threshold: float = DEFAULT_THRESHOLD) → dict[str, torch.Tensor]¶

Every per-target metric, over valid positions.

Parameters:

probability (torch.Tensor) – Predicted probabilities.
target (torch.Tensor) – Binary targets.
mask (torch.Tensor) – True where the target is real supervision.
threshold (float) – Decision threshold.

Returns:

Keyed by METRIC_NAMES. Every value is 0 when the epoch contained no valid position, rather than NaN, which would poison the epoch average and any checkpoint comparison.

Return type:

dict[str, torch.Tensor]

blinklinmult.train.metrics.frame_group_ids(sample_ids: collections.abc.Sequence[str]) → torch.Tensor | None¶

Map each sample to an integer id shared by the eyes of the same frames.

OmniLoader forwards only the sample id into a batch, so the recording and the frame group travel encoded inside it (see build_sample_id()). This strips the eye side and interns what remains, so the two eye-wise samples cut from one window receive the same id and frame_level_metrics() can recombine them.

The ids are stable across batches, not batch-local. They are a hash of the group key rather than an enumeration, because the accumulator persists across every batch of an epoch: ids that restarted at 0 each batch would make window 0 of batch 1 collide with window 0 of batch 0, and the aggregation would silently max together frames from different recordings. Measured on RN30’s test split, that collapsed 329 040 positions into a degenerate set and reported frame_max/f1 = 1.0 against a true 0.57.

Parameters:: sample_ids (Sequence[str]) – One id per sample in the batch.
Returns:: (B,) int64 ids, or None when the batch carries no ids or any of them is unparseable — in which case the per-eye metrics still run and only frame-level scoring is skipped.
Return type:: torch.Tensor | None

blinklinmult.train.metrics.aggregate_by_group(values: torch.Tensor, groups: torch.Tensor, reducer: str) → tuple[torch.Tensor, torch.Tensor]¶

Reduce per-position values to one value per frame group.

Parameters:

values (torch.Tensor) – (N,) per-position values.
groups (torch.Tensor) – (N,) int64 group id per position.
reducer (str) – One of FRAME_REDUCERS.

Returns:

(reduced, unique_groups), both (G,), ordered by group id.

Return type:

tuple[torch.Tensor, torch.Tensor]

Raises:

ValueError – If the reducer is unknown or the shapes disagree.

blinklinmult.train.metrics.frame_level_metrics(metric: BlinkMetrics) → dict[str, torch.Tensor]¶

Score one target at frame level, aggregating the eyes of each frame.

This is the eye state recognition (ESR) protocol: one decision per frame. Training is eye-wise because that is how the corpora annotate, but the benchmark reports per frame, so the two eye-wise predictions of a frame are recombined here. The same aggregation runs on validation as on test, so model selection optimises the reported number rather than a proxy.

A position is aggregated only when it is validly supervised; a frame whose every eye was masked out contributes nothing rather than a default.

Parameters:: metric (BlinkMetrics) – An accumulator for one target.
Returns:: frame_<reducer>/<metric> for every reducer in FRAME_REDUCERS. Empty when no group ids were accumulated, which is what a caller checks to know whether frame-level scoring ran.
Return type:: dict[str, torch.Tensor]

blinklinmult.train.metrics.window_level_metrics(metric: BlinkMetrics) → dict[str, torch.Tensor]¶

Score one target at window level: did this window contain a blink?

This is the blink presence detection (BPD) protocol, and it is a genuinely different task from frame-wise eye-state recognition rather than a coarser view of it. A window is positive when any of its frames is annotated as blinking, and a prediction is positive when the model’s sequence output crosses the threshold at any timestep — a max over the window.

Aggregating a single sequence head, rather than training a second clip-level head, is what keeps the two tasks consistent: a window cannot come out “no frame shows a closed eye” and “a blink occurred”, which two independent heads would permit.

BPD is the more error-tolerant protocol, and that is why it is reported alongside ESR: the corpora do not annotate blink boundaries precisely, since a closure is motion that begins before and ends after the frames marked as blinking. A frame-wise score punishes a one-frame boundary disagreement that a window-wise score absorbs.

Parameters:: metric (BlinkMetrics) – An accumulator for one target, whose group ids identify the sample each position came from.
Returns:: window/<metric>. Empty when no group ids were accumulated.
Return type:: dict[str, torch.Tensor]

class blinklinmult.train.metrics.BlinkMetrics(target_name: str, threshold: float = DEFAULT_THRESHOLD)¶

Bases: torchmetrics.Metric

Accumulates an epoch of masked predictions for one target.

A torchmetrics.Metric rather than plain lists, so state is gathered across devices automatically. Appending to Python lists on the module would, under DDP, silently report only rank 0’s shard.

Parameters:

target_name (str) – The target these metrics describe.
threshold (float) – Decision threshold.

Raises:

ValueError – If the threshold is outside (0, 1).

update(probability: torch.Tensor, target: torch.Tensor, mask: torch.Tensor, group: torch.Tensor | None = None) → None¶

Accumulate one batch.

Parameters:

probability (torch.Tensor) – Predicted probabilities, (B, T).
target (torch.Tensor) – Binary targets, same shape.
mask (torch.Tensor) – True where the target is real supervision.
group (torch.Tensor | None) – (B,) integer frame-group id per sample, broadcast across its timesteps. None accumulates a sentinel, which disables frame-level aggregation for this epoch but leaves the per-eye metrics intact.

Raises:

ValueError – If the shapes disagree.

restore(probability: torch.Tensor, target: torch.Tensor, mask: torch.Tensor, group: torch.Tensor, window_length: int) → None¶

Re-accumulate already-flattened state from a previous pass.

update() derives the group ids from the shapes it is handed, which is right for a live batch but wrong for state read back from disk: the ids there already encode the frames they came from and must survive verbatim, or every consumer that groups by frame would regroup the restored positions incorrectly.

Every metric is an order-independent reduction over per-position values, so state appended here scores identically to state computed in place – which is what makes resuming an interrupted test pass exact rather than approximate.

Parameters:

probability (torch.Tensor) – (N,) predicted probabilities.
target (torch.Tensor) – (N,) binary targets.
mask (torch.Tensor) – (N,) validity mask.
group (torch.Tensor) – (N,) frame-group id per position.
window_length (int) – Timesteps per sample in the restored pass.

Raises:

ValueError – If the shapes disagree.

groups() → torch.Tensor | None¶

Frame-group id of every accumulated position.

Returns:: (N,) int64, or None when nothing was accumulated or no group ids were supplied.
Return type:: torch.Tensor | None

property valid_count: int¶

Number of validly-supervised positions accumulated this epoch.

Logged so a run makes visible how much supervision each target actually received — a joint run where one head saw almost nothing is a data problem, not a modelling one, and this is what reveals it.

Returns:: Valid position count.
Return type:: int

compute() → dict[str, torch.Tensor]¶

Compute every metric over the accumulated epoch.

Returns:: Keyed "<target>/<metric>". Zeros throughout when the target received no supervision this epoch, which is a legitimate state for a joint run whose batch mix happened to exclude the annotating corpus.
Return type:: dict[str, torch.Tensor]

predictions() → tuple[torch.Tensor, torch.Tensor, torch.Tensor]¶

Return the accumulated predictions, targets, and masks.

Used to dump per-sample test predictions as a run artifact, so downstream threshold tuning and significance testing never need the model re-run.

Returns:: (probability, target, mask), each (N,); empty tensors when nothing was accumulated.
Return type:: tuple

class blinklinmult.train.metrics.MultiTargetMetrics(target_names: list[str], threshold: float = DEFAULT_THRESHOLD)¶

Bases: torch.nn.Module

Holds one BlinkMetrics per supervised target.

A Module rather than a plain dict so Lightning moves the child metrics’ state to the right device along with the model.

Parameters:

target_names (list[str]) – Targets to track, in head order.
threshold (float) – Decision threshold, shared by every target.

__contains__(target_name: object) → bool¶

Whether this accumulator holds one target.

Parameters:: target_name (object) – The target to look for.
Returns:: Whether it is supervised here. Lets a caller skip an eval-only target on the training split, where no accumulator for it exists, without catching KeyError.
Return type:: bool

__getitem__(target_name: str) → BlinkMetrics¶

Look up one target’s accumulator.

ModuleDict.__getitem__ is typed as returning Module, which loses every BlinkMetrics method. Narrowing here once means callers — including the prediction-writing callbacks — get the real type instead of casting at each use.

Parameters:: target_name (str) – Which target’s accumulator to return.
Returns:: That target’s accumulator.
Return type:: BlinkMetrics
Raises:: KeyError – If this run does not supervise that target.

update(target_name: str, probability: torch.Tensor, target: torch.Tensor, mask: torch.Tensor, group: torch.Tensor | None = None) → None¶

Accumulate one batch for one target.

Parameters:

target_name (str) – Which target to update.
probability (torch.Tensor) – Predicted probabilities.
target (torch.Tensor) – Binary targets.
mask (torch.Tensor) – Validity mask.
group (torch.Tensor | None) – (B,) frame-group id per sample.

compute() → dict[str, torch.Tensor]¶

Compute every target’s metrics plus the cross-target summary.

Reports two families of scores. The eye-level ones describe what the model was trained on — one prediction per eye. The frame-level ones aggregate the two eyes of a frame into a single decision, which is how the corpora are actually scored, and are what drives model selection when they are available.

Returns:: <target>/<metric> per eye, <target>/frame_<reducer>/<metric> per frame, a <target>/valid_positions count, and PRIMARY_METRIC.
Return type:: dict[str, torch.Tensor]

reset() → None¶: Clear every target’s accumulated state.