Vision Modules

Composable vision modules over a live video stream, giving robots, systems, and agents perception they can act on.

Perception here is modular: every module solves one perceptual job over the same incoming stream and exposes it behind a uniform interface (start() / stop() / latest()). Richer perception means switching on another module, not rewriting the pipeline. Shared work (finding the hand, for instance) is done once in a shared stage and fanned out to every module that needs it, and each node runs on its own thread at its own framerate.

The library perceives; the caller decides. Modules report what is seen (a gesture label, a hand's position and confidence), never what it means. Turning a stop gesture into a command is the job of the robot or agent consuming the results.

What's in the box

Concept	Public API	Role	Spec
Stream	StreamProvider, Frame, FrameLike, FrameSource, OpenCVSource	Single capture entry point: one thread reads a webcam, video file, URL, or injected source and publishes the newest frame	specs/stream.md
Pipeline	Stage, Module, Result, Pipeline, LatestValue, Upstream	Staged-graph runtime: latest-value sampling, per-node threads and framerates, ordered start/stop	specs/pipeline.md
Hand	HandStage, HandResult, Hand, HandDetector, DetectedHand, MediaPipeHandDetector	Shared stage: detects and crops hands once per frame for every hand-based module	specs/hand.md
Gesture classifier	GestureClassifier, Gesture, HandGesture, ImageClassifier, HaGRIDViTClassifier, select_device	Perception module: names the gesture in each hand crop (18 HaGRID classes)	specs/gesture_classifier.md
Face	FaceDetector, DetectedFace, MediaPipeFaceDetector	Detector seam (not a stage): faces with a box, nose, eyes, head orientation and landmarks, for a caller running its own loop	specs/face.md
Snapshot	save_frame, save_crops, save_input	Write what a node last published or last consumed to a local image file, from the caller's thread	specs/snapshot.md
Hand demo	examples/hand_demo.py	Runnable Gradio browser app wiring the whole graph end to end	specs/hand_demo.md

Everything in the table is implemented and exported from vision_modules.

How it works

Every node samples the newest output of the node above it, processes it on its own thread, and publishes its newest result for the nodes below. No queues, no backlog: a slow consumer always sees the latest value and never falls behind, and a node never reprocesses a frame it has already seen (stale skip on frame_id).

StreamProvider                 root: capture thread, publishes the latest Frame
  └─ HandStage                 shared stage: detect + crop once per frame (default 30 fps)
       └─ GestureClassifier    module: classifies each crop at its own rate (default 5 fps)
            (add more modules here, or another shared stage next to HandStage)

Key rules every node obeys:

• Every published value carries frame_id and ts from the originating Frame, so results from different modules can be matched on the same input frame.
• present=False is a real result. A module publishes one Result per processed frame; "no hand in this frame" is present=False, and only None means "not started yet".
• A stage's real rate is capped by its upstream's. target_fps is a ceiling, not a guarantee. Stage.published_count is the exact way to measure a stage's achieved rate, and StreamProvider.published_count gives the same counter for the root (source_fps is only what the camera or file claims).
• Frames are shared by reference and read-only. Anything that draws on a frame copies it first; anything that publishes a sub-region (the hand crop) copies it too.
• Errors in process() never kill the graph. They are recorded in last_error, logged, and the loop moves on to the next frame.
• Threads, not processes. OpenCV, NumPy, MediaPipe and PyTorch release the GIL during native work. Non-thread-safe resources (a MediaPipe instance, a loaded model) are created lazily on the worker thread and touched only there.
• Every node is restartable. stop() then start() runs a fresh worker; latest() and the counters carry over. Pipeline.start() is all-or-nothing: if a node fails to start, the ones already started are stopped again before the error propagates.
• Owned vs. borrowed backends. A stage closes and recreates the detector or classifier it created itself. One you pass in (detector=, classifier=) is borrowed: never closed by the stage, reused across restarts, yours to close.

Installation

Python 3.12 or newer.

pip install vision-modules                    # core: numpy only — stream + pipeline runtime, for a feed of your own
pip install "vision-modules[opencv]"          # + webcams, video files, URLs, snapshots
pip install "vision-modules[hand]"            # + hand stage and gesture classifier
pip install "vision-modules[face-mediapipe]"  # + MediaPipe face detector
pip install "vision-modules[opencv,hand]"     # what the Quickstart below needs

Runtime dependencies are split by feature. The core package needs only numpy: it runs the stream and pipeline runtime over any FrameSource you provide, and lets you write your own stages. The opencv extra adds opencv-python, which OpenCVSource (so StreamProvider(0) / StreamProvider("clip.mp4")) and the snapshot functions use. Any OpenCV wheel (-headless, -contrib, …) provides the same cv2 module, so if you already have one you can skip the extra. Without OpenCV, those two paths raise an ImportError naming the extra. The hand extra adds mediapipe (hand detection) plus torch, transformers and pillow (gesture classification). The face-mediapipe extra adds mediapipe for the face detector; face extras are named per backend (face-<backend>), since later face backends bring unrelated dependencies. Each feature family gets its own extra, so you install only the perception you enable.

Supported platforms are macOS on Apple silicon, Linux x86_64 and Windows x86_64. On Intel Macs and on Linux arm64 (Raspberry Pi, Jetson), only the core and the opencv extra install: MediaPipe ships no wheels there, so hand and face-mediapipe cannot install. Those two platforms are not tested in CI.

On Linux, MediaPipe's native library (the hand and face-mediapipe extras) loads the system's OpenGL ES and EGL libraries, which a desktop has but a minimal or server image may lack (OSError: libGLESv2.so.2: cannot open shared object file). Install them with sudo apt-get install libgles2 libegl1 (Debian and Ubuntu).

For development, uv sync --dev installs every extra plus the tooling and the demo's dependencies.

Quickstart

import time

from vision_modules import GestureClassifier, HandStage, Pipeline, StreamProvider

provider = StreamProvider(0, mirror=True)  # webcam index 0, selfie view
hands = HandStage(provider, target_fps=30)  # detect + crop once per frame
gestures = GestureClassifier(hands, target_fps=5, threshold=0.55)

# Pipeline starts nodes upstream-first and stops them downstream-first.
with Pipeline([provider, hands, gestures]):
    seen = None
    while True:
        result = gestures.latest()
        if result is None or result.frame_id == seen:
            time.sleep(0.01)  # nothing new yet
            continue
        seen = result.frame_id
        hand = result.first
        if hand is not None and hand.label is not None:
            print(f"frame {result.frame_id}: {hand.label} ({hand.confidence:.2f})")

The first run downloads the models (see Models and caches).

StreamProvider also accepts a file path or URL (StreamProvider("clip.mp4")) or any object implementing the FrameSource protocol, so the same graph runs over recordings or frames pushed from elsewhere. provider.ended turns True when a file source runs out.

API overview

Stream

StreamProvider(source: int | str | FrameSource = 0, *, mirror: bool = False)
  .start() -> StreamProvider   # opens the source, spawns the capture thread; restartable
  .latest() -> Frame | None    # newest frame, or None before the first read
  .ended -> bool               # True once the source ran out on its own (not after stop())
  .published_count -> int      # frames published so far, ever (across restarts); exact achieved-rate counter
  .source_fps -> float | None  # rate the source claims (CAP_PROP_FPS); None when unknown or not open
  .stop() -> None              # signals the thread, closes the source, joins the thread
  # also a context manager

Frame is a frozen dataclass: frame_id (monotonic, from 1), ts (time.monotonic() at capture), image ((H, W, 3) uint8 BGR, OpenCV's native order). BGR is the in-library convention; consumers convert as needed and get a new array.

FrameLike is the protocol frame consumers are typed against — the same three members, declared as read-only properties. HandStage takes any Upstream[FrameLike], and save_frame / save_input accept any FrameLike, so a host that already has a frame class and a latest-value slot of its own (a robot's camera reader, a ROS topic, a dataset reader) feeds the shipped stages directly, pyright-clean, without re-publishing its frames through a second StreamProvider:

@dataclass(frozen=True)
class CameraFrame:  # yours; imports nothing from vision_modules
    frame_id: int
    ts: float
    image: npt.NDArray[np.uint8]  # (H, W, 3) BGR


class CameraFeed:  # yours; anything with latest()
    def latest(self) -> CameraFrame | None: ...


hands = HandStage(CameraFeed(), target_fps=30)  # an Upstream[FrameLike]

FrameSource is the seam for injecting frames without a camera:

class FrameSource(Protocol):
    def open(self) -> None: ...  # called by start(); a restart re-opens
    def read(self) -> np.ndarray | None: ...  # next BGR image, or None when exhausted
    def close(self) -> None: ...  # called by stop(); must unblock a pending read()
    def fps(self) -> float | None: ...  # nominal rate the device reports, or None

The provider drives the source's lifecycle: open() on every start(), close() on every stop(). close() is called while the capture thread may be inside read(), which is what interrupts a read waiting on a frame that never comes, so a source must tolerate that (OpenCVSource serializes the two with a lock).

Pipeline

Stage[TIn, TOut](upstream, target_fps: float | None, name: str | None = None)
  .start() / .stop() / .latest()
  .process(item) -> TOut | None   # subclass hook, runs on the worker thread only
  .close()                        # subclass hook, releases owned resources
  .target_fps                     # settable while running; None or > 0, else ValueError
  .published_count                # exact number of publishes so far
  .last_input                     # the item last handed to process(); set before it runs
  .last_error                     # last exception raised by process(), if any

Module[TIn, TOut: Result](Stage)  # a Stage whose output is a perception Result
Result: frame_id, ts, present     # frozen base class of every published value
Pipeline(nodes).start() / .stop() # ordered lifecycle, also a context manager

target_fps=None means "as fast as the upstream delivers". Setting it at or above the upstream's rate sees every item; setting it lower deliberately down-samples.

Hand

HandStage(provider, target_fps=30, pad=0.35, max_hands=2, rank="area", hysteresis=0.2,
          detector: HandDetector | None = None)

Publishes a HandResult per frame, also when no hand is found (present=False):

• hands: tuple[Hand, ...], at most max_hands. When a detector offers more candidates than that, the biggest (rank="area") win, and a published hand is only displaced by one more than hysteresis bigger; the order is slot-stable, so hands[i] stays the same physical hand while it is visible. max_hands is also the detector's capacity: the shipped MediaPipe detector never returns more hands than that, so with max_hands=1 it keeps the first hand it tracked.
• first: Hand | None

Each Hand has bbox ((x0, y0, x1, y1) in full-frame pixels, padded by pad and clamped), crop (an independent BGR copy of that box, or None for a zero-area box at the frame edge), and score.

HandDetector is a boxes-only protocol (detect(image_bgr, ts) -> tuple[DetectedHand, ...]), so any detector can sit behind the stage. The shipped MediaPipeHandDetector uses the MediaPipe Tasks HandLandmarker in VIDEO mode, taking the box as the extremes of the 21 landmarks. Landmarks and handedness stay internal for now.

Gesture classifier

GestureClassifier(hand_stage, target_fps=5, threshold=0.55,
                  classifier: ImageClassifier | None = None, device: str | None = None)

Publishes a Gesture per processed HandResult, with hands: tuple[HandGesture, ...] index-aligned with HandResult.hands and a first convenience. Each HandGesture has:

• label: str | None, the top class when its score clears threshold, else None
• confidence: float, the top score
• scores: dict[str, float], every class's score, so you can apply your own rule

Labels are HaGRID's 18 class names, verbatim and never remapped:

call, dislike, fist, four, like, mute, ok, one, palm, peace, peace_inverted, rock, stop, stop_inverted, three, three2, two_up, two_up_inverted

The runtime authority is the classifier's labels property, so a fine-tuned backend with another vocabulary works unchanged. ImageClassifier is the seam (labels, classify(image_bgr) -> dict[str, float], close()); the shipped HaGRIDViTClassifier wraps the Hugging Face model dima806/hand_gestures_image_detection, a ViT fine-tuned on HaGRID.

threshold and target_fps are plain attributes you can retune while the module runs.

Face

MediaPipeFaceDetector(max_faces=1, min_face_detection_confidence=0.5,
                      min_face_presence_confidence=0.5, min_tracking_confidence=0.5,
                      model_path=None)
detector.detect(image_bgr, ts) -> tuple[DetectedFace, ...]
detector.close()

A detector, not a stage: you call it from your own loop with a BGR image and its capture time in seconds (monotonic), and it returns every face it found, in no particular order. Each DetectedFace has coordinates normalised to the image you passed (x, y in [0, 1], y down):

• box: (x0, y0, x1, y1), the extremes of the face mesh (tighter than a detector box)
• nose: the nose tip; eyes: the iris centres, the subject's right eye first
• orientation: (roll, pitch, yaw) in radians. Roll is positive when the eye line turns clockwise in the image, pitch when the head tilts down, yaw when the face turns toward the image's right. A frontal, upright face reads (0, 0, 0).
• landmarks: MediaPipe's 478-point mesh, (478, 3) float32
• score: always None here (the landmarker exposes no per-face score)

It tracks between calls (MediaPipe VIDEO mode), so use one instance per video. It is not thread-safe: call it from one thread at a time. Successive calls may come from different threads, a thread pool for instance. The constructor downloads the model on first use, so build it off any latency-sensitive thread. FaceDetector is the seam for other backends.

Snapshot

save_frame(frame: FrameLike | None, path) -> Path
save_crops(result: HandResult | None, path) -> tuple[Path | None, ...]
save_input(stage: Stage, path) -> Path | tuple[Path | None, ...]

Pure functions over a node's newest output (latest()) or newest input (last_input), meant to be called right where you read those (main thread, request handler, agent loop), never from a worker:

from vision_modules import save_frame, save_crops, save_input

f = provider.latest()
save_frame(f, f"out/{f.frame_id:06d}.png")  # full BGR frame, format by suffix
paths = save_crops(
    hands.latest(), "out/hand.png"
)  # hand_0.png, hand_1.png, ... aligned with .hands
save_input(hands, "out/hand_in.png")  # the frame the hand stage last detected on
save_input(gestures, "out/gesture_in.png")  # the crop(s) the classifier last classified

save_input is how a module that publishes no pixels (GestureClassifier) gets its "current frame": the exact input it last processed, pinned by the stage itself, so it stays right even when the upstream has already moved on. None (nothing published or processed yet) raises ValueError; missing directories are created; an unsupported suffix raises before anything is written.

Writing your own module

Subclass Module, pick an upstream, implement process(), and return a frozen Result subclass carrying the upstream's frame_id and ts:

from dataclasses import dataclass

from vision_modules import HandResult, Module, Result


@dataclass(frozen=True)
class HandHeight(Result):
    y_center: float | None


class HandHeightModule(Module[HandResult, HandHeight]):
    def process(self, item: HandResult) -> HandHeight:
        hand = item.first
        if hand is None:
            return HandHeight(item.frame_id, item.ts, present=False, y_center=None)
        _, y0, _, y1 = hand.bbox
        return HandHeight(item.frame_id, item.ts, present=True, y_center=(y0 + y1) / 2)

Add it to the Pipeline list after its upstream and read it with latest() like any other node. Heavy resources go in a lazily-initialized attribute used only inside process(), and are released and dropped in close() so a restarted run creates its own.

Models and caches

Nothing is bundled in the wheel. Models download on first use and are cached afterwards:

Model	Used by	Size	Cache
MediaPipe hand_landmarker.task	MediaPipeHandDetector	a few MB	~/.cache/vision-modules/, overridable with VISION_MODULES_CACHE or model_path=
MediaPipe face_landmarker.task	MediaPipeFaceDetector	a few MB	same as above
dima806/hand_gestures_image_detection	HaGRIDViTClassifier	~340 MB	Hugging Face's own cache

Inference device: select_device() picks mps if available, else cuda, else cpu. Pass device= to override. The library never sets PYTORCH_ENABLE_MPS_FALLBACK; export it yourself if an MPS op is missing.

Demo

A Gradio browser app runs the hand stage and gesture classifier live over your webcam:

uv run python examples/hand_demo.py --mirror

Open the printed URL and grant the browser camera access. Flags: --mirror, --hand-fps (default 30), --classifier-fps (default 5), --threshold (default 0.55), --max-hands (default 2; one classification panel per hand).

The page shows the raw feed, the same frame with a numbered box per detected hand in its slot colour, and one panel per hand (same colour) with that hand's three best classes and the validated label marked; a hand keeps its panel when the two hands' sizes cross. Sliders retune the threshold and both stages' target fps live, and a readout shows each stage's real achieved fps next to its target. Two buttons save what the hand stage and the classifier last consumed (the full frame, the classified crop) as snapshot_<stage>_<timestamp>.jpg in a folder you pick, the script's own directory by default. The demo needs the demo dependency group, which uv sync --dev installs; it is not part of the package.

Development

This project is built spec-first: every concept is designed in specs/ before it is built, and each spec carries a status (Draft, Stable, Implemented, Updated) that tracks whether the code matches it. Implementation plans in plans/ turn settled specs into buildable steps. Start with AGENTS.md, then specs/_index.md and plans/_index.md.

uv sync --dev
uv run ruff check .
uv run ruff format .
uv run pyright
uv run pytest

Lint, type check, and tests must all pass before a change is done. pyright covers the tests and examples/ too.

Tests

Two tiers, separated by directory:

Tier	Directory	Network	Runs by default
Unit / integration	tests/	never	yes
Live / e2e	tests-e2e/	real models and devices	no

tests/ mirrors the src/vision_modules/ layout and drives every stage through its seam with scripted sources, detectors and classifiers, so no camera, model, or network is needed (the one OpenCVSource test reads a clip it wrote itself). The demo's pure helpers are tested in tests/test_hand_demo.py. tests/test_project_map.py guards the docs: every module must appear in the AGENTS.md project map and be governed by a spec whose frontmatter paths exist.

The live tier runs the real MediaPipe and ViT models against photos in tests-e2e/fixtures/ (one hand_<gesture>.jpg per gesture, plus no_hand.jpg) and asserts the full pipeline predicts the photographed gesture:

uv run pytest tests-e2e

Tests gated on a credential skip instead of failing; the model-backed tests need network on their first run to download weights and fail if that download fails. Environment variables:

• VISION_MODULES_CAMERA: camera index for the webcam capture test (required for it; never guessed, because opening an unavailable camera can crash the process natively)
• VISION_MODULES_HAND_IMAGE: overrides the bundled hand photo
• VISION_MODULES_CACHE: where the MediaPipe model bundle is cached

License

MIT, see LICENSE.

The models downloaded on first use are not part of this repository and come under their own licenses: the MediaPipe hand_landmarker.task bundle is Apache-2.0, and dima806/hand_gestures_image_detection is published as Apache-2.0 by its author. That model was fine-tuned on the HaGRID dataset, which is licensed CC BY-SA 4.0; check that this fits your use if you redistribute the model.