3

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.

Tom
  • 4,070
  • 4
  • 22
  • 50
  • maybe you could look into rospy, which is the python implementation of the Robot Operating System. Contrary to its name its not actually an operating system but an asynchronous communication network with message publishers and subscribers. You can have a look here: http://wiki.ros.org/rospy and http://wiki.ros.org/ROS/Tutorials/WritingPublisherSubscriber%28python%29 There might however be simpler and leaner solutions I am not aware of. – Manumerous Sep 06 '21 at 22:01
  • It seems you have to make "set_stream" and "detect_with_video" async (or run the latter one in another thread) to avoid blocking. To retrieve the generated data, see https://stackoverflow.com/questions/37549846/how-to-use-yield-inside-async-function Alternatively you can use an asynchronous queue to transfer the data. – Michael Butscher Sep 06 '21 at 22:03

0 Answers0