Vision Pipeline

Reading frames off the Pengo Grabber in Python, preparing them for a vision LLM, and feeding them into an agent loop.

The shape of the problem

A vision-capable LLM (Claude, GPT-4o, Gemini, Llama 3.2 Vision, Qwen2-VL) expects images as either:

• A base64-encoded JPEG / PNG embedded in the JSON request, or
• A URL to an image you've uploaded somewhere

For a live capture loop, you want option 1. So the pipeline is:

Pengo /dev/video0 ──► Decode MJPG/YUYV ──► Resize/centre-crop ──► Encode JPEG ──► Base64 ──► LLM API
       ▲                                                                                      │
       │                                                                                      ▼
       └─────────────── Optional: read at fixed rate (e.g. 1 fps for "see now") ◄──────────────┘

There are four reasonable Python ways to get the frames off the device:

Method 1: PyAV (recommended)

PyAV wraps FFmpeg's libav* libraries directly. Fastest, most reliable, no subprocess.

pip install av pillow

import av
import io
from PIL import Image

container = av.open("/dev/video0", format="v4l2", options={
    "input_format": "mjpeg",
    "framerate":   "30",        # ask for 30 fps from the device
    "video_size":  "1920x1080",
})

for frame in container.decode(video=0):
    img = frame.to_image()      # PIL.Image
    # downscale to keep API tokens and latency reasonable
    img.thumbnail((1280, 1280))
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=85)
    jpeg_bytes = buf.getvalue()
    # ready to send to any vision API

To grab a single frame on demand (no continuous loop):

container = av.open("/dev/video0", format="v4l2", options={
    "input_format": "mjpeg",
    "framerate":   "30",
    "video_size":  "1920x1080",
})
frame = next(container.decode(video=0))
img = frame.to_image()
img.save("/tmp/snap.jpg", "JPEG", quality=90)

Method 2: OpenCV

OpenCV's VideoCapture is simple but its MJPEG support can be flaky on Linux. Stick with YUYV here.

pip install opencv-python-headless

import cv2

cap = cv2.VideoCapture(0, cv2.CAP_V4L2)
cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*"YUYV"))
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1920)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 1080)
cap.set(cv2.CAP_PROP_FPS, 60)

ok, frame = cap.read()
if ok:
    cv2.imwrite("/tmp/snap.jpg", frame)

OpenCV reads BGR

OpenCV stores frames as BGR. When sending to a vision API, encode via cv2.imencode(".jpg", cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)) to get correct colours.

Method 3: ffmpeg-python

Thin wrapper over the FFmpeg CLI. Useful when you want FFmpeg's codec and filter graph without writing the pipeline yourself.

pip install ffmpeg-python

import ffmpeg
import numpy as np

proc = (
    ffmpeg
    .input("/dev/video0", format="v4l2", input_format="mjpeg",
           framerate="30", video_size="1920x1080")
    .output("pipe:", format="rawvideo", pix_fmt="rgb24")
    .run_async(pipe_stdout=True)
)

while True:
    in_bytes = proc.stdout.read(1920 * 1080 * 3)
    if not in_bytes:
        break
    frame = np.frombuffer(in_bytes, np.uint8).reshape([1080, 1920, 3])
    # ... use frame ...

Method 4: GStreamer via PyGObject

Best when you want to leverage the full GStreamer pipeline (e.g. mixing in overlays, motion detection, hardware decode).

import gi
gi.require_version("Gst", "1.0")
from gi.repository import Gst, GLib

Gst.init(None)

pipeline = Gst.parse_launch(
    "v4l2src device=/dev/video0 ! "
    "image/jpeg,width=1920,height=1080,framerate=30/1 ! "
    "jpegdec ! videoconvert ! "
    "appsink name=sink emit-signals=true sync=false max-buffers=1 drop=true"
)

pipeline.set_state(Gst.State.PLAYING)
sink = pipeline.get_by_name("sink")

while True:
    sample = sink.emit("pull-sample")
    buf = sample.get_buffer()
    caps = sample.get_caps()
    # extract a numpy array from buf, then PIL.Image, then JPEG

Encoding for an LLM

Whatever method you use, you'll end up with either:

• A PIL.Image, or
• A numpy.ndarray of shape (H, W, 3) (RGB), or
• Raw BGR bytes

For vision APIs, the practical path is:

1. Resize so the longest side is ≤ 1568 px for Claude, ≤ 2048 px for OpenAI. Bigger images are accepted but cost more tokens.
2. Centre-crop if you have a known region of interest.
3. Encode as JPEG quality 85 — a good quality/size compromise.

def frame_for_api(img: Image.Image, max_side: int = 1568, quality: int = 85) -> bytes:
    img = img.convert("RGB")
    img.thumbnail((max_side, max_side))
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=quality)
    return buf.getvalue()

Rate limiting

Most "see what's happening on this HDMI output" agent loops don't need 60 fps. 1 frame every 1–5 s is usually enough and saves API costs.

import time

INTERVAL = 2.0  # seconds between vision requests

last_send = 0.0
for frame in container.decode(video=0):
    now = time.monotonic()
    if now - last_send < INTERVAL:
        continue
    last_send = now

    jpeg_bytes = frame_for_api(frame.to_image())
    response = ask_vision_model(jpeg_bytes, prompt="What is on screen?")
    print(response)

Local vs cloud vision models

For privacy / cost reasons, you can run a vision model locally:

Model	Size	Runs on	Quality
Llama 3.2 Vision 11B	~6 GB	GPU with ≥ 8 GB VRAM	Good for general description
Qwen2-VL 7B	~5 GB	GPU with ≥ 8 GB VRAM	Strong OCR + UI understanding
MiniCPM-V 2.6	~5 GB	GPU with ≥ 8 GB VRAM	Lightweight, optimised for low VRAM
moondream2	~1.7 GB	CPU or any GPU	Fast, decent for basic questions

For cloud:

Provider	Models	Notes
Anthropic	Claude Haiku 4.5, Sonnet 5, Opus 5	Excellent OCR + UI understanding, good at agentic workflows
OpenAI	GPT-4o, GPT-4.1	Fast, good vision
Google	Gemini 2.0 Flash	Cheap and fast

See Example scripts → for a complete, working agent loop.