X86 pedestrian detection smart distancing

config-skeleton.ini

@@ -8,7 +8,7 @@ Resolution: 640,480
 [Detector]
 ; Supported devices: Jetson , EdgeTPU, Dummy
 Device: EdgeTPU
-; Detector's Name can be either "mobilenet_ssd_v2", "pedestrian_ssdlite_mobilenet_v2" or "pedestrian_ssdlite_mobilenet_v2"
+; Detector's Name can be either "mobilenet_ssd_v2", "pedestrian_ssd_mobilenet_v2" or "pedestrian_ssdlite_mobilenet_v2"
 ; the first one is trained on COCO dataset and next two are trained on Oxford Town Center dataset to detect pedestrians
 Name: pedestrian_ssdlite_mobilenet_v2
 ;ImageSize should be 3 numbers seperated by commas, no spaces: 300,300,3

config-x86.ini

@@ -8,8 +8,11 @@ Resolution: 640,480
 [Detector]
 ; Supported devices: Jetson , EdgeTPU, Dummy
 Device: x86
-Name: mobilenet_ssd_v2
+;Name can be either ssd_mobilenet_v2 or pedestrian_ssd_mobilenet_v2 or pedestrian_ssdlite_mobilenet_v2 or pedestrian_ssdlite_mobilenet_v2
+;The first one is trained on COCO dataset and next three are trained on Oxford Town Center dataset to detect pedestrians
+Name: pedestrian_ssd_mobilenet_v2
 ;ImageSize should be 3 numbers seperated by commas, no spaces: 300,300,3
+;For the pedestrian_ssdlite_mobilenet_v3: 320,320,3
 ImageSize: 300,300,3
 ModelPath: 
 ClassID: 1

libs/detectors/x86/detector.py

@@ -13,12 +13,21 @@ def __init__(self, config):
         self.config = config
         self.name = self.config.get_section_dict('Detector')['Name']
 
-        if self.name == 'mobilenet_ssd_v2':
-            from libs.detectors.x86 import mobilenet_ssd
-            self.net = mobilenet_ssd.Detector(self.config)
+        if self.name == 'ssd_mobilenet_v2':
+            from libs.detectors.x86 import ssd_mobilenet_v2
+            self.net = ssd_mobilenet_v2.Detector(self.config)
         elif self.name == "openvino":
             from libs.detectors.x86 import openvino
             self.net = openvino.Detector(self.config)
+        elif self.name == "pedestrian_ssd_mobilenet_v2":
+            from libs.detectors.x86 import pedestrian_ssd_mobilenet_v2
+            self.net = pedestrian_ssd_mobilenet_v2.Detector(self.config)
+        elif self.name == "pedestrian_ssdlite_mobilenet_v2":
+            from libs.detectors.x86 import pedestrian_ssdlite_mobilenet_v2
+            self.net = pedestrian_ssdlite_mobilenet_v2.Detector(self.config)
+        elif self.name == "pedestrian_ssdlite_mobilenet_v3":
+            from libs.detectors.x86 import pedestrian_ssdlite_mobilenet_v3
+            self.net = pedestrian_ssdlite_mobilenet_v3.Detector(self.config)
         else:
             raise ValueError('Not supported network named: ', self.name)
 

libs/detectors/x86/pedestrian_ssd_mobilenet_v2.py

@@ -0,0 +1,77 @@
+import pathlib
+import time
+import os
+import numpy as np
+import wget
+import tensorflow as tf
+
+from libs.detectors.utils.fps_calculator import convert_infr_time_to_fps
+
+
+def load_model(model_name):
+    base_url = 'https://raw.githubusercontent.com/neuralet/neuralet-models/master/amd64/'
+    model_file = model_name + "/saved_model/saved_model.pb"
+    base_dir = "libs/detectors/x86/data/"
+    model_dir = os.path.join(base_dir, model_name)
+    if not os.path.isdir(model_dir):
+        os.makedirs(os.path.join(model_dir, "saved_model"), exist_ok=True)
+        print('model does not exist under: ', model_dir, 'downloading from ', base_url + model_file)
+        wget.download(base_url + model_file, os.path.join(model_dir, "saved_model"))
+
+    model_dir = pathlib.Path(model_dir) / "saved_model"
+
+    model = tf.saved_model.load(str(model_dir))
+    model = model.signatures['serving_default']
+
+    return model
+
+
+class Detector:
+    """
+    Perform object detection with the given model. The model is a quantized tflite
+    file which if the detector can not find it at the path it will download it
+    from neuralet repository automatically.
+
+    :param config: Is a ConfigEngine instance which provides necessary parameters.
+    """
+
+    def __init__(self, config):
+        self.config = config
+        # Get the model name from the config
+        self.model_name = self.config.get_section_dict('Detector')['Name']
+        # Frames Per Second
+        self.fps = None
+
+        self.detection_model = load_model('ped_ssd_mobilenet_v2')
+
+    def inference(self, resized_rgb_image):
+        """
+        inference function sets input tensor to input image and gets the output.
+        The interpreter instance provides corresponding detection output which is used for creating result
+        Args:
+            resized_rgb_image: uint8 numpy array with shape (img_height, img_width, channels)
+
+        Returns:
+            result: a dictionary contains of [{"id": 0, "bbox": [x1, y1, x2, y2], "score":s%}, {...}, {...}, ...]
+        """
+        input_image = np.expand_dims(resized_rgb_image, axis=0)
+        input_tensor = tf.convert_to_tensor(input_image)
+        t_begin = time.perf_counter()
+        output_dict = self.detection_model(input_tensor)
+        inference_time = time.perf_counter() - t_begin  # Seconds
+
+        # Calculate Frames rate (fps)
+        self.fps = convert_infr_time_to_fps(inference_time)
+
+        boxes = output_dict['detection_boxes']
+        labels = output_dict['detection_classes']
+        scores = output_dict['detection_scores']
+
+        class_id = int(self.config.get_section_dict('Detector')['ClassID'])
+        score_threshold = float(self.config.get_section_dict('Detector')['MinScore'])
+        result = []
+        for i in range(boxes.shape[1]):  # number of boxes
+            if labels[0, i] == class_id and scores[0, i] > score_threshold:
+                result.append({"id": str(class_id) + '-' + str(i), "bbox": boxes[0, i, :], "score": scores[0, i]})
+
+        return result

libs/detectors/x86/pedestrian_ssdlite_mobilenet_v2.py

@@ -0,0 +1,77 @@
+import pathlib
+import time
+import os
+import numpy as np
+import wget
+import tensorflow as tf
+
+from libs.detectors.utils.fps_calculator import convert_infr_time_to_fps
+
+
+def load_model(model_name):
+    base_url = 'https://raw.githubusercontent.com/neuralet/neuralet-models/master/amd64/'
+    model_file = model_name + "/saved_model/saved_model.pb"
+    base_dir = "libs/detectors/x86/data/"
+    model_dir = os.path.join(base_dir, model_name)
+    if not os.path.isdir(model_dir):
+        os.makedirs(os.path.join(model_dir, "saved_model"), exist_ok=True)
+        print('model does not exist under: ', model_dir, 'downloading from ', base_url + model_file)
+        wget.download(base_url + model_file, os.path.join(model_dir, "saved_model"))
+
+    model_dir = pathlib.Path(model_dir) / "saved_model"
+
+    model = tf.saved_model.load(str(model_dir))
+    model = model.signatures['serving_default']
+
+    return model
+
+
+class Detector:
+    """
+    Perform object detection with the given model. The model is a quantized tflite
+    file which if the detector can not find it at the path it will download it
+    from neuralet repository automatically.
+
+    :param config: Is a ConfigEngine instance which provides necessary parameters.
+    """
+
+    def __init__(self, config):
+        self.config = config
+        # Get the model name from the config
+        self.model_name = self.config.get_section_dict('Detector')['Name']
+        # Frames Per Second
+        self.fps = None
+
+        self.detection_model = load_model('ped_ssdlite_mobilenet_v2')
+
+    def inference(self, resized_rgb_image):
+        """
+        inference function sets input tensor to input image and gets the output.
+        The interpreter instance provides corresponding detection output which is used for creating result
+        Args:
+            resized_rgb_image: uint8 numpy array with shape (img_height, img_width, channels)
+
+        Returns:
+            result: a dictionary contains of [{"id": 0, "bbox": [x1, y1, x2, y2], "score":s%}, {...}, {...}, ...]
+        """
+        input_image = np.expand_dims(resized_rgb_image, axis=0)
+        input_tensor = tf.convert_to_tensor(input_image)
+        t_begin = time.perf_counter()
+        output_dict = self.detection_model(input_tensor)
+        inference_time = time.perf_counter() - t_begin  # Seconds
+
+        # Calculate Frames rate (fps)
+        self.fps = convert_infr_time_to_fps(inference_time)
+
+        boxes = output_dict['detection_boxes']
+        labels = output_dict['detection_classes']
+        scores = output_dict['detection_scores']
+
+        class_id = int(self.config.get_section_dict('Detector')['ClassID'])
+        score_threshold = float(self.config.get_section_dict('Detector')['MinScore'])
+        result = []
+        for i in range(boxes.shape[1]):  # number of boxes
+            if labels[0, i] == class_id and scores[0, i] > score_threshold:
+                result.append({"id": str(class_id) + '-' + str(i), "bbox": boxes[0, i, :], "score": scores[0, i]})
+
+        return result

libs/detectors/x86/pedestrian_ssdlite_mobilenet_v3.py

@@ -0,0 +1,77 @@
+import pathlib
+import time
+import os
+import numpy as np
+import wget
+import tensorflow as tf
+
+from libs.detectors.utils.fps_calculator import convert_infr_time_to_fps
+
+
+def load_model(model_name):
+    base_url = 'https://raw.githubusercontent.com/neuralet/neuralet-models/master/amd64/'
+    model_file = model_name + "/saved_model/saved_model.pb"
+    base_dir = "libs/detectors/x86/data/"
+    model_dir = os.path.join(base_dir, model_name)
+    if not os.path.isdir(model_dir):
+        os.makedirs(os.path.join(model_dir, "saved_model"), exist_ok=True)
+        print('model does not exist under: ', model_dir, 'downloading from ', base_url + model_file)
+        wget.download(base_url + model_file, os.path.join(model_dir, "saved_model"))
+
+    model_dir = pathlib.Path(model_dir) / "saved_model"
+
+    model = tf.saved_model.load(str(model_dir))
+    model = model.signatures['serving_default']
+
+    return model
+
+
+class Detector:
+    """
+    Perform object detection with the given model. The model is a quantized tflite
+    file which if the detector can not find it at the path it will download it
+    from neuralet repository automatically.
+
+    :param config: Is a ConfigEngine instance which provides necessary parameters.
+    """
+
+    def __init__(self, config):
+        self.config = config
+        # Get the model name from the config
+        self.model_name = self.config.get_section_dict('Detector')['Name']
+        # Frames Per Second
+        self.fps = None
+
+        self.detection_model = load_model('ped_ssdlite_mobilenet_v3')
+
+    def inference(self, resized_rgb_image):
+        """
+        inference function sets input tensor to input image and gets the output.
+        The interpreter instance provides corresponding detection output which is used for creating result
+        Args:
+            resized_rgb_image: uint8 numpy array with shape (img_height, img_width, channels)
+
+        Returns:
+            result: a dictionary contains of [{"id": 0, "bbox": [x1, y1, x2, y2], "score":s%}, {...}, {...}, ...]
+        """
+        input_image = np.expand_dims(resized_rgb_image, axis=0)
+        input_tensor = tf.convert_to_tensor(input_image)
+        t_begin = time.perf_counter()
+        output_dict = self.detection_model(input_tensor)
+        inference_time = time.perf_counter() - t_begin  # Seconds
+
+        # Calculate Frames rate (fps)
+        self.fps = convert_infr_time_to_fps(inference_time)
+
+        boxes = output_dict['detection_boxes']
+        labels = output_dict['detection_classes']
+        scores = output_dict['detection_scores']
+
+        class_id = int(self.config.get_section_dict('Detector')['ClassID'])
+        score_threshold = float(self.config.get_section_dict('Detector')['MinScore'])
+        result = []
+        for i in range(boxes.shape[1]):  # number of boxes
+            if labels[0, i] == class_id and scores[0, i] > score_threshold:
+                result.append({"id": str(class_id) + '-' + str(i), "bbox": boxes[0, i, :], "score": scores[0, i]})
+
+        return result

libs/detectors/x86/mobilenet_ssd.py → libs/detectors/x86/ssd_mobilenet_v2

@@ -1,27 +1,31 @@
 import pathlib
 import time
-
+import os
 import numpy as np
-
+import tarfile
+import wget
 import tensorflow as tf
 
 from libs.detectors.utils.fps_calculator import convert_infr_time_to_fps
 
 
 def load_model(model_name):
-  base_url = 'http://download.tensorflow.org/models/object_detection/'
-  model_file = model_name + '.tar.gz'
-  model_dir = tf.keras.utils.get_file(
-    fname=model_name,
-    origin=base_url + model_file,
-    untar=True)
-
-  model_dir = pathlib.Path(model_dir) / "saved_model"
-
-  model = tf.saved_model.load(str(model_dir))
-  model = model.signatures['serving_default']
-
-  return model
+    base_url = 'http://download.tensorflow.org/models/object_detection/'
+    model_file = model_name + '.tar.gz'
+    base_dir = "libs/detectors/x86/data/"
+    model_dir = os.path.join(base_dir, model_name)
+    if not os.path.isdir(model_dir):
+        print('model does not exist under: ', model_dir, 'downloading from ', base_url + model_file)
+        wget.download(base_url + model_file, base_dir)
+        with tarfile.open(base_dir + model_file, "r") as tar:
+            tar.extractall(path=base_dir)
+
+    model_dir = pathlib.Path(model_dir) / "saved_model"
+
+    model = tf.saved_model.load(str(model_dir))
+    model = model.signatures['serving_default']
+
+    return model
 
 
 class Detector: