refactor: video processing

[mxmaxime]

Today we process at a very low fps rate to limit the CPU load, because every frame is analyzed by the model to check whether or not people are presents in the frame (picture).

It works well today because we don't need to do anything with the video. But this will change with #33 (video stream) and #58 (video save). So we need to refactor this but we still want to analyze (with the model) at a low fps rate (1 fps for example), because otherwise it is a waste of resource.

More importantly, the process that reads frames from PiCamera and the tensorflow processing are on the same process. That means that we cannot have 30-60 fps, because this processing is cpu intensive, it takes time!

Architecture

If your goal is to read as many (new) frames as possible and then process them, then this is a standard producer/consumer relationship. Your “producer” is the frame reader (single thread) which only grabs new frames and sends them to the consumer. The consumers should be a set of processes that look for new frames in the queue and process them. There are many, many ways to accomplish this in Python, but as long as you use a producer/consumer relationship, you’ll be okay.

Why Processes and not Threads?

Because Python.

Python can actually use all available CPU cores through the multiprocessing module. It's limited to one core only when using multiple threads in parallel due to​ Python supports multi-threading but the global interpreter lock (GIL) prevents us from utilising all CPU cores for CPU heavy tasks. The recommended approach is to use Python’s multiprocessing library to work around the GIL, but that has its own set of challenges, notably the ability to share data between sub-processes is limited.

Communication between FrameProducer and Processing

Queue or Pipe? At the first glance, I implemented a "simple" Queue. The issue is that a Queue is a one way communication channel.

ℹ️ And why not using Event?

• Frame producer -> frames -> processing
• Processing -> someone is here, record -> frame producer

We need duplex (two-way) communication channel.

Sources:

• Exchanging objects between processes
• Pipe
• Queue
• How does Event work?

Empty BytesIO

I kept the same structure as before, so the capture was in a BytesIO. Then, its BytesIO was pushed into the Queue. Then, the consumer was trying to get it and transform it to PIL Image but I was getting this kind of error: OSError: cannot identify image file <_io.BytesIO object at [memory address]>. The issue was that the BytesIO was empty because the seek(0) and truncate() was done. It is a shared memory.

Lock due to Queues

⚠️ If you fill a Queue with a maxsize, the producer will be locked.

Queue constructor - maxsize is an integer that sets the upperbound limit on the number of items that can be placed in the queue. Insertion will block once this size has been reached, until queue items are consumed. If maxsize is less than or equal to zero, the queue size is infinite.

Queue.put() - If optional args block is true and timeout is None (the default), block if necessary until a free slot is available. Source.

⚠️ make sure you are using mp.Queue (Queue from multiprocessing python package), otherwise it won't work as expected.

Python version

In Python 3.7.4 I got this error when I was trying to queue.get the PIL Image.

12: Put image <PIL.Image.Image image mode=RGB size=640x480 at 0xB66F0710> with size 640x480 in queue
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/multiprocessing/queues.py", line 236, in _feed
    obj = _ForkingPickler.dumps(obj)
  File "/usr/local/lib/python3.7/multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
TypeError: can't pickle ImagingCore objects
^CProcess Process-2:
Traceback (most recent call last):
  File "multiprocess_video.py", line 105, in <module>
    proc.join()
  File "/usr/local/lib/python3.7/multiprocessing/process.py", line 140, in join
    res = self._popen.wait(timeout)
  File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 48, in wait
    return self.poll(os.WNOHANG if timeout == 0.0 else 0)
  File "/usr/local/lib/python3.7/multiprocessing/popen_fork.py", line 28, in poll
    pid, sts = os.waitpid(self.pid, flag)Traceback (most recent call last):

It was working well with Python 3.7.3. I didn't find any change related to BytesIO or Queue in release notes.

Warning

PiCamera can be accessed by only one process. As soon as we instantiate PiCamera the resource is locked! Then, if you try to do something like capture_continuous in another process, it will be locked, without any crash.

Issue: I instantiated a class that was instantiating PiCamera in the constructor. Then, I was starting a method inside a process (mp.Process(target=class.run)), which was doing self.picamera.capture_continuous(...). And... everything was locked.

Orphans process

The issue is: the parent process is getting terminate() because the alarm is turn off. But! The process has created some other processes: one to take frames, one to process it. These two processes keep running and are orphans, and so the alarm is not switched off, which is a big issue.

Read about it.

Research

PiCamera multiprocessing

The idea is to use a process to take frames for picamera and one for processing with tensorflow.

• Picamera documentation example.

Sources

• https://picamera.readthedocs.io/en/release-1.13/faq.html#camera-locks-up-with-multiprocessing
• https://picamera.readthedocs.io/en/release-1.13/recipes2.html#capturing-images-whilst-recording
• Python Process and exceptions
• https://www.cloudcity.io/blog/2019/02/27/things-i-wish-they-told-me-about-multiprocessing-in-python/
• how-to-process-sigterm-signal-gracefully
• Python BytesIO binary streams