updating model options

updating stimulation options

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.generic_output import DigitalPiBoardDevice
+            device = DigitalPiBoardDevice(port)
+
+        if type == 'RASP_NETWORK':
+            from experiments.utils.generic_output import DigitalPiBoardDevice
+            if ip is not None:
+                device = DigitalPiBoardDevice(port, ip)
+            else:
+                raise ValueError('IP required for remote GPIO control.')
+
+        if type == 'ARDUINO':
+            from experiments.utils.generic_output 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.generic_output import DigitialPiBoardDevice
+            device = DigitialPiBoardDevice(port)
+
+        if type == 'RASP_NETWORK':
+            from experiments.utils.generic_output 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 from DAQ, PIN from Raspberry
 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 from DAQ, PIN from Raspberry or USB PORT for ARDUINO
 STIM_PORT = Dev1/PFI6
 REMOVAL_PORT = Dev1/PFI5
 STIM_TIME = 3.5

experiments/utils/gpio_raspberry.py experiments/utils/generic_output.pyexperiments/utils/gpio_raspberry.py renamed to experiments/utils/generic_output.py

@@ -1,19 +1,22 @@
 from gpiozero import DigitalOutputDevice
 from gpiozero.pins.pigpio import PiGPIOFactory
+
+import serial
 BOARD_IP = '192.168.1.2'
 
 
 
-class DigitialPiBoardDevice:
+class DigitalPiBoardDevice:
     """
     Digital modulated devices in combination with Raspberry Pi GPIO
     """
 
-    def __init__(self, PIN, BOARD_IP: str = None,  remote: bool = False):
+    def __init__(self, PIN, BOARD_IP: str = None):
+
         """
         :param BOARD_IP:  IP adress of board connected to the Device
         """
-        if remote:
+        if BOARD_IP is not None:
             self._factory = PiGPIOFactory(host = BOARD_IP)
             self._device = DigitalOutputDevice(PIN= PIN, pin_factory = self._factory)
         else:
@@ -34,3 +37,28 @@ def toggle(self):
         self._running = self._device.is_active
 
 
+class DigitalArduinoDevice:
+    """
+    Digital modulated devices in combination with Arduino boards connected via USB
+    """
+
+    def __init__(self, PORT):
+        """
+        :param PORT:  USB PORT of the arduino board
+        """
+        self._device = serial.Serial(PORT, baudrate=19200)
+        self._running = False
+
+    def turn_on(self):
+        self._device.write(b'1')
+        self._running = True
+
+    def turn_off(self):
+        self._device.write(b'0')
+        self._running = False
+
+    def toggle(self):
+        if self._running:
+            self.turn_off()
+        else:
+            self.turn_on()

utils/VideoAnalyzer.py

@@ -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_dpp.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()