video_helper.faces package

Submodules

Module contents

video_helper.faces

Reusable, HuggingFace-free computer-vision primitives for face-anchored speaker identity on video, built on the same local-first ethos as the rest of video-helper (ffmpeg + OpenCV + ONNX, no cloud, no HuggingFace at runtime).

Pieces (each usable on its own):

FaceDetector — YuNet detection + 5 landmarks (detect).
FaceRecognizer — SFace 128-d embeddings (recognize).
track_faces() — IoU/ByteTrack-family tracking into FaceTrack.
mouth_roi() — lip-centred ROI for the ASD visual stream (align).
get_engine() — an ASDEngine (lip-motion proxy or Light-ASD ONNX).
active_speaker_map() — the smart-sampling harness that ties it all together: shots + speaker-turns + a cheap face census pick a small set of clips, heavy ASD runs only there and grows until every speaker is covered with certainty (sampling).
ensure_model() — the sovereign model downloader (models).

Install the extra: pip install "video-helper[faces]" (adds onnxruntime + scenedetect; opencv-python is already a core dependency).

class video_helper.faces.ASDEngine[source]

Bases: object

Interface: score each track’s speaking likelihood within one clip window.

available()[source]

Return type:: bool

name = 'base'

score_tracks(frames, tracks, audio_16k, fps)[source]

Parameters:

frames (list[ndarray])
tracks (list[FaceTrack])
audio_16k (ndarray)
fps (float)

Return type:

dict[int, float]

class video_helper.faces.DigestSegment(digest_start, digest_end, source_start, source_end)[source]

Bases: object

One clip placed in the digest, with its two timelines reconciled.

Parameters:

digest_start (float)
digest_end (float)
source_start (float)
source_end (float)

digest_start

Start time (seconds) of this clip within the digest video.

Type:: float

digest_end

End time (seconds) of this clip within the digest video.

Type:: float

source_start

Start time (seconds) of this clip within the original source video.

Type:: float

source_end

End time (seconds) of this clip within the original source video.

Type:: float

digest_end: float

digest_start: float

source_end: float

source_start: float

to_source_time(digest_time)[source]

Map a timestamp inside this segment’s digest span back to source time.

Parameters:: digest_time (float) – A timestamp (seconds), in the digest’s own timeline.
Returns:: The corresponding source-video timestamp, or None when digest_time does not fall inside this segment.
Return type:: float or None

Examples

>>> seg = DigestSegment(digest_start=0.0, digest_end=12.0,
...                     source_start=340.0, source_end=352.0)
>>> seg.to_source_time(3.0)
343.0
>>> seg.to_source_time(99.0) is None
True

class video_helper.faces.Face(box, landmarks, score, raw)[source]

Bases: object

One detected face in one frame.

Parameters:

box (tuple[float, float, float, float])
landmarks (ndarray)
score (float)
raw (ndarray)

box

(x, y, w, h) in pixels.

Type:: tuple[float, float, float, float]

landmarks

(5, 2) float array — see _LANDMARK_NAMES.

Type:: np.ndarray

score

Detector confidence in [0, 1].

Type:: float

raw

The full 15-float YuNet row (box + 10 landmark coords + score), kept so cv2.FaceRecognizerSF.alignCrop can be fed the exact detector output.

Type:: np.ndarray

box: tuple[float, float, float, float]

landmarks: ndarray

raw: ndarray

score: float

class video_helper.faces.FaceDetector(*, score_threshold=0.6, min_size=40)[source]

Bases: object

Lazy, reusable YuNet detector.

The underlying cv2.FaceDetectorYN is created on first use and its input size is reset per frame (YuNet requires the exact frame dimensions). Construction never downloads; the first detect() does.

Parameters:

score_threshold (float)
min_size (int)

detect(frame_bgr)[source]

Detect faces in a single BGR uint8 frame (OpenCV convention).

Parameters:: frame_bgr (ndarray)
Return type:: list[Face]

class video_helper.faces.FaceRecognizer[source]

Bases: object

Lazy, reusable SFace embedder.

emb_dim

Embedding dimensionality (128 for SFace). Read this rather than assuming.

Type:: int

embed(frame_bgr, face)[source]

Return the L2-normalised 128-d embedding for one face, or None.

The face is aligned+cropped from frame_bgr using its raw YuNet row, then run through SFace. Returns None on any failure so callers can skip a bad crop rather than poison an average.

Parameters:

frame_bgr (ndarray)
face (Face)

Return type:

ndarray | None

embed_track(frames, faces, *, top_k=12)[source]

Aggregate one persistent embedding for a face track.

Quality-gates the crops (highest detector score first — the face analogue of picking the longest, most-confident turns for a voiceprint), embeds up to top_k of them, and returns the L2-normalised mean. None if no crop yields a usable embedding.

Parameters:

frames (list[ndarray])
faces (list[Face])
top_k (int)

Return type:

ndarray | None

class video_helper.faces.FaceTrack(track_id, frame_idx=<factory>, faces=<factory>)[source]

Bases: object

A temporally coherent sequence of one face’s detections.

Parameters:

track_id (int)
frame_idx (list[int])
faces (list[Face])

track_id

Stable id across the video (subject to stitching upstream).

Type:: int

frame_idx

Absolute frame indices where the face was seen.

Type:: list[int]

faces

The per-frame detection at each frame_idx.

Type:: list[Face]

add(frame_idx, face)[source]

Parameters:

frame_idx (int)
face (Face)

Return type:

None

faces: list[Face]

frame_idx: list[int]

property last_box: tuple[float, float, float, float]

span(fps)[source]

Track time span (t0, t1) in seconds given the sampling fps.

Parameters:: fps (float)
Return type:: tuple[float, float]

track_id: int

class video_helper.faces.LightASD[source]

Bases: ASDEngine

Light-ASD (Junhua-Liao et al., CVPR 2023) via PyTorch — accurate engine.

Loads the pretrained weights (light_asd.pth, research license) into the vendored _lightasd model and scores each face track with the model’s own audio-visual head. No ONNX: a faithful ONNX export of the MFCC front-end is fragile, so we run the real network. Degrades to unavailable (the harness swaps in the lip-motion proxy) if torch or the weights are missing.

Preprocessing matches the original exactly: audio → 13-cepstrum MFCC with fps-adjusted windows; visual → 112x112 grayscale mouth crops fed RAW (the model normalises (x/255 - 0.4161)/0.1688 internally — feeding an already /255 array would double-divide). Audio runs at ~4x the visual frame rate, so we align to 4 * T_visual MFCC frames; the two front-ends reduce to a common length, and we min-clip to be safe.

available()[source]

Return type:: bool

name = 'light-asd'

score_tracks(frames, tracks, audio_16k, fps)[source]

Parameters:

frames (list[ndarray])
tracks (list[FaceTrack])
audio_16k (ndarray)
fps (float)

Return type:

dict[int, float]

class video_helper.faces.LipMotionASD[source]

Bases: ASDEngine

Weights-free proxy: lip-motion variance × audio activity.

available()[source]

Return type:: bool

name = 'lip-motion'

score_tracks(frames, tracks, audio_16k, fps)[source]

Parameters:

frames (list[ndarray])
tracks (list[FaceTrack])
audio_16k (ndarray)
fps (float)

Return type:

dict[int, float]

class video_helper.faces.SpeakerFaceAssignment(speaker, face_id, coverage, margin, crops=<factory>)[source]

Bases: object

The face assigned to one diarization cluster.

Parameters:

speaker (int)
face_id (int)
coverage (float)
margin (float)
crops (list[tuple[ndarray, Face]])

speaker

Diarization cluster label.

Type:: int

face_id

Global face identity (stable across the whole video).

Type:: int

coverage

Fraction of the cluster’s sampled speech during which the assigned face was on screen and scored as speaking. Drives the fusion “face vs voice”.

Type:: float

margin

Vote margin over the runner-up face in [0, 1] — the certainty signal.

Type:: float

crops

Best (frame_bgr, Face) samples of the assigned face, for embedding.

Type:: list[tuple[np.ndarray, Face]]

coverage: float

crops: list[tuple[ndarray, Face]]

face_id: int

margin: float

speaker: int

video_helper.faces.active_speaker_map(video_path, audio_16k, speaker_turns, *, asd_engine='auto', clip_len=3.0, asd_fps=12.0, census_period=1.0, clip_budget=24, per_round=6, margin_tau=0.35, coverage_floor=0.3, asd_tau=0.4, rescue_budget=None)[source]

Assign each diarization cluster to a global on-screen face via sampled ASD.

Parameters mirror the design knobs in .private/face.md §4/§7: clip_len and asd_fps bound per-window cost; clip_budget caps total heavy work; margin_tau/coverage_floor define per-cluster certainty; clusters below it pull more windows until certain or the budget is spent.

rescue_budget adds a last-chance pass: if the shared clip_budget runs out while some clusters are still uncertain, resume ASD on just those, drawing from their remaining candidate windows, until each is certain, out of windows, or clearly not improving (a no-progress guard drops a genuinely off-screen speaker rather than burning the machine on it). None (the default) lets the rescue run until the finite window pool or the no-progress guard stops it; an int caps the extra windows.

Returns one SpeakerFaceAssignment per cluster that cleared the vote floor. Clusters left uncertain are logged and omitted (caller falls back to voiceprint for those).

Parameters:

video_path (str)
audio_16k (ndarray | None)
speaker_turns (list[dict])
asd_engine (str)
clip_len (float)
asd_fps (float)
census_period (float)
clip_budget (int)
per_round (int)
margin_tau (float)
coverage_floor (float)
asd_tau (float)
rescue_budget (int | None)

Return type:

list[SpeakerFaceAssignment]

video_helper.faces.build_asd_digest(video_path, anchor_times, output_video, *, window=6.0, merge_gap=12.0)[source]

Build a compact digest video from anchor-driven windows of video_path.

See the module docstring for the full design (why a digest, how windows are chosen, the splice-boundary caveat for consumers). In short: fuse nearby anchors to their midpoint, form a boundary window around each fused anchor and a mid-segment window between consecutive anchors (and at the two ends of the timeline), merge overlapping windows, then build the digest in three phases:

Transcode once — to_editing_intermediate() turns the source into an all-keyframe (GOP 1), PCM-audio intermediate. Every timestamp in it is a valid, lossless, frame-accurate cut point.
Cut + concat on the intermediate — each window is lifted with extract_video_chunk() (copy=True, pure stream-copy, no re-encode) and the clips are stitched with concat_videos() (reencode=False), safe because every chunk shares the intermediate’s exact codec/timebase.
Final encode — the stitched intermediate (huge, all-keyframe/PCM) is re-encoded once via video_converter() down to a normal delivery-sized file, audio included.

Doing the heavy per-window work as stream-copy and re-encoding only once, at the end, keeps the digest audio-safe and avoids paying a re-encode cost per window (see concat_videos() for why re-encoding a concat directly, reencode=True, used to silently drop audio).

Parameters:

video_path (str) – Path to the source video.
anchor_times (list of float) – Timestamps (seconds) where something diarization-worthy happens — the caller’s merged, sorted union of e.g. raw diarization speaker-change instants and shot-change instants. This function does not know or care where they came from.
output_video (str) – Path to write the concatenated digest video (with audio).
window (float, optional) – Half-width in seconds of every extracted window (default 6.0 — a full window is then 2 * window = 12s, comfortably covering the 1-6s duration range LR-ASD’s own evaluation methodology uses for reliability).
merge_gap (float, optional) – Anchors closer than this (seconds) fuse to one midpoint before windows are formed (default 12.0).

Returns:

One entry per clip actually placed in the digest, in digest-time order. Also written alongside output_video as <output_video>.manifest.json (a JSON array of {digest_start, digest_end, source_start, source_end}).

Return type:

list of DigestSegment

Raises:

AssertionError – If video_path is not a valid video, or no windows could be formed.

Notes

Splice boundaries (every digest_start/digest_end pair in the returned manifest) are real discontinuities: any frame-to-frame continuity assumption (face tracking, an ASD model’s own temporal context) must be reset there by the caller. This function only builds the video and the mapping; it runs no tracking or detection itself, so it has no opinion on which engine consumes the digest — Light-ASD, LR-ASD, or anything else conforming to video_helper.faces.asd.ASDEngine.

Examples

>>> segs = build_asd_digest(
...     "meeting.mp4", anchor_times=[12.0, 340.0, 341.0, 900.0],
...     output_video="digest.mp4",
... )
>>> segs[0].source_start
0.0

video_helper.faces.ensure_model(name)[source]

Resolve a model to a local path, downloading + caching on first use.

Parameters:: name (str) – A key in REGISTRY.
Returns:: Local filesystem path to the ready weight, or None if the model could not be fetched from any source (the caller degrades gracefully).
Return type:: str or None

video_helper.faces.get_engine(name='auto')[source]

Return an ASD engine by name, degrading to the proxy when needed.

"auto" prefers Light-ASD when its weights are hosted, else the proxy. "light-asd" forces the accurate engine (still falls back if unavailable). "lip-motion" forces the proxy.

Parameters:: name (str)
Return type:: ASDEngine

video_helper.faces.load_manifest(path)[source]

Read back a manifest written by build_asd_digest().

Parameters:: path (str) – Path to a <output_video>.manifest.json file.
Returns:: The segments, in the order they were written (digest-time order).
Return type:: list of DigestSegment

video_helper.faces.model_dir()[source]

Return (creating if needed) the local model cache directory.

Return type:: str

video_helper.faces.mouth_openness(frame_bgr, face)[source]

Cheap vertical-mouth-opening proxy in [0, 1] (weights-free ASD cue).

Uses the vertical gradient energy inside the lip ROI, normalised by the ROI size, as a stand-in for mouth opening/closing. It is deliberately crude — the lip-motion ASD proxy scores variance over time of this signal, not its absolute value, so only relative movement matters.

Parameters:

frame_bgr (ndarray)
face (Face)

Return type:

float

video_helper.faces.mouth_roi(frame_bgr, face, *, size=112, pad=1.6)[source]

Crop a square, lip-centred grayscale ROI for the ASD visual stream.

The crop is centred on the mouth-corner midpoint, sized to pad times the inter-corner distance (so the whole mouth plus a margin is captured), clamped to the frame, and resized to size``×``size. Returns a (size, size) uint8 grayscale array (zeros if the face falls entirely off-frame).

Parameters:

frame_bgr (ndarray)
face (Face)
size (int)
pad (float)

Return type:

ndarray

video_helper.faces.source_to_digest_window(segments, source_start, source_end)[source]

Map a [source_start, source_end] span back into digest-time, if covered.

The reverse of DigestSegment.to_source_time(): a caller with a window in the original video’s timeline (e.g. an ASD candidate window) uses this to find where — if anywhere — that span landed in the digest, so it can read frames from the small digest file instead of seeking into the original.

Parameters:

segments (list of DigestSegment) – The digest’s manifest, as returned by build_asd_digest().
source_start (float) – A time span (seconds) in the source video’s timeline.
source_end (float) – A time span (seconds) in the source video’s timeline.

Returns:

The corresponding (digest_start, digest_end) span, or None when no single digest segment fully covers [source_start, source_end] — this happens whenever the requested span was not anchor-driven into the digest (e.g. it falls between digest windows, or straddles a splice boundary). Callers should fall back to reading the original video in that case.

Return type:

(float, float) or None

Examples

>>> segs = [DigestSegment(0.0, 12.0, 340.0, 352.0)]
>>> source_to_digest_window(segs, 342.0, 345.0)
(2.0, 5.0)
>>> source_to_digest_window(segs, 400.0, 403.0) is None
True

video_helper.faces.track_faces(frame_dets, *, iou_threshold=0.3, max_gap=15)[source]

Link per-frame detections into tracks by greedy IoU association.

Parameters:

frame_dets (list[tuple[int, list[Face]]]) – (frame_idx, faces) in increasing frame order.
iou_threshold (float, optional) – Minimum IoU to attach a detection to an existing track.
max_gap (int, optional) – How many frames a track may coast unmatched before it is retired (lets a track survive a brief miss / occlusion).

Returns:

All tracks discovered, in creation order.

Return type:

list[FaceTrack]