Skip to content

Example Scripts

Complete, runnable Python scripts demonstrating the Pengo + AI vision loop. Drop these into a venv, set an API key, and they just work.

The standalone copies are checked in at examples/ in the repo.

0. Common setup

python3 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip

# Core video + image
pip install av pillow numpy

# Optional: AI clients (install only what you use)
pip install anthropic              # Claude
pip install openai                 # GPT-4o
# pip install ollama               # local Llama 3.2 Vision

# Optional: local STT for the multimodal script
# pip install faster-whisper

1. pengo_snap.py — capture a single JPEG on demand

The simplest possible script. Reads one frame off /dev/video0 and saves it as snap.jpg. Useful as a starting point for a screenshot-style agent.

#!/usr/bin/env python3
"""
pengo_snap.py — read one frame off the Pengo HDMI Grabber and save as JPEG.

Usage:
    python pengo_snap.py [output.jpg]

Requires: pip install av pillow
"""
import sys
import av
from PIL import Image

OUT = sys.argv[1] if len(sys.argv) > 1 else "snap.jpg"
DEVICE = "/dev/video0"

container = av.open(DEVICE, format="v4l2", options={
    "input_format": "mjpeg",
    "framerate":   "30",
    "video_size":  "1920x1080",
})

frame = next(container.decode(video=0))
img: Image.Image = frame.to_image()
img.save(OUT, "JPEG", quality=90)
print(f"saved {OUT} ({img.size[0]}x{img.size[1]})")

2. pengo_vision_loop.py — periodic vision loop → Claude

Continuously grabs a frame every INTERVAL seconds, downsizes it, encodes as JPEG, and sends it to Claude with a fixed prompt. Prints the model's response.

#!/usr/bin/env python3
"""
pengo_vision_loop.py — read frames off the Pengo at a fixed rate and ask
Claude (or any vision-capable model) about them.

Usage:
    export ANTHROPIC_API_KEY=sk-ant-...
    python pengo_vision_loop.py

Requires: pip install av pillow anthropic
"""
import io
import os
import sys
import time
import base64

import av
from PIL import Image

try:
    import anthropic
except ImportError:
    sys.exit("install anthropic: pip install anthropic")

INTERVAL  = float(os.getenv("INTERVAL", "2.0"))     # seconds between frames
PROMPT    = os.getenv("PROMPT",
    "Describe what is on screen in one or two short sentences. "
    "If you see any text, transcribe it verbatim. "
    "If you see an error or warning, call it out.")
DEVICE    = os.getenv("VIDEO_DEVICE", "/dev/video0")
MODEL     = os.getenv("MODEL", "claude-haiku-4-5")
MAX_SIDE  = int(os.getenv("MAX_SIDE", "1568"))

client = anthropic.Anthropic()

def frame_to_jpeg(img: Image.Image, max_side: int, 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()

def ask_claude(jpeg: bytes, prompt: str) -> str:
    msg = client.messages.create(
        model=MODEL,
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image",
                 "source": {"type": "base64",
                            "media_type": "image/jpeg",
                            "data": base64.b64encode(jpeg).decode()}},
                {"type": "text", "text": prompt},
            ],
        }],
    )
    return "".join(b.text for b in msg.content if b.type == "text").strip()

def main() -> None:
    print(f"opening {DEVICE}…", flush=True)
    container = av.open(DEVICE, format="v4l2", options={
        "input_format": "mjpeg",
        "framerate":   "30",
        "video_size":  "1920x1080",
    })

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

        try:
            jpeg = frame_to_jpeg(frame.to_image(), MAX_SIDE)
            reply = ask_claude(jpeg, PROMPT)
            print(f"[{time.strftime('%H:%M:%S')}] {reply}", flush=True)
        except Exception as e:
            print(f"[{time.strftime('%H:%M:%S')}] error: {e}", flush=True)

if __name__ == "__main__":
    main()

Variants

  • Local model (Ollama): swap ask_claude for a requests.post to http://localhost:11434/api/chat with model="llama3.2-vision".
  • OpenAI: swap ask_claude for client = openai.OpenAI() and use client.chat.completions.create(model="gpt-4o", ...).
  • Multiple frames in one prompt: keep a rolling buffer of the last N JPEG bytes and send them all in one request.

3. pengo_alert.py — vision + alerting

An offshoot of #2 that only prints (and optionally pages) when the model detects a state change worth noticing. Useful for monitoring dashboards, build pipelines, observability screens.

#!/usr/bin/env python3
"""
pengo_alert.py — vision loop that only fires on state changes / anomalies.
"""
import os, time, hashlib, sys
import av
from PIL import Image

# ... (import your favourite vision client) ...

INTERVAL  = float(os.getenv("INTERVAL", "5.0"))
PROMPT    = ("Look at this screen. Reply with JSON: "
             "{'state': '<short label>', 'alert': <true|false>, "
             "'summary': '<1 sentence>'}. Alert only if something is "
             "wrong, stuck, or unexpected.")

last_state = None

for frame in container.decode(video=0):
    if time.monotonic() - last_send < INTERVAL:
        continue
    jpeg = frame_to_jpeg(frame.to_image(), 1568)
    raw  = ask_claude(jpeg, PROMPT)

    try:
        import json
        result = json.loads(raw.strip("`\n "))
    except Exception:
        result = {"state": "?", "alert": True, "summary": raw}

    if result["state"] != last_state:
        print(f"STATE → {result['state']} — {result['summary']}", flush=True)
        last_state = result["state"]
    if result["alert"]:
        # hook your alerting here (page, webhook, log file, …)
        pass

4. pengo_conference.py — webcam in your own video calls

Use the Pengo as a virtual webcam in Zoom / Discord / Meet / any browser. PipeWire makes this a one-liner.

# Install v4l2loopback
sudo pacman -S v4l2-loopback-dkms v4l-utils
# or: sudo apt install v4l2loopback-dkms v4l-utils

# Load the loopback module with the right defaults
sudo modprobe v4l2loopback devices=1 video_nr=10 \
    card_label="Pengo Virtual" exclusive_caps=1

# Push frames into it
gst-launch-1.0 \
    v4l2src device=/dev/video0 ! image/jpeg,width=1920,height=1080,framerate=30/1 \
        ! jpegdec ! videoconvert \
        ! video/x-raw,format=YUY2,width=1280,height=720,framerate=30/1 \
        ! v4l2sink device=/dev/video10

Then in Zoom / Discord / browser, pick Pengo Virtual as the camera. It's the Pengo signal, optionally resized to a more conferencing-friendly 1280×720.

5. Reading audio for STT

Capturing the Pengo's USB audio in Python for downstream speech-to-text:

import av

container = av.open("hw:1,0", format="alsa", options={
    "sample_rate": "48000",
    "sample_format": "s16",
})

import numpy as np
chunks = []
for frame in container.decode(audio=0):
    chunks.append(frame.to_ndarray())   # shape (channels, samples), int16

audio = np.concatenate(chunks, axis=1) if chunks else np.zeros((0, 0), dtype=np.int16)
# save or stream into faster-whisper / openai.audio.transcriptions.create

Production tips

  • Use a venv — uv venv && source .venv/bin/activate works well.
  • Process per stream — a Pengo pipeline typically wants a dedicated thread or process for video, another for audio.
  • Drop frames aggressively. A queue with maxsize=2 and drop=True is your friend.
  • Backpressure on the API. If your request rate exceeds the API rate limit, batch frames or sleep — don't queue indefinitely.
  • Shutdown gracefully. Catch KeyboardInterrupt and container.close() to release the device.

Want more? See the upstream docs: