Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
#!/usr/bin/env python3
always getting the most recent frame of a camera
ESC - exit
# Python 2/3 compatibility
from __future__ import print_function
import os
import sys
import time
import threading
import numpy as np
import cv2 as cv
# also acts (partly) like a cv.VideoCapture
class FreshestFrame(threading.Thread):
def __init__(self, capture, name='FreshestFrame'):
self.capture = capture
assert self.capture.isOpened()
# this lets the read() method block until there's a new frame
self.cond = threading.Condition()
# this allows us to stop the thread gracefully
self.running = False
# keeping the newest frame around
self.frame = None
# passing a sequence number allows read() to NOT block
# if the currently available one is exactly the one you ask for
self.latestnum = 0
# this is just for demo purposes
self.callback = None
def start(self):
self.running = True
def release(self, timeout=None):
self.running = False
def run(self):
counter = 0
while self.running:
# block for fresh frame
(rv, img) =
assert rv
counter += 1
# publish the frame
with self.cond: # lock the condition for this operation
self.frame = img if rv else None
self.latestnum = counter
if self.callback:
def read(self, wait=True, seqnumber=None, timeout=None):
# with no arguments (wait=True), it always blocks for a fresh frame
# with wait=False it returns the current frame immediately (polling)
# with a seqnumber, it blocks until that frame is available (or no wait at all)
# with timeout argument, may return an earlier frame;
# may even be (0,None) if nothing received yet
with self.cond:
if wait:
if seqnumber is None:
seqnumber = self.latestnum+1
if seqnumber < 1:
seqnumber = 1
rv = self.cond.wait_for(lambda: self.latestnum >= seqnumber, timeout=timeout)
if not rv:
return (self.latestnum, self.frame)
return (self.latestnum, self.frame)
def main():
# these windows belong to the main thread
# on win32, imshow from another thread to this DOES work
# open some camera
cap = cv.VideoCapture(0)
cap.set(cv.CAP_PROP_FPS, 30)
# wrap it
fresh = FreshestFrame(cap)
# a way to watch the camera unthrottled
def callback(img):
cv.imshow("realtime", img)
# main thread owns windows, does waitkey
fresh.callback = callback
# main loop
# get freshest frame, but never the same one twice (cnt increases)
# see read() for details
cnt = 0
while True:
# test that this really takes NO time
# (if it does, the camera is actually slower than this loop and we have to wait!)
t0 = time.perf_counter()
cnt,img =
dt = time.perf_counter() - t0
if dt > 0.010: # 10 milliseconds
print("NOTICE: read() took {dt:.3f} secs".format(dt=dt))
# let's pretend we need some time to process this frame
print("processing {cnt}...".format(cnt=cnt), end=" ", flush=True)
cv.imshow("frame", img)
# this keeps both imshow windows updated during the wait (in particular the "realtime" one)
key = cv.waitKey(200)
if key == 27:
if __name__ == '__main__':
Copy link

dzulfiamien commented Aug 29, 2021

thanks for this solution, it helps me a lot!

Copy link

ahgsql commented May 9, 2022

it keep freezing, not responding :/

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment