4
\$\begingroup\$

My program intends to capture video streams (in mjpeg) with OpenCV and upload the captured frames into Google Cloud Storage for later processing. I am expecting to capture ~15-20 frames per second and expected to upload nearly all frames within a second.

The trick is, the script is running on Raspberry Pi 3, Model B, and together with Python, I found that it has some serious I/O limitation in terms of frames loading and uploading.

I use multiple threads to handle the loads and still find it is taking too long for uploading tasks. My current internet has a constant uplink speed of 30mbps. The received frame size is reduced to ~500KB.

I am thinking of instead of saving the frames locally, treat the loaded frames as an bytestring array and parse it accordingly in the server. But as of now, I would much prefer perhaps there is a much better solution which I am not awared of.

Below is the code I've written.

import os
import cv2
import sys
import time
import imutils
import datetime
from threading import Thread
from google.cloud import storage
from imutils.video import VideoStream

if sys.version_info < (3, 0):
    from Queue import Queue
else:
    from queue import Queue

BUCKET_NAME = "name"
RPID_FOLDER = "rpi1"
BY_DAY = datetime.datetime.now().strftime("%Y-%m-%d")

LOCAL_STREAM_MJPEG = "http://0.0.0.0:8080/stream/video.mjpeg"
REMOTE_STREAM_MJPEG = "http://address:8080/stream/video.mjpeg"  

QUEUE_SIZE = 120
grabbedFrames = Queue(QUEUE_SIZE)
processFrames = Queue(QUEUE_SIZE)


class FrameGraber(Thread):

    def __init__(self):
        Thread.__init__(self)
        self.vs = VideoStream(src=REMOTE_STREAM_MJPEG).start()

    def run(self):
        global grabbedFrames

        while True:
            frame = self.vs.read()
            if frame is not None:
                grabbedFrames.put(frame)
                time.sleep(0.05)


class GCSUploader(Thread):

    def __init__(self):
        Thread.__init__(self)
        self.start()
        self.client = storage.Client()
        self.bucket = self.client.get_bucket(BUCKET_NAME)

    def run(self):
        global processFrames

        while True:
            if (not processFrames.empty()):
                fileName = processFrames.get()
                try:
                    blob = self.bucket.blob("{}/{}/{}".format(RPID_FOLDER, BY_DAY, fileName))
                    blob.upload_from_filename(os.path.abspath(fileName))
                    print("{} uploaded to {}/{}".format(fileName, RPID_FOLDER, BY_DAY))
                    os.remove(fileName)
                except Exception as e:
                    print(e)


class FrameProcessor(Thread):

    def __init__(self):
        Thread.__init__(self)
        self.secondCount = 0
        self.startSecond = datetime.datetime.now().second

    def run(self):
        global grabbedFrames, processFrames
        while True:

            if datetime.datetime.now().second != self.startSecond:
                self.startSecond = datetime.datetime.now().second
                self.secondCount = 0

            if (not grabbedFrames.empty()):
                frame = grabbedFrames.get()
                self.secondCount = self.secondCount + 1
                frame = imutils.resize(frame, 800)
                fileName = "{}-{}.png".format(datetime.datetime.now().strftime("%H-%M-%S"), self.secondCount)
                cv2.imwrite(fileName, frame)
                time.sleep(0.1)
                processFrames.put(fileName)


if __name__ == "__main__":

    uploader1 = GCSUploader()
    uploader2 = GCSUploader()
    uploader3 = GCSUploader()
    uploader4 = GCSUploader()
    uploader5 = GCSUploader()
    uploader6 = GCSUploader()
    uploader7 = GCSUploader()
    uploader8 = GCSUploader()

    frameGrabber = FrameGraber()
    frameProcessor = FrameProcessor()

    frameGrabber.start()
    frameProcessor.start()

    frameProcessor.join()
    frameGrabber.join()
    uploader1.join()
    uploader2.join()
    uploader3.join()
    uploader4.join()
    uploader5.join()
    uploader6.join()
    uploader7.join()
    uploader8.join()
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.