Python Analysis API#

The analysis/ folder is a separate, uv-managed Python project (analysis/pyproject.toml) from the live-acquisition python/ project — see Architecture. It contains eight independent offline Analysis-tab plugins, each with a run_<plugin>.py CLI wrapper (User guide covers running them from the app), plus a ninth supporting package (Live Transcription (Real-time tab)) used by the Real-time tab’s live captions, not by any post-hoc plugin. This page documents each package’s importable library surface, not the CLI argument parsing — see each subsection’s linked math page for the algorithm behind its output.

🚶 Pose Estimation

2D COCO keypoints via YOLOv8-pose — the foundational signal every derived plugin below builds on.

Pose Estimation

🫥 Face Masking

Detect and blur/box faces for anonymized sharing outside the lab.

Face Masking

🗣️ Speaker Diarization

faster-whisper transcription + pyannote speaker turns, merged by max overlap.

Speaker Diarization

🙂 Facial Expression

3 backends: a transparent blendshape heuristic, FER+, and real FACS Action Units via py-feat.

Facial Expression

🎯 Multi-Camera Gaze Fusion

Per-camera 3D gaze rays, triangulated into one fused ray + target point.

Multi-Camera Gaze Fusion

🧍 3D Pose Reconstruction

Multi-view DLT triangulation and cross-camera person tracking, from the Pose plugin’s own 2D output.

3D Pose Reconstruction

❤️ Remote Heart Rate

Camera-based pulse-rate estimation (rPPG). Experimental.

Remote Heart Rate (rPPG)

🐭 Motion Tracking

Background-subtraction centroid tracking, run from the Session Browser.

Motion Tracking

💬 Live Transcription

Rolling-buffer streaming speech-to-text behind the Real-time tab’s captions.

Live Transcription (Real-time tab)

Pose Estimation #

YOLOv8-pose detection over a session’s videos, producing per-frame, per-subject 2D keypoints in COCO format. This is the foundational 2D signal every derived plugin consumes: Pose Kinematics’s speed/acceleration, and the 3D Pose Reconstruction plugin below.

from pose import HumanPoseEstimator

estimator = HumanPoseEstimator(model_name="yolov8n-pose.pt")
result = estimator.infer(frame_bgr, frame_index=0, timestamp_ns=0, camera_index=0)

Pose estimation backends for MOSAIC analysis.

`HumanPoseEstimator`	Wraps YOLOv8-pose for single-frame inference.
`PoseResult`	Normalised output from any pose estimator backend.
`SubjectPose`	Keypoints for a single detected subject.

class pose.HumanPoseEstimator(model_name='yolov8n-pose.pt', device=None, conf_threshold=0.4, iou_threshold=0.7)[source]#

Wraps YOLOv8-pose for single-frame inference.

Parameters:

model_name (str, default "yolov8n-pose.pt") – YOLOv8 model variant. Downloaded automatically from the Ultralytics CDN on first use (~4-87 MB).
device (str or None, default None) – Inference device — "cpu", "cuda:0", "mps" (Apple Silicon). None auto-detects (prefers CUDA → MPS → CPU).
conf_threshold (float, default 0.40) – Minimum detection confidence to include a subject.
iou_threshold (float, default 0.70) – NMS IoU threshold.

Raises:

ImportError – If ultralytics is not installed.

infer(frame, frame_index=0, timestamp_ns=0, camera_index=0)[source]#

Run pose estimation on one BGR frame.

Parameters:

frame (numpy.ndarray) – BGR frame, as returned by cv2.imread/cv2.VideoCapture.
frame_index (int, default 0) – Caller-supplied frame index, echoed into the result.
timestamp_ns (int, default 0) – Caller-supplied timestamp (ns), echoed into the result.
camera_index (int, default 0) – Caller-supplied camera index, echoed into the result.

Returns:

Structured result containing per-subject keypoints, echoing frame_index/timestamp_ns/camera_index back unchanged and reporting this call’s own inference_ms.

Return type:

PoseResult

class pose.PoseResult(frame_index, timestamp_ns, camera_index, subjects=<factory>, backend='unknown', inference_ms=0.0)[source]#

Normalised output from any pose estimator backend.

Parameters:

frame_index (int)
timestamp_ns (int)
camera_index (int)
subjects (list[SubjectPose])
backend (str)
inference_ms (float)

frame_index#

Caller-supplied frame index this result was computed from.

Type:: int

timestamp_ns#

Caller-supplied timestamp (ns) this result was computed from.

Type:: int

camera_index#

Caller-supplied camera index this result was computed from.

Type:: int

subjects#

One entry per detected subject in this frame.

Type:: list of SubjectPose

backend#

Identifies which estimator/model produced this result (e.g. "yolov8-pose/yolov8n-pose.pt").

Type:: str, default “unknown”

inference_ms#

Wall-clock inference time for this frame, in milliseconds.

Type:: float, default 0.0

class pose.SubjectPose(subject_id, confidence, keypoints, visibilities, bbox_xyxy=(0, 0, 0, 0))[source]#

Keypoints for a single detected subject.

Parameters:

subject_id (int)
confidence (float)
keypoints (list[tuple[float, float]])
visibilities (list[float])
bbox_xyxy (tuple[float, float, float, float])

subject_id#

Tracking ID, or -1 if no cross-frame tracking is performed.

Type:: int

confidence#

Overall detection confidence, in [0, 1].

Type:: float

keypoints#

(x_px, y_px) per keypoint, in the backend’s own keypoint order (e.g. COCO_KEYPOINTS).

Type:: list of tuple of float

visibilities#

Per-keypoint confidence, in [0, 1], parallel to keypoints.

Type:: list of float

bbox_xyxy#

Subject’s detection bounding box, (x1, y1, x2, y2) pixels.

Type:: tuple of float, default (0, 0, 0, 0)

See Pose Kinematics for the Speed/Acceleration math applied to this plugin’s output (implemented in C++, not here — see mosaic::compute_kinematics()).

Tip

run_pose.py has two operating modes: session mode (--session <dir>, the common post-processing path used by the Analysis tab) and pipe mode (--pipe, base64-JPEG frames over stdin/stdout), which is how MOSAIC’s live Real-time tab drives the same model internally.

Face Masking #

Anonymizes a session’s videos by detecting and blurring/boxing faces, writing the result into a sibling anonymized/ folder — the originals are never touched. Three interchangeable detector backends (MediaPipeFaceDetector, default; YoloFaceDetector; OpenCVDnnFaceDetector, YuNet) share one detect() -> list[Box] interface.

from facemask import make_detector, expand_and_clip, apply_mask

detector = make_detector("mediapipe", model=None, conf_threshold=0.5)
boxes = [expand_and_clip(b, 0.25, frame_w, frame_h) for b in detector.detect(frame_bgr)]
masked = apply_mask(frame_bgr, boxes, style="blur")

Face-detection backends and blur/box masking helpers for MOSAIC’s Face Masking analysis plugin.

facemask.Box is a plain type alias, tuple[float, float, float, float] (x1, y1, x2, y2 pixel coordinates) — the shape every detector’s detect() and both geometry functions below use.

`FaceDetector`	Structural interface every face-detection backend satisfies.
`MediaPipeFaceDetector`	MediaPipe Tasks `FaceLandmarker`-based face detector.
`YoloFaceDetector`	Ultralytics YOLO face detector, using a community face checkpoint.
`OpenCVDnnFaceDetector`	`cv2.FaceDetectorYN` (YuNet) ONNX face detector.
`make_detector`	Build a face-detection backend by name.
`expand_and_clip`	Pad a face box and clip it to the frame bounds.
`apply_mask`	Blur or solid-fill every box's region in `frame`, in place.

class facemask.FaceDetector(*args, **kwargs)[source]#

Structural interface every face-detection backend satisfies.

All three backends (MediaPipeFaceDetector, YoloFaceDetector, OpenCVDnnFaceDetector) implement this without inheriting from it — a plain duck-typing Protocol so run_face_mask.py doesn’t need to know which one is in use.

detect(frame_bgr)[source]#

Detect faces in one BGR frame.

Parameters:: frame_bgr (numpy.ndarray) – BGR frame, as returned by cv2.imread/cv2.VideoCapture.
Returns:: One (x1, y1, x2, y2) pixel box per detected face, already filtered by this backend’s own confidence threshold (box tuples carry no separate confidence score).
Return type:: list of Box

class facemask.MediaPipeFaceDetector(conf_threshold=0.5, max_faces=10)[source]#

MediaPipe Tasks FaceLandmarker-based face detector.

Best recall of the three backends, supports multiple faces. Downloads face_landmarker.task to models/ on first use.

Parameters:

conf_threshold (float, default 0.5) – Minimum face-detection/-presence confidence to keep a face.
max_faces (int, default 10) – Maximum simultaneous faces to detect per frame.

detect(frame_bgr)[source]#

See FaceDetector.detect().

Parameters:: frame_bgr (ndarray)
Return type:: list[tuple[float, float, float, float]]

class facemask.YoloFaceDetector(model=None, conf_threshold=0.5, device=None)[source]#

Ultralytics YOLO face detector, using a community face checkpoint.

There is no Ultralytics-official face-detection weights file (unlike yolov8n-pose.pt), so this downloads a named community checkpoint on first use (see the module docstring) unless model points at a local file.

Parameters:

model (str or None, default None) – A local checkpoint path, a known community checkpoint name (see _YOLO_FACE_URLS), or None for the default "yolov8n-face.pt".
conf_threshold (float, default 0.5) – Minimum detection confidence to keep a face.
device (str or None, default None) – Inference device (e.g. "cpu", "cuda:0"); None lets ultralytics choose.

detect(frame_bgr)[source]#

See FaceDetector.detect().

Parameters:: frame_bgr (ndarray)
Return type:: list[tuple[float, float, float, float]]

class facemask.OpenCVDnnFaceDetector(conf_threshold=0.5)[source]#

cv2.FaceDetectorYN (YuNet) ONNX face detector.

Replaces an earlier Caffe-based res10 SSD detector, which broke outright once OpenCV 5.0 removed cv2.dnn.readNetFromCaffe (and every other non-ONNX/TensorFlow DNN importer) — this project’s opencv-python>=4.8 constraint (analysis/pyproject.toml) has no upper bound, so a fresh uv sync now always resolves to 5.x. YuNet is OpenCV’s own currently-maintained face-detection model (opencv/opencv_zoo), ONNX-based (works with any OpenCV DNN build), and a real accuracy upgrade over the retired res10 model, not just a compatibility shim. No new ML framework dependency beyond opencv-python. Downloads the official OpenCV-hosted, sha256-verified ONNX model to models/ on first use.

Parameters:: conf_threshold (float, default 0.5) – Minimum detection confidence to keep a face.

detect(frame_bgr)[source]#

See FaceDetector.detect().

Parameters:: frame_bgr (ndarray)
Return type:: list[tuple[float, float, float, float]]

facemask.make_detector(backend, model, conf_threshold, device=None)[source]#

Build a face-detection backend by name.

Parameters:

backend ({"mediapipe", "yolov8", "opencv"}) – Which backend to construct.
model (str or None) – Forwarded to YoloFaceDetector as its model argument; ignored by the other two backends.
conf_threshold (float) – Minimum detection confidence to keep a face.
device (str or None, default None) – Forwarded to YoloFaceDetector as its device argument; ignored by the other two backends.

Returns:

A constructed, ready-to-use detector.

Return type:

FaceDetector

Raises:

ValueError – If backend isn’t one of the three known names.

facemask.expand_and_clip(box, margin_frac, frame_w, frame_h)[source]#

Pad a face box and clip it to the frame bounds.

Pads box by margin_frac of its own width/height (so hairline, ears, and chin are covered, not just the tight landmark/detector box), then clips the result to [0, frame_w] x [0, frame_h].

Parameters:

box (Box) – (x1, y1, x2, y2) pixel coordinates of the detected face box.
margin_frac (float) – Fraction of the box’s own width/height to pad on each side (e.g. 0.25 pads a 100 px-wide box by 25 px per side).
frame_w (int) – Frame dimensions in pixels, used as the clip bounds.
frame_h (int) – Frame dimensions in pixels, used as the clip bounds.

Returns:

The padded, clipped (x1, y1, x2, y2) box. Always a valid, non-negative box even when the input box touches the frame edge.

Return type:

Box

Notes

Padding is applied before clipping, so a face near the frame edge is padded first and only then clamped — it cannot produce a negative or out-of-bounds coordinate a caller could mis-slice. See Face Masking for the exact padding/clamp formula.

facemask.apply_mask(frame, boxes, style)[source]#

Blur or solid-fill every box’s region in frame, in place.

Parameters:

frame (numpy.ndarray) – BGR frame, modified in place.
boxes (list of Box) – Regions to mask, e.g. from expand_and_clip(). A degenerate box (x2<=x1 or y2<=y1) is silently skipped.
style ({"blur", "box"}) – "blur" applies a Gaussian blur with its kernel size scaled to the box’s own size; "box" fills the region with a solid color.

Returns:

frame, returned for convenient call-site chaining (it was already modified in place).

Return type:

numpy.ndarray

Notes

See Face Masking for the blur-kernel sizing formula.

See Face Masking.

Speaker Diarization #

Transcribes each microphone’s audio with faster-whisper, diarizes speaker turns with pyannote.audio, then assigns each transcript segment to whichever diarization turn overlaps it most (assign_speakers()) — the standard WhisperX-style recipe. Diarization is optional: without a Hugging Face token, transcription still runs and every segment’s speaker is left None.

Important

pyannote’s diarization models are gated on Hugging Face — you must accept both pyannote/speaker-diarization-3.1 and pyannote/segmentation-3.0’s terms of use and generate an access token before diarization (not just transcription) will work. See User guide for where to configure the token in the app.

from diarize import resolve_device, load_whisper_model, transcribe_audio

device = resolve_device(device_arg=None)
model = load_whisper_model("small", device)
segments, detected_language = transcribe_audio(model, audio_path, language=None)

Transcription + speaker-diarization pipeline for MOSAIC’s Speaker Diarization analysis plugin.

`resolve_device`	Resolve which device to run inference on.
`load_whisper_model`	Load a faster-whisper model once, for reuse across a whole session.
`transcribe_audio`	Transcribe one audio file with a pre-loaded whisper model.
`load_diarization_pipeline`	Load the pyannote diarization pipeline once, for reuse across a session.
`diarize_audio`	Run a pre-loaded diarization pipeline against one audio file.
`assign_speakers`	Label each transcribed segment with its best-overlapping speaker turn.

diarize.resolve_device(device_arg)[source]#

Resolve which device to run inference on.

Parameters:: device_arg (str or None) – Explicit device (e.g. "cuda", "cpu"), or None to auto-detect.
Returns:: device_arg unchanged if given, else "cuda" if a CUDA device is available, else "cpu".
Return type:: str

Notes

Shared by load_whisper_model()/load_diarization_pipeline() so they never disagree about which hardware to use for the same run.

diarize.load_whisper_model(model_size, device)[source]#

Load a faster-whisper model once, for reuse across a whole session.

Parameters:

model_size (str) – faster-whisper model size (e.g. "small", "large-v3").
device (str) – Inference device, typically from resolve_device().

Returns:

A loaded model, ready for repeated transcribe_audio() calls.

Return type:

faster_whisper.WhisperModel

Notes

Call this once per run and pass the result to every transcribe_audio() call in that run — building a fresh model per audio file reloads weights from disk/cache once per microphone instead of once per run.

diarize.transcribe_audio(model, audio_path, language)[source]#

Transcribe one audio file with a pre-loaded whisper model.

Parameters:

model (faster_whisper.WhisperModel) – A model from load_whisper_model().
audio_path (pathlib.Path) – Path to the audio file. faster-whisper decodes/resamples it internally (mono/16kHz) — no separate preprocessing step is needed regardless of the recording’s original sample rate/channel count.
language (str or None) – Force a language code (e.g. "en"), or None to auto-detect.

Returns:

(segments, detected_language).

Return type:

tuple of (list of WhisperSegment, str)

diarize.load_diarization_pipeline(hf_token, device)[source]#

Load the pyannote diarization pipeline once, for reuse across a session.

Parameters:

hf_token (str) – Hugging Face access token with the gated diarization models accepted (see the module docstring’s gating instructions).
device (str) – Inference device, typically from resolve_device().

Returns:

A loaded pipeline, ready for repeated diarize_audio() calls.

Return type:

pyannote.audio.Pipeline

Raises:

RuntimeError – If the token is missing/invalid or the gated models haven’t been accepted yet — raised with actionable setup instructions rather than letting a raw HTTP/auth exception propagate to the caller’s log.

Notes

Call this once per run and pass the result to every diarize_audio() call in that run — building a fresh pipeline per audio file reloads weights from disk/cache once per microphone instead of once per run.

diarize.diarize_audio(pipeline, audio_path, min_speakers, max_speakers)[source]#

Run a pre-loaded diarization pipeline against one audio file.

Parameters:

pipeline (pyannote.audio.Pipeline) – A pipeline from load_diarization_pipeline().
audio_path (pathlib.Path) – Path to the audio file.
min_speakers (int) – Optional pyannote hints for the expected speaker count; 0 means “no hint” and is omitted from the call.
max_speakers (int) – Optional pyannote hints for the expected speaker count; 0 means “no hint” and is omitted from the call.

Returns:

One entry per detected speaker turn.

Return type:

list of DiarizationTurn

Notes

pyannote.audio 4.x’s pipeline call now returns a DiarizeOutput dataclass, not the bare Annotation 3.x returned. Uses exclusive_speaker_diarization (turns with overlapping speech resolved to a single speaker) rather than speaker_diarization (keeps overlaps) — its own docstring calls this out as “adapted to downstream transcription”, exactly this function’s use case via assign_speakers()’s single-best-speaker-per-segment matching, where an overlapping turn would only add ambiguity.

diarize.assign_speakers(whisper_segments, diarization_turns)[source]#

Label each transcribed segment with its best-overlapping speaker turn.

Parameters:

whisper_segments (list of WhisperSegment) – Transcription segments from transcribe_audio(), start/ end in seconds.
diarization_turns (list of DiarizationTurn) – Speaker turns from diarize_audio(), start/end in seconds plus a speaker label. May be empty (diarization was skipped) — every output segment then gets speaker=None.

Returns:

One entry per input segment, in order, with start_ms/ end_ms (rounded to the nearest millisecond) and speaker set to whichever turn overlaps it most by duration, or None if no turn overlaps it at all.

Return type:

list of TranscriptSegment

Notes

Max-overlap interval matching, the standard WhisperX-style recipe. See Speaker Diarization for the exact overlap formula.

class diarize.WhisperSegment[source]#

One faster-whisper transcription segment.

start, end

Segment bounds, in seconds.

Type:: float

text#

Transcribed text for this segment.

Type:: str

class diarize.DiarizationTurn[source]#

One pyannote.audio speaker turn.

start, end

Turn bounds, in seconds.

Type:: float

speaker#

Speaker label (e.g. "SPEAKER_00").

Type:: str

class diarize.TranscriptSegment[source]#

One final, speaker-labeled transcript segment (assign_speakers()’s output).

start_ms, end_ms

Segment bounds, in milliseconds.

Type:: int

speaker#

Assigned speaker label, or None if no diarization turn overlapped this segment (or diarization wasn’t run at all).

Type:: str or None

text#

Transcribed text for this segment.

Type:: str

See Speaker Diarization.

Facial Expression #

Detects faces and their MediaPipe blendshapes per frame, then classifies a dominant expression via one of three interchangeable backends: a transparent, dependency-free weighted-blendshape heuristic (default), a pretrained FER+ ONNX model, or py-feat for real, continuous FACS Action Unit intensities (see the dedicated subsection below).

from expression import MediaPipeExpressionDetector, classify_expression, BLENDSHAPE_NAMES

detector = MediaPipeExpressionDetector()
faces = detector.detect(frame_bgr)
label, score = classify_expression(BLENDSHAPE_NAMES, faces[0].blendshape_scores)

Facial-expression detection and classification backends (rule-based heuristic + FER+ ONNX + py-feat) for MOSAIC’s Facial Expression analysis plugin.

`classify_expression`	Pick the dominant basic-emotion category from blendshape scores.
`MediaPipeExpressionDetector`	MediaPipe Tasks `FaceLandmarker`-based blendshape detector.
`crop_bbox`	Crop a frame to a bounding box, clamped to the frame bounds.
`FerPlusClassifier`	Microsoft FER+ ONNX model backend — 8-category emotion classification.

expression.classify_expression(blendshape_names, blendshape_scores)[source]#

Pick the dominant basic-emotion category from blendshape scores.

Parameters:

blendshape_names (list of str) – MediaPipe blendshape category names, parallel to blendshape_scores (see BLENDSHAPE_NAMES).
blendshape_scores (list of float) – Activation score in [0, 1] for each name. A name referenced by CATEGORY_WEIGHTS but missing here defaults to 0.0 (robust to a caller passing a truncated array).

Returns:

(dominant_expression, dominant_score); dominant_expression is one of CATEGORIES, dominant_score always in [0, 1].

Return type:

tuple of (str, float)

Notes

Each category’s score is the weighted mean (not sum) of its FACS-Action-Unit-derived blendshape weights, so a category listing many blendshapes isn’t unfairly favored over one listing few. Falls back to "Neutral" when the winning score is below _MIN_ACTIVATION. Ties keep the first-seen category in CATEGORIES ("Neutral" listed first) — a deliberate “when in doubt, don’t overclaim an emotion” default, consistent with assign_speakers() leaving speaker=None on a zero-overlap tie. See Facial Expression for the formula.

expression.CATEGORIES: list[str] = ["Neutral", "Happy", "Sad", "Surprised", "Angry", "Disgusted", "Fearful"]#: The 7 basic-emotion categories the heuristic backend classifies into, in argmax tie-break order ("Neutral" listed first — see classify_expression()’s Notes).

expression.CATEGORY_WEIGHTS: dict[str, dict[str, float]]#: {category: {blendshape_name: weight}}. Weighted MEAN (not sum) is taken per category at classification time, so a category listing 6 blendshapes isn’t unfairly favored over one listing 2 — every category’s score stays comparable in [0, 1] regardless of how many shapes it references.

class expression.MediaPipeExpressionDetector(max_faces=5, min_confidence=0.5)[source]#

MediaPipe Tasks FaceLandmarker-based blendshape detector.

Parameters:

max_faces (int, default 5) – Maximum simultaneous faces to detect per frame.
min_confidence (float, default 0.5) – Minimum face-detection/-presence confidence to keep a face.

detect(frame_bgr)[source]#

Detect faces and their blendshape scores in one BGR frame.

Parameters:: frame_bgr (numpy.ndarray) – BGR frame, as returned by cv2.imread/cv2.VideoCapture.
Returns:: One entry per detected face.
Return type:: list of FaceExpression

class expression.FaceExpression(bbox_xyxy, confidence, blendshape_scores)[source]#

One detected face’s bounding box and raw blendshape scores.

Parameters:

bbox_xyxy (tuple[float, float, float, float])
confidence (float)
blendshape_scores (list[float])

bbox_xyxy#

Detection bounding box, (x1, y1, x2, y2) pixels.

Type:: tuple of float

confidence#

Detection confidence. Always 1.0 — see the constructor site’s comment for why a constant is more honest here than a proxy metric.

Type:: float

blendshape_scores#

Per-category activation, in [0, 1], parallel to BLENDSHAPE_NAMES.

Type:: list of float

expression.BLENDSHAPE_NAMES: list[str]#: The standard ARKit-style blendshape category names MediaPipe’s FaceLandmarker outputs when output_face_blendshapes=True. Score lookup is always by category name against this list (never positional), so a future mediapipe version reordering its output categories can’t silently misalign names/scores.

expression.crop_bbox(frame_bgr, bbox_xyxy)[source]#

Crop a frame to a bounding box, clamped to the frame bounds.

Parameters:

frame_bgr (numpy.ndarray) – BGR frame to crop.
bbox_xyxy (tuple of float) – (x1, y1, x2, y2) pixel coordinates, e.g. from FaceExpression.bbox_xyxy.

Returns:

The cropped region. Possibly empty (shape (0, 0, 3)) if the box is degenerate (e.g. fully outside the frame) — callers must check .size before use.

Return type:

numpy.ndarray

Notes

Used by the FER+ backend (ferplus), which needs a tight face crop rather than the full frame.

class expression.FerPlusClassifier[source]#

Microsoft FER+ ONNX model backend — 8-category emotion classification.

Downloads and sha256-verifies emotion-ferplus-8.onnx to models/ on first use (see _ensure_download_verified()).

classify(face_crop_bgr)[source]#

Classify one face crop’s dominant emotion.

Parameters:: face_crop_bgr (numpy.ndarray) – A tight crop around one detected face (e.g. via crop_bbox()). The model card gives no explicit crop/alignment guidance, but FER2013/FER+’s source data is already tightly-cropped near-square face images, so a loose/full-scene frame would be a real preprocessing mismatch against training.
Returns:: (label, score) — label is one of FERPLUS_LABELS, score the softmax probability of that label.
Return type:: tuple of (str, float)

expression.FERPLUS_LABELS: list[str]#: Official FER+ label order (index 0-7). Verified against both the onnx/models model card and the upstream FERPlus training repo’s CSV column order — see the module’s own docstring for the two-source cross-check this pinned down.

py-feat backend (expression.pyfeat) — the third, most detailed backend: real FACS Action Units, not just a dominant-emotion label.

class expression.pyfeat.PyFeatClassifier(device='cpu')[source]#

py-feat Detectorv1 backend — 20-AU + 7-class emotion.

Loads/downloads Detectorv1’s model weights once per process (py-feat’s own cache, outside this project’s control — see module docstring).

Parameters:: device (str)

detect(face_crop_bgr)[source]#

Run AU + emotion detection on one already-cropped face image.

Parameters:: face_crop_bgr (numpy.ndarray) – A tight crop around one detected face (e.g. via crop_bbox()) — the same already-cropped-image contract FerPlusClassifier.classify() uses, so this backend can be wired into run_expression.py’s per-face dispatch identically to the FER+ backend.
Returns:: (dominant_emotion_label, dominant_score, au_values) — au_values keyed by AU_NAMES, each in [0, 1]. Returns ("Neutral", 0.0, {}) if py-feat’s own internal face detector fails to find a face in the crop (a real, if rare, possibility on an already-tightly-cropped image — see the module docstring) rather than raising and aborting the whole analysis run, the same fail-soft precedent run_expression.py’s own degenerate-bbox handling already uses.
Return type:: tuple of (str, float, dict[str, float])

expression.pyfeat.AU_NAMES: list[str]#: The 20 py-feat/Detectorv1 xgb-head Action Units (verified against feat/pretrained.py’s AU_LANDMARK_MAP["Feat"]). Values are a continuous [0, 1] calibrated probability, not the classic FACS 0–5 intensity scale.

expression.pyfeat._fex_row_to_result(au_values, emotion_values)[source]#

Pure argmax-over-emotions + AU-passthrough — the one testable piece of this backend, isolated from the actual Detectorv1 call exactly like ferplus.py’s _softmax_and_label() isolates softmax+argmax from the ONNX session call.

Parameters:

au_values (dict of str to float) – One row of py-feat’s Fex.aus output, as a plain dict (e.g. fex.aus.iloc[0].to_dict()).
emotion_values (dict of str to float) – One row of py-feat’s Fex.emotions output, as a plain dict.

Returns:

(dominant_emotion_label, dominant_score, au_values) — dominant_emotion_label is one of _EMOTION_COLUMN_TO_LABEL’s values (or the raw column name if unrecognized), au_values is keyed exactly as passed in, values rounded to 4 decimals.

Return type:

tuple of (str, float, dict[str, float])

Notes

NaN values (py-feat’s own per-row failure marker when its internal models can’t produce a score) are treated as 0.0 rather than propagating/crashing argmax — mirrors this codebase’s established “default missing/bad data to 0.0, never crash” convention (e.g. the BLENDSHAPE_NAMES lookup-by-name-with-default on the detection side).

Important

import feat unconditionally pulls in torchcodec at module load time, which needs a torchcodec-compatible FFmpeg (versions 4–8, a shared/DLL build) discoverable on PATH at runtime — even though this backend never touches video I/O. See the module’s own docstring and Facial Expression’s recommendations for the full diagnosis if this backend fails to construct.

See Facial Expression.

Multi-Camera Gaze Fusion #

For each camera that sees a face, solves a real (not weak-perspective) 3D head pose via cv2.solvePnP, perturbs it by a small iris-offset heuristic into one camera-local gaze ray, transforms every contributing camera’s ray into shared room coordinates, and triangulates them into one fused ray and (if a target plane is calibrated) a target point.

from gaze import transform_ray_to_room, closest_point_of_rays

rays_room = [transform_ray_to_room(origin, direction, cam.extrinsic_rt)
             for origin, direction, cam in per_camera_rays]
origins, directions = zip(*rays_room)
fused_point, residual_rms = closest_point_of_rays(origins, directions)

Per-camera 3D gaze-ray estimation and multi-camera ray-fusion math for MOSAIC’s Multi-Camera Gaze Fusion analysis plugin.

`camera_ray_from_pose`	Turn a solved head pose plus a 2D iris-offset into a 3D camera-space gaze ray.
`transform_ray_to_room`	Transform a camera-local ray into room coordinates.
`closest_point_of_rays`	Triangulate the point closest to a set of 3D rays (least squares).
`ray_plane_intersection`	Intersect a ray with a plane.
`MediaPipeGazeEstimator3D`	MediaPipe Tasks `FaceLandmarker` + `cv2.solvePnP`-based 3D gaze estimator.

gaze.camera_ray_from_pose(rotation, translation, gaze_dx, gaze_dy, eye_origin_model_mm, max_eye_yaw_deg=30.0, max_eye_pitch_deg=20.0)[source]#

Turn a solved head pose plus a 2D iris-offset into a 3D camera-space gaze ray.

Parameters:

rotation (numpy.ndarray) – 3x3 head-pose rotation, camera-local (from cv2.solvePnP).
translation (numpy.ndarray) – Length-3 head-pose translation, camera-local, mm.
gaze_dx (float) – 2D iris-offset heuristic, each in [-1, 1] — same convention as python/pose/gaze_estimator.py’s GazeResult.
gaze_dy (float) – 2D iris-offset heuristic, each in [-1, 1] — same convention as python/pose/gaze_estimator.py’s GazeResult.
eye_origin_model_mm (numpy.ndarray) – Length-3 eye-center point in the generic face model’s own coordinates, mm (see estimator.py’s EYE_ORIGIN_MODEL_MM).
max_eye_yaw_deg (float, default 30.0) – Eye-in-socket yaw bound, applied when gaze_dx = ±1.
max_eye_pitch_deg (float, default 20.0) – Eye-in-socket pitch bound, applied when gaze_dy = ±1.

Returns:

(origin, direction) — the ray’s origin (mm) and unit direction, both in the camera’s local coordinate frame. gaze_dx = gaze_dy = 0 returns the head’s own forward direction, unperturbed.

Return type:

tuple of (numpy.ndarray, numpy.ndarray)

Notes

The direction composes the head’s own forward axis (+Z of the face model, metric — solved via cv2.solvePnP against the camera’s real intrinsics) with a small eye-in-socket yaw/pitch perturbation derived from gaze_dx/gaze_dy. This deliberately does not claim true stereo eye depth (unobservable from monocular iris landmarks) — it only separates “which way is the head pointing” (metric, solved) from “which way are the eyes rotated within it” (heuristic, bounded by max_eye_yaw_deg/max_eye_pitch_deg), which is strictly more information than the live 2D-only gaze_dx/gaze_dy estimator this pipeline replaces. See Multi-Camera Gaze Fusion for the full derivation.

gaze.transform_ray_to_room(origin_cam, direction_cam, extrinsic_rt)[source]#

Transform a camera-local ray into room coordinates.

Parameters:

origin_cam (numpy.ndarray) – Ray origin, camera-local, mm.
direction_cam (numpy.ndarray) – Ray unit direction, camera-local.
extrinsic_rt (array_like) – This camera’s extrinsic pose: a flat length-16 sequence or a 4x4 numpy.ndarray (row-major — see the module docstring’s convention note).

Returns:

(origin_room, direction_room) — the same ray, in room coordinates.

Return type:

tuple of (numpy.ndarray, numpy.ndarray)

gaze.closest_point_of_rays(origins, directions)[source]#

Triangulate the point closest to a set of 3D rays (least squares).

Parameters:

origins (sequence of array_like) – One ray origin per contributing camera, room coordinates, mm.
directions (sequence of array_like) – One ray direction per contributing camera (need not be pre-normalized — renormalized internally), room coordinates.

Returns:

(point, residual_rms) — the fused 3D point and the RMS perpendicular distance from it to each contributing ray (always 0.0 for a single ray, which trivially “fuses” to a point on itself).

Return type:

tuple of (numpy.ndarray, float)

Notes

Standard closest-point-of-multiple-rays least squares: minimizes \(\sum_i \lVert (I - d_i d_i^\mathsf{T})(x - o_i) \rVert^2\), i.e. the point whose summed squared perpendicular distance to every ray is smallest. Falls back to a pseudo-inverse (rather than raising) when the accumulated normal matrix is near-singular — e.g. all rays nearly parallel — so a genuinely degenerate configuration still returns a best-effort point; the resulting (large) residual_rms is what should flag it as untrustworthy to a caller, not an exception. See Multi-Camera Gaze Fusion for the full derivation.

gaze.ray_plane_intersection(origin, direction, plane_point, plane_normal, eps=1e-09)[source]#

Intersect a ray with a plane.

Parameters:

origin (array_like) – Ray origin, room coordinates, mm.
direction (array_like) – Ray direction (need not be pre-normalized).
plane_point (array_like) – Any point on the plane, room coordinates, mm.
plane_normal (array_like) – Plane normal (need not be pre-normalized — renormalized internally).
eps (float, default 1e-9) – Below this, direction is treated as parallel to the plane.

Returns:

The 3D point where the ray (origin + t*direction, t >= 0) intersects the plane, or None if the ray is (near-)parallel to the plane or the intersection lies behind the ray’s origin (t < 0 — gaze pointing away from the surface). See Multi-Camera Gaze Fusion for the derivation.

Return type:

numpy.ndarray or None

class gaze.MediaPipeGazeEstimator3D(max_faces=1, min_confidence=0.5)[source]#

MediaPipe Tasks FaceLandmarker + cv2.solvePnP-based 3D gaze estimator.

Thread-unsafe — one instance per process, matching every other MediaPipe-backed detector in this codebase.

Parameters:

max_faces (int, default 1) – Maximum simultaneous faces to detect per frame. The Multi-Camera Gaze Fusion pipeline only ever consumes subject 0, so this is left at its single-subject default in practice.
min_confidence (float, default 0.5) – Minimum face-detection/-presence confidence to keep a face.

detect(frame_bgr, camera_matrix, dist_coeffs)[source]#

Detect faces and solve each one’s 3D head pose in one BGR frame.

Parameters:

frame_bgr (numpy.ndarray) – BGR frame, as returned by cv2.imread/cv2.VideoCapture.
camera_matrix (numpy.ndarray) – 3x3 camera intrinsic matrix — this camera’s real calibration (read from session_meta.json by the caller), required for a metric solvePnP solve.
dist_coeffs (numpy.ndarray) – Length-5 distortion coefficients, this camera’s real calibration.

Returns:

One entry per successfully-solved face. Faces missing iris landmarks, with a degenerate eye width, or a failed solvePnP fit are silently skipped — a documented, low-risk degrade (one fewer camera contributing to that frame’s fusion), not a crash.

Return type:

list of FaceGazeSample

class gaze.FaceGazeSample(bbox_xyxy, confidence, head_rotation, head_translation, gaze_dx, gaze_dy)[source]#

One detected face’s head pose and 2D iris-offset heuristic.

Parameters:

bbox_xyxy (tuple[float, float, float, float])
confidence (float)
head_rotation (ndarray)
head_translation (ndarray)
gaze_dx (float)
gaze_dy (float)

bbox_xyxy#

Detection bounding box, (x1, y1, x2, y2) pixels.

Type:: tuple of float

confidence#

Detection confidence. Always 1.0 (FaceLandmarkerResult carries no per-face detection score).

Type:: float

head_rotation#

3x3 head-pose rotation, camera-local (from cv2.solvePnP).

Type:: numpy.ndarray

head_translation#

Length-3 head-pose translation, camera-local, mm.

Type:: numpy.ndarray

gaze_dx, gaze_dy

2D iris-offset heuristic, each in [-1, 1] — same heuristic as python/pose/gaze_estimator.py.

Type:: float

See Multi-Camera Gaze Fusion and Room (Extrinsic) Calibration (this plugin consumes the room/extrinsic calibration solved there).

3D Pose Reconstruction #

Triangulates each camera’s already-computed 2D COCO keypoints (from the Pose plugin above) into one 3D skeleton per detected person, per frame. Three stages: multi-view DLT triangulation with one-shot reprojection- error outlier rejection (triangulate_with_rejection()), cross-camera person association by reusing that same triangulation cost as a matching score (cluster_people()), and greedy nearest-centroid tracking across frames (PersonTracker3D).

Important

Needs the Pose plugin to have already been run on at least 2 cameras in the session (for the 2D keypoints) and room/extrinsic calibration to have been solved (Room (Extrinsic) Calibration) — running this plugin against a session missing either prerequisite fails with a clear error rather than a fabricated result.

Multi-view DLT triangulation, cross-camera person association, and 3D track identity for MOSAIC’s 3D Pose Reconstruction analysis plugin.

`CameraGeom`	One calibrated camera's geometry, as needed for triangulation.
`invert_rt`	Invert a rigid-body (rotation + translation) transform.
`normalize_point`	Undistort a single pixel into ideal, K-free normalized coordinates.
`triangulate_point_dlt`	Classic linear multi-view DLT (Direct Linear Transform) triangulation.
`reproject_error_px`	Compute one view's reprojection error for a triangulated point.
`triangulate_with_rejection`	Triangulate one keypoint across cameras, rejecting bad views once.
`PersonObservation`
`pairwise_cost`	Average 2-view reprojection error (px) across every keypoint index visible (>=min_visibility) in BOTH a and b.
`match_camera_pair`	Hungarian-assigns obs_a<->obs_b by pairwise_cost(), then discards any assigned pair whose cost exceeds max_pair_cost_px — Hungarian minimizes TOTAL cost but doesn't itself refuse an individually-bad pair when one side has more/fewer detections than the other, so this threshold is the actual "not obviously the same person" gate.
`cluster_people`	observations: {camera_index: [PersonObservation, ...]}.
`PersonTracker3D`

class pose3d.CameraGeom(index, camera_matrix, dist_coeffs, extrinsic_rt)[source]#

One calibrated camera’s geometry, as needed for triangulation.

Parameters:

index (int)
camera_matrix (ndarray)
dist_coeffs (ndarray)
extrinsic_rt (ndarray)

index#

Camera index, matching the session’s camN numbering.

Type:: int

camera_matrix#

Real (distorted-space) intrinsic matrix, reshaped to (3, 3).

Type:: numpy.ndarray

dist_coeffs#

OpenCV distortion coefficients, reshaped to a flat vector.

Type:: numpy.ndarray

extrinsic_rt#

Row-major rigid transform, reshaped to (4, 4) — room-from-camera (point_room = R @ point_camera + t), matching CalibrationData::extrinsicRt exactly (see Room (Extrinsic) Calibration).

Type:: numpy.ndarray

camera_from_room#

Computed automatically as invert_rt(extrinsic_rt) — the inverse transform every projection in this module actually uses.

Type:: numpy.ndarray

pose3d.invert_rt(m)[source]#

Invert a rigid-body (rotation + translation) transform.

Parameters:: m (array_like) – A row-major rigid transform, as a (4, 4) matrix or a flat length-16 array — [R | t] with the bottom row implicitly [0, 0, 0, 1].
Returns:: The inverse transform, shape (4, 4): [R^T | -R^T @ t].
Return type:: numpy.ndarray

Notes

Must stay mathematically identical to room_frame::invert() (src/calibration/room_frame_solver.cpp) — Python cannot call the C++ function directly, so this is a small, deliberate reimplementation of the exact same formula. See 3D Pose Reconstruction for the room-frame convention this assumes.

pose3d.normalize_point(uv_px, cam)[source]#

Undistort a single pixel into ideal, K-free normalized coordinates.

Parameters:

uv_px (array_like) – A single pixel coordinate (u, v), in that camera’s real distorted pixel space.
cam (CameraGeom) – The camera whose intrinsics/distortion coefficients to undistort against.

Returns:

Length-2 array, the ideal (undistorted, K-free) normalized coordinate.

Return type:

numpy.ndarray

Notes

Uses cv2.undistortPoints() with no P= argument — K is deliberately left out of the result, since triangulate_point_dlt() only ever consumes already-normalized points and its projection matrices (see projection_matrix()) carry no K term either.

pose3d.projection_matrix(cam)[source]#

Build a camera’s K-free (3, 4) projection matrix.

Parameters:: cam (CameraGeom) – The camera to build a projection matrix for.
Returns:: P = camera_from_room[:3, :4], shape (3, 4) — no K term, matching normalize_point()’s already-K-free output, so the two combine directly in triangulate_point_dlt().
Return type:: numpy.ndarray

pose3d.triangulate_point_dlt(points_normalized, projection_matrices)[source]#

Classic linear multi-view DLT (Direct Linear Transform) triangulation.

Parameters:

points_normalized (sequence of array_like) – One ideal, K-free normalized (u, v) point per contributing view (see normalize_point()), same order as projection_matrices.
projection_matrices (sequence of array_like) – One (3, 4) K-free projection matrix per contributing view (see projection_matrix()), same order as points_normalized.

Returns:

The triangulated 3D point in room space (mm), shape (3,), or None if triangulation isn’t possible (see Notes).

Return type:

numpy.ndarray or None

Notes

Pure numpy, zero cv2. Each view i contributes 2 homogeneous rows (u_i * P_i[2,:] - P_i[0,:] and v_i * P_i[2,:] - P_i[1,:]) to one (2N, 4) system; the 3D point is the right-singular-vector for the smallest singular value, dehomogenized. Requires at least 2 views. Returns None on too few views, a failed SVD, or a degenerate (near-zero homogeneous coordinate) solution — never a fabricated point. See 3D Pose Reconstruction for the full derivation.

pose3d.project_point_px(point_room, cam)[source]#

Project a 3D room-space point into one camera’s real pixel space.

Parameters:

point_room (array_like) – A 3D point in room space (mm), shape (3,).
cam (CameraGeom) – The camera to project through — its real (distorted) intrinsics are used, not the K-free normalized-coordinate space the rest of this module operates in.

Returns:

Length-2 array, the projected pixel coordinate.

Return type:

numpy.ndarray

Notes

The shared primitive behind both reproject_error_px() (triangulation quality) and run_pose3d.py’s per-camera overlay precomputation (Skeleton3DPerson::reprojectedPx, consumed by mosaic::Skeleton3DResult with zero calibration math on the C++ side).

pose3d.reproject_error_px(point_room, cam, pixel_observed)[source]#

Compute one view’s reprojection error for a triangulated point.

Parameters:

point_room (array_like) – A triangulated 3D point in room space (mm), shape (3,).
cam (CameraGeom) – The camera to reproject through.
pixel_observed (array_like) – The originally observed 2D pixel coordinate this camera actually detected, to compare the reprojection against.

Returns:

Euclidean pixel distance between the reprojected point and pixel_observed — an interpretable, distortion-aware quality metric, consistent with RoomCalibrationManager’s own extrinsic-solve reprojection-RMS report (see Room (Extrinsic) Calibration).

Return type:

float

class pose3d.TriangulationResult(point_room, used_views, per_view_error_px)[source]#

The outcome of a single triangulate_with_rejection() call.

Parameters:

point_room (ndarray)
used_views (list)
per_view_error_px (dict)

point_room#

The triangulated 3D point in room space (mm), shape (3,).

Type:: numpy.ndarray

used_views#

Camera indices that actually contributed to point_room — after visibility filtering and, if a re-triangulation happened, after outlier rejection too.

Type:: list of int

per_view_error_px#

Each entry in used_views’ own reprojection error (px) against point_room, keyed by camera index.

Type:: dict of int to float

pose3d.triangulate_with_rejection(observations, cameras, max_reprojection_error_px=15.0, min_visibility=0.1)[source]#

Triangulate one keypoint across cameras, rejecting bad views once.

Parameters:

observations (dict of int to tuple) – {cam_idx: (pixel_uv, visibility)} — one observation per camera that detected this keypoint at all, regardless of confidence.
cameras (dict of int to CameraGeom) – Every calibrated camera available for this session, keyed by index.
max_reprojection_error_px (float, default 15.0) – A view whose reprojection error exceeds this (px) after the first triangulation pass is dropped before a single re-triangulation.
min_visibility (float, default 0.1) – Observations below this visibility/confidence are excluded before triangulation even starts.

Returns:

None if fewer than 2 views survive visibility filtering, the initial triangulation fails, or fewer than 2 views survive outlier rejection — never a fabricated point.

Return type:

TriangulationResult or None

Notes

Filters by min_visibility, requires >=2 remaining cameras, triangulates via normalize_point() + triangulate_point_dlt(), then drops any view whose real reprojection error exceeds max_reprojection_error_px and re-triangulates once from the remainder — no iterative chase, matching pose_kinematics.cpp’s “skip, don’t fabricate” discipline (see Pose Kinematics).

class pose3d.PersonObservation(camera_index: 'int', person_index: 'int', keypoints_px: 'np.ndarray', visibilities: 'np.ndarray')[source]#

Parameters:

camera_index (int)
person_index (int)
keypoints_px (ndarray)
visibilities (ndarray)

pose3d.pairwise_cost(a, b, cam_a, cam_b, min_shared_keypoints=4, min_visibility=0.1)[source]#

Average 2-view reprojection error (px) across every keypoint index visible (>=min_visibility) in BOTH a and b. inf if fewer than min_shared_keypoints such indices exist, or if 2-view triangulation fails for a majority of them (a near-parallel/degenerate configuration — not evidence of a real match either way, so treated as “can’t assess”, not “definitely different people”).

Parameters:

a (PersonObservation)
b (PersonObservation)
cam_a (CameraGeom)
cam_b (CameraGeom)
min_shared_keypoints (int)
min_visibility (float)

Return type:

float

pose3d.match_camera_pair(obs_a, obs_b, cam_a, cam_b, max_pair_cost_px=20.0, min_shared_keypoints=4)[source]#

Hungarian-assigns obs_a<->obs_b by pairwise_cost(), then discards any assigned pair whose cost exceeds max_pair_cost_px — Hungarian minimizes TOTAL cost but doesn’t itself refuse an individually-bad pair when one side has more/fewer detections than the other, so this threshold is the actual “not obviously the same person” gate. Returns (person_index_a, person_index_b, cost) triples for accepted pairs only.

Parameters:

cam_a (CameraGeom)
cam_b (CameraGeom)
max_pair_cost_px (float)
min_shared_keypoints (int)

pose3d.cluster_people(observations, cameras, max_pair_cost_px=20.0, min_shared_keypoints=4)[source]#

observations: {camera_index: [PersonObservation, …]}. Runs match_camera_pair() over EVERY camera pair (not just adjacent ones — a room-scale rig may have non-adjacent pairs with better mutual visibility than adjacent ones), unions accepted matches via union-find over (camera_index, person_index) nodes, and returns one cluster (list of (camera_index, person_index)) per connected component spanning >=2 distinct cameras. Singleton (1-camera, unmatched) clusters are dropped — a single view can’t be triangulated.

Known limitation, not fixed here: any single accepted pairwise match unions its two nodes regardless of how many OTHER camera pairs agree — with only 2 cameras total, two genuinely different people can (rarely) still pass max_pair_cost_px if their rays happen to nearly cross by coincidence (2 independent rays in 3D always have SOME closest point; reprojection error alone can’t rule out a coincidental near-intersection the way a 3rd independent view’s disagreement could). Rooms with 3+ calibrated cameras are far less exposed to this, since a false-positive 2-camera pairing only survives into the union if no other accepted pair disagrees — but it is not structurally impossible even then. A cost threshold tuned tighter than the default (max_pair_cost_px) is the practical mitigation; a full multi-pair consistency vote is future work.

Parameters:

max_pair_cost_px (float)
min_shared_keypoints (int)

class pose3d.TrackedPerson3D(track_id: 'int', last_tick: 'int', last_centroid_room: 'np.ndarray')[source]#

Parameters:

track_id (int)
last_tick (int)
last_centroid_room (ndarray)

class pose3d.PersonTracker3D(max_gap_ticks=5, max_jump_mm=400.0)[source]#

Parameters:

max_gap_ticks (int)
max_jump_mm (float)

update(tick, cluster_centroids)[source]#

Returns one track_id per input centroid, same order. Existing tracks within max_jump_mm are matched greedily (nearest pair first); unmatched existing tracks age by (tick - last_tick) and are dropped once that exceeds max_gap_ticks; unmatched new centroids get a fresh monotonically-increasing track_id.

Parameters:: tick (int)
Return type:: list

See 3D Pose Reconstruction. Consumes the room/extrinsic calibration solved in Room (Extrinsic) Calibration, and the Pose plugin’s own per-camera .pose.json output (Pose Kinematics’s data source) as its 2D input.

Remote Heart Rate (rPPG)#

Important

EXPERIMENTAL — research-grade heart-rate estimate only, not a medical device, not clinically validated. See Remote Heart Rate (rPPG) for the full accuracy discussion before relying on any output.

Extracts a forehead/cheek skin-color signal per frame (MediaPipeFaceRoiExtractor), combines its RGB channels into one pulse signal via a selectable backend (naive Green, or the more motion-robust CHROM/POS chrominance methods — POS is the default), then bandpass-filters and Welch-periodogram-analyzes each sliding time window to estimate BPM and a pulse-SNR quality score. Deliberately offers no frame-skip option, unlike every sibling plugin — skipping frames would downsample the pulse signal itself below what Nyquist needs for the physiological frequency band.

from rppg import BACKENDS, bandpass_filter, estimate_hr_welch

pulse_signal = BACKENDS["pos"](rgb_means)          # (N, 3) -> (N,)
filtered = bandpass_filter(pulse_signal, fs=frame_rate_hz)
bpm, snr_db = estimate_hr_welch(filtered, fs=frame_rate_hz)

Face-ROI extraction, classical signal-combination algorithms (Green/ CHROM/POS), and Welch-periodogram heart-rate estimation for MOSAIC’s Remote Heart Rate (rPPG) analysis plugin — EXPERIMENTAL, research-grade only, not a medical device.

`MediaPipeFaceRoiExtractor`	MediaPipe Tasks `FaceLandmarker`-based skin-ROI RGB sampler.
`FaceRoiSample`	One frame's detected face ROI and its mean RGB color.
`normalize_channels`	Divide each channel by its own temporal mean.
`green_signal`	Naive baseline: the raw green channel, mean-centered.
`chrom_signal`	CHROM — chrominance-based pulse extraction.
`pos_signal`	POS — Plane-Orthogonal-to-Skin pulse extraction (default backend).
`bandpass_filter`	Zero-phase Butterworth bandpass filter.
`estimate_hr_welch`	Estimate heart rate via Welch's periodogram peak frequency.
`median_smooth`	Centered, NaN-aware median filter over per-window BPM estimates.

class rppg.MediaPipeFaceRoiExtractor(min_confidence=0.5)[source]#

MediaPipe Tasks FaceLandmarker-based skin-ROI RGB sampler.

Parameters:: min_confidence (float, default 0.5) – Minimum face-detection/-presence confidence to keep a detection.

extract(frame_bgr)[source]#

Detect a face and sample its lower-face ROI’s mean RGB color.

Parameters:: frame_bgr (numpy.ndarray) – BGR frame, as returned by cv2.VideoCapture.
Returns:: None if no face was detected this frame — callers must treat this as a real gap, not fabricate a sample (matches this codebase’s established “skip missing samples” discipline, e.g. pose_kinematics.hpp).
Return type:: FaceRoiSample or None

class rppg.FaceRoiSample(roi_bbox_px, rgb_mean)[source]#

One frame’s detected face ROI and its mean RGB color.

Parameters:

roi_bbox_px (tuple[int, int, int, int])
rgb_mean (tuple[float, float, float])

roi_bbox_px#

(x, y, w, h) pixel bounding box of the sampled ROI polygon — kept for the debug overlay, not used by the signal-processing math.

Type:: tuple of int

rgb_mean#

Mean (R, G, B) pixel value inside the ROI polygon, in [0,255].

Type:: tuple of float

rppg.normalize_channels(rgb)[source]#

Divide each channel by its own temporal mean.

Cn = C / mean(C) — the standard first step shared by CHROM and POS: it removes each channel’s own DC/brightness level so that illumination differences between R, G, and B don’t dominate over the much smaller pulse-induced color variation the whole method exists to recover.

Parameters:: rgb (numpy.ndarray) – Shape (N, 3), columns R, G, B, rows are per-frame temporal samples within one analysis window.
Returns:: Same shape, each column divided by its own mean.
Return type:: numpy.ndarray
Raises:: ValueError – If any channel’s temporal mean is exactly zero (a degenerate all-black window) — dividing would produce inf/nan.

rppg.green_signal(rgb)[source]#

Naive baseline: the raw green channel, mean-centered.

Verkruysse, Svaasand & Nelson (2008), “Remote plethysmographic imaging using ambient light” — the original, simplest rPPG method. Green has the strongest hemoglobin-absorption response of the three channels, but this method has no motion/illumination compensation at all — kept as a fast baseline for comparison/debugging, not the default backend.

Parameters:: rgb (numpy.ndarray) – Shape (N, 3), columns R, G, B.
Returns:: Shape (N,), the mean-centered green channel.
Return type:: numpy.ndarray

rppg.chrom_signal(rgb)[source]#

CHROM — chrominance-based pulse extraction.

de Haan & Jeanne, IEEE TBME 2013, “Robust Pulse Rate From Chrominance-Based rPPG” — chrominance signals Ω = 3R − 2G, Φ = 1.5R + G − 1.5B, alpha-tuned combination.

Applies the standard temporal-mean normalization (via normalize_channels()) before the chrominance projection — the theoretically-required step for CHROM’s motion/illumination cancellation to hold: without it, the DC brightness term dominates and the paper’s own skin-reflection-model argument for why the chrominance signals cancel illumination changes no longer applies.

Notes

One real ambiguity, surfaced honestly rather than silently resolved during verification: a reference implementation inspected while building this (phuselab/pyVHR, cpu_CHROM) applies the 3R−2G / 1.5R+G−1.5B formula directly to raw (non-normalized) RGB in the one function body actually retrieved — normalization may happen upstream in that library’s own RGB-extraction stage, which wasn’t independently confirmed. If CHROM’s output quality looks wrong in practice, re-verify this normalization choice against the original IEEE paper’s own equations directly, not just a secondary implementation, before assuming a bug elsewhere.

Parameters:: rgb (numpy.ndarray) – Shape (N, 3), columns R, G, B.
Returns:: Shape (N,), the combined chrominance pulse signal. All-zero if the window is degenerate (Φ’s temporal std is ~0).
Return type:: numpy.ndarray

rppg.pos_signal(rgb)[source]#

POS — Plane-Orthogonal-to-Skin pulse extraction (default backend).

Wang, den Brinker, Stuijk & de Haan, IEEE TBME 2017, “Algorithmic Principles of Remote-PPG” — generally regarded as the best classical (non-deep-learning) rPPG method. Verified directly against a real, cited reference implementation (pavisj/rppg-pos, pos_face_seg.py) rather than reconstructed from memory: temporal normalization, then the fixed projection matrix [[0,1,-1], [-2,1,1]] applied to Cn = [Rn; Gn; Bn], i.e. Xs = Gn − Bn, Ys = −2·Rn + Gn + Bn, combined as α = std(Xs)/std(Ys), S = Xs + α·Ys.

Notes

Deliberate simplification vs. the reference implementation: the reference runs this projection over short (~1.6s) overlapping windows with overlap-add reconstruction, tuned for real-time streaming use. This implementation applies one projection per (longer, caller-supplied) HR-analysis window instead — a documented, understood divergence from that streaming-specific implementation detail, not a misunderstanding of the underlying algorithm.

Parameters:: rgb (numpy.ndarray) – Shape (N, 3), columns R, G, B.
Returns:: Shape (N,), the combined POS pulse signal. All-zero if the window is degenerate (Ys’s temporal std is ~0).
Return type:: numpy.ndarray

rppg.BACKENDS: dict[str, Callable]#: {"green": green_signal, "chrom": chrom_signal, "pos": pos_signal} — the backend-name → pure-function dispatch table run_rppg.py’s --backend argument resolves against.

rppg.bandpass_filter(signal, fs, low_hz=0.7, high_hz=3.0, order=4)[source]#

Zero-phase Butterworth bandpass filter.

Restricts signal to the physiological pulse band. This also removes DC/slow drift, so no separate detrending stage is needed for windows this short — the bandpass’s own low cutoff subsumes it.

Parameters:

signal (numpy.ndarray) – 1-D pulse signal, uniformly sampled at fs Hz.
fs (float) – Sample rate in Hz.
low_hz (float) – Passband edges in Hz. Defaults (0.7-3.0 Hz = 42-180 BPM) cover adult resting-to-moderate-exertion heart rate.
high_hz (float) – Passband edges in Hz. Defaults (0.7-3.0 Hz = 42-180 BPM) cover adult resting-to-moderate-exertion heart rate.
order (int, default 4) – Butterworth filter order.

Returns:

The filtered signal, same shape as input. If the signal is too short for scipy.signal.filtfilt’s required padding length, degrades gracefully to a mean-centered (unfiltered) copy rather than raising — matches this codebase’s “skip/degrade, don’t crash” discipline for too-little-data cases.

Return type:

numpy.ndarray

rppg.estimate_hr_welch(signal, fs, low_hz=0.7, high_hz=3.0)[source]#

Estimate heart rate via Welch’s periodogram peak frequency.

Parameters:

signal (numpy.ndarray) – 1-D, already-bandpass-filtered pulse signal (see bandpass_filter()), uniformly sampled at fs Hz.
fs (float) – Sample rate in Hz.
low_hz (float) – Physiological band to search for the dominant peak.
high_hz (float) – Physiological band to search for the dominant peak.

Returns:

bpm (float or None) – Estimated heart rate, or None if the signal is too short or no dominant frequency exists in the physiological band.
snr_db (float or None) – A pulse-signal-quality metric: the ratio (in dB) of spectral power near the detected peak frequency and its first harmonic versus the remaining power in the analyzed band. Higher is more confident. This is a standard, documented SNR definition in the spirit of the rPPG literature’s “pulse SNR” concept (signal power concentrated at the pulse frequency + its harmonic vs. spread elsewhere) — it is not a verified reproduction of one specific paper’s exact formula (unlike the CHROM/POS projections in rppg.algorithms, which were checked against primary sources), since no single canonical SNR formula was independently confirmed during this feature’s research pass.

Return type:

tuple[float | None, float | None]

Notes

Minimum usable length is enforced implicitly by scipy.signal.welch (returns coarser resolution rather than raising for short input); a signal shorter than roughly 2 seconds at typical camera frame rates will usually not resolve a meaningful frequency at all, in which case both return values are None.

rppg.median_smooth(values, window=3)[source]#

Centered, NaN-aware median filter over per-window BPM estimates.

Parameters:

values (numpy.ndarray) – 1-D array of per-hop BPM estimates; NaN marks a window with no reliable estimate (never fabricated — see estimate_hr_welch()).
window (int, default 3) – Filter width in windows. 1 disables smoothing (returned unchanged). Even values are silently rounded up to the next odd number, matching this project’s established defensive convention for smoothing-window parity (item 16, pose kinematics).

Returns:

Same length as values. A position’s output is the median of the valid (non-NaN) values within its centered window — one bad neighboring window doesn’t blank out an otherwise-good estimate, and a position with no valid values in range stays NaN.

Return type:

numpy.ndarray

See Remote Heart Rate (rPPG).

Motion Tracking #

Run from the Session Browser, not a live Analysis-tab plugin — see User guide. Background-subtraction blob detection (CentroidTracker) plus greedy nearest-centroid tracking across frames, independent of the Pose plugin’s keypoint-based approach — see Motion Tracking for an explicit contrast between the two.

from motion import CentroidTracker

tracker = CentroidTracker(mm_per_px=1.0)
tracks = tracker.update(frame_bgr, timestamp_ns=0, fps=30.0)

Centroid tracking and heatmap/trajectory visualization for MOSAIC’s Motion analysis (run from the Session Browser, not a live Analysis-tab plugin).

`CentroidTracker`	Greedy nearest-neighbour multi-object tracker over MOG2 foreground blobs.
`Track`	One tracked animal's recent position/area history.
`draw_tracks`	Overlay tracks, centroids, IDs, and velocity on a BGR frame.
`generate_heatmap`	Render a Gaussian-smoothed position density map.
`generate_trajectory_plot`	Plain trajectory line plot, one colour per animal.
`generate_velocity_histogram`	Histogram of per-frame velocities, with median/mean markers.

class motion.CentroidTracker(min_area=400, max_area=7000, max_distance=80.0, max_lost=20, learning_rate=-1, close_kernel=9, open_kernel=5, history=500, var_threshold=16.0, mm_per_px=1.0, n_animals=0)[source]#

Greedy nearest-neighbour multi-object tracker over MOG2 foreground blobs.

See the module docstring for the full 7-step pipeline.

Parameters:

min_area (int) – Contour area thresholds (px²). Tune to your arena / camera height. Typical mouse at ~40 cm camera height: 500-6000 px².
max_area (int) – Contour area thresholds (px²). Tune to your arena / camera height. Typical mouse at ~40 cm camera height: 500-6000 px².
max_distance (float, default 80.0) – Maximum centroid movement between frames (px) for assignment. See Motion Tracking for the assignment rule.
max_lost (int, default 20) – Frames a track can be missing before deletion.
learning_rate (float, default -1) – MOG2 learning rate (-1 = automatic).
close_kernel (int) – Morphological structuring element sizes (px).
open_kernel (int) – Morphological structuring element sizes (px).
history (int, default 500) – MOG2 frame history for the background model.
var_threshold (float, default 16.0) – MOG2 Mahalanobis distance threshold (lower = more sensitive).
mm_per_px (float, default 1.0) – Manual pixel-to-real-world scale factor, calibrated from arena dimensions. 1.0 means outputs stay in pixels.
n_animals (int, default 0) – Expected number of animals. Limits track creation to prevent phantom tracks from lighting artefacts. 0 = unlimited.

update(frame, timestamp_ns=0, fps=30.0)[source]#

Process one frame and return the currently active tracks.

Parameters:

frame (numpy.ndarray) – BGR or grayscale frame from OpenCV.
timestamp_ns (int, default 0) – Wall-clock nanoseconds for velocity computation.
fps (float, default 30.0) – Frames per second (used only for velocity in mm/s).

Returns:

Every currently active (not-yet-pruned) track, including ones not matched this frame.

Return type:

list of Track

reset()[source]#

Clear all tracks and rebuild the background model from scratch.

Return type:: None

set_roi(mask)[source]#

Restrict detection to a region of interest.

Parameters:: mask (numpy.ndarray) – Binary uint8 array (255 = keep, 0 = ignore), the same size as the input frame.
Return type:: None

class motion.Track(id, positions=<factory>, timestamps_ns=<factory>, areas=<factory>, last_frame=0, lost_count=0)[source]#

One tracked animal’s recent position/area history.

Parameters:

id (int)
positions (deque)
timestamps_ns (deque)
areas (deque)
last_frame (int)
lost_count (int)

id#

Stable track ID, assigned once at creation.

Type:: int

positions#

Recent (cx, cy) centroid history, most recent last, capped to the last 90 frames.

Type:: collections.deque

timestamps_ns#

Recent per-frame timestamps (ns), parallel to positions.

Type:: collections.deque

areas#

Recent per-frame contour areas (px²), parallel to positions.

Type:: collections.deque

last_frame#

Internal frame index this track was last updated at.

Type:: int, default 0

lost_count#

Consecutive frames since this track was last matched to a detection; pruned once this exceeds CentroidTracker.max_lost.

Type:: int, default 0

property position: tuple[float, float]#

Most recent (cx, cy) centroid, or (0.0, 0.0) if none yet.

Type:: tuple of float

property velocity_px_per_frame: float#

Euclidean distance between the last two centroids, px/frame.

Type:: float

velocity_mm_per_s(mm_per_px, fps)[source]#

Convert velocity_px_per_frame to a real-world speed.

Parameters:

mm_per_px (float) – Manual pixel-to-real-world scale factor.
fps (float) – Assumed nominal frame rate.

Returns:

Speed in mm/s.

Return type:

float

Notes

Uses a fixed nominal fps rather than each sample’s real elapsed time, unlike the more careful real-Δt derivative in Pose Kinematics. See Motion Tracking for the contrast.

property mean_area: float#

Mean contour area (px²) over this track’s recent history.

Type:: float

motion.draw_tracks(frame, tracks, trail_length=30, mm_per_px=1.0, fps=30.0)[source]#

Overlay tracks, centroids, IDs, and velocity on a BGR frame.

Parameters:

frame (numpy.ndarray) – BGR frame; not modified — a copy is drawn on and returned.
tracks (list of Track) – Tracks to draw, e.g. from CentroidTracker.update().
trail_length (int, default 30) – Number of recent positions to draw as a fading trail per track.
mm_per_px (float, default 1.0) – Forwarded to Track.velocity_mm_per_s() for the on-frame velocity label.
fps (float, default 30.0) – Forwarded to Track.velocity_mm_per_s() for the on-frame velocity label.

Returns:

A new BGR frame (copy of frame) with tracks overlaid.

Return type:

numpy.ndarray

motion.generate_heatmap(trajectories, frame_size, output_path, sigma=20.0, cmap='hot', title='Position Density Heatmap', show_trails=True, dpi=150)[source]#

Render a Gaussian-smoothed position density map.

Parameters:

trajectories (dict of int to list of tuple of float) – {animal_id: [(cx, cy), ...]} in pixel coordinates.
frame_size (tuple of int) – (width, height) of the original video frame.
output_path (str) – Destination PNG/PDF path.
sigma (float, default 20.0) – Gaussian blur radius (px) for density smoothing.
cmap (str, default "hot") – Matplotlib colormap name (e.g. "hot", "viridis", "plasma").
title (str, default "Position Density Heatmap") – Plot title.
show_trails (bool, default True) – If True, overlay per-animal trajectory lines.
dpi (int, default 150) – Output image resolution.

Return type:

None

motion.generate_trajectory_plot(trajectories, frame_size, output_path, title='Trajectory Plot', dpi=150)[source]#

Plain trajectory line plot, one colour per animal.

Parameters:

trajectories (dict of int to list of tuple of float) – {animal_id: [(cx, cy), ...]} in pixel coordinates.
frame_size (tuple of int) – (width, height) of the original video frame.
output_path (str) – Destination PNG/PDF path.
title (str, default "Trajectory Plot") – Plot title.
dpi (int, default 150) – Output image resolution.

Return type:

None

Notes

Each trajectory is colour-coded by time (start = transparent, end = opaque); start marker is a circle, end marker is a square.

motion.generate_velocity_histogram(velocities, output_path, mm_per_px=1.0, title='Velocity Distribution', dpi=150)[source]#

Histogram of per-frame velocities, with median/mean markers.

Parameters:

velocities (list of float) – Per-frame velocity samples; non-positive values are excluded.
output_path (str) – Destination PNG/PDF path.
mm_per_px (float, default 1.0) – Only used to pick the x-axis unit label ("mm/s" if not 1.0, else "px/frame") — velocities themselves must already be pre-scaled by the caller.
title (str, default "Velocity Distribution") – Plot title.
dpi (int, default 150) – Output image resolution.

Return type:

None

See Motion Tracking.

Live Transcription (Real-time tab)#

Not a post-hoc Analysis-tab plugin — this package backs the Real-time tab’s live-captions panel (analysis/run_live_transcribe.py, a persistent subprocess started by mosaic::TranscriptWorker; see User guide). Documented here because it’s real, importable, independently-testable library code, not because it fits the “run a session, get a result file” shape every plugin above does.

from transcribe import pcm16_to_mono_float32, resample_to_16k, confirm_segments

mono = resample_to_16k(pcm16_to_mono_float32(pcm_bytes, channels=2), source_rate_hz=48000)
# ... run whisper over the rolling buffer, then split its segments ...
confirmed, tentative_text, watermark_sec = confirm_segments(
    segments, buffer_duration_sec=8.0, trailing_margin_sec=1.0)

Pure windowing/confirmation logic and PCM resampling for the Real-time tab’s live transcription worker (analysis/run_live_transcribe.py).

`Segment`
`confirm_segments`	Splits one whisper pass's segments into confirmed vs. tentative.
`trim_buffer_samples`	Convert a confirmation watermark into a sample count to trim.
`pcm16_to_mono_float32`	Interleaved int16 LE PCM -> mono float32 in [-1, 1].
`resample_to_16k`	No-op if already 16kHz (the common case is NOT 16kHz — see audio_recorder.cpp's device-negotiation doc comment, e.g. this dev machine's real mic negotiates 48000Hz/2ch).

class transcribe.Segment(start_sec, end_sec, text)[source]#

Parameters:

start_sec (float)
end_sec (float)
text (str)

start_sec: float#: Alias for field number 0

end_sec: float#: Alias for field number 1

text: str#: Alias for field number 2

transcribe.confirm_segments(segments, buffer_duration_sec, trailing_margin_sec)[source]#

Splits one whisper pass’s segments into confirmed vs. tentative.

Parameters:

segments (list of Segment) – This pass’s segments, in buffer-relative seconds, chronological (faster-whisper always returns them in order).
buffer_duration_sec (float) – Current rolling buffer’s total length in seconds.
trailing_margin_sec (float) – A segment is confirmable only if it ends at least this long before buffer_duration_sec.

Returns:

confirmed (list of Segment) – The newly-final segments from this pass (still buffer-relative — the caller adds its own running offset to get absolute time).
tentative_text (str) – Everything after the last confirmed segment, space-joined. Confirmed segments’ text is NOT included.
watermark_sec (float) – confirmed[-1].end_sec, or 0.0 if nothing was confirmed this pass — the caller trims the buffer’s front up to this point (see trim_buffer_samples()).

Return type:

tuple[list[Segment], str, float]

transcribe.trim_buffer_samples(confirmed_watermark_sec, sample_rate_hz)[source]#

Convert a confirmation watermark into a sample count to trim.

Parameters:

confirmed_watermark_sec (float) – How far (in seconds) audio has been confirmed — typically confirm_segments()’s watermark_sec return value.
sample_rate_hz (int) – The rolling buffer’s sample rate.

Returns:

Sample count to drop from the front of the rolling buffer.

Return type:

int

Notes

Pure arithmetic, split out only so the “seconds -> sample index” conversion has one, tested home rather than being re-derived at each call site.

transcribe.pcm16_to_mono_float32(payload, channels)[source]#

Interleaved int16 LE PCM -> mono float32 in [-1, 1].

Parameters:

payload (bytes)
channels (int)

Return type:

ndarray

transcribe.resample_to_16k(mono, source_rate_hz)[source]#

No-op if already 16kHz (the common case is NOT 16kHz — see audio_recorder.cpp’s device-negotiation doc comment, e.g. this dev machine’s real mic negotiates 48000Hz/2ch).

Parameters:

mono (ndarray)
source_rate_hz (int)

Return type:

ndarray

Trailing-margin confirmation. Whisper (tiny model, by default) is re-run over the entire rolling audio buffer on every pass rather than incrementally. A segment is confirmed — final, never revised again — once its end lies at least TRAILING_MARGIN_SEC before the buffer’s current end, giving it a margin of trailing audio context on both the previous pass and this one; everything after that point is “tentative” text, replaced wholesale each pass. Confirmed audio is then trimmed off the buffer’s front so growth stays bounded. This intentionally skips more elaborate cross-pass textual-agreement (“LocalAgreement-n”) policies some streaming-ASR projects use — VAD-anchored segment boundaries are already stable in practice for the confirmed prefix, and the simpler rule is sufficient for a tiny-model live-captions v1.