From e37eba49ff3e6d8e5a3b570024183c5e4511c23b Mon Sep 17 00:00:00 2001 From: Blake Blackshear Date: Fri, 13 Mar 2020 16:13:01 -0500 Subject: [PATCH] make object processor resilient to plasma failures --- Dockerfile | 6 - detect_objects.py | 5 +- frigate/object_processing.py | 227 +++++++++++++++++++---------------- 3 files changed, 127 insertions(+), 111 deletions(-) diff --git a/Dockerfile b/Dockerfile index 33df7a4e1..cb81052af 100755 --- a/Dockerfile +++ b/Dockerfile @@ -51,12 +51,6 @@ RUN wget -q https://storage.googleapis.com/download.tensorflow.org/models/tflite mv /detect.tflite /cpu_model.tflite && \ rm /cpu_model.zip -RUN apt -qq update && apt -qq install --no-install-recommends -y \ - gdb \ - python3.7-dbg \ - && rm -rf /var/lib/apt/lists/* \ - && (apt-get autoremove -y; apt-get autoclean -y) - WORKDIR /opt/frigate/ ADD frigate frigate/ COPY detect_objects.py . diff --git a/detect_objects.py b/detect_objects.py index 71d82e805..afbc6f057 100644 --- a/detect_objects.py +++ b/detect_objects.py @@ -71,13 +71,12 @@ def start_plasma_store(): return plasma_process class CameraWatchdog(threading.Thread): - def __init__(self, camera_processes, config, tflite_process, tracked_objects_queue, object_processor, plasma_process): + def __init__(self, camera_processes, config, tflite_process, tracked_objects_queue, plasma_process): 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.object_processor = object_processor self.plasma_process = plasma_process def run(self): @@ -202,7 +201,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, object_processor, plasma_process) + camera_watchdog = CameraWatchdog(camera_processes, CONFIG['cameras'], tflite_process, tracked_objects_queue, plasma_process) camera_watchdog.start() # create a flask app that encodes frames a mjpeg on demand diff --git a/frigate/object_processing.py b/frigate/object_processing.py index 034c3dff1..a06824246 100644 --- a/frigate/object_processing.py +++ b/frigate/object_processing.py @@ -1,6 +1,7 @@ import json import hashlib import datetime +import time import copy import cv2 import threading @@ -44,109 +45,131 @@ class TrackedObjectProcessor(threading.Thread): def get_current_frame(self, camera): return self.camera_data[camera]['current_frame'] - - def run(self): + + def connect_plasma_client(self): while True: try: self.plasma_client = plasma.connect("/tmp/plasma") - while True: - camera, frame_time, tracked_objects = self.tracked_objects_queue.get() - - config = self.config[camera] - best_objects = self.camera_data[camera]['best_objects'] - current_object_status = self.camera_data[camera]['object_status'] - self.camera_data[camera]['tracked_objects'] = tracked_objects - - ### - # Draw tracked objects on the frame - ### - 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 - for obj in tracked_objects.values(): - thickness = 2 - color = COLOR_MAP[obj['label']] - - if obj['frame_time'] != frame_time: - thickness = 1 - color = (255,0,0) - - # draw the bounding boxes on the frame - box = obj['box'] - draw_box_with_label(current_frame, box[0], box[1], box[2], box[3], obj['label'], f"{int(obj['score']*100)}% {int(obj['area'])}", thickness=thickness, color=color) - # draw the regions on the frame - region = obj['region'] - cv2.rectangle(current_frame, (region[0], region[1]), (region[2], region[3]), (0,255,0), 1) - - if config['snapshots']['show_timestamp']: - time_to_show = datetime.datetime.fromtimestamp(frame_time).strftime("%m/%d/%Y %H:%M:%S") - cv2.putText(current_frame, time_to_show, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, fontScale=.8, color=(255, 255, 255), thickness=2) - - ### - # Set the current frame as ready - ### - self.camera_data[camera]['current_frame'] = current_frame - - # store the object id, so you can delete it at the next loop - previous_object_id = self.camera_data[camera]['object_id'] - if not previous_object_id is None: - 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 - ### - for obj in tracked_objects.values(): - # if the object wasn't seen on the current frame, skip it - if obj['frame_time'] != frame_time: - continue - if obj['label'] in best_objects: - now = datetime.datetime.now().timestamp() - # if the object is a higher score than the current best score - # or the current object is more than 1 minute old, use the new object - if obj['score'] > best_objects[obj['label']]['score'] or (now - best_objects[obj['label']]['frame_time']) > 60: - obj['frame'] = np.copy(self.camera_data[camera]['current_frame']) - best_objects[obj['label']] = obj - else: - obj['frame'] = np.copy(self.camera_data[camera]['current_frame']) - best_objects[obj['label']] = obj - - ### - # Report over MQTT - ### - # count objects with more than 2 entries in history by type - obj_counter = Counter() - for obj in tracked_objects.values(): - if len(obj['history']) > 1: - obj_counter[obj['label']] += 1 - - # report on detected objects - for obj_name, count in obj_counter.items(): - new_status = 'ON' if count > 0 else 'OFF' - if new_status != current_object_status[obj_name]: - current_object_status[obj_name] = new_status - self.client.publish(f"{self.topic_prefix}/{camera}/{obj_name}", new_status, retain=False) - # send the best snapshot over mqtt - best_frame = cv2.cvtColor(best_objects[obj_name]['frame'], cv2.COLOR_RGB2BGR) - 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) - - # expire any objects that are ON and no longer detected - expired_objects = [obj_name for obj_name, status in current_object_status.items() if status == 'ON' and not obj_name in obj_counter] - for obj_name in expired_objects: - current_object_status[obj_name] = 'OFF' - self.client.publish(f"{self.topic_prefix}/{camera}/{obj_name}", 'OFF', retain=False) - # send updated snapshot over mqtt - best_frame = cv2.cvtColor(best_objects[obj_name]['frame'], cv2.COLOR_RGB2BGR) - 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) + return except: - pass \ No newline at end of file + print(f"TrackedObjectProcessor: unable to connect plasma client") + time.sleep(10) + + def get_from_plasma(self, object_id): + while True: + try: + return self.plasma_client.get(object_id, timeout_ms=0) + except: + self.connect_plasma_client() + time.sleep(1) + + def delete_from_plasma(self, object_ids): + while True: + try: + self.plasma_client.delete(object_ids) + return + except: + self.connect_plasma_client() + time.sleep(1) + + def run(self): + self.connect_plasma_client() + while True: + camera, frame_time, tracked_objects = self.tracked_objects_queue.get() + + config = self.config[camera] + best_objects = self.camera_data[camera]['best_objects'] + current_object_status = self.camera_data[camera]['object_status'] + self.camera_data[camera]['tracked_objects'] = tracked_objects + + ### + # Draw tracked objects on the frame + ### + 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.get_from_plasma(object_id) + + if not current_frame is plasma.ObjectNotAvailable: + # draw the bounding boxes on the frame + for obj in tracked_objects.values(): + thickness = 2 + color = COLOR_MAP[obj['label']] + + if obj['frame_time'] != frame_time: + thickness = 1 + color = (255,0,0) + + # draw the bounding boxes on the frame + box = obj['box'] + draw_box_with_label(current_frame, box[0], box[1], box[2], box[3], obj['label'], f"{int(obj['score']*100)}% {int(obj['area'])}", thickness=thickness, color=color) + # draw the regions on the frame + region = obj['region'] + cv2.rectangle(current_frame, (region[0], region[1]), (region[2], region[3]), (0,255,0), 1) + + if config['snapshots']['show_timestamp']: + time_to_show = datetime.datetime.fromtimestamp(frame_time).strftime("%m/%d/%Y %H:%M:%S") + cv2.putText(current_frame, time_to_show, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, fontScale=.8, color=(255, 255, 255), thickness=2) + + ### + # Set the current frame as ready + ### + self.camera_data[camera]['current_frame'] = current_frame + + # store the object id, so you can delete it at the next loop + previous_object_id = self.camera_data[camera]['object_id'] + if not previous_object_id is None: + self.delete_from_plasma([previous_object_id]) + self.camera_data[camera]['object_id'] = object_id + + ### + # Maintain the highest scoring recent object and frame for each label + ### + for obj in tracked_objects.values(): + # if the object wasn't seen on the current frame, skip it + if obj['frame_time'] != frame_time: + continue + if obj['label'] in best_objects: + now = datetime.datetime.now().timestamp() + # if the object is a higher score than the current best score + # or the current object is more than 1 minute old, use the new object + if obj['score'] > best_objects[obj['label']]['score'] or (now - best_objects[obj['label']]['frame_time']) > 60: + obj['frame'] = np.copy(self.camera_data[camera]['current_frame']) + best_objects[obj['label']] = obj + else: + obj['frame'] = np.copy(self.camera_data[camera]['current_frame']) + best_objects[obj['label']] = obj + + ### + # Report over MQTT + ### + # count objects with more than 2 entries in history by type + obj_counter = Counter() + for obj in tracked_objects.values(): + if len(obj['history']) > 1: + obj_counter[obj['label']] += 1 + + # report on detected objects + for obj_name, count in obj_counter.items(): + new_status = 'ON' if count > 0 else 'OFF' + if new_status != current_object_status[obj_name]: + current_object_status[obj_name] = new_status + self.client.publish(f"{self.topic_prefix}/{camera}/{obj_name}", new_status, retain=False) + # send the best snapshot over mqtt + best_frame = cv2.cvtColor(best_objects[obj_name]['frame'], cv2.COLOR_RGB2BGR) + 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) + + # expire any objects that are ON and no longer detected + expired_objects = [obj_name for obj_name, status in current_object_status.items() if status == 'ON' and not obj_name in obj_counter] + for obj_name in expired_objects: + current_object_status[obj_name] = 'OFF' + self.client.publish(f"{self.topic_prefix}/{camera}/{obj_name}", 'OFF', retain=False) + # send updated snapshot over mqtt + best_frame = cv2.cvtColor(best_objects[obj_name]['frame'], cv2.COLOR_RGB2BGR) + 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)