Skip to content

Instantly share code, notes, and snippets.

@moodoki
Created September 19, 2021 05:25
Show Gist options
  • Save moodoki/5dda3bdb5264c3511b473c7a2d1038e1 to your computer and use it in GitHub Desktop.
Save moodoki/5dda3bdb5264c3511b473c7a2d1038e1 to your computer and use it in GitHub Desktop.
Barebones Camera Thread
import threading
import numpy as np
import cv2
class CameraThread(threading.Thread):
def __init__(self, name='CameraThread'):
super().__init__(name=name, daemon=True)
self.stop_event = False
self.open_camera()
self.setup_inference_engine()
self._frame, self.results = np.zeros((300,300,3), dtype=np.uint8), [] #initial empty frame
self.lock = threading.Lock()
self.log_counter = 0
def open_camera(self):
self.webcam = cv2.VideoCapture(0) #integer index or a video filename
def setup_inference_engine(self):
pass #saved for later
def process_frame(self, img):
#saved for later
return []
def run(self):
while not self.stop_event:
ret, img = self.webcam.read()
if not ret: #re-open camera if read fails. Useful for looping test videos
self.open_camera()
continue
results = self.process_frame(img)
with self.lock: self.results, self._frame = results, img.copy()
self.log_frame()
def log_frame(self):
pass #saved for later
def stop(self): self.stop_event = True
def read(self):
with self.lock: return self._frame.copy(), self.results
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment