Merge remote-tracking branch 'origin/dev' into dev

Author: JensBlack

DeepLabStream.py

@@ -302,7 +302,19 @@ def get_pose_mp(input_q, output_q):
 
                     output_q.put((index, peaks))
         elif MODEL_ORIGIN == 'DEEPPOSEKIT':
-            print('Not here yet...')
+            from deepposekit.models import load_model
+            from utils.configloader import MODEL_PATH
+            model = load_model(MODEL_PATH)
+            predict_model = model.predict_model
+            while True:
+                if input_q.full():
+                    index, frame = input_q.get()
+                    frame = frame[..., 1][..., None]
+                    st_frame = np.stack([frame])
+                    prediction = predict_model.predict(st_frame, batch_size=1, verbose=True)
+                    peaks= prediction[0,:,:2]
+                    output_q.put((index, peaks))
+
 
         else:
             raise ValueError(f'Model origin {MODEL_ORIGIN} not available.')

experiments/base/stimulation.py

@@ -19,18 +19,35 @@ def __init__(self):
         self._name = 'BaseStimulation'
         self._parameter_dict = dict(TYPE='str',
                                     PORT='str',
+                                    IP = 'str',
                                     STIM_TIME='float')
         self._settings_dict = get_stimulation_settings(self._name, self._parameter_dict)
         self._running = False
-        self._stim_device = self._setup_device(self._settings_dict['TYPE'], self._settings_dict['PORT'])
+        self._stim_device = self._setup_device(self._settings_dict['TYPE'], self._settings_dict['PORT'], self._settings_dict['IP'])
 
     @staticmethod
-    def _setup_device(type, port):
+    def _setup_device(type, port, ip):
         device = None
         if type == 'NI':
             from experiments.utils.DAQ_output import DigitalModDevice
             device = DigitalModDevice(port)
 
+        if type == 'RASPBERRY':
+            from experiments.utils.gpio_control import DigitalPiDevice
+            device = DigitalPiDevice(port)
+
+        if type == 'RASP_NETWORK':
+            from experiments.utils.gpio_control import DigitalPiDevice
+            if ip is not None:
+                device = DigitalPiDevice(port, ip)
+            else:
+                raise ValueError('IP required for remote GPIO control.')
+
+        if type == 'ARDUINO':
+            from experiments.utils.gpio_control import DigitalArduinoDevice
+            device = DigitalArduinoDevice(port)
+
+
         return device
 
     def stimulate(self):
@@ -72,23 +89,38 @@ class RewardDispenser(BaseStimulation):
     def __init__(self):
         self._name = 'RewardDispenser'
         self._parameter_dict = dict(TYPE = 'str',
+                                    IP = 'str',
                                     STIM_PORT= 'str',
                                     REMOVAL_PORT = 'str',
                                     STIM_TIME = 'float',
                                     REMOVAL_TIME = 'float')
         self._settings_dict = get_stimulation_settings(self._name, self._parameter_dict)
         self._running = False
-        self._stim_device = self._setup_device(self._settings_dict['TYPE'], self._settings_dict['STIM_PORT'])
-        self._removal_device = self._setup_device(self._settings_dict['TYPE'], self._settings_dict['REMOVAL_PORT'])
+        self._stim_device = self._setup_device(self._settings_dict['TYPE'], self._settings_dict['STIM_PORT'],
+                                               self._settings_dict['IP'])
+        self._removal_device = self._setup_device(self._settings_dict['TYPE'], self._settings_dict['REMOVAL_PORT'],
+                                                  self._settings_dict['IP'])
 
 
     @staticmethod
-    def _setup_device(type, port):
+    def _setup_device(type, port, ip):
         device = None
         if type == 'NI':
             from experiments.utils.DAQ_output import DigitalModDevice
             device = DigitalModDevice(port)
 
+        if type == 'RASPBERRY':
+            from experiments.utils.gpio_control import DigitialPiBoardDevice
+            device = DigitialPiBoardDevice(port)
+
+        if type == 'RASP_NETWORK':
+            from experiments.utils.gpio_control import DigitialPiBoardDevice
+            if ip is not None:
+                device = DigitialPiBoardDevice(port, ip)
+            else:
+                raise ValueError('IP required for remote GPIO control.')
+
+
         return device
 
     def stimulate(self):

experiments/configs/default_config.ini

@@ -98,12 +98,20 @@ STIMULATION = BaseStimulation
 ;[STIMULATION]
 
 [BaseStimulation]
+; can be NI, RASPBERRY or RASP_NETWORK
 TYPE = NI
+;only used in RASP_NETWORK
+IP = None
+;PORT parameter is used for all (Port from DAQ, PIN from Raspberry, or serial port from Arduino)
 PORT = Dev1/PFI6
 STIM_TIME = 3.5
 
 [RewardDispenser]
+; can be NI, RASPBERRY, RASP_NETWORK or ARDUINO
 TYPE = NI
+;only used in RASP_NETWORK
+IP = None
+;PORT parameter is used for all (Port from DAQ, PIN from Raspberry, or serial port from Arduino)
 STIM_PORT = Dev1/PFI6
 REMOVAL_PORT = Dev1/PFI5
 STIM_TIME = 3.5

experiments/utils/gpio_control.py

@@ -0,0 +1,66 @@
+from gpiozero import DigitalOutputDevice
+from gpiozero.pins.pigpio import PiGPIOFactory
+
+import serial
+
+
+
+class DigitalPiDevice:
+    """
+    Digital modulated devices in combination with Raspberry Pi GPIO
+    Setup: https://gpiozero.readthedocs.io/en/stable/remote_gpio.html
+    """
+
+    def __init__(self, PIN, BOARD_IP: str = None):
+
+        """
+        :param BOARD_IP:  IP adress of board connected to the Device
+        """
+        if BOARD_IP is not None:
+            self._factory = PiGPIOFactory(host = BOARD_IP)
+            self._device = DigitalOutputDevice(PIN, pin_factory = self._factory)
+        else:
+            self._factory = None
+            self._device = DigitalOutputDevice(PIN)
+        self._running = False
+
+    def turn_on(self):
+        self._device.on()
+        self._running = True
+
+    def turn_off(self):
+        self._device.off()
+        self._running = False
+
+    def toggle(self):
+        self._device.toggle()
+        self._running = self._device.is_active
+
+
+class DigitalArduinoDevice:
+    """
+    Digital modulated devices in combination with Arduino boards connected via USB
+        setup: https://pythonforundergradengineers.com/python-arduino-LED.html
+
+    """
+
+    def __init__(self, PORT):
+        """
+        :param PORT:  USB PORT of the arduino board
+        """
+        self._device = serial.Serial(PORT, baudrate=9600)
+        self._running = False
+
+    def turn_on(self):
+        self._device.write(b'H')
+        self._running = True
+
+    def turn_off(self):
+        self._device.write(b'L')
+        self._running = False
+
+    def toggle(self):
+        if self._running:
+            self.turn_off()
+        else:
+            self.turn_on()

experiments/utils/gpio_raspberry.py

@@ -1,8 +1,11 @@
+gpiozero
+pigpio
+pyserial
 nidaqmx>=0.5.7
 click>=7.0
 opencv-python>=3.4.5.20
 numpy>=1.14.5
 pandas>=0.21.0
 matplotlib>=3.0.3
 scikit-image>=0.14.2
-scipy>=1.1.0
+scipy>=1.1.0

requirements.txt

@@ -18,6 +18,8 @@
 from utils.configloader import VIDEO_SOURCE, OUT_DIR, ANIMALS_NUMBER
 from experiments.custom.experiments import ExampleExperiment
 
+from deepposekit.models import load_model
+
 
 def create_dataframes(data_output):
     """
@@ -43,7 +45,7 @@ def create_dataframes(data_output):
 
 def start_videoanalyser():
     print("Starting DeepLabCut")
-    config, sess, inputs, outputs = load_deeplabcut()
+    model = load_model(r"D:\DeepPoseKit-Data-master\datasets\fly\best_model_densenet.h5")
 
     experiment_enabled = False
     video_output = True

utils/VideoAnalyzer.py

@@ -240,7 +240,7 @@ def extract_to_animal_skeleton(coords):
     return animal_skeletons
 
 
-"""DLC LIVE"""
+"""DLC LIVE & DeepPoseKit"""
 
 
 
@@ -286,11 +286,11 @@ def calculate_skeletons(peaks: dict, animals_number: int) -> list:
     elif MODEL_ORIGIN == 'MADLC':
         animal_skeletons = calculate_ma_skeletons(peaks, animals_number)
 
-    elif MODEL_ORIGIN == 'DLC-LIVE':
+    elif MODEL_ORIGIN == 'DLC-LIVE' or MODEL_ORIGIN == 'DEEPPOSEKIT':
         animal_skeletons = calculate_skeletons_dlc_live(peaks,  animals_number= 1)
         if animals_number != 1:
             raise ValueError('Multiple animals are currently not supported by DLC-LIVE.'
-                             ' If you are using differently colored animals, please refere to the bodyparts directly.')
+                             ' If you are using differently colored animals, please refer to the bodyparts directly.')
 
     return animal_skeletons
 

utils/poser.py

@@ -0,0 +1,75 @@
+"""
+DeepLabStream
+© J.Schweihoff, M. Loshakov
+University Bonn Medical Faculty, Germany
+https://github.com/SchwarzNeuroconLab/DeepLabStream
+Licensed under GNU General Public License v3.0
+"""
+
+import cv2
+from deepposekit.models import load_model
+from deepposekit.io import VideoReader
+import numpy as np
+import tensorflow as tf
+
+
+
+def plot_dlc_bodyparts(image, bodyparts):
+    """
+    Plots dlc bodyparts on given image
+    adapted from plotter
+    """
+
+    for bp in bodyparts:
+        center = tuple(bp.astype(int))
+        cv2.circle(image, center=center, radius=3, color=(255, 0, 0), thickness=2)
+    return image
+
+def start_videoanalyser():
+    print("Starting DeepPoseKit")
+    video = cv2.VideoCapture(r"D:\DeepPoseKit-Data-master\datasets\fly\video.avi")
+    resolution = int(video.get(cv2.CAP_PROP_FRAME_WIDTH)), int(video.get(cv2.CAP_PROP_FRAME_HEIGHT))
+
+    path_model = r"D:\DeepPoseKit-Data-master\datasets\fly\best_model_densenet.h5"
+    model = load_model(path_model)
+
+
+    predict_model = model.predict_model
+    # predict_model.layers.pop(0)  # remove current input layer
+    #
+    # inputs = tf.keras.layers.Input((resolution[0], resolution[1], 3))
+    # outputs = predict_model(inputs)
+    # predict_model = tf.keras.Model(inputs, outputs)
+
+
+    experiment_enabled = False
+
+    index = 0
+    while video.isOpened():
+        ret, frame = video.read()
+        if ret is not None:
+            org_frame = frame
+            frame = frame[..., 1][..., None]
+            st_frame = np.stack([frame])
+            prediction = predict_model.predict(st_frame, batch_size= 1,  verbose=True)
+            x, y, confidence = np.split(prediction, 3, -1)
+
+            print(prediction.shape)
+            predi = prediction[0,:,:2]
+            pre = predi[:, :2]
+            print(pre)
+            out_frame = plot_dlc_bodyparts(org_frame, predi)
+            # out_frame = org_frame
+            cv2.imshow('stream', out_frame)
+            index += 1
+        else:
+            break
+
+        if cv2.waitKey(1) & 0xFF == ord('q'):
+            break
+
+    video.release()
+
+
+if __name__ == "__main__":
+    start_videoanalyser()