Hi, I’ll provide you two codes i.e DQN and Evader. I want you to make the following modifications: I got this termination condition in place if np.linalg.norm( np.array(self.final_destination) - np.array(self.current_location)[:2]) < 0.5. If this condition satisfies then the car speed will become zero and it rests. So, for a DQN to learn what’s happening in the env I want you to implement this steps: 1. Update the scan_callback function such that for every single episode. You’ve implement no of episodes and max steps or time for each episode depending on our env since it’s somewhat a big grid with atleast 30 obstacles. the car moves from origin to the final point minimizing the following np.linalg.norm( np.array(self.final_destination) - np.array(self.current_location)[:2]). If it moves farther away from the point decrease the reward and if it moves closer to the final point increase the reward. And also if it hits an obstacle decrease the reward by some value and if avoids obstacles increase the reward by some value. Finally , by the time car reaches the final point, then agent should save the updated weights and rest. In test Mode, Implement the steps to load the saved weights and the car should choose the actions based on learned policy. The agent can learn on steering angles and details about obstacles, when to choose the best action etc. Based on this Modify the DQN which I’ll provide. First, I’ll provide Evader and later I provide DQN. Please read and understand each and every point mentioned carefully. Then proceed with the modifications. I’m always here to help you. Feel free to make modifications. Provide the updated modified end to end evader code. Let’s start! Here is the code for Evader Node: #! /usr/bin/env python3


from sensor_msgs.msg import LaserScan
from ackermann_msgs.msg import AckermannDrive
from nav_msgs.msg import Odometry
from DQN import DQNAgentPytorch

FORWARD = 0
LEFT = 1
RIGHT = 2
REVERSE = 3
ACTION_SPACE = [FORWARD, LEFT, RIGHT, REVERSE]
class EvaderNode:
def init(self, agent):

rospy.init_node(‘evader_node’, anonymous=True)

rospy.Subscriber(‘car_1/scan’, LaserScan, self.scan_callback)
rospy.Subscriber(‘car_1/base/odom’, Odometry, self.odom_callback)

self.drive_pub = rospy.Publisher(‘car_1/command’, AckermannDrive, queue_size=10)

self.init_location = None # Set the initial location in odom_callback when available
self.current_location = None # Will be updated in update_current_location method

self.distance_threshold = 2 # meters
self.angle_range = [(-150, -30), (30, 150)]

self.agent = agent
self.collision_threshold = 0.2

self.is_training = True # Flag to indicate if the agent is being trained

self.init_odom_received = False
self.latest_odom_msg = None

self.reached_destination = False

self.final_destination = (10, 10) # Set the final destination here


def process_scan(self, scan):
# Process the laser scan data to get the state input for the DQN
state = np.array(scan.ranges)[::10] # Downsample the scan data for faster processing
# return np.concatenate((state, self.current_location))
return state

def check_collision(self, scan):
for i in range(len(scan.ranges)):
if scan.ranges[i] < self.collision_threshold:
return True
return False

def action_to_drive_msg(self, action):
drive_msg = AckermannDrive()
if action == FORWARD:
drive_msg.speed = 2.0
drive_msg.steering_angle = 0.0
elif action == LEFT:
drive_msg.speed = 2.0
drive_msg.steering_angle = float(random.uniform(math.pi / 4, math.pi / 2))
elif action == RIGHT:
drive_msg.speed = 2.0
drive_msg.steering_angle = float(random.uniform(-math.pi / 2, -math.pi / 4))
# elif action == REVERSE:
# drive_msg.speed = -2.0
# drive_msg.steering_angle = 0.0
return drive_msg

def reverse_car(self, reverse_distance):
reverse_drive_msg = AckermannDrive()
reverse_drive_msg.speed = -2.0
reverse_drive_msg.steering_angle = 0.0
reverse_duration = reverse_distance / (-reverse_drive_msg.speed)

start_time = time.time()
while time.time() - start_time < reverse_duration:
self.drive_pub.publish(reverse_drive_msg)
rospy.sleep(0.1)

reverse_drive_msg.speed = 0.0
self.drive_pub.publish(reverse_drive_msg)

def initialize_final_destination(self):

init_x, init_y = self.current_location[:2]
dest_x, dest_y = self.final_destination

angle_to_dest = math.atan2(dest_y - init_y, dest_x - init_x)

return angle_to_dest


def scan_callback(self, scan):
state = self.process_scan(scan)
action = self.agent.choose_action(state, test=not self.is_training)

collision_detected = self.check_collision(scan)

# steer_angle = None
if collision_detected and not self.reached_destination:
print(‘Collision Detected’)
self.reverse_car(0.2) # Set the reverse distance here

#float(random.uniform(0, math.pi/2))

action = self.agent.choose_action(state, test=not self.is_training)

# steer_angle = self.initialize_final_destination()


drive_msg = self.action_to_drive_msg(action)

# if steer_angle is not None:

# drive_msg.steering_angle -= steer_angle

# if self.latest_odom_msg is not None:
# self.update_current_location(self.latest_odom_msg)





# if not self.is_training:
# if np.linalg.norm(
# np.array(self.final_destination) - np.array(self.current_location)) < self.distance_threshold:
# self.current_location = copy.deepcopy(self.init_location)
# self.drive_pub.publish(drive_msg)
# return

# min_distance = min(state)
# reward_collision = collision_detected and not self.reached_destination
# reward = self.agent.calculate_reward(state, action, min_distance, reward_collision)

done = False # set this flag to True if a termination condition is met
# self.agent.add_experience(state, action, reward, state, done)
# self.agent.train(32) # Set the batch size here

self.drive_pub.publish(drive_msg)

# Set the termination condition for training

if np.linalg.norm( np.array(self.final_destination) - np.array(self.current_location)[:2]) < 0.5:

if not self.reached_destination:
print(‘Reached final destination!’)
self.reached_destination = True
# drive_msg = AckermannDrive()
drive_msg.speed = 0.0
self.drive_pub.publish(drive_msg)
# self.align_to_initial_point()
# else:
# print(‘Reached the initial point!’)
# self.reached_destination = False

# if self.is_training:
# self.is_training = False

# self.agent.save_weights(‘model_weights.pth’)
# self.agent.load_weights(‘model_weights.pth’)

def update_current_location(self, odom_msg):
position = odom_msg.pose.pose.position
orientation = odom_msg.pose.pose.orientation
euler_angles = tf.transformations.euler_from_quaternion(
(orientation.x, orientation.y, orientation.z, orientation.w))

self.current_location = ( position.x, position.y, euler_angles[2]) # (x, y, yaw)

print(“Current Location of the Car:”, self.current_location[:2])

def odom_callback(self, odom_msg):

if not self.init_odom_received:
self.init_location = (odom_msg.pose.pose.position.x,
odom_msg.pose.pose.position.y)
print(f"Intial Location of the Car: {self.init_location}")
self.init_odom_received = True

self.update_current_location(odom_msg)
self.latest_odom_msg = odom_msg


if name == ‘main’:
try:
# Initialize the DQNAgentPytorch instance with the necessary environment and parameters
action_space = ACTION_SPACE

observation_space = np.zeros((360,)) # 360 laser scan equidistant points + (x, y) position + grid

agent = DQNAgentPytorch(action_space, observation_space)

# Initialize the EvaderNode instance with the DQNAgent
node = EvaderNode(agent)

# Wait for initial odometry information
while not node.init_odom_received:
rospy.sleep(0.1)

rospy.spin()

except rospy.ROSInterruptException:
pass