<commit_before>import numpy as np
import collections

"""Base class for controllers. 

Instantiates a controller and forces the user to pass a 
maximum acceleration to the controller. Provides the method
safe_action to ensure that controls are never made that could
cause the system to crash. 

"""
class BaseController:
    def __init__(self, veh_id, controller_params):
        """
        Arguments:
            veh_id {string} -- ID of the vehicle this controller is used for
            controller_params {Dictionary} -- Dictionary that optionally 
            contains 'delay', the delay, and must contain 
            'max_deaccel', the maximum deacceleration as well as all 
            other parameters that dictate the driving behavior. 
        """
        self.d = 0
        self.veh_id = veh_id
        self.controller_params = controller_params
        if not controller_params['delay']:
            self.delay = 0
        else:
            self.delay = controller_params['delay']
        # max deaccel should always be a positive
        self.max_deaccel = np.abs(controller_params['max_deaccel'])
        self.acc_queue = collections.deque() 

    def reset_delay(self, env):
        raise NotImplementedError

    def get_action(self, env):
        """ Returns the acceleration of the controller """
        raise NotImplementedError

    def get_safe_action(self, env, action):
        """ USE THIS INSTEAD OF GET_ACTION for computing the actual controls.
        Checks if the computed acceleration would put us above safe velocity.
        If it would, output the acceleration that would put at to safe velocity. 
        """
        safe_velocity = self.safe_velocity(env)

        #this is not being used?
        this_lane = env.vehicles[self.veh_id]['lane']

        this_vel = env.vehicles[self.veh_id]['speed']
        time_step = env.time_step

        if this_vel + action*time_step > safe_velocity:
            return (safe_velocity - this_vel)/time_step
        else:
            return action

    def safe_velocity(self, env):
        """Finds maximum velocity such that if the lead vehicle breaks
        with max acceleration, we can bring the following vehicle to rest
        at the point at which the headway is zero.
        """
        this_lane = env.vehicles[self.veh_id]['lane']
        lead_id = env.get_leading_car(self.veh_id, this_lane)

        lead_pos = env.get_x_by_id(lead_id)
        lead_vel = env.vehicles[lead_id]['speed']
        lead_length = env.vehicles[lead_id]['length']

        this_pos = env.get_x_by_id(self.veh_id)

        # need to account for the position being reset around the length
        if lead_pos > this_pos: 
            dist = (this_pos + lead_length) - lead_pos
        else:
            loop_length = env.scenario.net_params["length"]
            dist = (this_pos + lead_length) - (lead_pos + loop_length)
        self.last_d = self.d
        d = dist - np.power((lead_vel - self.max_deaccel * env.time_step),2)/(2*self.max_deaccel)
        self.d = d

        if -2*d+self.max_deaccel*self.delay**2 < 0:
            v_safe = 0
        else:<commit_msg>diff failsafe<commit_after>import numpy as np
import collections

"""Base class for controllers. 

Instantiates a controller and forces the user to pass a 
maximum acceleration to the controller. Provides the method
safe_action to ensure that controls are never made that could
cause the system to crash. 

"""
class BaseController:
    def __init__(self, veh_id, controller_params):
        """
        Arguments:
            veh_id {string} -- ID of the vehicle this controller is used for
            controller_params {Dictionary} -- Dictionary that optionally 
            contains 'delay', the delay, and must contain 
            'max_deaccel', the maximum deacceleration as well as all 
            other parameters that dictate the driving behavior. 
        """
        self.d = 0
        self.veh_id = veh_id
        self.controller_params = controller_params
        if not controller_params['delay']:
            self.delay = 0
        else:
            self.delay = controller_params['delay']
        # max deaccel should always be a positive
        self.max_deaccel = np.abs(controller_params['max_deaccel'])
        self.acc_queue = collections.deque() 

    def reset_delay(self, env):
        raise NotImplementedError

    def get_action(self, env):
        """ Returns the acceleration of the controller """
        raise NotImplementedError

    def get_safe_action(self, env, action):
        """ USE THIS INSTEAD OF GET_ACTION for computing the actual controls.
        Checks if the computed acceleration would put us above safe velocity.
        If it would, output the acceleration that would put at to safe velocity. 
        """
        this_lane = env.vehicles[self.veh_id]['lane']
        lead_id = env.get_leading_car(self.veh_id, this_lane)
        lead_pos = env.get_x_by_id(lead_id)
        lead_vel = env.vehicles[lead_id]['speed']
        lead_length = env.vehicles[lead_id]['length']

        this_pos = env.get_x_by_id(self.veh_id)
        this_vel = env.vehicles[self.veh_id]['speed']
        time_step = env.time_step


        h = (lead_pos - lead_length - this_pos) % env.scenario.length

        # need to account for the position being reset around the length

        if h < 1:
            print(self.veh_id, 'distance too small: dist = ', h)
            return -this_vel / time_step
        else:
            return action


        # safe_velocity = self.safe_velocity(env)

        # #this is not being used?
        # this_lane = env.vehicles[self.veh_id]['lane']

        # this_vel = env.vehicles[self.veh_id]['speed']
        # time_step = env.time_step

        # if this_vel + action*time_step > safe_velocity:
        #     return (safe_velocity - this_vel)/time_step
        # else:
        #     return action

    def safe_velocity(self, env):
        """Finds maximum velocity such that if the lead vehicle breaks
        with max acceleration, we can bring the following vehicle to rest
        at the point at which the headway is zero.
        """
        this_lane = env.vehicles[self.veh_id]['lane']
        lead_id = env.get_leading_car(self.veh_id, this_lane)

        lead_pos = env.get_x_by_id(lead_id)
        lead_vel = env.vehicles[lead_id]['speed']
        lead_length = env.vehicles[lead_id]['length']

        this_pos = env.get_x_by_id(self.veh_id)

        # need to account for the position being reset around the length
        if lead_pos > this_pos: 
            dist = (this_pos + lead_length) - lead_pos
        else:
            loop_length = env.scenario.net_params["length"]
            dist = (this_pos + lead_length) - (lead_pos + loop_length)
        self.last_d = self.d
        d = dist - np.power((lead_vel - self.max_deaccel * env.time_step),2)/(2*self.max_deaccel)
        self.d = d

        if -2*d+self.max_deaccel*self.delay**2 < 0:
            v_safe = 0
        else:
{'fraction_non_alphanumeric': 0.08326192516423879, 'fraction_numerical': 0.00242787774921451, 'mean_word_length': 2.3635926993275698, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5277363', 'n_tokens_mistral': 2088, 'n_tokens_neox': 1974, 'n_words': 602}