I have written a Python script that runs on a Raspberry Pi and utilizes the PiCamera library to capture video, the Python Image Library (PIL) to extract individual frames, and then does some image processing on it using DIPLib and OpenCV. The goal of the program is to continuously capture frames of 3D printer filament and return the diameter value. The actual image processing portion works just fine- it's the frame capture that is causing me issues.

I am following the PiCamera Rapid Capture and processing tutorial from PiCamera and using the Python Threading library as they have done to ideally utilize more of th Pi's processor for processing and not getting bottlenecked and falling behind.

The implementation of this code is built to "Drop Frames" when there are not any threads available for processing. As I understand it, this should prevent the Pi from storing any extra frames in the buffer for processing, thus preventing a memory overflow (not sure if that's the correct terminology) from happening. Unfortunately this is exactly what is happening.

I am running the PiCamera at about 3 frames-per-second which gives ~10 threads the ability to process all the incoming images, that is until the memory starts to overflow. However, if I leave the script running for 5-10 minutes, the memory (as shown using htop) slowly compounds until it reaches maximum capacity- at which point the script basically drops all incoming frames.

UPDATE: here is the error it shows:

Exception has occurred: MemoryError exception: no description File "/home/pi/Desktop/FilamentPuller_01/pi_camera_threading_03.py", line 45, in run img = np.array(Image.open(self.stream)

My theory is that the video recording functionality of PiCamera is holding a buffer of some sort, but I am not sure how to see it or how to stop it from doing that. I've been using VSCode on the Pi to debug, and each thread doesn't seem to holding any more data at a time than they should- essentially there should be no reason for them to compound more data from one cycle to the next as all the variables are reused.

I have included my code below, please let me know what other information I can provide to help with solving this issue. Thank you for any insight you might have

import io
import sys
import time
import threading
import cv2
import numpy as np
import os
import picamera
from PIL import Image
import csv
from diameter_detection import diameter_detection
from moving_average import MovingAverageFilter
from serial_write import serial_write


##### CAMERA SETTINGS ######
focalValue = 40 # focus
cameraResolution = (1080, 1000)
cameraZoom = (0.3,0,0.3,0.8)
cameraFrameRate = 1

# create an array for storing filtered diameter values
filtered_dia_meas = []

# create moving average filter
ma5_filter = MovingAverageFilter(2)

class ImageProcessor(threading.Thread):
    def __init__(self, owner):
        super(ImageProcessor, self).__init__()
        self.stream = io.BytesIO()
        self.event = threading.Event()
        self.terminated = False
        self.owner = owner
        self.start()

    def run(self):
        # This method runs in a separate thread
        while not self.terminated:
            # Wait for an image to be written to the stream
            if self.event.wait(1):
                try:
                    self.stream.seek(0)
                    # Read the image and do some processing on it
                    img = np.array(Image.open(self.stream))
                    try:
                        diameter = diameter_detection(img)
                    except:
                        serial_write(0)
                        print('Could not read diameter, pausing and retrying...')
                        time.sleep(0.1)
                        
                    # add the diameter to the filter
                    ma5_filter.step(diameter)


                    #filtered_dia_meas.append(ma5_filter.current_state())

                    # display the current filtered diameter to the Terminal
                    print(ma5_filter.current_state())
                    
                    try:
                        # attempt to send the diameter to the connected serial device
                        serial_write(ma5_filter.current_state())
                    except:
                        print('Serial write failed!')
                            
                    # Set done to True if you want the script to terminate
                    # at some point
                    #self.owner.done=True
                finally:
                    # Reset the stream and event
                    self.stream.seek(0)
                    self.stream.truncate()
                    self.event.clear()
                    # Return ourselves to the available pool
                    with self.owner.lock:
                        self.owner.pool.append(self)

class ProcessOutput(object):
    def __init__(self):
        self.done = False
        # Construct a pool of 10 image processors along with a lock
        # to control access between threads
        self.lock = threading.Lock()
        self.pool = [ImageProcessor(self) for i in range(10)]
        print('Threaded processes created')
        self.processor = None

    def write(self, buf):
        if buf.startswith(b'\xff\xd8'):
            # New frame; set the current processor going and grab
            # a spare one
            if self.processor:
                self.processor.event.set()
            with self.lock:
                if self.pool:
                    self.processor = self.pool.pop()
                else:
                    # No processor's available, we'll have to skip this frame
                    print('Frame Skipped!')
                    self.processor = None
        if self.processor:
            self.processor.stream.write(buf)

    def flush(self):
        # When told to flush (end of recording), shut
        # down in an orderly fashion. First, add the current processor
        # back to the pool
        if self.processor:
            with self.lock:
                self.pool.append(self.processor)
                self.processor = None
        # Now, empty the pool, joining each thread as we go
        while True:
            with self.lock:
                try:
                    proc = self.pool.pop()
                except IndexError:
                    pass # pool is empty
            proc.terminated = True
            proc.join()

with picamera.PiCamera(resolution=cameraResolution) as camera:
    print('Succesfully created camera object')
    camera.framerate = cameraFrameRate
    camera.zoom = (0.3,0,0.3,0.8)
    
    # set focus motor
    os.system("i2cset -y 0 0x0c %d %d" % (focalValue,0))
    print('Camera focus set')
    time.sleep(2)
    print('Starting recording...')
    output = ProcessOutput()
    camera.start_recording(output, format='mjpeg')
    while not output.done:
        camera.wait_recording(1)
    camera.stop_recording()