blinklinmult.pipeline

Blink detection on a video, as seven observable stages.

The corpora hand a model eye crops that were extracted months earlier. A video hands it pixels, and everything between – finding a face, keeping it the same face, reading its pose, locating the eyes, deciding which are even visible – has to happen first. This module is that path.

Staged, not streamed. Each stage consumes all frames from the previous one before the next begins, because that is how the work actually batches, and each reports its own wall-clock time. A slow run is then immediately attributable rather than a single opaque wait.

#	stage	produces
1	face detection + tracking	one box per frame, the same person throughout
2	head pose	[yaw, pitch, roll] degrees per frame
3	landmarks + eye boxes	478 points and two eye boxes per frame
4	eye selection	per frame: left, right, both or neither
5	inference	per-frame closeness, per eye
6	labelling	open/closed per frame, plus blink events
7	visualisation	handled by the caller, from these results

run() is a generator yielding a Stage as each completes, so a UI can show progress without the pipeline knowing anything about a UI.

Attributes

logger	Module-level logger.
MAX_SECONDS	Longest video processed, in seconds.
LOCALISATION_FAST	Locate eyes from the face detector's own keypoints.
LOCALISATION_PRECISE	Locate eyes from FaceMesh landmarks, and extract the iris descriptor.
LOCALISATIONS	Valid localisation arguments to run().
POSE_GEOMETRIC	Estimate head rotation from the detector's five keypoints.
POSE_6DREPNET	Estimate head rotation with a dedicated pose network.
HEAD_POSES	Valid head_pose arguments to run().
YAW_LIMIT	Degrees of yaw past which one eye is treated as occluded.
PROGRESS_EVERY	Frames between progress lines within a stage.
SIDES	The two eyes, in the order they are reported.

Exceptions

PipelineError	Raised when a video cannot be processed at all.
NoFaceError	Raised when no frame of the video holds a face.

Classes

Extraction	How a per-frame eye-state curve becomes blink events.
Stage	One completed stage, for a progress display.
FrameResult	Everything known about one frame, for the overlay.
Result	The finished analysis of one video.

Functions

ground_truth(→ dict[str, numpy.ndarray] | None)	Per-eye annotated eye state, for a corpus clip that ships a .tag.
occluded_side(→ str | None)	Which eye a head turn hides, if any.
run(→ collections.abc.Iterator[Stage | Result])	Analyse a video, yielding each stage as it completes.

Module Contents

blinklinmult.pipeline.logger

Module-level logger.

blinklinmult.pipeline.MAX_SECONDS = 10.0

Longest video processed, in seconds.

Seven stages over several models per frame is minutes of CPU for seconds of video. Longer input is truncated with a warning rather than silently accepted and left to time out.

blinklinmult.pipeline.LOCALISATION_FAST = 'fast'

Locate eyes from the face detector’s own keypoints.

YOLO11 returns both eye centres in the same forward pass as the box, so the 478-point FaceMesh pass afterwards recomputes what is already in hand. Dropping it, and shrinking the detector’s input to POSE_IMAGE_SIZE, is what takes the streaming demos from 0.53x realtime to over 40 fps on CPU.

blinklinmult.pipeline.LOCALISATION_PRECISE = 'precise'

Locate eyes from FaceMesh landmarks, and extract the iris descriptor.

Slower, and the only route that can feed a model with needs_features: the 160-d stream comes from the iris landmarker, which the fast path never runs. Also what every published corpus was built with.

blinklinmult.pipeline.LOCALISATIONS

Valid localisation arguments to run().

blinklinmult.pipeline.POSE_GEOMETRIC = 'geometric'

Estimate head rotation from the detector’s five keypoints.

0.05 ms against 26.7 ms for 6DRepNet, because the keypoints are already computed – there is no second forward pass. See pose_from_keypoints() for the accuracy this trades away: roll and pitch correlate at 0.94, yaw is unvalidated.

blinklinmult.pipeline.POSE_6DREPNET = '6drepnet'

Estimate head rotation with a dedicated pose network.

A full forward pass per frame. What the corpora were built with, and what the two-stream model’s descriptor encodes.

blinklinmult.pipeline.HEAD_POSES

Valid head_pose arguments to run().

blinklinmult.pipeline.YAW_LIMIT = 45.0

Degrees of yaw past which one eye is treated as occluded.

Positive yaw turns the nose toward image-left, so it occludes the viewer’s right eye; negative yaw occludes the left. That sign is not arbitrary: exordium draws the yaw axis as x = size * sin(-radians(yaw)), which puts the nose at negative x – image-left – for positive yaw. Our side labels are the viewer’s, fixed by the same TalkingFace measurement that pins the region mirror in FaceMeshLocator.

Getting this backwards suppresses the visible eye and keeps the occluded one, which still produces a plausible-looking result. tests.test_pipeline asserts both directions.

blinklinmult.pipeline.PROGRESS_EVERY = 50

Frames between progress lines within a stage.

One line per frame would be noise, and yielding per frame would cost a meaningful share of the runtime it is reporting.

blinklinmult.pipeline.SIDES

The two eyes, in the order they are reported.

class blinklinmult.pipeline.Extraction

How a per-frame eye-state curve becomes blink events.

ESR gives a continuous closeness signal; blink presence is a decision laid on top of it, and this is that decision made explicit. It is separated from the model because the two are independently wrong: a shipped operating point is only fitted for the corpus it was fitted on, and applying it elsewhere is a transfer, not a measurement.

Parameters:

• high (float) – A run must peak above this to count as a blink.

• low (float | None) – It extends while above this. None is a single cut, where the same value has to both reject noise and catch a closure’s shallow onset – two jobs one number does badly.

classmethod fitted(model_spec: blinklinmult.registry.ModelSpec) → Extraction

The operating point registered for a model.

Parameters:

model_spec (ModelSpec) – The model’s registry entry.

Returns:

Its thresholds. For the 1.x models this is 0.5, a neutral sigmoid midpoint rather than a measurement – that work reported eye state, not events, so no operating point was ever fitted for them. Supply your own rather than inheriting it.

Return type:

Extraction

validate() → None

Check the thresholds can separate anything.

Raises:

PipelineError – If a threshold is outside (0, 1), or the low cut is not below the high one – at or above it, hysteresis stops extending runs and silently degrades to a single threshold.

describe() → str

One line naming the rule, for the log.

Returns:

Human-readable summary.

Return type:

str

exception blinklinmult.pipeline.PipelineError

Bases: Exception

Raised when a video cannot be processed at all.

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

exception blinklinmult.pipeline.NoFaceError

Bases: PipelineError

Raised when no frame of the video holds a face.

Separate from PipelineError because it is a property of the input, not a fault: the message tells a user what to upload instead, where a decode failure tells them something is broken.

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

class blinklinmult.pipeline.Stage

One completed stage, for a progress display.

Parameters:

• index (int) – 1-based stage number.

• total (int) – How many stages there are.

• name (str) – Short human label.

• detail (str) – What it found – counts, not prose.

• seconds (float) – Wall-clock time the stage took.

line() → str

Render as one log line.

Returns:

[n/7] name ... detail   1.2 s

Return type:

str

class blinklinmult.pipeline.FrameResult

Everything known about one frame, for the overlay.

Parameters:

• index (int) – Frame number in the source video.

• face_box (tuple | None) – (x1, y1, x2, y2), or None when the tracked subject was not found here.

• landmarks (np.ndarray | None) – (478, 2) FaceMesh points.

• pose (np.ndarray | None) – [yaw, pitch, roll] in degrees.

• eyes (dict[str, EyeBox | None]) – Box per side.

• used (dict[str, bool]) – Whether each side was scored.

• score (dict[str, float | None]) – Closeness per side; None where the eye was not scored, which is not the same as zero.

class blinklinmult.pipeline.Result

The finished analysis of one video.

Parameters:

• frames (list[FrameResult]) – Per-frame findings, in order.

• signal (dict[str, np.ndarray]) – Per-side closeness, NaN where the eye was not scored. NaN rather than zero so a gap cannot be mistaken for a confident “open”.

• events (dict[str, list[Interval]]) – Blink intervals per side.

• fps (float) – Source frame rate.

• model_id (str) – Which model produced this.

• truncated (bool) – Whether the video was cut to MAX_SECONDS.

• truth (dict[str, np.ndarray] | None) – Annotated per-eye state, when the clip came with a .tag. None for an uploaded video.

• extraction (Extraction | None) – The rule that produced events.

blinklinmult.pipeline.ground_truth(tag_path: str | pathlib.Path, frames: int, offset: int = 0) → dict[str, numpy.ndarray] | None

Per-eye annotated eye state, for a corpus clip that ships a .tag.

Only the .tag corpora have this; an uploaded video does not. It is drawn on the plot so a prediction can be read against what was annotated rather than judged by eye.

Parameters:

• tag_path (str | Path) – The .tag file beside the video.

• frames (int) – How many frames were analysed.

• offset (int) – Index of the first analysed frame in the source video, so a mid-video segment lines its annotation up correctly.

Returns:

Per-side 0/1 arrays of length frames, or None when the file is absent or unreadable.

Return type:

dict[str, np.ndarray] | None

blinklinmult.pipeline.occluded_side(yaw: float, limit: float = YAW_LIMIT) → str | None

Which eye a head turn hides, if any.

Parameters:

• yaw (float) – Head yaw in degrees.

• limit (float) – Magnitude past which one eye counts as occluded.

Returns:

The hidden side, or None when the head is frontal enough for both.

Return type:

str | None

blinklinmult.pipeline.run(video_path: str | pathlib.Path, detector: blinklinmult.detector.BlinkDetector, stats: dict[str, list[float]] | None = None, tag_path: str | pathlib.Path | None = None, extraction: Extraction | None = None, start: float = 0.0, duration: float = MAX_SECONDS, localisation: str = LOCALISATION_FAST, head_pose: str = POSE_GEOMETRIC) → collections.abc.Iterator[Stage | Result]

Analyse a video, yielding each stage as it completes.

Yields six Stage objects and then a final Result. The caller decides what to do with the stages – a UI prints them, a test ignores them.

Parameters:

• video_path (str | Path) – The video to analyse.

• detector (BlinkDetector) – A loaded model.

• stats (dict | None) – Corpus feature statistics, {"mean": ..., "std": ...}. Required by blinklinmult-union, ignored by the others.

• tag_path (str | Path | None) – A .tag annotation to plot alongside the prediction. Only the corpus clips have one.

• extraction (Extraction | None) – How to turn the eye-state curve into events. Defaults to the model’s registered operating point.

• start (float) – Where in the video to begin, in seconds.

• duration (float) – How much of it to analyse, in seconds.

• localisation (str) – LOCALISATION_FAST (default) reads the eye centres from the face detector’s keypoints; LOCALISATION_PRECISE runs FaceMesh and the iris landmarker. A model needing the descriptor forces the precise route – see _resolve_pipeline().

• head_pose (str) – POSE_GEOMETRIC (default) derives the angles from those same keypoints; POSE_6DREPNET runs the pose network. Forced to 6DRepNet alongside the precise route.

Yields:

Stage | Result – One per completed stage, then the result.

Raises:

PipelineError – If the video cannot be read, the two-stream model was given no feature statistics, or localisation/head_pose is not a recognised value.