Barebones Camera Thread

barebones_camera_thread.py

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