upgrade to python3.8 and switch from plasma store to shared_memory

Dockerfile

@@ -1,4 +1,4 @@
-FROM ubuntu:18.04
+FROM ubuntu:20.04
 LABEL maintainer "[email protected]"
 
 ENV DEBIAN_FRONTEND=noninteractive
@@ -11,38 +11,37 @@ 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 - \
  && apt -qq update \
  && 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"]

detect_objects.py

@@ -63,23 +63,13 @@
 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 @@ def run(self):
  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 on_connect(client, userdata, flags, rc):
  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 on_connect(client, userdata, flags, rc):
 
  # 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 on_connect(client, userdata, flags, rc):
  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 on_connect(client, userdata, flags, rc):
  }
 
  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 on_connect(client, userdata, flags, rc):
  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 receiveSignal(signalNumber, frame):
  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 stats():
  '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 imagestream(camera_name, fps, height):
  app.run(host='0.0.0.0', port=WEB_PORT, debug=False)
 
  object_processor.join()
-
- plasma_process.terminate()
 
 if __name__ == '__main__':
  main()

frigate/edgetpu.py

@@ -2,12 +2,14 @@
 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 @@ def detect_raw(self, tensor_input):
 
  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 @@ def detect(self, tensor_input, threshold=.4):
  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

frigate/object_processing.py

@@ -10,9 +10,8 @@
 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 @@ def __init__(self, name, config, frame_manager):
  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 @@ def update(self, frame_time, tracked_objects):
  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 @@ def __init__(self, camera_config, client, topic_prefix, tracked_objects_queue, e
  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 @@ def object_status(camera, object_name, status):
  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)