Save a camera frame every n minutes in Python without stale frames

A logger that sleeps five minutes, calls read() and saves whatever comes back shows you the recent past: the capture object has been queueing frames the whole time, and read() returns the oldest one still buffered. This how-to is for anyone building an interval snapshot recorder in Python with OpenCV — a process monitor, a time-lapse of a fixture, an evidence trail for a station. You will drain the buffer with grab() before every save, schedule ticks with time.monotonic(), and write timestamped grayscale PNGs. You will also put a proper title on the preview window, scale the same pattern to three cameras, and prove with a burned-in clock that each file shows what the camera saw when the timer fired. The worked numbers assume 30 fps and a 300 s interval; the formulas re-price them for any camera and interval.

Why does read() return an old frame after a long sleep?

OpenCV's VideoCapture keeps filling a small internal buffer while your script sleeps. After a 300 s wait on a 30 fps camera — 9,000 frames produced between saves — read() hands you the oldest frame left in that buffer, not the scene at the tick.

The queue lives inside the backend the capture opened. It is a few frames deep on typical desktop setups, and when it is full the oldest entry is dropped. The frames your sleep produced have long since cycled through, and what survives is the residue. If the queue holds five frames at 30 fps, that residue is 5 × 33 ms ≈ 170 ms old at the moment read() returns. Backends that queue deeper leave a bigger gap, and on some Windows configurations the saved image lands seconds behind the tick.

The same startup path hides a second trap. The first grab() after open() consistently takes hundreds of milliseconds while the backend starts delivering frames, an effect logged in the OpenCV issue tracker. A logger that saves on its first tick therefore writes either a failed retrieve or a half-initialized frame.

Worked example, 30 fps camera saving every 300 s: frames between saves = 30 × 300 = 9,000; frame period = 1 / 30 ≈ 0.033 s ≈ 33 ms. A slower camera at 5 fps gives 5 × 300 = 1,500 frames per interval and a frame period of 1 / 5 = 0.2 s, which also makes any per-tick drain six times more expensive in wall time.

Real cameras sit on both rows: the LS-CMR-0166 is a 12 MP USB3 area scan camera streaming 30 fps at 4000 × 3000, so the first row is its arithmetic.

How do you grab the newest frame at the tick instead of the oldest buffered one?

Split the call: grab() flushes stale frames and retrieve() decodes the newest one. A drain of 6 grabs at 30 fps costs about 0.2 s at the tick and leaves the saved frame within one frame period, about 33 ms, of the timer.

grab() moves the next frame into the capture's internal storage without any conversion or copy to NumPy; retrieve() turns the last grabbed frame into the array. At the tick, the first grabs pop whatever stale frames are queued — those return almost immediately — and once the queue is empty, the next grab blocks until a fresh frame arrives. That blocking grab is the guarantee: retrieve() then returns a frame that reached the camera during the drain, so its age is bounded by one frame period, not by how long you slept.

Set DRAIN_GRABS above the queue depth. Start at 6 for desktop backends and let the probe in the proof section confirm it; on a backend that queues deeper, raise the value — each extra stale pop costs milliseconds. Worst case at 5 fps, all six grabs block and the drain costs 6 × 0.2 s = 1.2 s; at 30 fps the same six cost about 0.2 s.

import time
import cv2

DRAIN_GRABS = 6  # int, initial drain depth; raise it if the freshness probe fails

def save_fresh_snapshot(cap: cv2.VideoCapture, drain_grabs: int = DRAIN_GRABS) -> bool:
    '''Pop stale frames, then write the newest one as a grayscale PNG.'''
    for _ in range(drain_grabs):
        cap.grab()                 # fast on queued frames, blocking on an empty queue
    ok, frame = cap.retrieve()     # decodes the last grabbed frame
    if not ok:
        return False
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    stamp = time.strftime('%Y%m%d_%H%M%S')   # sorts correctly in any file browser
    return bool(cv2.imwrite(f'snapshot_{stamp}.png', gray))

cv2.CAP_PROP_BUFFERSIZE exists as a settable property, and treating it as the fix is a trap: support is partial across backends, so the guarantee would silently hinge on which backend opened the camera. The drain needs no support at all. The companion symptom — a consumer that asks for the latest image and keeps receiving duplicates — is this same queue in another costume: drain the buffer queue when latest-image mode returns duplicates.

What does the full snapshot script look like, step by step?

Open the camera and throw away the first handful of frames, because the first grab() after open() consistently takes hundreds of milliseconds. Then loop on time.monotonic(): sleep to the deadline, drain, retrieve, convert to gray, and write one timestamped PNG per tick.

1. Pick the camera index and backend: cv2.VideoCapture(index, cv2.CAP_ANY) lets OpenCV choose; pass cv2.CAP_DSHOW on Windows or cv2.CAP_V4L2 on Linux when you need identical behavior on every machine.
2. Open the capture and refuse to run if isOpened() is False — a logger that never opened a camera must not look like a working one.
3. Warm up with 30 grabs: it covers the slow first grab and gives auto-exposure about one second of frames at 30 fps.
4. Set the first deadline with time.monotonic(), which ignores wall-clock changes such as NTP corrections during the night.
5. Sleep for the remaining time, then drain DRAIN_GRABS frames and call retrieve().
6. Convert with cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) and write the file with a stamp from time.strftime('%Y%m%d_%H%M%S').
7. Check the boolean return of cv2.imwrite() and print a line when it fails.
8. Advance the deadline with += INTERVAL_S rather than re-reading the clock, so tick cost cannot accumulate drift, and release() the camera in a finally block.

import time
import cv2

CAMERA_INDEX = 0        # int, first device the backend enumerates
BACKEND = cv2.CAP_ANY   # int, 0 lets OpenCV choose; CAP_DSHOW on Windows, CAP_V4L2 on Linux
INTERVAL_S = 300.0      # float, one snapshot every 5 minutes
DRAIN_GRABS = 6         # int, must exceed the backend queue depth
WARMUP_GRABS = 30       # int, about 1 s at 30 fps

cap = cv2.VideoCapture(CAMERA_INDEX, BACKEND)
if not cap.isOpened():
    raise RuntimeError(f'camera index {CAMERA_INDEX} did not open')

for _ in range(WARMUP_GRABS):
    cap.grab()

next_tick = time.monotonic() + INTERVAL_S
try:
    while True:
        pause = next_tick - time.monotonic()
        if pause > 0:
            time.sleep(pause)
        for _ in range(DRAIN_GRABS):
            cap.grab()                  # pops stale frames, blocks for the last fresh one
        ok, frame = cap.retrieve()
        next_tick += INTERVAL_S         # keeps the average interval exact
        if not ok:
            print('retrieve failed')
            continue
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        stamp = time.strftime('%Y%m%d_%H%M%S')
        if not cv2.imwrite(f'snapshot_{stamp}.png', gray):
            print(f'imwrite failed at {stamp}')
except KeyboardInterrupt:
    pass
finally:
    cap.release()

How do you show a live grayscale view and put a title on the window?

Convert every frame with cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY), show it with imshow(), and set the caption with cv2.setWindowTitle(). Note that a running display loop keeps the buffer fresh, so the tick needs no drain and retrieve() returns a frame at most one period old.

highgui only paints when waitKey() runs, so the loop below services the window every frame; the title line keeps the snapshot count in front of the operator. Gray display also makes the on-disk choice explicit: raw gray data is one byte per pixel instead of three, a third of the bytes before compression.

saved_count = 0                                # int, snapshots written so far
WINDOW = 'camera 0 - snapshot every 5 min'     # str, caption on the preview window

cv2.namedWindow(WINDOW, cv2.WINDOW_AUTOSIZE)
while True:
    ok, frame = cap.read()                     # continuous reads keep the queue fresh
    if not ok:
        break
    cv2.setWindowTitle(WINDOW, f'camera {CAMERA_INDEX} - {saved_count} snapshots saved')
    cv2.imshow(WINDOW, cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY))
    if time.monotonic() >= next_tick:
        stamp = time.strftime('%Y%m%d_%H%M%S')
        cv2.imwrite(f'snapshot_{stamp}.png', frame)   # keep the color file on disk
        saved_count += 1
        next_tick += INTERVAL_S
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break
cv2.destroyAllWindows()

When the preview must live inside a GUI rather than a highgui window, display OpenCV video in a Tkinter window without flicker shows the thread split; on an embedded board without a host filesystem, save grayscale camera frames as BMP to an SD card is the same write path with no display at all.

Does the same pattern work for three cameras at once?

Open one VideoCapture per device index — 0, 1 and 2 for a three-camera rig. Run each capture in its own thread with its own cam{index}_ filename prefix, and the same drain-then-retrieve tick runs per camera without the captures blocking each other.

Keep everything in one process rather than three: the warm-up, drain and stamping code stays shared, and the timestamps line up across cameras. Threads matter because grab() blocks until that camera delivers; with one thread per camera, a slow camera cannot delay another's tick.

import threading

running = True   # bool, set False from the main thread to stop all cameras

def camera_worker(index: int, interval_s: float) -> None:
    cap = cv2.VideoCapture(index, cv2.CAP_ANY)
    if not cap.isOpened():
        print(f'camera {index} did not open')
        return
    for _ in range(WARMUP_GRABS):
        cap.grab()
    next_tick = time.monotonic() + interval_s
    while running:
        if time.monotonic() < next_tick:
            time.sleep(0.2)
            continue
        for _ in range(DRAIN_GRABS):
            cap.grab()
        ok, frame = cap.retrieve()
        next_tick += interval_s
        if ok:
            stamp = time.strftime('%Y%m%d_%H%M%S')
            cv2.imwrite(f'cam{index}_{stamp}.png',
                        cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY))

threads = [threading.Thread(target=camera_worker, args=(i, 300.0), daemon=True)
           for i in range(3)]      # cameras 0, 1, 2
for t in threads:
    t.start()

Device index order can change across reboots, so pin each camera to a physical port and re-check after any re-plug. Three color USB cameras on one controller can also starve each other; spread them across controllers, drop the saved data to gray, or move the station to GigE cameras on their own switch, where each camera's frames arrive on a dedicated cable. For a fixed multi-camera station, the MV-CB016-10GM-M-W is a 1.6 MP GigE board-level camera with a global-shutter IMX296 and an M12 mount.

Through a vendor SDK the function names differ but the rule is identical: keep a grab running per camera and decode only the frame you intend to write. The area scan cameras and board-level cameras pages filter by interface, shutter and resolution when the camera choice is still open.

When is a timer snapshot the wrong tool for the job?

Switch tools when the moment matters more than the clock. Use a hardware trigger for frames at exact instants, a recorded stream when nothing between ticks may be lost, and a smart camera that decides on-device when the goal is a pass/fail verdict rather than an archive.

• Archive at fixed times — the script above; nothing more is needed.
• Frame must coincide with a part sitting in position — wire the camera's trigger input and use the hardware trigger grab loop in Python; the strobed light fires with it, so exposure is deterministic too.
• Nothing between ticks may be lost — record continuously with cv2.VideoWriter, or hand the timing to the camera itself: HTTP snapshot polling on a timer covers network cameras that serve a still image URL.
• A decision per part on the line — the smart cameras & vision sensors category runs the inspection on the device, no PC in the loop.

Triggered and high-rate capture also want a camera built for it: the LS-BCM-0003 is a 3.1 MP USB3 board-level camera with a 1/1.8" global shutter that reads out at 120 fps.

At 120 fps the frame period is 1 / 120 ≈ 8 ms, so even a six-grab drain bounds staleness to single-digit milliseconds — and the global shutter means a moving part is frozen without the smear a rolling sensor would add.

How do you prove the saved frames are fresh before you trust them?

Burn the save time into the image with cv2.putText, save a frame every second for a minute, and compare each burned clock with the filename. A lag beyond one frame period, about 33 ms at 30 fps, means your drain_grabs value is too low.

probe_until = time.monotonic() + 60.0   # float, one minute of probe frames
while time.monotonic() < probe_until:
    for _ in range(DRAIN_GRABS):
        cap.grab()
    ok, frame = cap.retrieve()
    if ok:
        clock = time.strftime('%H:%M:%S')
        cv2.putText(frame, clock, (10, 40),
                    cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 255, 255), 2)
        cv2.imwrite(f'probe_{time.strftime("%Y%m%d_%H%M%S")}.png', frame)
    time.sleep(1.0)

Open a few probe files and read the burned clock against the stamp in the filename: the stale-queue bug shows up immediately as a burned clock that trails the wall clock by seconds. Once the drain is deep enough, the remaining error is bounded by construction — the drain's final grab returns only after a new frame has arrived — so the worst case is one frame period. If a camera delivers fewer frames than configured because USB bandwidth or a long exposure stretches the period, the drain takes proportionally longer; measure the real rate with camera frame rate measurement from timestamped frames and re-price the arithmetic before blaming the drain.

What do you check before the first unattended night?

Check five things: the freshness test passing at the real interval, disk headroom computed as saves per day times file size, a pinned camera index, a warm-up pass so the first snapshot is neither dark nor stale, and an imwrite() return value you actually log.

Before the first run:

1. Run the probe at the real interval and fps — queue depth differs per backend, and the probe is the only proof your DRAIN_GRABS holds.
2. Compute disk as saves per day × file size: one save a minute is 1,440 files a day, so a 5 MB PNG means about 7 GB per camera — measure one day, then set retention.
3. Log the boolean from cv2.imwrite(); a False nobody sees looks exactly like a working logger.
4. Pin each camera to a physical port, since enumeration order can change across reboots.
5. Keep the warm-up grabs — the first grab takes hundreds of milliseconds and auto-exposure needs frames to converge.

While it runs:

6. Run headless where possible — a 30 fps display loop spends a CPU core for nothing between saves.
7. Save gray only if gray answers the questions you will ask later; color costs a third more raw bytes but preserves evidence.

When the camera side of the station is still open, the camera finder filters by interface, shutter and resolution, and the build a system configurator pairs the camera with a lens and lighting; more capture-loop patterns are collected on the systems topic page.