From ec4d0489050f44e3156abd9b6d74e695f6278cda Mon Sep 17 00:00:00 2001
From: Blake Blackshear <blakeb@blakeshome.com>
Date: Mon, 21 Sep 2020 21:02:00 -0500
Subject: [PATCH] upgrade to python3.8 and switch from plasma store to
 shared_memory

---
 Dockerfile                   | 25 +++++-----
 detect_objects.py            | 47 ++++++------------
 frigate/edgetpu.py           | 69 ++++++++++++++------------
 frigate/object_processing.py | 11 ++---
 frigate/util.py              | 94 +++++++++++++++---------------------
 frigate/video.py             | 28 +++++------
 process_clip.py              |  8 ++-
 7 files changed, 128 insertions(+), 154 deletions(-)

diff --git a/Dockerfile b/Dockerfile
index f261fcb45..5fd62ec48 100755
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM ubuntu:18.04
+FROM ubuntu:20.04
 LABEL maintainer "blakeb@blakeshome.com"
 
 ENV DEBIAN_FRONTEND=noninteractive
@@ -11,27 +11,26 @@ RUN apt -qq update && apt -qq install --no-install-recommends -y \
     # libcap-dev \
     && add-apt-repository ppa:deadsnakes/ppa -y \
     && apt -qq install --no-install-recommends -y \
-        python3.7 \
-        python3.7-dev \
+        python3.8 \
+        python3.8-dev \
         python3-pip \
         ffmpeg \
         # VAAPI drivers for Intel hardware accel
         libva-drm2 libva2 i965-va-driver vainfo \
-    && python3.7 -m pip install -U pip \
-    && python3.7 -m pip install -U wheel setuptools \
-    && python3.7 -m pip install -U \
+    && python3.8 -m pip install -U pip \
+    && python3.8 -m pip install -U wheel setuptools \
+    && python3.8 -m pip install -U \
         opencv-python-headless \
         # python-prctl \
         numpy \
         imutils \
         scipy \
         psutil \
-    && python3.7 -m pip install -U \
+    && python3.8 -m pip install -U \
         Flask \
         paho-mqtt \
         PyYAML \
         matplotlib \
-        pyarrow \
         click \
     && echo "deb https://packages.cloud.google.com/apt coral-edgetpu-stable main" > /etc/apt/sources.list.d/coral-edgetpu.list \
     && wget -q -O - https://packages.cloud.google.com/apt/doc/apt-key.gpg | apt-key add - \
@@ -39,10 +38,10 @@ RUN apt -qq update && apt -qq install --no-install-recommends -y \
     && echo "libedgetpu1-max libedgetpu/accepted-eula boolean true" | debconf-set-selections \
     && apt -qq install --no-install-recommends -y \
         libedgetpu1-max \
-    ## Tensorflow lite (python 3.7 only)
-    && wget -q https://dl.google.com/coral/python/tflite_runtime-2.1.0.post1-cp37-cp37m-linux_x86_64.whl \
-    && python3.7 -m pip install tflite_runtime-2.1.0.post1-cp37-cp37m-linux_x86_64.whl \
-    && rm tflite_runtime-2.1.0.post1-cp37-cp37m-linux_x86_64.whl \
+    ## Tensorflow lite
+    && wget -q https://dl.google.com/coral/python/tflite_runtime-2.1.0.post1-cp38-cp38-linux_x86_64.whl \
+    && python3.8 -m pip install tflite_runtime-2.1.0.post1-cp38-cp38-linux_x86_64.whl \
+    && rm tflite_runtime-2.1.0.post1-cp38-cp38-linux_x86_64.whl \
     && rm -rf /var/lib/apt/lists/* \
     && (apt-get autoremove -y; apt-get autoclean -y)
 
@@ -60,4 +59,4 @@ COPY detect_objects.py .
 COPY benchmark.py .
 COPY process_clip.py .
 
-CMD ["python3.7", "-u", "detect_objects.py"]
+CMD ["python3.8", "-u", "detect_objects.py"]
diff --git a/detect_objects.py b/detect_objects.py
index c069d5c21..ce391b086 100644
--- a/detect_objects.py
+++ b/detect_objects.py
@@ -63,23 +63,13 @@ WEB_PORT = CONFIG.get('web_port', 5000)
 DEBUG = (CONFIG.get('debug', '0') == '1')
 TENSORFLOW_DEVICE = CONFIG.get('tensorflow_device')
 
-def start_plasma_store():
-    plasma_cmd = ['plasma_store', '-m', '400000000', '-s', '/tmp/plasma']
-    plasma_process = sp.Popen(plasma_cmd, stdout=sp.DEVNULL, stderr=sp.DEVNULL)
-    time.sleep(1)
-    rc = plasma_process.poll()
-    if rc is not None:
-        return None
-    return plasma_process
-
 class CameraWatchdog(threading.Thread):
-    def __init__(self, camera_processes, config, tflite_process, tracked_objects_queue, plasma_process, stop_event):
+    def __init__(self, camera_processes, config, tflite_process, tracked_objects_queue, stop_event):
         threading.Thread.__init__(self)
         self.camera_processes = camera_processes
         self.config = config
         self.tflite_process = tflite_process
         self.tracked_objects_queue = tracked_objects_queue
-        self.plasma_process = plasma_process
         self.stop_event = stop_event
 
     def run(self):
@@ -93,12 +83,6 @@ class CameraWatchdog(threading.Thread):
                 break
 
             now = datetime.datetime.now().timestamp()
-            
-            # check the plasma process
-            rc = self.plasma_process.poll()
-            if rc != None:
-                print(f"plasma_process exited unexpectedly with {rc}")
-                self.plasma_process = start_plasma_store()
 
             # check the detection process
             detection_start = self.tflite_process.detection_start.value
@@ -172,8 +156,6 @@ def main():
     client.connect(MQTT_HOST, MQTT_PORT, 60)
     client.loop_start()
 
-    plasma_process = start_plasma_store()
-
     ##
     # Setup config defaults for cameras
     ##
@@ -189,11 +171,16 @@ def main():
 
     # Queue for clip processing
     event_queue = mp.Queue()
+
+    # create the detection pipes
+    detection_pipes = {}
+    for name in CONFIG['cameras'].keys():
+        detection_pipes[name] = mp.Pipe(duplex=False)
     
     # Start the shared tflite process
-    tflite_process = EdgeTPUProcess(TENSORFLOW_DEVICE)
+    tflite_process = EdgeTPUProcess(result_connections={ key:value[1] for (key,value) in detection_pipes.items() }, tf_device=TENSORFLOW_DEVICE)
 
-    # start the camera processes
+    # create the camera processes
     camera_processes = {}
     for name, config in CONFIG['cameras'].items():
         # Merge the ffmpeg config with the global config
@@ -236,6 +223,8 @@ def main():
             frame_shape = (config['height'], config['width'], 3)
         else:
             frame_shape = get_frame_shape(ffmpeg_input)
+        
+        config['frame_shape'] = frame_shape
 
         frame_size = frame_shape[0] * frame_shape[1] * frame_shape[2]
         take_frame = config.get('take_frame', 1)
@@ -275,12 +264,13 @@ def main():
         }
 
         camera_process = mp.Process(target=track_camera, args=(name, config, frame_queue, frame_shape,
-            tflite_process.detection_queue, tracked_objects_queue, camera_processes[name]['process_fps'], 
+            tflite_process.detection_queue, detection_pipes[name][0], tracked_objects_queue, camera_processes[name]['process_fps'], 
             camera_processes[name]['detection_fps'], 
             camera_processes[name]['read_start'], camera_processes[name]['detection_frame'], stop_event))
         camera_process.daemon = True
         camera_processes[name]['process'] = camera_process
 
+    # start the camera_processes
     for name, camera_process in camera_processes.items():
         camera_process['process'].start()
         print(f"Camera_process started for {name}: {camera_process['process'].pid}")
@@ -291,7 +281,7 @@ def main():
     object_processor = TrackedObjectProcessor(CONFIG['cameras'], client, MQTT_TOPIC_PREFIX, tracked_objects_queue, event_queue, stop_event)
     object_processor.start()
     
-    camera_watchdog = CameraWatchdog(camera_processes, CONFIG['cameras'], tflite_process, tracked_objects_queue, plasma_process, stop_event)
+    camera_watchdog = CameraWatchdog(camera_processes, CONFIG['cameras'], tflite_process, tracked_objects_queue, stop_event)
     camera_watchdog.start()
 
     def receiveSignal(signalNumber, frame):
@@ -300,11 +290,9 @@ def main():
         event_processor.join()
         object_processor.join()
         camera_watchdog.join()
-        for name, camera_process in camera_processes.items():
+        for camera_process in camera_processes.values():
             camera_process['capture_thread'].join()
-        rc = camera_watchdog.plasma_process.poll()
-        if rc == None:
-            camera_watchdog.plasma_process.terminate()
+        tflite_process.stop()
         sys.exit()
     
     signal.signal(signal.SIGTERM, receiveSignal)
@@ -368,9 +356,6 @@ def main():
             'pid': tflite_process.detect_process.pid
         }
 
-        rc = camera_watchdog.plasma_process.poll()
-        stats['plasma_store_rc'] = rc
-
         return jsonify(stats)
 
     @app.route('/<camera_name>/<label>/best.jpg')
@@ -448,8 +433,6 @@ def main():
     app.run(host='0.0.0.0', port=WEB_PORT, debug=False)
 
     object_processor.join()
-    
-    plasma_process.terminate()
 
 if __name__ == '__main__':
     main()
diff --git a/frigate/edgetpu.py b/frigate/edgetpu.py
index 0fd7b4323..52ef4be80 100644
--- a/frigate/edgetpu.py
+++ b/frigate/edgetpu.py
@@ -2,12 +2,14 @@ import os
 import datetime
 import hashlib
 import multiprocessing as mp
+import queue
+from multiprocessing.connection import Connection
 from abc import ABC, abstractmethod
+from typing import Dict
 import numpy as np
-import pyarrow.plasma as plasma
 import tflite_runtime.interpreter as tflite
 from tflite_runtime.interpreter import load_delegate
-from frigate.util import EventsPerSecond, listen
+from frigate.util import EventsPerSecond, listen, SharedMemoryFrameManager
 
 def load_labels(path, encoding='utf-8'):
   """Loads labels from file (with or without index numbers).
@@ -100,73 +102,77 @@ class LocalObjectDetector(ObjectDetector):
         
         return detections
 
-def run_detector(detection_queue, avg_speed, start, tf_device):
+def run_detector(detection_queue, result_connections: Dict[str, Connection], avg_speed, start, tf_device):
     print(f"Starting detection process: {os.getpid()}")
     listen()
-    plasma_client = plasma.connect("/tmp/plasma")
+    frame_manager = SharedMemoryFrameManager()
     object_detector = LocalObjectDetector(tf_device=tf_device)
 
     while True:
-        object_id_str = detection_queue.get()
-        object_id_hash = hashlib.sha1(str.encode(object_id_str))
-        object_id = plasma.ObjectID(object_id_hash.digest())
-        object_id_out = plasma.ObjectID(hashlib.sha1(str.encode(f"out-{object_id_str}")).digest())
-        input_frame = plasma_client.get(object_id, timeout_ms=0)
+        connection_id = detection_queue.get()
+        input_frame = frame_manager.get(connection_id, (1,300,300,3))
 
-        if input_frame is plasma.ObjectNotAvailable:
+        if input_frame is None:
             continue
 
         # detect and put the output in the plasma store
         start.value = datetime.datetime.now().timestamp()
-        plasma_client.put(object_detector.detect_raw(input_frame), object_id_out)
+        # TODO: what is the overhead for pickling this result vs writing back to shared memory?
+        #       I could try using an Event() and waiting in the other process before looking in memory...
+        detections = object_detector.detect_raw(input_frame)
+        result_connections[connection_id].send(detections)
         duration = datetime.datetime.now().timestamp()-start.value
         start.value = 0.0
 
         avg_speed.value = (avg_speed.value*9 + duration)/10
         
 class EdgeTPUProcess():
-    def __init__(self, tf_device=None):
+    def __init__(self, result_connections, tf_device=None):
+        self.result_connections = result_connections
         self.detection_queue = mp.Queue()
         self.avg_inference_speed = mp.Value('d', 0.01)
         self.detection_start = mp.Value('d', 0.0)
         self.detect_process = None
         self.tf_device = tf_device
         self.start_or_restart()
+    
+    def stop(self):
+        self.detect_process.terminate()
+        print("Waiting for detection process to exit gracefully...")
+        self.detect_process.join(timeout=30)
+        if self.detect_process.exitcode is None:
+            print("Detection process didnt exit. Force killing...")
+            self.detect_process.kill()
+            self.detect_process.join()
 
     def start_or_restart(self):
         self.detection_start.value = 0.0
         if (not self.detect_process is None) and self.detect_process.is_alive():
-            self.detect_process.terminate()
-            print("Waiting for detection process to exit gracefully...")
-            self.detect_process.join(timeout=30)
-            if self.detect_process.exitcode is None:
-                print("Detection process didnt exit. Force killing...")
-                self.detect_process.kill()
-                self.detect_process.join()
-        self.detect_process = mp.Process(target=run_detector, args=(self.detection_queue, self.avg_inference_speed, self.detection_start, self.tf_device))
+            self.stop()
+        self.detect_process = mp.Process(target=run_detector, args=(self.detection_queue, self.result_connections, self.avg_inference_speed, self.detection_start, self.tf_device))
         self.detect_process.daemon = True
         self.detect_process.start()
 
 class RemoteObjectDetector():
-    def __init__(self, name, labels, detection_queue):
+    def __init__(self, name, labels, detection_queue, result_connection: Connection):
         self.labels = load_labels(labels)
         self.name = name
         self.fps = EventsPerSecond()
-        self.plasma_client = plasma.connect("/tmp/plasma")
         self.detection_queue = detection_queue
+        self.result_connection = result_connection
+        self.shm = mp.shared_memory.SharedMemory(name=self.name, create=True, size=300*300*3)
+        self.np_shm = np.ndarray((1,300,300,3), dtype=np.uint8, buffer=self.shm.buf)
     
     def detect(self, tensor_input, threshold=.4):
         detections = []
 
-        now = f"{self.name}-{str(datetime.datetime.now().timestamp())}"
-        object_id_frame = plasma.ObjectID(hashlib.sha1(str.encode(now)).digest())
-        object_id_detections = plasma.ObjectID(hashlib.sha1(str.encode(f"out-{now}")).digest())
-        self.plasma_client.put(tensor_input, object_id_frame)
-        self.detection_queue.put(now)
-        raw_detections = self.plasma_client.get(object_id_detections, timeout_ms=10000)
-
-        if raw_detections is plasma.ObjectNotAvailable:
-            self.plasma_client.delete([object_id_frame])
+        # copy input to shared memory
+        # TODO: what if I just write it there in the first place?
+        self.np_shm[:] = tensor_input[:]
+        self.detection_queue.put(self.name)
+        if self.result_connection.poll(10):
+            raw_detections = self.result_connection.recv()
+        else:
             return detections
 
         for d in raw_detections:
@@ -177,6 +183,5 @@ class RemoteObjectDetector():
                 float(d[1]),
                 (d[2], d[3], d[4], d[5])
             ))
-        self.plasma_client.delete([object_id_frame, object_id_detections])
         self.fps.update()
         return detections
diff --git a/frigate/object_processing.py b/frigate/object_processing.py
index 5cf3c8cad..6cf9ee4ef 100644
--- a/frigate/object_processing.py
+++ b/frigate/object_processing.py
@@ -10,9 +10,8 @@ import copy
 import numpy as np
 from collections import Counter, defaultdict
 import itertools
-import pyarrow.plasma as plasma
 import matplotlib.pyplot as plt
-from frigate.util import draw_box_with_label, PlasmaFrameManager
+from frigate.util import draw_box_with_label, SharedMemoryFrameManager
 from frigate.edgetpu import load_labels
 from typing import Callable, Dict
 from statistics import mean, median
@@ -59,7 +58,7 @@ class CameraState():
         self.object_status = defaultdict(lambda: 'OFF')
         self.tracked_objects = {}
         self.zone_objects = defaultdict(lambda: [])
-        self.current_frame = np.zeros((720,1280,3), np.uint8)
+        self.current_frame = np.zeros(self.config['frame_shape'], np.uint8)
         self.current_frame_time = 0.0
         self.previous_frame_id = None
         self.callbacks = defaultdict(lambda: [])
@@ -88,7 +87,7 @@ class CameraState():
         self.current_frame_time = frame_time
         # get the new frame and delete the old frame
         frame_id = f"{self.name}{frame_time}"
-        self.current_frame = self.frame_manager.get(frame_id)
+        self.current_frame = self.frame_manager.get(frame_id, self.config['frame_shape'])
         if not self.previous_frame_id is None:
             self.frame_manager.delete(self.previous_frame_id)
         self.previous_frame_id = frame_id
@@ -238,7 +237,7 @@ class TrackedObjectProcessor(threading.Thread):
         self.event_queue = event_queue
         self.stop_event = stop_event
         self.camera_states: Dict[str, CameraState] = {}
-        self.plasma_client = PlasmaFrameManager(self.stop_event)
+        self.frame_manager = SharedMemoryFrameManager()
 
         def start(camera, obj):
             # publish events to mqtt
@@ -273,7 +272,7 @@ class TrackedObjectProcessor(threading.Thread):
             self.client.publish(f"{self.topic_prefix}/{camera}/{object_name}", status, retain=False)
 
         for camera in self.camera_config.keys():
-            camera_state = CameraState(camera, self.camera_config[camera], self.plasma_client)
+            camera_state = CameraState(camera, self.camera_config[camera], self.frame_manager)
             camera_state.on('start', start)
             camera_state.on('update', update)
             camera_state.on('end', end)
diff --git a/frigate/util.py b/frigate/util.py
index f42ee82c0..fc111b422 100755
--- a/frigate/util.py
+++ b/frigate/util.py
@@ -9,7 +9,8 @@ import cv2
 import threading
 import matplotlib.pyplot as plt
 import hashlib
-import pyarrow.plasma as plasma
+from multiprocessing import shared_memory
+from typing import AnyStr
 
 def draw_box_with_label(frame, x_min, y_min, x_max, y_max, label, info, thickness=2, color=None, position='ul'):
     if color is None:
@@ -148,12 +149,16 @@ def listen():
     signal.signal(signal.SIGUSR1, print_stack)
 
 class FrameManager(ABC):
+    @abstractmethod
+    def create(self, name, size) -> AnyStr:
+        pass
+
     @abstractmethod
     def get(self, name, timeout_ms=0):
         pass
 
     @abstractmethod
-    def put(self, name, frame):
+    def close(self, name):
         pass
 
     @abstractmethod
@@ -164,66 +169,45 @@ class DictFrameManager(FrameManager):
     def __init__(self):
         self.frames = {}
     
-    def get(self, name, timeout_ms=0):
-        return self.frames.get(name)
+    def create(self, name, size) -> AnyStr:
+        mem = bytearray(size)
+        self.frames[name] = mem
+        return mem
     
-    def put(self, name, frame):
-        self.frames[name] = frame
+    def get(self, name, shape):
+        mem = self.frames[name]
+        return np.ndarray(shape, dtype=np.uint8, buffer=mem)
+    
+    def close(self, name):
+        pass
     
     def delete(self, name):
         del self.frames[name]
 
-class PlasmaFrameManager(FrameManager):
-    def __init__(self, stop_event=None):
-        self.stop_event = stop_event
-        self.connect()
+class SharedMemoryFrameManager(FrameManager):
+    def __init__(self):
+        self.shm_store = {}
     
-    def connect(self):
-        while True:
-            if self.stop_event != None and self.stop_event.is_set():
-                return
-            try:
-                self.plasma_client = plasma.connect("/tmp/plasma")
-                return
-            except:
-                print(f"TrackedObjectProcessor: unable to connect plasma client")
-                time.sleep(10)
+    def create(self, name, size) -> AnyStr:
+        shm = shared_memory.SharedMemory(name=name, create=True, size=size)
+        self.shm_store[name] = shm
+        return shm.buf
 
-    def get(self, name, timeout_ms=0):
-        object_id = plasma.ObjectID(hashlib.sha1(str.encode(name)).digest())
-        while True:
-            if self.stop_event != None and self.stop_event.is_set():
-                return
-            try:
-                frame = self.plasma_client.get(object_id, timeout_ms=timeout_ms)
-                if frame is plasma.ObjectNotAvailable:
-                    return None
-                return frame
-            except:
-                self.connect()
-                time.sleep(1)
+    def get(self, name, shape):
+        if name in self.shm_store:
+            shm = self.shm_store[name]
+        else:
+            shm = shared_memory.SharedMemory(name=name)
+            self.shm_store[name] = shm
+        return np.ndarray(shape, dtype=np.uint8, buffer=shm.buf)
 
-    def put(self, name, frame):
-        object_id = plasma.ObjectID(hashlib.sha1(str.encode(name)).digest())
-        while True:
-            if self.stop_event != None and self.stop_event.is_set():
-                return
-            try:
-                self.plasma_client.put(frame, object_id)
-                return
-            except Exception as e:
-                print(f"Failed to put in plasma: {e}")
-                self.connect()
-                time.sleep(1)
+    def close(self, name):
+        if name in self.shm_store:
+            self.shm_store[name].close()
+            del self.shm_store[name]
 
     def delete(self, name):
-        object_id = plasma.ObjectID(hashlib.sha1(str.encode(name)).digest())
-        while True:
-            if self.stop_event != None and self.stop_event.is_set():
-                return
-            try:
-                self.plasma_client.delete([object_id])
-                return
-            except:
-                self.connect()
-                time.sleep(1)
\ No newline at end of file
+        if name in self.shm_store:
+            self.shm_store[name].close()
+            self.shm_store[name].unlink()
+            del self.shm_store[name]
\ No newline at end of file
diff --git a/frigate/video.py b/frigate/video.py
index aa816dbd9..2e51fbbd6 100755
--- a/frigate/video.py
+++ b/frigate/video.py
@@ -5,7 +5,6 @@ import cv2
 import queue
 import threading
 import ctypes
-import pyarrow.plasma as plasma
 import multiprocessing as mp
 import subprocess as sp
 import numpy as np
@@ -15,7 +14,7 @@ import json
 import base64
 from typing import Dict, List
 from collections import defaultdict
-from frigate.util import draw_box_with_label, area, calculate_region, clipped, intersection_over_union, intersection, EventsPerSecond, listen, FrameManager, PlasmaFrameManager
+from frigate.util import draw_box_with_label, area, calculate_region, clipped, intersection_over_union, intersection, EventsPerSecond, listen, FrameManager, SharedMemoryFrameManager
 from frigate.objects import ObjectTracker
 from frigate.edgetpu import RemoteObjectDetector
 from frigate.motion import MotionDetector
@@ -154,11 +153,10 @@ def capture_frames(ffmpeg_process, camera_name, frame_shape, frame_manager: Fram
             continue
 
         # put the frame in the frame manager
-        frame_manager.put(f"{camera_name}{current_frame.value}",
-                np
-                    .frombuffer(frame_bytes, np.uint8)
-                    .reshape(frame_shape)
-            )
+        frame_buffer = frame_manager.create(f"{camera_name}{current_frame.value}", frame_size)
+        frame_buffer[:] = frame_bytes[:]
+        frame_manager.close(f"{camera_name}{current_frame.value}")
+
         # add to the queue
         frame_queue.put(current_frame.value)
         last_frame = current_frame.value
@@ -173,7 +171,7 @@ class CameraCapture(threading.Thread):
         self.take_frame = take_frame
         self.fps = fps
         self.skipped_fps = EventsPerSecond()
-        self.plasma_client = PlasmaFrameManager(stop_event)
+        self.frame_manager = SharedMemoryFrameManager()
         self.ffmpeg_process = ffmpeg_process
         self.current_frame = mp.Value('d', 0.0)
         self.last_frame = 0
@@ -182,10 +180,10 @@ class CameraCapture(threading.Thread):
 
     def run(self):
         self.skipped_fps.start()
-        capture_frames(self.ffmpeg_process, self.name, self.frame_shape, self.plasma_client, self.frame_queue, self.take_frame,
+        capture_frames(self.ffmpeg_process, self.name, self.frame_shape, self.frame_manager, self.frame_queue, self.take_frame,
             self.fps, self.skipped_fps, self.stop_event, self.detection_frame, self.current_frame)
 
-def track_camera(name, config, frame_queue, frame_shape, detection_queue, detected_objects_queue, fps, detection_fps, read_start, detection_frame, stop_event):
+def track_camera(name, config, frame_queue, frame_shape, detection_queue, result_connection, detected_objects_queue, fps, detection_fps, read_start, detection_frame, stop_event):
     print(f"Starting process for {name}: {os.getpid()}")
     listen()
 
@@ -218,13 +216,13 @@ def track_camera(name, config, frame_queue, frame_shape, detection_queue, detect
         mask[:] = 255
 
     motion_detector = MotionDetector(frame_shape, mask, resize_factor=6)
-    object_detector = RemoteObjectDetector(name, '/labelmap.txt', detection_queue)
+    object_detector = RemoteObjectDetector(name, '/labelmap.txt', detection_queue, result_connection)
 
     object_tracker = ObjectTracker(10)
 
-    plasma_client = PlasmaFrameManager()
+    frame_manager = SharedMemoryFrameManager()
 
-    process_frames(name, frame_queue, frame_shape, plasma_client, motion_detector, object_detector,
+    process_frames(name, frame_queue, frame_shape, frame_manager, motion_detector, object_detector,
         object_tracker, detected_objects_queue, fps, detection_fps, detection_frame, objects_to_track, object_filters, mask, stop_event)
 
     print(f"{name}: exiting subprocess")
@@ -281,7 +279,7 @@ def process_frames(camera_name: str, frame_queue: mp.Queue, frame_shape,
         
         current_frame_time.value = frame_time
 
-        frame = frame_manager.get(f"{camera_name}{frame_time}")
+        frame = frame_manager.get(f"{camera_name}{frame_time}", frame_shape)
 
         if frame is None:
             print(f"{camera_name}: frame {frame_time} is not in memory store.")
@@ -364,3 +362,5 @@ def process_frames(camera_name: str, frame_queue: mp.Queue, frame_shape,
         detected_objects_queue.put((camera_name, frame_time, object_tracker.tracked_objects))
 
         detection_fps.value = object_detector.fps.eps()
+
+        frame_manager.close(f"{camera_name}{frame_time}")
diff --git a/process_clip.py b/process_clip.py
index f754e4687..4ac4a8673 100644
--- a/process_clip.py
+++ b/process_clip.py
@@ -4,7 +4,7 @@ import os
 import datetime
 from unittest import TestCase, main
 from frigate.video import process_frames, start_or_restart_ffmpeg, capture_frames, get_frame_shape
-from frigate.util import DictFrameManager, EventsPerSecond, draw_box_with_label
+from frigate.util import DictFrameManager, SharedMemoryFrameManager, EventsPerSecond, draw_box_with_label
 from frigate.motion import MotionDetector
 from frigate.edgetpu import LocalObjectDetector
 from frigate.objects import ObjectTracker
@@ -19,6 +19,7 @@ class ProcessClip():
         self.frame_shape = frame_shape
         self.camera_name = 'camera'
         self.frame_manager = DictFrameManager()
+        # self.frame_manager = SharedMemoryFrameManager()
         self.frame_queue = mp.Queue()
         self.detected_objects_queue = mp.Queue()
         self.camera_state = CameraState(self.camera_name, config, self.frame_manager)
@@ -72,13 +73,15 @@ class ProcessClip():
             for obj in self.camera_state.tracked_objects.values():
                 print(f"{frame_time}: {obj['id']} - {obj['computed_score']} - {obj['score_history']}")
         
+        self.frame_manager.delete(self.camera_state.previous_frame_id)
+        
         return {
             'object_detected': obj_detected,
             'top_score': top_computed_score
         }
     
     def save_debug_frame(self, debug_path, frame_time, tracked_objects):
-        current_frame = self.frame_manager.get(f"{self.camera_name}{frame_time}")
+        current_frame = self.frame_manager.get(f"{self.camera_name}{frame_time}", self.frame_shape)
         # draw the bounding boxes on the frame
         for obj in tracked_objects:
             thickness = 2
@@ -132,6 +135,7 @@ def process(path, label, threshold, debug_path):
     results = []
     for c in clips:
         frame_shape = get_frame_shape(c)
+        config['frame_shape'] = frame_shape
         process_clip = ProcessClip(c, frame_shape, config)
         process_clip.load_frames()
         process_clip.process_frames(objects_to_track=config['objects']['track'])