call the restart function and handle errors better in the detection process

Dockerfile

@@ -25,6 +25,7 @@ RUN apt -qq update && apt -qq install --no-install-recommends -y \
         imutils \
         scipy \
     && python3.7 -m pip install -U \
+        SharedArray \
         Flask \
         paho-mqtt \
         PyYAML \
@@ -37,9 +38,9 @@ RUN apt -qq update && apt -qq install --no-install-recommends -y \
     && 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 \
+    && wget -q https://dl.google.com/coral/python/tflite_runtime-2.1.0-cp37-cp37m-linux_x86_64.whl \
+    && python3.7 -m pip install tflite_runtime-2.1.0-cp37-cp37m-linux_x86_64.whl \
+    && rm tflite_runtime-2.1.0-cp37-cp37m-linux_x86_64.whl \
     && rm -rf /var/lib/apt/lists/* \
     && (apt-get autoremove -y; apt-get autoclean -y)
 

detect_objects.py

@@ -1,7 +1,4 @@
 import os
-import sys
-import traceback
-import signal
 import cv2
 import time
 import datetime
@@ -12,10 +9,10 @@ import multiprocessing as mp
 import subprocess as sp
 import numpy as np
 import logging
-from flask import Flask, Response, make_response, jsonify, request
+from flask import Flask, Response, make_response, jsonify
 import paho.mqtt.client as mqtt
 
-from frigate.video import track_camera, get_ffmpeg_input, get_frame_shape, CameraCapture, start_or_restart_ffmpeg
+from frigate.video import track_camera
 from frigate.object_processing import TrackedObjectProcessor
 from frigate.util import EventsPerSecond
 from frigate.edgetpu import EdgeTPUProcess
@@ -61,72 +58,41 @@ GLOBAL_OBJECT_CONFIG = CONFIG.get('objects', {})
 WEB_PORT = CONFIG.get('web_port', 5000)
 DEBUG = (CONFIG.get('debug', '0') == '1')
 
-def start_plasma_store():
-    plasma_cmd = ['plasma_store', '-m', '400000000', '-s', '/tmp/plasma']
-    plasma_process = sp.Popen(plasma_cmd, stdout=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):
+    def __init__(self, camera_processes, config, tflite_process, tracked_objects_queue, object_processor):
         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.object_processor = object_processor
 
     def run(self):
         time.sleep(10)
         while True:
             # wait a bit before checking
-            time.sleep(10)
-            
-            # 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()
+            time.sleep(30)
 
-            # check the detection process
             if (self.tflite_process.detection_start.value > 0.0 and 
                 datetime.datetime.now().timestamp() - self.tflite_process.detection_start.value > 10):
                 print("Detection appears to be stuck. Restarting detection process")
                 self.tflite_process.start_or_restart()
-            elif not self.tflite_process.detect_process.is_alive():
-                print("Detection appears to have stopped. Restarting detection process")
-                self.tflite_process.start_or_restart()
+                time.sleep(30)
 
-            # check the camera processes
             for name, camera_process in self.camera_processes.items():
                 process = camera_process['process']
                 if not process.is_alive():
-                    print(f"Track process for {name} is not alive. Starting again...")
+                    print(f"Process for {name} is not alive. Starting again...")
                     camera_process['fps'].value = float(self.config[name]['fps'])
                     camera_process['skipped_fps'].value = 0.0
                     camera_process['detection_fps'].value = 0.0
-                    camera_process['read_start'].value = 0.0
-                    process = mp.Process(target=track_camera, args=(name, self.config[name], GLOBAL_OBJECT_CONFIG, camera_process['frame_queue'],
-                        camera_process['frame_shape'], self.tflite_process.detection_queue, self.tracked_objects_queue, 
-                        camera_process['fps'], camera_process['skipped_fps'], camera_process['detection_fps'],
-                        camera_process['read_start']))
+                    process = mp.Process(target=track_camera, args=(name, self.config[name], FFMPEG_DEFAULT_CONFIG, GLOBAL_OBJECT_CONFIG, 
+                        self.tflite_process.detection_queue, self.tracked_objects_queue, 
+                        camera_process['fps'], camera_process['skipped_fps'], camera_process['detection_fps']))
                     process.daemon = True
                     camera_process['process'] = process
                     process.start()
-                    print(f"Track process started for {name}: {process.pid}")
-                
-                if not camera_process['capture_thread'].is_alive():
-                    frame_shape = camera_process['frame_shape']
-                    frame_size = frame_shape[0] * frame_shape[1] * frame_shape[2]
-                    ffmpeg_process = start_or_restart_ffmpeg(camera_process['ffmpeg_cmd'], frame_size)
-                    camera_capture = CameraCapture(name, ffmpeg_process, frame_shape, camera_process['frame_queue'], 
-                        camera_process['take_frame'], camera_process['camera_fps'])
-                    camera_capture.start()
-                    camera_process['ffmpeg_process'] = ffmpeg_process
-                    camera_process['capture_thread'] = camera_capture
+                    print(f"Camera_process started for {name}: {process.pid}")
 
 def main():
     # connect to mqtt and setup last will
@@ -151,7 +117,14 @@ def main():
     client.connect(MQTT_HOST, MQTT_PORT, 60)
     client.loop_start()
 
-    plasma_process = start_plasma_store()
+    # start plasma store
+    plasma_cmd = ['plasma_store', '-m', '400000000', '-s', '/tmp/plasma']
+    plasma_process = sp.Popen(plasma_cmd, stdout=sp.DEVNULL)
+    time.sleep(1)
+    rc = plasma_process.poll()
+    if rc is not None:
+        raise RuntimeError("plasma_store exited unexpectedly with "
+                            "code %d" % (rc,))
 
     ##
     # Setup config defaults for cameras
@@ -162,7 +135,7 @@ def main():
         }
 
     # Queue for cameras to push tracked objects to
-    tracked_objects_queue = mp.SimpleQueue()
+    tracked_objects_queue = mp.Queue()
     
     # Start the shared tflite process
     tflite_process = EdgeTPUProcess()
@@ -170,54 +143,14 @@ def main():
     # start the camera processes
     camera_processes = {}
     for name, config in CONFIG['cameras'].items():
-        # Merge the ffmpeg config with the global config
-        ffmpeg = config.get('ffmpeg', {})
-        ffmpeg_input = get_ffmpeg_input(ffmpeg['input'])
-        ffmpeg_global_args = ffmpeg.get('global_args', FFMPEG_DEFAULT_CONFIG['global_args'])
-        ffmpeg_hwaccel_args = ffmpeg.get('hwaccel_args', FFMPEG_DEFAULT_CONFIG['hwaccel_args'])
-        ffmpeg_input_args = ffmpeg.get('input_args', FFMPEG_DEFAULT_CONFIG['input_args'])
-        ffmpeg_output_args = ffmpeg.get('output_args', FFMPEG_DEFAULT_CONFIG['output_args'])
-        ffmpeg_cmd = (['ffmpeg'] +
-                ffmpeg_global_args +
-                ffmpeg_hwaccel_args +
-                ffmpeg_input_args +
-                ['-i', ffmpeg_input] +
-                ffmpeg_output_args +
-                ['pipe:'])
-        
-        if 'width' in config and 'height' in config:
-            frame_shape = (config['height'], config['width'], 3)
-        else:
-            frame_shape = get_frame_shape(ffmpeg_input)
-
-        frame_size = frame_shape[0] * frame_shape[1] * frame_shape[2]
-        take_frame = config.get('take_frame', 1)
-
-        ffmpeg_process = start_or_restart_ffmpeg(ffmpeg_cmd, frame_size)
-        frame_queue = mp.SimpleQueue()
-        camera_fps = EventsPerSecond()
-        camera_fps.start()
-        camera_capture = CameraCapture(name, ffmpeg_process, frame_shape, frame_queue, take_frame, camera_fps)
-        camera_capture.start()
-
         camera_processes[name] = {
-            'camera_fps': camera_fps,
-            'take_frame': take_frame,
             'fps': mp.Value('d', float(config['fps'])),
             'skipped_fps': mp.Value('d', 0.0),
-            'detection_fps': mp.Value('d', 0.0),
-            'read_start': mp.Value('d', 0.0),
-            'ffmpeg_process': ffmpeg_process,
-            'ffmpeg_cmd': ffmpeg_cmd,
-            'frame_queue': frame_queue,
-            'frame_shape': frame_shape,
-            'capture_thread': camera_capture
+            'detection_fps': mp.Value('d', 0.0)
         }
-
-        camera_process = mp.Process(target=track_camera, args=(name, config, GLOBAL_OBJECT_CONFIG, frame_queue, frame_shape,
-            tflite_process.detection_queue, tracked_objects_queue, camera_processes[name]['fps'], 
-            camera_processes[name]['skipped_fps'], camera_processes[name]['detection_fps'], 
-            camera_processes[name]['read_start']))
+        camera_process = mp.Process(target=track_camera, args=(name, config, FFMPEG_DEFAULT_CONFIG, GLOBAL_OBJECT_CONFIG, 
+            tflite_process.detection_queue, tracked_objects_queue, 
+            camera_processes[name]['fps'], camera_processes[name]['skipped_fps'], camera_processes[name]['detection_fps']))
         camera_process.daemon = True
         camera_processes[name]['process'] = camera_process
 
@@ -228,7 +161,7 @@ def main():
     object_processor = TrackedObjectProcessor(CONFIG['cameras'], client, MQTT_TOPIC_PREFIX, tracked_objects_queue)
     object_processor.start()
     
-    camera_watchdog = CameraWatchdog(camera_processes, CONFIG['cameras'], tflite_process, tracked_objects_queue, plasma_process)
+    camera_watchdog = CameraWatchdog(camera_processes, CONFIG['cameras'], tflite_process, tracked_objects_queue, object_processor)
     camera_watchdog.start()
 
     # create a flask app that encodes frames a mjpeg on demand
@@ -241,23 +174,6 @@ def main():
         # return a healh
         return "Frigate is running. Alive and healthy!"
 
-    @app.route('/debug/stack')
-    def processor_stack():
-        frame = sys._current_frames().get(object_processor.ident, None)
-        if frame:
-            return "<br>".join(traceback.format_stack(frame)), 200
-        else:
-            return "no frame found", 200
-
-    @app.route('/debug/print_stack')
-    def print_stack():
-        pid = int(request.args.get('pid', 0))
-        if pid == 0:
-            return "missing pid", 200
-        else:
-            os.kill(pid, signal.SIGUSR1)
-            return "check logs", 200
-
     @app.route('/debug/stats')
     def stats():
         stats = {}
@@ -269,22 +185,21 @@ def main():
             stats[name] = {
                 'fps': round(camera_stats['fps'].value, 2),
                 'skipped_fps': round(camera_stats['skipped_fps'].value, 2),
-                'detection_fps': round(camera_stats['detection_fps'].value, 2),
-                'read_start': camera_stats['read_start'].value,
-                'pid': camera_stats['process'].pid,
-                'ffmpeg_pid': camera_stats['ffmpeg_process'].pid
+                'detection_fps': round(camera_stats['detection_fps'].value, 2)
             }
         
         stats['coral'] = {
             'fps': round(total_detection_fps, 2),
             'inference_speed': round(tflite_process.avg_inference_speed.value*1000, 2),
-            'detection_start': tflite_process.detection_start.value,
-            'pid': tflite_process.detect_process.pid
+            'detection_queue': tflite_process.detection_queue.qsize(),
+            'detection_start': tflite_process.detection_start.value
         }
 
-        rc = camera_watchdog.plasma_process.poll()
+        rc = plasma_process.poll()
         stats['plasma_store_rc'] = rc
 
+        stats['tracked_objects_queue'] = tracked_objects_queue.qsize()
+
         return jsonify(stats)
 
     @app.route('/<camera_name>/<label>/best.jpg')
@@ -303,33 +218,28 @@ def main():
 
     @app.route('/<camera_name>')
     def mjpeg_feed(camera_name):
-        fps = int(request.args.get('fps', '3'))
-        height = int(request.args.get('h', '360'))
         if camera_name in CONFIG['cameras']:
             # return a multipart response
-            return Response(imagestream(camera_name, fps, height),
+            return Response(imagestream(camera_name),
                             mimetype='multipart/x-mixed-replace; boundary=frame')
         else:
             return "Camera named {} not found".format(camera_name), 404
 
-    def imagestream(camera_name, fps, height):
+    def imagestream(camera_name):
         while True:
-            # max out at specified FPS
-            time.sleep(1/fps)
+            # max out at 1 FPS
+            time.sleep(1)
             frame = object_processor.get_current_frame(camera_name)
             if frame is None:
-                frame = np.zeros((height,int(height*16/9),3), np.uint8)
-
-            frame = cv2.resize(frame, dsize=(int(height*16/9), height), interpolation=cv2.INTER_LINEAR)
+                frame = np.zeros((720,1280,3), np.uint8)
             frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
-
             ret, jpg = cv2.imencode('.jpg', frame)
             yield (b'--frame\r\n'
                 b'Content-Type: image/jpeg\r\n\r\n' + jpg.tobytes() + b'\r\n\r\n')
 
     app.run(host='0.0.0.0', port=WEB_PORT, debug=False)
 
-    object_processor.join()
+    camera_watchdog.join()
     
     plasma_process.terminate()
 

frigate/edgetpu.py

@@ -3,10 +3,11 @@ import datetime
 import hashlib
 import multiprocessing as mp
 import numpy as np
+import SharedArray as sa
 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
 
 def load_labels(path, encoding='utf-8'):
   """Loads labels from file (with or without index numbers).
@@ -63,7 +64,6 @@ class ObjectDetector():
 
 def run_detector(detection_queue, avg_speed, start):
     print(f"Starting detection process: {os.getpid()}")
-    listen()
     plasma_client = plasma.connect("/tmp/plasma")
     object_detector = ObjectDetector()
 
@@ -75,6 +75,7 @@ def run_detector(detection_queue, avg_speed, start):
         input_frame = plasma_client.get(object_id, timeout_ms=0)
 
         if input_frame is plasma.ObjectNotAvailable:
+            plasma_client.put(np.zeros((20,6), np.float32), object_id_out)
             continue
 
         # detect and put the output in the plasma store
@@ -87,7 +88,7 @@ def run_detector(detection_queue, avg_speed, start):
         
 class EdgeTPUProcess():
     def __init__(self):
-        self.detection_queue = mp.SimpleQueue()
+        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
@@ -123,11 +124,7 @@ class RemoteObjectDetector():
         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])
-            return detections
+        raw_detections = self.plasma_client.get(object_id_detections)
 
         for d in raw_detections:
             if d[1] < threshold:

frigate/object_processing.py

@@ -1,7 +1,6 @@
 import json
 import hashlib
 import datetime
-import time
 import copy
 import cv2
 import threading
@@ -9,8 +8,9 @@ import numpy as np
 from collections import Counter, defaultdict
 import itertools
 import pyarrow.plasma as plasma
+import SharedArray as sa
 import matplotlib.pyplot as plt
-from frigate.util import draw_box_with_label, PlasmaManager
+from frigate.util import draw_box_with_label
 from frigate.edgetpu import load_labels
 
 PATH_TO_LABELS = '/labelmap.txt'
@@ -29,6 +29,7 @@ class TrackedObjectProcessor(threading.Thread):
         self.client = client
         self.topic_prefix = topic_prefix
         self.tracked_objects_queue = tracked_objects_queue
+        self.plasma_client = plasma.connect("/tmp/plasma")
         self.camera_data = defaultdict(lambda: {
             'best_objects': {},
             'object_status': defaultdict(lambda: defaultdict(lambda: 'OFF')),
@@ -36,7 +37,6 @@ class TrackedObjectProcessor(threading.Thread):
             'current_frame': np.zeros((720,1280,3), np.uint8),
             'object_id': None
         })
-        self.plasma_client = PlasmaManager()
         
     def get_best(self, camera, label):
         if label in self.camera_data[camera]['best_objects']:
@@ -59,7 +59,10 @@ class TrackedObjectProcessor(threading.Thread):
             ###
             # Draw tracked objects on the frame
             ###
-            current_frame = self.plasma_client.get(f"{camera}{frame_time}")
+            object_id_hash = hashlib.sha1(str.encode(f"{camera}{frame_time}"))
+            object_id_bytes = object_id_hash.digest()
+            object_id = plasma.ObjectID(object_id_bytes)
+            current_frame = self.plasma_client.get(object_id, timeout_ms=0)
 
             if not current_frame is plasma.ObjectNotAvailable:
                 # draw the bounding boxes on the frame
@@ -88,10 +91,10 @@ class TrackedObjectProcessor(threading.Thread):
                 self.camera_data[camera]['current_frame'] = current_frame
 
                 # store the object id, so you can delete it at the next loop
-                previous_object_id = f"{camera}{frame_time}"
+                previous_object_id = self.camera_data[camera]['object_id']
                 if not previous_object_id is None:
-                    self.plasma_client.delete(f"{camera}{frame_time}")
-                self.camera_data[camera]['object_id'] = f"{camera}{frame_time}"
+                    self.plasma_client.delete([previous_object_id])
+                self.camera_data[camera]['object_id'] = object_id
             
             ###
             # Maintain the highest scoring recent object and frame for each label
@@ -143,4 +146,4 @@ class TrackedObjectProcessor(threading.Thread):
                 ret, jpg = cv2.imencode('.jpg', best_frame)
                 if ret:
                     jpg_bytes = jpg.tobytes()
-                    self.client.publish(f"{self.topic_prefix}/{camera}/{obj_name}/snapshot", jpg_bytes, retain=True)
+                    self.client.publish(f"{self.topic_prefix}/{camera}/{obj_name}/snapshot", jpg_bytes, retain=True)

frigate/util.py

@@ -1,14 +1,9 @@
 import datetime
-import time
-import signal
-import traceback
 import collections
 import numpy as np
 import cv2
 import threading
 import matplotlib.pyplot as plt
-import hashlib
-import pyarrow.plasma as plasma
 
 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:
@@ -132,52 +127,3 @@ class EventsPerSecond:
         now = datetime.datetime.now().timestamp()
         seconds = min(now-self._start, last_n_seconds)
         return len([t for t in self._timestamps if t > (now-last_n_seconds)]) / seconds
-
-def print_stack(sig, frame):
-    traceback.print_stack(frame)
-
-def listen():
-    signal.signal(signal.SIGUSR1, print_stack)
-
-class PlasmaManager:
-    def __init__(self):
-        self.connect()
-    
-    def connect(self):
-        while True:
-            try:
-                self.plasma_client = plasma.connect("/tmp/plasma")
-                return
-            except:
-                print(f"TrackedObjectProcessor: unable to connect plasma client")
-                time.sleep(10)
-
-    def get(self, name, timeout_ms=0):
-        object_id = plasma.ObjectID(hashlib.sha1(str.encode(name)).digest())
-        while True:
-            try:
-                return self.plasma_client.get(object_id, timeout_ms=timeout_ms)
-            except:
-                self.connect()
-                time.sleep(1)
-
-    def put(self, name, obj):
-        object_id = plasma.ObjectID(hashlib.sha1(str.encode(name)).digest())
-        while True:
-            try:
-                self.plasma_client.put(obj, object_id)
-                return
-            except Exception as e:
-                print(f"Failed to put in plasma: {e}")
-                self.connect()
-                time.sleep(1)
-
-    def delete(self, name):
-        object_id = plasma.ObjectID(hashlib.sha1(str.encode(name)).digest())
-        while True:
-            try:
-                self.plasma_client.delete([object_id])
-                return
-            except:
-                self.connect()
-                time.sleep(1)

frigate/video.py

@@ -5,15 +5,17 @@ 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
+import hashlib
+import pyarrow.plasma as plasma
+import SharedArray as sa
 import copy
 import itertools
 import json
 from collections import defaultdict
-from frigate.util import draw_box_with_label, area, calculate_region, clipped, intersection_over_union, intersection, EventsPerSecond, listen, PlasmaManager
+from frigate.util import draw_box_with_label, area, calculate_region, clipped, intersection_over_union, intersection, EventsPerSecond
 from frigate.objects import ObjectTracker
 from frigate.edgetpu import RemoteObjectDetector
 from frigate.motion import MotionDetector
@@ -102,62 +104,33 @@ def start_or_restart_ffmpeg(ffmpeg_cmd, frame_size, ffmpeg_process=None):
         ffmpeg_process.terminate()
         try:
             print("Waiting for ffmpeg to exit gracefully...")
-            ffmpeg_process.communicate(timeout=30)
+            ffmpeg_process.wait(timeout=30)
         except sp.TimeoutExpired:
             print("FFmpeg didnt exit. Force killing...")
             ffmpeg_process.kill()
-            ffmpeg_process.communicate()
-        ffmpeg_process = None
+            ffmpeg_process.wait()
 
     print("Creating ffmpeg process...")
     print(" ".join(ffmpeg_cmd))
-    process = sp.Popen(ffmpeg_cmd, stdout = sp.PIPE, stdin = sp.DEVNULL, bufsize=frame_size*10, start_new_session=True)
-    return process
+    return sp.Popen(ffmpeg_cmd, stdout = sp.PIPE, bufsize=frame_size*10)
 
-class CameraCapture(threading.Thread):
-    def __init__(self, name, ffmpeg_process, frame_shape, frame_queue, take_frame, fps):
-        threading.Thread.__init__(self)
-        self.name = name
-        self.frame_shape = frame_shape
-        self.frame_size = frame_shape[0] * frame_shape[1] * frame_shape[2]
-        self.frame_queue = frame_queue
-        self.take_frame = take_frame
-        self.fps = fps
-        self.plasma_client = PlasmaManager()
-        self.ffmpeg_process = ffmpeg_process
-
-    def run(self):
-        frame_num = 0
-        while True:
-            if self.ffmpeg_process.poll() != None:
-                print(f"{self.name}: ffmpeg process is not running. exiting capture thread...")
-                break
-
-            frame_bytes = self.ffmpeg_process.stdout.read(self.frame_size)
-            frame_time = datetime.datetime.now().timestamp()
-
-            if len(frame_bytes) == 0:
-                print(f"{self.name}: ffmpeg didnt return a frame. something is wrong.")
-                continue
-
-            frame_num += 1
-            if (frame_num % self.take_frame) != 0:
-                continue
-
-            # put the frame in the plasma store
-            self.plasma_client.put(f"{self.name}{frame_time}",
-                    np
-                        .frombuffer(frame_bytes, np.uint8)
-                        .reshape(self.frame_shape)
-                )
-            # add to the queue
-            self.frame_queue.put(frame_time)
-
-            self.fps.update()
-
-def track_camera(name, config, global_objects_config, frame_queue, frame_shape, detection_queue, detected_objects_queue, fps, skipped_fps, detection_fps, read_start):
+def track_camera(name, config, ffmpeg_global_config, global_objects_config, detection_queue, detected_objects_queue, fps, skipped_fps, detection_fps):
     print(f"Starting process for {name}: {os.getpid()}")
-    listen()
+
+    # Merge the ffmpeg config with the global config
+    ffmpeg = config.get('ffmpeg', {})
+    ffmpeg_input = get_ffmpeg_input(ffmpeg['input'])
+    ffmpeg_global_args = ffmpeg.get('global_args', ffmpeg_global_config['global_args'])
+    ffmpeg_hwaccel_args = ffmpeg.get('hwaccel_args', ffmpeg_global_config['hwaccel_args'])
+    ffmpeg_input_args = ffmpeg.get('input_args', ffmpeg_global_config['input_args'])
+    ffmpeg_output_args = ffmpeg.get('output_args', ffmpeg_global_config['output_args'])
+    ffmpeg_cmd = (['ffmpeg'] +
+            ffmpeg_global_args +
+            ffmpeg_hwaccel_args +
+            ffmpeg_input_args +
+            ['-i', ffmpeg_input] +
+            ffmpeg_output_args +
+            ['pipe:'])
 
     # Merge the tracked object config with the global config
     camera_objects_config = config.get('objects', {})    
@@ -172,8 +145,21 @@ def track_camera(name, config, global_objects_config, frame_queue, frame_shape,
         object_filters[obj] = {**global_object_filters.get(obj, {}), **camera_object_filters.get(obj, {})}
 
     expected_fps = config['fps']
+    take_frame = config.get('take_frame', 1)
 
-    frame = np.zeros(frame_shape, np.uint8)
+    if 'width' in config and 'height' in config:
+        frame_shape = (config['height'], config['width'], 3)
+    else:
+        frame_shape = get_frame_shape(ffmpeg_input)
+
+    frame_size = frame_shape[0] * frame_shape[1] * frame_shape[2]
+
+    try:
+        sa.delete(name)
+    except:
+        pass
+
+    frame = sa.create(name, shape=frame_shape, dtype=np.uint8)
 
     # load in the mask for object detection
     if 'mask' in config:
@@ -189,8 +175,10 @@ def track_camera(name, config, global_objects_config, frame_queue, frame_shape,
     object_detector = RemoteObjectDetector(name, '/labelmap.txt', detection_queue)
 
     object_tracker = ObjectTracker(10)
-
-    plasma_client = PlasmaManager()
+    
+    ffmpeg_process = start_or_restart_ffmpeg(ffmpeg_cmd, frame_size)
+    
+    plasma_client = plasma.connect("/tmp/plasma")
     frame_num = 0
     avg_wait = 0.0
     fps_tracker = EventsPerSecond()
@@ -199,23 +187,36 @@ def track_camera(name, config, global_objects_config, frame_queue, frame_shape,
     skipped_fps_tracker.start()
     object_detector.fps.start()
     while True:
-        read_start.value = datetime.datetime.now().timestamp()
-        frame_time = frame_queue.get()
-        duration = datetime.datetime.now().timestamp()-read_start.value
-        read_start.value = 0.0
+        start = datetime.datetime.now().timestamp()
+        frame_bytes = ffmpeg_process.stdout.read(frame_size)
+        duration = datetime.datetime.now().timestamp()-start
         avg_wait = (avg_wait*99+duration)/100
 
+        if not frame_bytes:
+            rc = ffmpeg_process.poll()
+            if rc is not None:
+                print(f"{name}: ffmpeg_process exited unexpectedly with {rc}")
+                ffmpeg_process = start_or_restart_ffmpeg(ffmpeg_cmd, frame_size, ffmpeg_process)
+                time.sleep(10)
+            else:
+                print(f"{name}: ffmpeg_process is still running but didnt return any bytes")
+            continue
+
+        # limit frame rate
+        frame_num += 1
+        if (frame_num % take_frame) != 0:
+            continue
+
         fps_tracker.update()
         fps.value = fps_tracker.eps()
         detection_fps.value = object_detector.fps.eps()
-        
-        # Get frame from plasma store
-        frame = plasma_client.get(f"{name}{frame_time}")
 
-        if frame is plasma.ObjectNotAvailable:
-            skipped_fps_tracker.update()
-            skipped_fps.value = skipped_fps_tracker.eps()
-            continue
+        frame_time = datetime.datetime.now().timestamp()
+        
+        # Store frame in numpy array
+        frame[:] = (np
+                    .frombuffer(frame_bytes, np.uint8)
+                    .reshape(frame_shape))
         
         # look for motion
         motion_boxes = motion_detector.detect(frame)
@@ -224,7 +225,6 @@ def track_camera(name, config, global_objects_config, frame_queue, frame_shape,
         if frame_num > 100 and fps.value < expected_fps-1 and duration < 0.5*avg_wait:
             skipped_fps_tracker.update()
             skipped_fps.value = skipped_fps_tracker.eps()
-            plasma_client.delete(f"{name}{frame_time}")
             continue
         
         skipped_fps.value = skipped_fps_tracker.eps()
@@ -328,7 +328,7 @@ def track_camera(name, config, global_objects_config, frame_queue, frame_shape,
 
                 for index in idxs:
                     obj = group[index[0]]
-                    if clipped(obj, frame_shape):
+                    if clipped(obj, frame_shape): #obj['clipped']:
                         box = obj[2]
                         # calculate a new region that will hopefully get the entire object
                         region = calculate_region(frame_shape, 
@@ -368,6 +368,9 @@ def track_camera(name, config, global_objects_config, frame_queue, frame_shape,
         # now that we have refined our detections, we need to track objects
         object_tracker.match_and_update(frame_time, detections)
 
+        # put the frame in the plasma store
+        object_id = hashlib.sha1(str.encode(f"{name}{frame_time}")).digest()
+        plasma_client.put(frame, plasma.ObjectID(object_id))
         # add to the queue
         detected_objects_queue.put((name, frame_time, object_tracker.tracked_objects))