vigilo-stream

Zero-copy multi-modal stream fusion engine for real-time AI pipelines in Python.

vigilo-stream provides Python bindings for the stream fusion engine in vigilo-core. It gives Python vision and proctoring pipelines direct access to video frames and temporal rule evaluation without copying memory across the FFI boundary.

• Zero-copy buffer sharing: Frame memory allocated in Rust is exposed directly to NumPy and PyTorch through __array_interface__ and the buffer protocol.
• Lock-free frame exchange: Capture workers publish frames through ArcSwap slots, discarding stale frames automatically instead of building queues.
• Deterministic temporal fusion: The FusionEngine processes detection signals through configurable hysteresis bands, hold timers, and score accumulators. Given the same input, replay produces identical events.
• Multimodal detection: Wraps the vigilo-core inference pipeline for face detection (YuNet), head pose (MobileNetV3), gaze estimation (MobileGaze), object detection (YOLOX-Nano), and identity matching (ArcFace).

Documentation

Comprehensive usage guides and API references are available at the documentation site:

https://abdullah-masood-05.github.io/vigilo-stream/

• Getting started and installation
• Zero-copy memory sharing
• Pipeline lifecycle
• Fusion engine and replay
• Live OpenCV HUD
• Python API reference
• Hardware and GPU acceleration

Installation

pip install vigilo-stream

You can import the library using either vigilo_stream or the rustream alias.

Quick start

import vigilo_stream
import numpy as np

# 1. Zero-copy frame operations (no neural model files required)
frame = vigilo_stream.create_synthetic_frame(1280, 720, seq=1, r=255, g=0, b=0)
print(frame.width, frame.height, frame.shape) # 1280 720 (720, 1280, 3)

# Expose Rust memory directly as a NumPy array without copying
arr = np.asarray(frame)
assert arr.__array_interface__["data"][0] == frame.__array_interface__["data"][0]

# 2. Vision and proctoring pipeline
# Pipeline automatically downloads default model weights on first run
with vigilo_stream.Pipeline(models_dir="models") as pipe:
    pipe.start("camera:0")  # Accepts "camera:0", "file:clip.mp4", or "dir:frames/"

    while pipe.is_running():
        frame = pipe.poll_frame()
        if frame:
            img = np.asarray(frame)

        snapshot = pipe.snapshot()
        if snapshot:
            print(f"Faces: {snapshot.face_count}, Pose: {snapshot.head_pose}")

        events = pipe.events()
        for event in events:
            print(f"Violation: {event}")

# 3. Headless deterministic stream fusion (no neural models or camera required)
engine = vigilo_stream.FusionEngine()
events = engine.replay("recorded_session.jsonl")
print(f"Replayed session produced {len(events)} events.")

Real-time OpenCV visualization with HUD

You can stream frames directly into OpenCV with zero-copy buffer access, overlay bounding boxes with confidence scores, draw 5 facial landmarks, project a 3D head pose orientation gizmo, render gaze direction rays, tag prohibited objects, and display proctoring telemetry cards and violation status pills matching the desktop viewer.

Complete example

import cv2
import math
import time
import numpy as np
import vigilo_stream

# Initialize pipeline with automatic model download
pipe = vigilo_stream.Pipeline(models_dir="models", auto_download=True)
pipe.start("camera:0")  # Accepts "camera:0", "file:clip.mp4", or "dir:frames/"

cv2.namedWindow("Vigilo Stream Viewer", cv2.WINDOW_NORMAL)
cv2.resizeWindow("Vigilo Stream Viewer", 1280, 720)

active_violations = set()
held_objects = []
last_object_time = 0.0

try:
    while pipe.is_running():
        # 1. Zero-copy frame access: raw pointer shared directly with NumPy
        frame = pipe.poll_frame()
        if frame is None:
            continue

        rgb = np.asarray(frame)
        img = cv2.cvtColor(rgb, cv2.COLOR_RGB2BGR)

        # 2. Instantaneous model snapshot and temporal fusion events
        snap = pipe.snapshot()
        for ev in pipe.events():
            if ev.event_type == "ViolationStarted" and ev.violation:
                active_violations.add(ev.violation.kind)
            elif ev.event_type == "ViolationEnded" and ev.violation:
                active_violations.discard(ev.violation.kind)

        # 3. Draw face bounding boxes, 5 landmarks, 3D pose, and gaze rays
        if snap:
            for i, face in enumerate(snap.faces):
                bx = int(face.bbox.x)
                by = int(face.bbox.y)
                bw = int(face.bbox.w)
                bh = int(face.bbox.h)

                # Bounding box with score badge
                cv2.rectangle(img, (bx, by), (bx + bw, by + bh), (94, 197, 34), 2)
                cv2.putText(
                    img,
                    f"FACE {int(face.score * 100)}%",
                    (bx, max(18, by - 8)),
                    cv2.FONT_HERSHEY_SIMPLEX,
                    0.45,
                    (140, 255, 100),
                    1,
                    cv2.LINE_AA,
                )

                # 5 Facial landmarks (right eye, left eye, nose, right mouth, left mouth)
                palette = [
                    (94, 197, 34),
                    (94, 197, 34),
                    (21, 204, 250),
                    (113, 113, 248),
                    (113, 113, 248),
                ]
                for pt, col in zip(face.landmarks, palette):
                    cv2.circle(img, (int(pt[0]), int(pt[1])), 4, col, -1, cv2.LINE_AA)

                # 3D Head pose orientation axes (Euler projection)
                if i == 0 and snap.head_pose:
                    cx, cy = int(bx + bw * 0.5), int(by + bh * 0.5)
                    s = min(bw, bh) * 0.45
                    rad = math.pi / 180.0
                    y_rad = -snap.head_pose.yaw_deg * rad
                    p_rad = snap.head_pose.pitch_deg * rad
                    r_rad = snap.head_pose.roll_deg * rad

                    cyaw, syaw = math.cos(y_rad), math.sin(y_rad)
                    cpit, spit = math.cos(p_rad), math.sin(p_rad)
                    crol, srol = math.cos(r_rad), math.sin(r_rad)

                    x_end = (
                        int(cx + s * (cyaw * crol)),
                        int(cy + s * (cpit * srol + crol * spit * syaw)),
                    )
                    y_end = (
                        int(cx + s * (-cyaw * srol)),
                        int(cy + s * (cpit * crol - spit * syaw * srol)),
                    )
                    z_end = (int(cx + s * syaw), int(cy + s * (-cyaw * spit)))

                    cv2.arrowedLine(img, (cx, cy), x_end, (68, 68, 239), 2, tipLength=0.2)
                    cv2.arrowedLine(img, (cx, cy), y_end, (94, 197, 34), 2, tipLength=0.2)
                    cv2.arrowedLine(img, (cx, cy), z_end, (250, 165, 96), 2, tipLength=0.2)

                # Gaze direction ray originating between the eyes
                if i == 0 and snap.gaze and len(face.landmarks) >= 2:
                    ox = int((face.landmarks[0][0] + face.landmarks[1][0]) * 0.5)
                    oy = int((face.landmarks[0][1] + face.landmarks[1][1]) * 0.5)
                    glen = bw * 1.1
                    gdx = -glen * math.sin(snap.gaze.yaw_rad) * math.cos(snap.gaze.pitch_rad)
                    gdy = -glen * math.sin(snap.gaze.pitch_rad)
                    cv2.arrowedLine(
                        img,
                        (ox, oy),
                        (int(ox + gdx), int(oy + gdy)),
                        (252, 171, 240),
                        2,
                        tipLength=0.15,
                    )

            # Prohibited objects (phones, books, secondary devices)
            # The object worker runs at 1 Hz to save compute, while face models run at 30 Hz.
            # Hold the latest detected objects for 1.2s to render a steady bounding box.
            if snap.objects:
                held_objects = snap.objects
                last_object_time = time.time()
            elif "prohibited_object" not in active_violations and (time.time() - last_object_time > 1.2):
                held_objects = []

            for obj in held_objects:
                ox = int(obj.bbox.x)
                oy = int(obj.bbox.y)
                ow = int(obj.bbox.w)
                oh = int(obj.bbox.h)
                cv2.rectangle(img, (ox, oy), (ox + ow, oy + oh), (68, 68, 239), 2)
                cv2.putText(
                    img,
                    f"{obj.label.upper()} {int(obj.score * 100)}%",
                    (ox, max(18, oy - 6)),
                    cv2.FONT_HERSHEY_SIMPLEX,
                    0.45,
                    (255, 200, 200),
                    1,
                    cv2.LINE_AA,
                )

        # 4. Display frame and handle interactive keys
        cv2.imshow("Vigilo Stream Viewer", img)
        key = cv2.waitKey(1) & 0xFF
        if key in (27, ord("q")):
            break
        elif key == ord("e"):
            pipe.enrol()

finally:
    pipe.stop()
    cv2.destroyAllWindows()

You can also run the full modular viewer script with HUD telemetry cards and violation status pills in examples/live_opencv_hud.py:

python examples/live_opencv_hud.py --source camera:0

Model weights

The neural pipeline uses ONNX Runtime models:

• Face detection: YuNet (face_detection_yunet_2023mar.onnx)
• Head pose: MobileNetV3 (headpose_mobilenetv3_small.onnx)
• Gaze estimation: MobileGaze (mobileone_s0_gaze.onnx)
• Object detection: YOLOX-Nano (yolox_nano.onnx)

By default, Pipeline(models_dir="models") downloads missing models on first use. You can also download them explicitly:

import vigilo_stream
vigilo_stream.download_models("models")

Alternatively, download them using curl:

mkdir -p models
curl -sSL -o models/face_detection_yunet_2023mar.onnx https://github.com/opencv/opencv_zoo/raw/main/models/face_detection_yunet/face_detection_yunet_2023mar.onnx
curl -sSL -o models/headpose_mobilenetv3_small.onnx https://github.com/yakhyo/head-pose-estimation/releases/download/weights/mobilenetv3_small.onnx
curl -sSL -o models/mobileone_s0_gaze.onnx https://github.com/yakhyo/gaze-estimation/releases/download/weights/mobileone_s0_gaze.onnx
curl -sSL -o models/yolox_nano.onnx https://github.com/Megvii-BaseDetection/YOLOX/releases/download/0.1.1rc0/yolox_nano.onnx

Architecture

Camera / Video File / Image Directory
                 │
                 ▼
  FrameSource (DirectShow / FFmpeg)
                 │
                 ▼
     ArcSwap Latest-Frame Slot  ◄── Zero-copy pointer sharing with NumPy
        ┌────────┴────────┐
        ▼                 ▼
   Face Worker      Object Worker
  YuNet+Pose+Gaze     YOLOX-Nano
        └────────┬────────┘
                 ▼
              Signals ──► FusionEngine ──► Events / Violations

Building from source

Requirements:

• Rust 1.80 or newer
• Python 3.9 or newer
• C++ build tools (MSVC on Windows, GCC/Clang on Linux and macOS)

# Set up a virtual environment and install build tools
uv venv
uv pip install maturin pytest numpy

# Build and install the extension into the active environment
uv run maturin develop

# Run the test suite
uv run pytest -v tests/

Release notes

v1.0.0

• Added cross-platform on-demand GPU acceleration architecture:
  • Windows: DirectML (DirectX 12) acceleration for NVIDIA, AMD, Intel Arc, and Qualcomm GPUs.
  • Linux: NVIDIA CUDA acceleration.
  • macOS: CoreML / Metal acceleration for Apple Silicon.
• Introduced dynamic hardware detection (detect_gpu_support()) and on-demand GPU backend downloading (download_gpu_backend()), keeping the default PyPI package lightweight (~18 MB).
• Added device="auto", device="gpu", and device="cpu" parameters to Pipeline.
• Added runtime provider inspection via device_info().
• Added full VitePress documentation website with dark/light themes and published to GitHub Pages.

v0.1.1

• Added download_models() helper to fetch default ONNX model weights automatically.
• Enhanced Pipeline to download missing model files automatically on first use (auto_download=True).
• Added MODEL_URLS mapping and updated documentation.

v0.1.0

• Initial release of vigilo-stream (with rustream backward-compatibility alias) targeting Python 3.9 through 3.13.
• Implemented Frame with __array_interface__ and memoryview() support for zero-copy NumPy interop.
• Implemented FusionEngine with single-frame stepping and deterministic JSONL log replay.
• Implemented Pipeline context manager wrapping camera capture, detection workers, and event polling.
• Added data bindings for BBox, FaceDetection, HeadPose, Gaze, ObjectDetection, Signals, Violation, and Event.
• Multi-platform CI testing across Windows, Ubuntu, and macOS.

License

AGPL-3.0. See LICENSE for details.