new SocialInteractionExperiment

JensBlack · JensBlack · commit ea44e8885844 · 2021-02-22T13:56:22.000+01:00
diff --git a/DeepLabStream.py b/DeepLabStream.py
@@ -19,10 +19,10 @@
 
 from utils.generic import VideoManager, WebCamManager, GenericManager
 from utils.configloader import RESOLUTION, FRAMERATE, OUT_DIR, MODEL_NAME, MULTI_CAM, STACK_FRAMES, \
-    ANIMALS_NUMBER, STREAMS, STREAMING_SOURCE, MODEL_ORIGIN, CROP, CROP_X, CROP_Y
+    ANIMALS_NUMBER, FLATTEN_MA, STREAMS, STREAMING_SOURCE, MODEL_ORIGIN, CROP, CROP_X, CROP_Y
 from utils.plotter import plot_bodyparts,plot_metadata_frame
-from utils.poser import load_deeplabcut,load_dpk,load_dlc_live,get_pose,calculate_skeletons, \
-    find_local_peaks_new,get_ma_pose,load_sleap
+from utils.poser import load_deeplabcut,load_dpk,load_dlc_live,load_sleap, get_pose,calculate_skeletons, \
+    find_local_peaks_new,get_ma_pose
 
 def create_video_files(directory, devices, resolution, framerate, codec):
     """
@@ -439,9 +439,7 @@ def get_analysed_frames(self) -> tuple:
                         self._experiment_running = False
 
                     if self._experiment_running and not self._experiment.experiment_finished:
-                        #TODO: Update to work for multiple animal and single animal experiments
-                        #Shift responsibility to experiments
-                        if ANIMALS_NUMBER > 1:
+                        if ANIMALS_NUMBER > 1 and not FLATTEN_MA:
                             self._experiment.check_skeleton(analysed_image,skeletons)
                         else:
                             for skeleton in skeletons:
diff --git a/experiments/custom/experiments.py b/experiments/custom/experiments.py
@@ -10,33 +10,41 @@
 import time
 from functools import partial
 from collections import Counter
-from experiments.custom.stimulus_process import ClassicProtocolProcess, SimpleProtocolProcess,Timer, ExampleProtocolProcess
-from experiments.custom.triggers import ScreenTrigger, RegionTrigger, OutsideTrigger, DirectionTrigger, SpeedTrigger
+from experiments.custom.stimulus_process import ClassicProtocolProcess, SimpleProtocolProcess,Timer\
+    , ExampleProtocolProcess
+from experiments.custom.triggers import ScreenTrigger, RegionTrigger, OutsideTrigger, DirectionTrigger\
+    , SpeedTrigger, SocialInteractionTrigger
 from utils.plotter import plot_triggers_response
 from utils.analysis import angle_between_vectors
 from experiments.custom.stimulation import show_visual_stim_img,laser_switch
 
 
-class ExampleSocialExperiment:
+"""Social or multiple animal experiments in combination with SLEAP or non-flattened maDLC pose estimation"""
+
+class ExampleSocialInteractionExperiment:
     """
+    In this experiment the skeleton/instance of each animal will be considers for the trigger,
+    any animal can trigger the stimulation (the first one to result in TRUE).
+
     Simple class to contain all of the experiment properties
     Uses multiprocess to ensure the best possible performance and
         to showcase that it is possible to work with any type of equipment, even timer-dependent
     """
     def __init__(self):
         self.experiment_finished = False
         self._process = ExampleProtocolProcess()
-        self._green_point = (313, 552)
-        self._radius = 80
+        self._proximity_threshold = 30
+        self._min_animals = 2
         self._event = None
         self._current_trial = None
+        self._max_reps = 999
         self._trial_count = {trial: 0 for trial in self._trials}
         self._trial_timers = {trial: Timer(10) for trial in self._trials}
         self._exp_timer = Timer(600)
 
     def check_skeleton(self, frame, skeletons):
         """
-        Checking each passed animal skeleton for a pre-defined set of conditions
+        Checking passed animal skeletons for a pre-defined set of conditions
         Outputting the visual representation, if exist
         Advancing trials according to inherent logic of an experiment
         :param frame: frame, on which animal skeleton was found
@@ -45,40 +53,161 @@ def check_skeleton(self, frame, skeletons):
         self.check_exp_timer()  # checking if experiment is still on
         for trial in self._trial_count:
             # checking if any trial hit a predefined cap
-            if self._trial_count[trial] >= 10:
+            if self._trial_count[trial] >= self._max_reps:
                 self.stop_experiment()
 
         if not self.experiment_finished:
             result, response = False, None
+            #checking if enough animals were detected
+            if len(skeletons) >= self._min_animals:
+                for trial in self._trials:
+                    # check if social interaction trigger is true
+                    result, response = self._trials[trial]['trigger'](skeletons=skeletons)
+                    plot_triggers_response(frame, response)
+                    if result:
+                        if self._current_trial is None:
+                            if not self._trial_timers[trial].check_timer():
+                                self._current_trial = trial
+                                self._trial_timers[trial].reset()
+                                self._trial_count[trial] += 1
+                                print(trial, self._trial_count[trial])
+                    else:
+                        if self._current_trial == trial:
+                            self._current_trial = None
+                            self._trial_timers[trial].start()
+
+                self._process.set_trial(self._current_trial)
+            else:
+                pass
+            return result, response
+
+    @property
+    def _trials(self):
+        """
+        Defining the trials
+        """
+        identification_dict = dict(active={'animal': 1
+                                            , 'bp': ['bp0']
+                                           }
+                                ,passive = {'animal': 0
+                                            , 'bp': ['bp2']
+                                                }
+                                )
+
+        interaction_trigger = SocialInteractionTrigger(threshold= self._proximity_threshold
+                                                       , identification_dict = identification_dict
+                                                       , interaction_type = 'proximity'
+                                                       , debug=True
+                                                       )
+
+        trials = {'DLStream_test': dict(trigger=interaction_trigger.check_skeleton,
+                                             count=0)}
+        return trials
+
+    def check_exp_timer(self):
+        """
+        Checking the experiment timer
+        """
+        if not self._exp_timer.check_timer():
+            print("Experiment is finished")
+            print("Time ran out.")
+            self.stop_experiment()
+
+    def start_experiment(self):
+        """
+        Start the experiment
+        """
+        self._process.start()
+        if not self.experiment_finished:
+            self._exp_timer.start()
+
+    def stop_experiment(self):
+        """
+        Stop the experiment and reset the timer
+        """
+        self.experiment_finished = True
+        print('Experiment completed!')
+        self._exp_timer.reset()
+        # don't forget to end the process!
+        self._process.end()
+
+    def get_trial(self):
+        """
+        Check which trial is going on right now
+        """
+        return self._current_trial
+
+
+class ExampleMultipleAnimalExperiment:
+    """
+    In this experiment the skeleton/instance of each animal will be considers for the trigger,
+    any animal can trigger the stimulation (the first one to result in TRUE).
+
+    Simple class to contain all of the experiment properties
+    Uses multiprocess to ensure the best possible performance and
+        to showcase that it is possible to work with any type of equipment, even timer-dependent
+    """
+
+    def __init__(self):
+        self.experiment_finished = False
+        self._process = ExampleProtocolProcess()
+        self._green_point = (550, 163)
+        self._radius = 40
+        self._dist_threshold = 80
+        self._event = None
+        self._current_trial = None
+        self._max_reps = 10
+        self._trial_count = {trial: 0 for trial in self._trials}
+        self._trial_timers = {trial: Timer(10) for trial in self._trials}
+        self._exp_timer = Timer(600)
+
+    def check_skeleton(self,frame,skeletons):
+        """
+        Checking each passed animal skeleton for a pre-defined set of conditions
+        Outputting the visual representation, if exist
+        Advancing trials according to inherent logic of an experiment
+        :param frame: frame, on which animal skeleton was found
+        :param skeletons: skeletons, consisting of multiple joints of an animal
+        """
+        self.check_exp_timer()  # checking if experiment is still on
+        for trial in self._trial_count:
+            # checking if any trial hit a predefined cap
+            if self._trial_count[trial] >= self._max_reps:
+                self.stop_experiment()
+
+        if not self.experiment_finished:
+            result,response = False,None
             for trial in self._trials:
                 # check for all trials if condition is met
                 result_list = []
                 for skeleton in skeletons:
-                    result, response = self._trials[trial]['trigger'](skeleton=skeleton)
+                    # checking each skeleton for trigger success
+                    result,response = self._trials[trial]['trigger'](skeleton=skeleton)
+                    # if one of the triggers is true, break the loop and continue (the first True)
                     if result:
                         break
-                plot_triggers_response(frame, response)
+                plot_triggers_response(frame,response)
                 if result:
                     if self._current_trial is None:
                         if not self._trial_timers[trial].check_timer():
                             self._current_trial = trial
                             self._trial_timers[trial].reset()
                             self._trial_count[trial] += 1
-                            print(trial, self._trial_count[trial])
+                            print(trial,self._trial_count[trial])
                 else:
                     if self._current_trial == trial:
                         self._current_trial = None
                         self._trial_timers[trial].start()
 
             self._process.set_trial(self._current_trial)
-            return result, response
+            return result,response
 
     @property
     def _trials(self):
         """
         Defining the trials
         """
-        green_roi = RegionTrigger('circle', self._green_point, self._radius * 2 + 7.5, 'bp1')
+        green_roi = RegionTrigger('circle',self._green_point,self._radius * 2 + 7.5,'bp1')
         trials = {'Greenbar_whiteback': dict(trigger=green_roi.check_skeleton,
                                              count=0)}
         return trials
@@ -117,6 +246,10 @@ def get_trial(self):
         return self._current_trial
 
 
+"""Single animal or flattened multi animal pose estimation experiments (e.g. different fur color) 
+or by use of the FLATTEN_MA parameter in advanced settings"""
+
+
 class ExampleExperiment:
     """
     Simple class to contain all of the experiment properties
diff --git a/experiments/custom/stimulation.py b/experiments/custom/stimulation.py
@@ -51,7 +51,8 @@ def show_visual_stim_img(type='background', name='vistim'):
     # Show image when called
     visual = {'background': dict(path=r"./experiments/src/whiteback_1920_1080.png"),
               'Greenbar_whiteback': dict(path=r"./experiments/src/greenbar_whiteback_1920_1080.png"),
-              'Bluebar_whiteback': dict(path=r"./experiments/src/bluebar_whiteback_1920_1080.png")}
+              'Bluebar_whiteback': dict(path=r"./experiments/src/bluebar_whiteback_1920_1080.png"),
+              'DLStream_test': dict(path=r"./experiments/src/stuckinaloop.jpg")}
     # load image unchanged (-1), greyscale (0) or color (1)
     img = cv2.imread(visual[type]['path'], -1)
     converted_image = np.uint8(img)
diff --git a/experiments/custom/stimulus_process.py b/experiments/custom/stimulus_process.py
@@ -73,10 +73,10 @@ def example_protocol_run(condition_q: mp.Queue):
         if condition_q.full():
             current_trial = condition_q.get()
         if current_trial is not None:
-            show_visual_stim_img(type=current_trial, name='inside')
+            show_visual_stim_img(type=current_trial, name='DlStream')
             #dmod_device.toggle()
         else:
-            show_visual_stim_img(name='inside')
+            show_visual_stim_img(name='DlStream')
             #dmod_device.turn_off()
 
         if cv2.waitKey(1) & 0xFF == ord('q'):
diff --git a/experiments/custom/triggers.py b/experiments/custom/triggers.py
@@ -14,7 +14,106 @@
 
 import time
 
-"""Single posture triggers"""
+"""Multiple Animal, Social Interaction Triggers
+can only be used with multiple animals/skeletons (e.g. with SLEAP)
+Note that currently"""
+
+
+class SocialInteractionTrigger:
+    """Trigger to check if one animal's body part is close to (Proximity trigger) far away from (distance trigger) another animal's body part
+    This trigger can be easily adapted to incorporate more complicated social interaction matching.
+    Note that this Trigger can be used without realtime identity tracking (e.g. with maDLC or DeepPoseKit) but is less effective
+     because no distinction between active/passive interaction partners can be made."""
+
+    def __init__(self, threshold: float
+                 , identification_dict: dict
+                 , interaction_type: str = 'proximity',  debug: bool = False):
+        """
+        Initialising trigger with following parameters:
+        :param float threshold: minimum distance between selected body parts for trigger to activate
+        :param dict identification_dict: nested dictionary that specifies which animal/instance is taken as "active"/acting interaction partner and which is "passive"/recieving.
+                    Identification of animal ('name') has to match DLStream naming parameters for calculate skeleton (default is 'Animal1', 'Animal2', etc.).
+                    Idenification of relevant bodyparts ('bp') has to match Dlstream autonaming (default is 'bp1', 'bp2', etc.) or model specific bodypart naming,
+                    whichever applies. This can be a single body part (e.g. ['nose']) or a list of body parts (e.g. ['nose', 'paw_left', 'paw_right']
+
+                    Example: identification_dict = dict(
+                                                        active = {'animal': 0,
+                                                                   'bp': ['nose']]
+                                                                   }
+                                                        ,passive = {'animal': 1,
+                                                                   'bp': ['center','tail_root']
+                                                                   }
+                                                        )
+
+        :param str interaction_type: Type of interaction ('proximity' or 'distance'). Proximity is checking for distances lower than threshold,
+                while distance is checking for distances higher than threshold. Default: 'proximity'
+        :param debug: Not used in this trigger
+         """
+
+        self._threshold = threshold
+        self._identification_dict = identification_dict
+        #for easier use
+        self._active_animal = self._identification_dict['active']['animal']
+        self._active_bp = self._identification_dict['active']['bp']
+        self._passive_animal = self._identification_dict['passive']['animal']
+        self._passive_bp = self._identification_dict['passive']['bp']
+
+        self._interaction_type = interaction_type
+        self._debug = debug
+
+    def check_skeleton(self, skeletons: dict):
+        """
+        Checking skeletons for trigger
+        :param skeletons: a skeleton dictionary, returned by calculate_skeletons() from poser file
+        :return: response, a tuple of result (bool) and response body
+        Response body is used for plotting and outputting results to trials dataframes
+        """
+
+        results = []
+        for active_bp in self._identification_dict['active']['bp']:
+            active_coords = skeletons[self._active_animal][active_bp]
+            for passive_bp in self._identification_dict['passive']['bp']:
+                passive_coords = skeletons[self._passive_animal][passive_bp]
+                #calculate distance for all combinations
+                distance = calculate_distance(active_coords, passive_coords)
+                temp_result = False
+                if distance >= self._threshold and self._interaction_type == 'distance':
+                        temp_result = True
+                elif distance < self._threshold and self._interaction_type == 'proximity':
+                        temp_result = True
+                else:
+                    pass
+                results.append(temp_result)
+
+        result = any(results)
+
+        color = (0, 255, 0) if result else (0, 0, 255)
+
+        if self._debug:
+            active_point_x, active_point_y = skeletons[self._active_animal][self._active_bp[0]]
+            passive_point_x,  passive_point_y= skeletons[self._passive_animal][self._passive_bp[0]]
+
+            response_body = {'plot': {'text': dict(text=f'{self._interaction_type}: {result}',
+                                                   org= (20 , 20),
+                                                   color= color),
+                                      'line': dict(pt1=(int(active_point_x), int(active_point_y)),
+                                                   pt2=(int(passive_point_x), int(passive_point_y)),
+                                                   color=color),
+                                      }
+                             }
+        else:
+            response_body = {'plot': {'text': dict(text=f'{self._interaction_type}: {result}',
+                                                   org=(20 , 20),
+                                                   color= color)
+                                      }
+                             }
+        response = (result, response_body)
+        return response
+
+
+"""Single posture triggers
+can be used in single animal or multiple animal experiments
+"""
 
 class HeaddirectionROITrigger:
     """Trigger to check if animal is turning its head in a specific angle to a reference point (center of the region of interest)
@@ -390,6 +489,7 @@ def check_skeleton(self, skeleton: dict):
 
         return response
 
+
 class SpeedTrigger:
     """
     Trigger to check if animal is moving above a certain speed
diff --git a/utils/advanced_settings.ini b/utils/advanced_settings.ini
diff --git a/utils/poser.py b/utils/poser.py
