Single sine generator

built_in_tasks/rotation_matrices.py

@@ -34,6 +34,28 @@
 
 exp_rotations = dict(
     none = np.identity(4),
+
+    mirror_x = np.array(
+        [[-1, 0, 0, 0], 
+        [0, 1, 0, 0], 
+        [0, 0, 1, 0], 
+        [0, 0, 0, 1]]
+    ), 
+
+    mirror_z = np.array(
+        [[1, 0, 0, 0], 
+        [0, 1, 0, 0], 
+        [0, 0, -1, 0], 
+        [0, 0, 0, 1]]
+    ), 
+
+    mirror_45 = np.array(
+        [[0, 0, 1, 0], 
+        [0, 1, 0, 0], 
+        [1, 0, 0, 0], 
+        [0, 0, 0, 1]]
+    ),
+
     about_x_90 = np.array(
         [[1, 0, 0, 0], 
         [0, 0, 1, 0], 

built_in_tasks/target_tracking_task.py

@@ -392,7 +392,7 @@ class ScreenTargetTracking(TargetTracking, Window):
     limit1d = traits.Bool(True, desc="Limit cursor movement to 1D")
 
     sequence_generators = [
-        'tracking_target_chain', 'tracking_target_debug', 'tracking_target_training'
+        'tracking_target_chain', 'tracking_target_debug', 'tracking_target_training', 'single_sine_chain', 'circle_chain', 'figure8_chain'
     ]
 
     hidden_traits = ['cursor_color', 'trajectory_color', 'cursor_bounds', 'cursor_radius', 'plant_hide_rate', 'starting_pos']
@@ -575,7 +575,7 @@ def setup_start_wait(self):
             self.trajectory = VirtualSnakeTarget(target_radius=self.trajectory_radius, target_color=target_colors[self.trajectory_color], trajectory=next_trajectory)
         elif self.trajectory_type == '2d':
             self.trajectory = VirtualSnakeTarget(target_radius=self.trajectory_radius, target_color=target_colors[self.trajectory_color], trajectory=self.targs)
-            self.trajectory.update_mask(self.frame_index, self.frame_index+self.lookahead)
+            self.trajectory.update_mask(0, self.lookahead)
         else: # 'none'
             next_trajectory = np.zeros((self.lookahead, 3))
             self.trajectory = VirtualCircularTarget()
@@ -1104,6 +1104,7 @@ def generate_trajectory(primes, base_period, ramp = 0.0):
         normalized_trajectory = trajectory/np.sum(a)
         return normalized_trajectory
 
+
     ### Generator functions ####
     @staticmethod
     def tracking_target_chain(nblocks=1, ntrials=500, time_length=20, ramp=1.5, ramp_down=1.5, num_primes=8, seed=40, sample_rate=120, dimensions = 1, disturbance=True, boundaries=(-10,10,-10,10), decay_rate = None):
@@ -1238,4 +1239,221 @@ def tracking_target_training(nblocks=1, ntrials=2, time_length=5, frequencies =
                 trajectory[:,2] = 5*np.concatenate((sum_of_sins_path[0]*np.ones(buffer_space_bef),sum_of_sins_path,sum_of_sins_path[-1]*np.ones(buffer_space_aft)))
                 pts.append(trajectory)
                 yield idx, pts, disturbance, disturbance_path, None
+                idx += 1
+
+    @staticmethod
+    def single_sine_chain(nblocks=1, ntrials=500, time_length=20, ramp=1.5, ramp_down=1.5, 
+                          ref_y_freq = 0.35, ref_x_freq = 0.5, dis_y_freq = 0.1, dis_x_freq = 0.15, seed=40, 
+                          sample_rate=120, dimensions = 1, disturbance=True, ref_x_phase = 0, ref_y_phase = 0):
+        '''
+        Generates a single sine wave for the target trajectory  
+
+        Parameters
+        ----------
+        nblocks : int
+            The number of tracking trials in the sequence.
+        ntrials : int
+            The number trials in a block
+        time_length : int
+            Lenght of each trial 
+        ramp: float 
+            Ramp up time 
+        ramp_down: float
+            Ramp down time
+        ref_y_freq: float
+            Frequency of the reference trajectory in the y dimension
+        ref_x_freq: float
+            Frequency of the reference trajectory in the x dimension (only used if dimensions = 2)
+        dis_y_freq: float
+            Frequency of the disturbance trajectory in the y dimension
+        dis_x_freq: float
+            Frequency of the disturbance trajectory in the x dimension (only used if dimensions = 2)
+        seed: int
+            The seed for the random generator
+        sample_rate: int
+            The sample rate of the generated trajectories
+        dimensions: int
+            Number of dimensions to generate trajectories for (1 or 2)
+        disturbance: boolean
+            Whether to add disturbance to the cursor 
+        ref_x_phase: float
+            Phase of the reference trajectory in the x dimension 
+        ref_y_phase: float
+            Phase of the reference trajectory in the y dimension
+
+        Returns
+        -------
+        [nblocks*ntrials x 1] array of tuples containing trial indices and [time_length*60 x 3] target coordinates
+        '''
+
+        dt = 1/sample_rate 
+        t = np.arange(0, time_length+ramp+ramp_down, dt)
+        idx = 0
+        N = t.size
+
+        ref_x_phase = np.deg2rad(ref_x_phase)/ (2*np.pi) #convert phase to degrees and then to cycles for calc_sum_sines
+        ref_y_phase = np.deg2rad(ref_y_phase) / (2*np.pi) #convert phase to degrees
+
+        #generate phase shifts for reference and disturbance trajectories
+
+        np.random.seed(seed)
+
+        ref_amp = 1
+        dis_amp = 1
+
+        if dimensions == 2: 
+            phase_shifts = 2*np.pi*np.random.rand(ntrials, 2)
+            phase_dis = phase_shifts*0.8
+        else:
+            phase_shifts = 2*np.pi*np.random.rand(ntrials)
+            phase_dis = phase_shifts*0.8
+
+
+        trials = dict(
+            id=np.arange(ntrials), times=np.tile(t,(ntrials,1)), ref_x=np.zeros((ntrials,N)), dis_x=np.zeros((ntrials,N)),ref_y = np.zeros((ntrials,N)), dis_y = np.zeros((ntrials,N))) 
+
+        for block_id in range(nblocks):  
+
+            for trial_id in range(ntrials):
+                targs = [] 
+                ref_trajectory = np.zeros((int((time_length+ramp+ramp_down)*sample_rate),3))
+                dis_trajectory = ref_trajectory.copy()
+
+
+                if dimensions == 1:
+
+                    ref_phase = ref_y_phase
+                    dis_phase = phase_dis[trial_id]
+
+                    ref_traj, A_ref = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([ref_y_freq]), np.array([ref_amp]), np.array([ref_phase]))
+                    dis_traj, A_dis = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([dis_y_freq]), np.array([dis_amp]), np.array([dis_phase]))
+
+                    ref_trajectory[:,2] = ref_traj/A_ref
+                    dis_trajectory[:,2] = dis_traj/A_dis
+
+                elif dimensions == 2: 
+
+                    dis_x_phase = phase_dis[trial_id][0]
+                    dis_y_phase = phase_dis[trial_id][1]
+
+                    ref_x_traj, A_ref_x = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([ref_x_freq]), np.array([ref_amp]), np.array([ref_x_phase]))
+                    ref_y_traj, A_ref_y = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([ref_y_freq]), np.array([ref_amp]), np.array([ref_y_phase]))
+                    dis_x_traj, A_dis_x = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([dis_x_freq]), np.array([dis_amp]), np.array([dis_x_phase]))
+                    dis_y_traj, A_dis_y = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([dis_y_freq]), np.array([dis_amp]), np.array([dis_y_phase]))
+
+                    ref_trajectory[:,0] = ref_x_traj/A_ref_x
+                    ref_trajectory[:,2] = ref_y_traj/A_ref_y
+                    dis_trajectory[:,0] = dis_x_traj/A_dis_x
+                    dis_trajectory[:,2] = dis_y_traj/A_dis_y
+
+                yield idx, [ref_trajectory], disturbance, dis_trajectory, sample_rate, ramp, ramp_down
+                idx += 1
+
+    @staticmethod
+    def circle_chain(nblocks=1, ntrials=500, time_length=20, ramp=1.5, ramp_down=1.5, 
+                          ref_freq = 0.5,  dis_y_freq = 0.1, dis_x_freq = 0.15, seed=40, 
+                          sample_rate=120, disturbance=True):
+
+        '''
+        Generate circle trajectory for the target trajectory 
+        '''
+
+        dt = 1/sample_rate 
+        t = np.arange(0, time_length+ramp+ramp_down, dt)
+        idx = 0
+        N = t.size
+
+        ref_x_phase = 0 
+        ref_y_phase = 0.25 # a quarter of a cycle out of phase (calc_sum_sines uses cylces not radians)
+
+        #generate phase shifts for reference and disturbance trajectories
+
+        np.random.seed(seed)
+
+        ref_amp = 1
+        dis_amp = 1
+
+        phase_shifts = 2*np.pi*np.random.rand(ntrials, 2)
+        phase_dis = phase_shifts*0.8
+
+
+        trials = dict(
+            id=np.arange(ntrials), times=np.tile(t,(ntrials,1)), ref_x=np.zeros((ntrials,N)), dis_x=np.zeros((ntrials,N)),ref_y = np.zeros((ntrials,N)), dis_y = np.zeros((ntrials,N))) 
+
+        for block_id in range(nblocks):  
+
+            for trial_id in range(ntrials):
+                targs = [] 
+                ref_trajectory = np.zeros((int((time_length+ramp+ramp_down)*sample_rate),3))
+                dis_trajectory = ref_trajectory.copy()
+
+                dis_x_phase = phase_dis[trial_id][0]
+                dis_y_phase = phase_dis[trial_id][1]
+
+                ref_x_traj, A_ref_x = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([ref_freq]), np.array([ref_amp]), np.array([ref_x_phase]))
+                ref_y_traj, A_ref_y = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([ref_freq]), np.array([ref_amp]), np.array([ref_y_phase]))
+                dis_x_traj, A_dis_x = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([dis_x_freq]), np.array([dis_amp]), np.array([dis_x_phase]))
+                dis_y_traj, A_dis_y = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([dis_y_freq]), np.array([dis_amp]), np.array([dis_y_phase]))
+
+                #max_A = np.max(A_ref_x, A_ref_y) 
+                ref_trajectory[:,0] = ref_x_traj/A_ref_x
+                ref_trajectory[:,2] = ref_y_traj/A_ref_y
+                dis_trajectory[:,0] = dis_x_traj/A_dis_x
+                dis_trajectory[:,2] = dis_y_traj/A_dis_y
+
+                print(A_ref_x, A_ref_y)
+                yield idx, [ref_trajectory], disturbance, dis_trajectory, sample_rate, ramp, ramp_down
+                idx += 1
+
+    @staticmethod
+    def figure8_chain(nblocks=1, ntrials=500, time_length=20, ramp=1.5, ramp_down=1.5, 
+                          ref_freq = 0.5,  dis_y_freq = 0.1, dis_x_freq = 0.15, seed=40, 
+                          sample_rate=120, disturbance=True, ref_phase = 0):
+
+        '''
+        Generate figure-8 trajectory for the target trajectory 
+        '''
+
+        dt = 1/sample_rate 
+        t = np.arange(0, time_length+ramp+ramp_down, dt)
+        idx = 0
+        N = t.size
+
+
+        #generate phase shifts for reference and disturbance trajectories
+
+        np.random.seed(seed)
+
+        ref_amp = 1
+        dis_amp = 1
+
+
+        phase_shifts = 2*np.pi*np.random.rand(ntrials, 2)
+        phase_dis = phase_shifts*0.8
+
+
+        trials = dict(
+            id=np.arange(ntrials), times=np.tile(t,(ntrials,1)), ref_x=np.zeros((ntrials,N)), dis_x=np.zeros((ntrials,N)),ref_y = np.zeros((ntrials,N)), dis_y = np.zeros((ntrials,N))) 
+
+        for block_id in range(nblocks):  
+
+            for trial_id in range(ntrials):
+                targs = [] 
+                ref_trajectory = np.zeros((int((time_length+ramp+ramp_down)*sample_rate),3))
+                dis_trajectory = ref_trajectory.copy()
+
+                dis_x_phase = phase_dis[trial_id][0]
+                dis_y_phase = phase_dis[trial_id][1]
+
+                ref_x_traj, A_ref_x = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([2*ref_freq]), np.array([ref_amp]), np.array([ref_phase]))
+                ref_y_traj, A_ref_y = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([ref_freq]), np.array([ref_amp]), np.array([ref_phase]))
+                dis_x_traj, A_dis_x = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([dis_x_freq]), np.array([dis_amp]), np.array([dis_x_phase]))
+                dis_y_traj, A_dis_y = ScreenTargetTracking.calc_sum_of_sines_ramp(t, ramp, ramp_down, np.array([dis_y_freq]), np.array([dis_amp]), np.array([dis_y_phase]))
+
+                ref_trajectory[:,0] = ref_x_traj/A_ref_x
+                ref_trajectory[:,2] = ref_y_traj/A_ref_y
+                dis_trajectory[:,0] = dis_x_traj/A_dis_x
+                dis_trajectory[:,2] = dis_y_traj/A_dis_y
+
+                yield idx, [ref_trajectory], disturbance, dis_trajectory, sample_rate, ramp, ramp_down
                 idx += 1

tests/test_tasks.py

@@ -79,11 +79,12 @@ def test_example_task(self):
     def test_tracking(self):
         print("Running tracking task test")
         seq = TrackingTask.tracking_target_chain(nblocks=1, ntrials=2, time_length=5, ramp=1, ramp_down=1, 
-                                                 num_primes=8, seed=42, sample_rate=60, 
+                                                 num_primes=2, seed=42, sample_rate=60, 
                                                  disturbance=False, boundaries=(-10,10,-10,10))
         exp = init_exp(TrackingTask, [HideLeftTrajectory, MouseControl, Window2D], seq, window_size=(1000,800), fullscreen=False,
                        lookahead_time=1, screen_half_height=10)
         exp.rotation = 'xzy'
+        exp.exp_rotation = 'mirror_45'
         exp.trajectory_type = '1d'
         exp.trajectory_amplitude = 5
         exp.trajectory_radius = 0.2
@@ -92,13 +93,56 @@ def test_tracking(self):
     @unittest.skip("")
     def test_tracking_2d(self):
         print("Running tracking task test")
-        seq = TrackingTask.tracking_target_chain(nblocks=1, ntrials=2, time_length=20, ramp=1, ramp_down=1, 
-                                                 num_primes=10, seed=42, sample_rate=60, dimensions=2, 
-                                                 disturbance=True, boundaries=(-10,10,-10,10), decay_rate = 0.1)
-        exp = init_exp(TrackingTask, [Window2D, MouseControl], seq, window_size=(1000,800), fullscreen=False, 
+        seq = TrackingTask.tracking_target_chain(nblocks=1, ntrials=20, time_length=20, ramp=1, ramp_down=1, 
+                                                 num_primes=4, seed=42, sample_rate=60, dimensions=2, 
+                                                 disturbance=True, boundaries=(-10,10,-10,10), decay_rate = None)
+        exp = init_exp(TrackingTask, [Window2D, MouseControl], seq, window_size=(1000,800), fullscreen=False, tracking_out_time = 10, 
                        limit1d=False, trajectory_amplitude=5, lookahead_time=1)
+        #exp.stereo_mode = 'projection'
+        exp.rotation = 'xzy'
+        exp.trajectory_type = '2d'
+        exp.run()
+
+    @unittest.skip("")
+    def test_sine_trajectory(self):
+        print("Running tracking task test")
+        seq = TrackingTask.single_sine_chain(nblocks=1, ntrials=2, time_length=20, ramp=1, ramp_down=0, 
+                          ref_y_freq = 0.5, ref_x_freq = 0.5, dis_y_freq = 0.1, dis_x_freq = 0.15, seed=40, 
+                          sample_rate=60, dimensions = 2, disturbance=False, ref_x_phase = 0, ref_y_phase = 270)
+        exp = init_exp(TrackingTask, [Window2D, MouseControl], seq, window_size=(1000,800), fullscreen=False, 
+                       limit1d=False, trajectory_amplitude=5, lookahead_time=20)
+        exp.stereo_mode = 'projection'
+        exp.rotation = 'xzy'
+        #exp.exp_rotation = 'mirror_z'
+        exp.trajectory_type = '2d'
+        exp.run()
+
+    @unittest.skip("")
+    def test_circle_trajectory(self):
+        print("Running tracking task test")
+        seq = TrackingTask.circle_chain(nblocks=1, ntrials=2, time_length=20, ramp=0, ramp_down=0, 
+                          ref_freq = 0.5,  dis_y_freq = 0, dis_x_freq = 0, seed=40, 
+                          sample_rate=60, disturbance=False)
+        exp = init_exp(TrackingTask, [Window2D, MouseControl], seq, window_size=(1000,800), fullscreen=False, 
+                       limit1d=False, trajectory_amplitude=5, lookahead_time=20)
         exp.stereo_mode = 'projection'
         exp.rotation = 'xzy'
+        exp.exp_rotation = 'mirror_z'
+        exp.trajectory_type = '2d'
+        exp.run()
+
+
+    @unittest.skip("")
+    def test_figure8_trajectory(self):
+        print("Running tracking task test")
+        seq = TrackingTask.figure8_chain(nblocks=1, ntrials=2, time_length=20, ramp=0, ramp_down=0, 
+                          ref_freq = 0.5,  dis_y_freq = 0.1, dis_x_freq = 0.15, seed=40, 
+                          sample_rate=60, disturbance=True, ref_phase = 0)
+        exp = init_exp(TrackingTask, [Window2D, MouseControl], seq, window_size=(1000,800), fullscreen=False, 
+                       limit1d=False, trajectory_amplitude=6, lookahead_time=20)
+        exp.stereo_mode = 'projection'
+        exp.rotation = 'xzy'
+        #exp.exp_rotation = 'mirror_z'
         exp.trajectory_type = '2d'
         exp.run()
 
@@ -213,6 +257,37 @@ def test_tracking_2d(self):
         plt.tight_layout()
         plt.show()
 
+    @unittest.skip("")
+    def test_single_sine(self):
+        seq = TrackingTask.single_sine_chain(nblocks=1, ntrials=2, time_length=20, base_period = 20, ramp=1, ramp_down=0, 
+                          ref_y_freq = 0.35, ref_x_freq = 0.5, dis_y_freq = 0.85, dis_x_freq = 0.15, seed=40, 
+                          sample_rate=60, dimensions = 1, disturbance=True)
+        trajectories = [t[1][0] for t in seq] # pulls out trajectory. Can use t[3] to get disturbance array
+        print("Test-------")
+        print(np.shape(trajectories))
+        print("Test-------")
+        fig, axs = plt.subplots(2,1, figsize=(10,8))
+        for idx, trial in enumerate(trajectories): 
+            ax = axs[idx] 
+            trialx = np.fft.fft(trial[60:,0]) #ignore ramp period for FFT
+            trial_length = np.shape(trialx)[0]
+            freq = np.fft.fftfreq(trial_length, d=1./60)
+            non_neg_freq = freq[freq >= 0] #get positive frequencies 
+            non_neg_x = trialx[freq >= 0] / complex(trial_length, 0) #normalize 
+            non_neg_x[1:] = 2*non_neg_x[1:] #account for negative frequencies
+            trialy = np.fft.fft(trial[60:,2]) #ignore ramp period for FFT
+            non_neg_y = trialy[freq >= 0] / complex(trial_length, 0) #normalize 
+            non_neg_y[1:] = 2*non_neg_y[1:] #account for negative frequencies
+            ax.plot(non_neg_freq, np.abs(non_neg_x), 'o-', label = 'X')
+            ax.plot(non_neg_freq, np.abs(non_neg_y), 'o-', label = 'Y')
+            ax.set_title(f'Trial {idx}')
+            ax.set_xlim(0, 3)
+            ax.set_xlabel('Frequency (Hz)')
+        plt.legend()
+        plt.tight_layout()
+        plt.show()
+
+
 class TestYouTube(unittest.TestCase):
 
     @unittest.skip("")