simulation_ws/src/rl-agent/markov/environments/mars_env.py [135:254]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ''' DO NOT EDIT - Function called by rl_coach to instruct the agent to take an action ''' def step(self, action): # initialize rewards, next_state, done self.reward = None self.done = False self.next_state = None steering = float(action[0]) throttle = float(action[1]) self.steps += 1 self.send_action(steering, throttle) time.sleep(SLEEP_BETWEEN_ACTION_AND_REWARD_CALCULATION_TIME_IN_SECOND) self.call_reward_function(action) info = {} # additional data, not to be used for training return self.next_state, self.reward, self.done, info ''' DO NOT EDIT - Function called at the conclusion of each episode to reset episodic values ''' def reset(self): print('Total Episodic Reward=%.2f' % self.reward_in_episode, 'Total Episodic Steps=%.2f' % self.steps) self.send_reward_to_cloudwatch(self.reward_in_episode) # Reset global episodic values self.reward = None self.done = False self.next_state = None self.ranges= None self.send_action(0, 0) # set the throttle to 0 self.rover_reset() self.call_reward_function([0, 0]) return self.next_state ''' DO NOT EDIT - Function called to send the agent's chosen action to the simulator (Gazebo) ''' def send_action(self, steering, throttle): speed = Twist() speed.linear.x = throttle speed.angular.z = steering self.ack_publisher.publish(speed) ''' DO NOT EDIT - Function to reset the rover to the starting point in the world ''' def rover_reset(self): # Reset Rover-related Episodic variables rospy.wait_for_service('gazebo/set_model_state') self.x = INITIAL_POS_X self.y = INITIAL_POS_Y # Put the Rover at the initial position model_state = ModelState() model_state.pose.position.x = INITIAL_POS_X model_state.pose.position.y = INITIAL_POS_Y model_state.pose.position.z = INITIAL_POS_Z model_state.pose.orientation.x = INITIAL_ORIENT_X model_state.pose.orientation.y = INITIAL_ORIENT_Y model_state.pose.orientation.z = INITIAL_ORIENT_Z model_state.pose.orientation.w = INITIAL_ORIENT_W model_state.twist.linear.x = 0 model_state.twist.linear.y = 0 model_state.twist.linear.z = 0 model_state.twist.angular.x = 0 model_state.twist.angular.y = 0 model_state.twist.angular.z = 0 model_state.model_name = 'rover' # List of joints to reset (this is all of them) joint_names_list = ["rocker_left_corner_lb", "rocker_right_corner_rb", "body_rocker_left", "body_rocker_right", "rocker_right_bogie_right", "rocker_left_bogie_left", "bogie_left_corner_lf", "bogie_right_corner_rf", "corner_lf_wheel_lf", "imu_wheel_lf_joint", "bogie_left_wheel_lm", "imu_wheel_lm_joint", "corner_lb_wheel_lb", "imu_wheel_lb_joint", "corner_rf_wheel_rf", "imu_wheel_rf_joint", "bogie_right_wheel_rm", "imu_wheel_rm_joint", "corner_rb_wheel_rb", "imu_wheel_rb_joint"] # Angle to reset joints to joint_positions_list = [0 for _ in range(len(joint_names_list))] self.gazebo_model_state_service(model_state) self.gazebo_model_configuration_service(model_name='rover', urdf_param_name='rover_description', joint_names=joint_names_list, joint_positions=joint_positions_list) self.last_collision_threshold = sys.maxsize self.last_position_x = self.x self.last_position_y = self.y time.sleep(SLEEP_AFTER_RESET_TIME_IN_SECOND) self.distance_travelled = 0 self.current_distance_to_checkpoint = INITIAL_DISTANCE_TO_CHECKPOINT self.steps = 0 self.reward_in_episode = 0 self.collision = False self.closer_to_checkpoint = False self.power_supply_range = MAX_STEPS - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - simulation_ws/src/rl-agent/markov/environments/training_env.py [136:255]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ''' DO NOT EDIT - Function called by rl_coach to instruct the agent to take an action ''' def step(self, action): # initialize rewards, next_state, done self.reward = None self.done = False self.next_state = None steering = float(action[0]) throttle = float(action[1]) self.steps += 1 self.send_action(steering, throttle) time.sleep(SLEEP_BETWEEN_ACTION_AND_REWARD_CALCULATION_TIME_IN_SECOND) self.call_reward_function(action) info = {} # additional data, not to be used for training #return dict(xy=np.array([self.next_state[0]]), observation=np.array([self.next_state[1]]), theta=np.array([self.next_state[2]]) ), self.reward, self.done, {} return self.next_state, self.reward, self.done, info ''' DO NOT EDIT - Function called at the conclusion of each episode to reset episodic values ''' def reset(self): print('Total Episodic Reward=%.2f' % self.reward_in_episode, 'Total Episodic Steps=%.2f' % self.steps) self.send_reward_to_cloudwatch(self.reward_in_episode) # Reset global episodic values self.reward = None self.done = False self.next_state = None self.ranges= None self.send_action(0, 0) # set the throttle to 0 self.rover_reset() self.call_reward_function([0, 0]) return self.next_state ''' DO NOT EDIT - Function called to send the agent's chosen action to the simulator (Gazebo) ''' def send_action(self, steering, throttle): speed = Twist() speed.linear.x = throttle speed.angular.z = steering self.ack_publisher.publish(speed) ''' DO NOT EDIT - Function to reset the rover to the starting point in the world ''' def rover_reset(self): # Reset Rover-related Episodic variables rospy.wait_for_service('gazebo/set_model_state') self.x = INITIAL_POS_X self.y = INITIAL_POS_Y # Put the Rover at the initial position model_state = ModelState() model_state.pose.position.x = INITIAL_POS_X model_state.pose.position.y = INITIAL_POS_Y model_state.pose.position.z = INITIAL_POS_Z model_state.pose.orientation.x = INITIAL_ORIENT_X model_state.pose.orientation.y = INITIAL_ORIENT_Y model_state.pose.orientation.z = INITIAL_ORIENT_Z model_state.pose.orientation.w = INITIAL_ORIENT_W model_state.twist.linear.x = 0 model_state.twist.linear.y = 0 model_state.twist.linear.z = 0 model_state.twist.angular.x = 0 model_state.twist.angular.y = 0 model_state.twist.angular.z = 0 model_state.model_name = 'rover' # List of joints to reset (this is all of them) joint_names_list = ["rocker_left_corner_lb", "rocker_right_corner_rb", "body_rocker_left", "body_rocker_right", "rocker_right_bogie_right", "rocker_left_bogie_left", "bogie_left_corner_lf", "bogie_right_corner_rf", "corner_lf_wheel_lf", "imu_wheel_lf_joint", "bogie_left_wheel_lm", "imu_wheel_lm_joint", "corner_lb_wheel_lb", "imu_wheel_lb_joint", "corner_rf_wheel_rf", "imu_wheel_rf_joint", "bogie_right_wheel_rm", "imu_wheel_rm_joint", "corner_rb_wheel_rb", "imu_wheel_rb_joint"] # Angle to reset joints to joint_positions_list = [0 for _ in range(len(joint_names_list))] self.gazebo_model_state_service(model_state) self.gazebo_model_configuration_service(model_name='rover', urdf_param_name='rover_description', joint_names=joint_names_list, joint_positions=joint_positions_list) self.last_collision_threshold = sys.maxsize self.last_position_x = self.x self.last_position_y = self.y time.sleep(SLEEP_AFTER_RESET_TIME_IN_SECOND) self.distance_travelled = 0 self.current_distance_to_checkpoint = INITIAL_DISTANCE_TO_CHECKPOINT self.steps = 0 self.reward_in_episode = 0 self.collision = False self.closer_to_checkpoint = False self.power_supply_range = MAX_STEPS - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -