non-threaded.py

from shared import camera_producer, cpu_bound, display


def main():
    frames = camera_producer()
    cpu_bound_op = cpu_bound()
    for frame in frames:
        cpu_bound_op(frame)
        display(frame)


if __name__ == '__main__':
    main()

shared.py

import cv2
import time
import numpy as np

LOG_EVERY_X_FRAMES = 30


class FPS:
    def __init__(self, length: int, title: str):
        self.length = length
        self.counter = 0
        self.start = time.time_ns()
        self.title = title

    def tick(self):
        self.counter += 1
        if (self.counter % self.length == 0):
            print(self.title, round(self.counter/round((time.time_ns() -
                                                        self.start)/1e9)), "FPS", round((time.time_ns() -
                                                                                         self.start)/(self.counter*1e3)), 'us')


def scale_frame(frame, wanted_frame_width: int):
    '''Scales frame to wanted width if smaller than frame or returns original'''
    raw_frame_height, raw_frame_width, _ = frame.shape
    if wanted_frame_width >= raw_frame_width:
        return frame
    scale_factor = wanted_frame_width / raw_frame_width
    width = int(raw_frame_width * scale_factor)
    height = int(raw_frame_height * scale_factor)
    dim = (width, height)
    frame_scaled = cv2.resize(
        frame, dim, interpolation=cv2.INTER_AREA)
    return frame_scaled


display_ticker = FPS(LOG_EVERY_X_FRAMES, 'Display')


def display(image_frame):
    cv2.imshow('display', scale_frame(image_frame, 200))
    cv2.waitKey(1)
    display_ticker.tick()


def cpu_bound():
    face_cascade = cv2.CascadeClassifier('haarcascade_frontalface_default.xml')
    cpu_bound_ticker = FPS(LOG_EVERY_X_FRAMES, 'CPU Bound')

    def inner(frame):
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        face_cascade.detectMultiScale(gray, 1.01, 5)
        cpu_bound_ticker.tick()

    return inner


def camera_producer():
    camera_ticker = FPS(LOG_EVERY_X_FRAMES, 'Camera')
    cam = cv2.VideoCapture(0)
    while True:
        ok, frame = cam.read()
        if ok:
            camera_ticker.tick()
            yield frame