diff --git a/built_in_tasks/rotation_matrices.py b/built_in_tasks/rotation_matrices.py index 0fb31e05..0a4636c4 100644 --- a/built_in_tasks/rotation_matrices.py +++ b/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], diff --git a/built_in_tasks/target_tracking_task.py b/built_in_tasks/target_tracking_task.py index b58fdf78..2a22c3ee 100644 --- a/built_in_tasks/target_tracking_task.py +++ b/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 \ No newline at end of file diff --git a/tests/test_tasks.py b/tests/test_tasks.py index 1612a91e..65acf001 100644 --- a/tests/test_tasks.py +++ b/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("")