Passing data asynchronously between functions

Viewed 64

I'm writing a program that looks for objects in a video stream using a Google Coral device.

I have a function that continuously detects objects in the video.

I want to 'stream' the data from that function to two other functions/classes:

  1. VideoViewer: This is a class that manages a debug window, showing the bounding boxes, etc.

  2. An API endpoint: This part of the application has not been implemented yet, but I want to expose the data from the object detection function, via a websocket, later on.

The following is what I have so far. It yields the results

from PIL import Image
from cv2 import VideoCapture

def detect_with_video(self, video: VideoCapture):
    """Detect objects in a video"""
    streaming = True
    frame_id = 0
    results = []
    detection_delay = 4

    while streaming:
        _, frame = video.read()

        if frame_id % detection_delay == 0:
            results = self._detect_image(Image.fromarray(frame))

        for result in results:
            yield result

In my app.py I would like to be able to do something like this:

class App:
    """The application"""
    def __init__(self):
        self.video: VideoCapture = cv2.VideoCapture(0)

    def run(self):
        # Initialize video
        VideoViewer().new_window()

        generator = Detector().detect_with_video(self.video)

        VideoViewer().set_stream(generator)
        HttpApi().set_stream(generator)

class VideoViewer:
    """VideoViewer launches video windows"""

   def set_stream(self, generator):
       for item in generator:
            print(item)

My guess is that I have to make the set_stream methods async, but yeah.

How can I make both consumers non blocking?

I'm new to python and aren't sure how to do this. Any help is highly appreciated.

0 Answers
Related