<jupyter_start><jupyter_text>1. Car and Road  1.1 Initialization of car object and road grid<jupyter_code># Defince car object, include whether the Cell has a car(exist), whether it is autonomous(auto),
# ...its speed(speed), and driving direction（direct）
class car(object):
    def __init__(self, exist = 0, auto = 0, speed = 0, accel = 0, direct = "S"):
        self.exist = exist
        self.auto = auto
        self.speed = speed
        self.accel = accel
        self.direct = direct


# Create road grid
def make2DArray(cols, rows):
    arr = np.ndarray((cols, rows), dtype=np.object)
    for i in range(0, cols):
        for j in range(0, rows): 
            arr[i][j] = car() 
    return arr

# An array used to check if the cells are updated during each time step
def generateUpdateArray(cols, rows):
    arr = np.zeros((cols, rows))
    return arr<jupyter_output><empty_output><jupyter_text>1.2 Micro Rules<jupyter_code># Rule for generation of cars at t = 0
def rule_generate(grid):

    n_cell = grid.shape[0] * grid.shape[1]
    n_car = road_density * road_length / 1000  * 2 # two side
    spacing_ini = (n_cell / n_car) - 1 # initial spacing, empirical value
    
    if spacing_ini <= 1:
        raise ValueError('The density setting is too high. Please lower the density.') 
        
    auto_choices = [0, 1]
    p_auto_choices = [1 - ratio_auto, ratio_auto]
    count_car = 0
    
    p_auto = [ratio_auto, 1 - ratio_auto]
    choice_auto = [1, 0]
    
    count_car_all = 0
    for i in range(0, lane_number): 
        x = i
        j = 0
        count_car_lane = 0
        
        while count_car_lane < int(n_car / lane_number):    
            y = int(spacing_ini * j)
            grid[x][y].exist = 1
            grid[x][y].auto = np.random.choice(choice_auto, 1, p = p_auto)[0]
            grid[x][y].speed = int(random.gauss(vm, dv)) # init speed
            
            j += 1
            count_car_lane += 1
            count_car_all += 1
    
    # If not enough cars are generated (due to numerical issue), add to enough here
    for i in range(0, lane_number):
        x = i
        for j in range(0, rows):
            y = j
            if count_car_all < n_car:
                if grid[x][y].exist == 0: 
                    grid[x][y].exist == 1
                    grid[x][y].auto = np.random.choice(choice_auto, 1, p = p_auto)[0]
                    grid[x][y].speed = int(random.gauss(vm, dv)) # 初始速度
                    count_car_all += 1
            else:
                break
                        
    return grid


# Rule for moving cars forward
def rule_forward(grid):
                    
    total_speed = 0
    total_traffic = 0
    
    for i in range(0, lane_number): 
        x = i
        for j in range(0, rows):
            y = rows - j - 1 # Backward update

            if grid[x][y].exist == 1 and update[x][y] == 0: # If there is a car in the cell, and it is not updated
                
                # Calculate its distance with the car infront of it, and return the position index of the front car
                d_current, yf = check_forward((x, y)) 

                # Calculate the safe distance based on whether current car is autonomous
                if grid[x][y].auto == 0:
                    d_safe = safe_distance(grid[x][y].speed, th)
                else:
                    d_safe = safe_distance(grid[x][y].speed, ta)

                # Calculate the displacement current car can move forward
                if (d_current - grid[x][y].speed * dt - car_size) >= d_safe:
                    displacement = int(grid[x][y].speed * dt)
                else:
                    displacement = max(min(int(grid[x][y].speed * dt), d_current - car_size - safe_margin), 0)

                # Update the vehicle to the new cell (pay attention to the boundary)
                if (y + displacement) >= rows:
                    y_new = y + displacement - rows
                else:
                    y_new = y + displacement

                grid[x][y_new] = update_cell(grid[x][y], grid[x][yf], d_safe, d_current)
                update[x][y_new] = 1

                # Count speed and traffic
                total_speed += grid[x][y_new].speed
                total_traffic += 1


                # If displacement is actually happened, set current cell to have no car
                if displacement != 0:
                    grid[x][y].exist = 0
        
    return grid, total_speed, total_traffic

# Rule for lane changing
def rule_changeLane(grid):

    for i in range(0, lane_number): 
        x = i
        for j in range(0, rows):
            y = rows - j - 1 # Backward update
            if grid[x][y].exist == 1:
                
                # Check left and right to determine if lane changing is applicable
                left, right = check_changability((x, y))  

                if left and right: # Both left and right are good for lane changing
                    directs = [-1, 0, 1]
                    ps = [pc / 2, 1 - pc, pc /2]

                elif left and (not right):
                    directs = [-1, 0]
                    ps = [pc, 1 - pc]

                elif (not left) and right:
                    directs = [1, 0]
                    ps = [pc, 1 - pc]

                else:
                    directs = [0]
                    ps = [1]

                dx = np.random.choice(directs, 1, p=ps)
                x_new = x + dx[0]
                grid[x_new][y] = copy_cell(grid[x][y])

                # If lane changing is actually happened, set current cell to have no car
                if dx[0] != 0:
                    grid[x][y].exist = 0
    
    return grid<jupyter_output><empty_output><jupyter_text>1.3 Utility functions<jupyter_code># copy Cell，used for lane changing
def copy_cell(car_to_copy):
    car_new = car()
    car_new.exist = car_to_copy.exist
    car_new.auto = car_to_copy.auto
    car_new.speed = car_to_copy.speed
    car_new.accel = car_to_copy.accel
    car_new.direct = car_to_copy.direct
    
    return car_new

# update Cell, used for forwarding
def update_cell(car_self, car_ahead, d_safe, d_current):
    car_new = car()
    car_new.exist = 1
    car_new.auto = car_self.auto
    car_new.direct = car_self.direct
    
    # update speed based on acceleration
    if car_self.accel >= 0:
        car_new.speed = max(min(car_self.speed + car_self.accel * dt, vmax) , vmin)
    else:
        car_new.speed = min(max(car_self.speed + car_self.accel * dt, vmin) , vmax)
    
    # If there is no car in front of the current car, set the reference fore speed to be vmax
    # ... set the reference acceleration to be 0
    if car_ahead.exist == 0:
        car_ahead.speed == vmax
        car_ahead.accel = 0
    
    # update acceleration based on whether current car/fore car is autonomous
    if car_self.auto == 0: # current car is not auto
        car_new.accel = alpha * (d_current - d_safe)
    else: 
        if car_ahead.auto == 0: # fore car is not auto
            car_new.accel = alpha * (d_current - d_safe) + beta * (car_ahead.speed - car_self.speed)
        else: 
            car_new.accel = alpha * (d_current - d_safe) + beta * (car_ahead.speed - car_self.speed) + gamma * car_ahead.accel
    
    return car_new

# Detect whether there is a car in front of current car
def check_forward(position):
    x = position[0]
    y = position[1]
    
    ymax = y + max_forward + 1 # maximum distance to detect
    d_current = max_forward
    
    # check for car in each cell, one by one
    for yf in range(y + 1, ymax):            
        if yf >= rows:
            yf_temp = yf - rows
        else:
            yf_temp = yf
        # if there is a car    
        if grid[x][yf_temp].exist == 1:
            d_current = max(car_size, yf - y) # current distance, which can not be smaller than the car size
            return d_current, yf_temp
    
    return d_current, yf_temp # current distance with the fore car and the position index of the fore car

# Detect whether current car can/should change lane
def check_changability(position):
    x = position[0]
    y = position[1]
    left = 0
    right = 0
    
    d_current, yf = check_forward(position)
    
    # Calculate the safe distance based on whether current car is autonomous
    if grid[x][y].auto == 0:
        d_safe = safe_distance(grid[x][y].speed, th)
    else:
        d_safe = safe_distance(grid[x][y].speed, ta)
    
    cond1 = (d_current < d_safe) # current spacing is too small -> triggers necessity for lane change
    cond2 = ((x - 1) >= 0) # left side is within the boundary
    cond3 = ((x + 1) <= int((lane_number / 2) - 1)) # right side is within the boundary
    
    if cond1:
        if cond2: 
            if grid[x - 1][y].exist == 0: # no car on the left
                safe = check_safe((x - 1, y), grid[x][y].speed)
                if safe: # meet safe distance requirement
                    left = 1
        if cond3:
            if grid[x + 1][y].exist == 0: # no car on the right
                safe = check_safe((x + 1, y), grid[x][y].speed)
                if safe:
                    right = 1
    
    return left, right

# Detect whether safe distance requirement is met before lane changing
def check_safe(position, speed):
    x = position[0]
    y = position[1]
    
    safe = 1 # 1 means safe
    
    # Calculate the safe distance (with the fore car) based on whether current car is autonomous
    if grid[x][y].auto == 0:
        d_safe_f = safe_distance(grid[x][y].speed, th)
    else:
        d_safe_f = safe_distance(grid[x][y].speed, ta)
    
    ymax = y + int(d_safe_f) # look forward d_safe_f
    ymin = y - int(speed * th) # look back d_safe_b
    
    # look forward until d_safe_f
    for yf in range(y + 1, ymax):
        # accomodate for boundary
        if yf >= rows:
            yf = yf - rows
        # if ther is a fore car
        if grid[x][yf].exist == 1:
            safe = 0 # not safe
            return safe
    
    # look back until d_safe_b
    for yb in range(ymin, y):
        if yb < 0:
            yb = rows + yb
        if grid[x][yb].exist == 1:
            
            # if the rear car is not auto
            if grid[x][yb].auto == 0:
                # calculate safe distance with the rear car
                d_safe_b = safe_distance(grid[x][y].speed, th) + safe_margin
                if abs(yb - y) < d_safe_b:
                    safe = 0 # not safe
                    return safe
                
            # if the rear car is auto
            if grid[x][yb].auto == 1:
                d_safe_b = safe_distance(grid[x][y].speed, ta) + safe_margin
                if abs(yb - y) < d_safe_b:
                    safe = 0 # not safe
                    return safe
                
    return safe

# Calculate safe distance based on speed and reaction time
def safe_distance(speed, t_reaction):
    
    safe_distance = max(speed, t_reaction, car_size + safe_margin)
    
    return safe_distance<jupyter_output><empty_output><jupyter_text>2. Visualization<jupyter_code>def init_everything():
    global step, grid, average_speed, time_passing, density, flow, time, update
    step = 0
    grid = make2DArray(cols, rows)
    time = [0]
    average_speed = [0]
    time_passing = [0]
    flow = [0]
    density = [0]
    update = generateUpdateArray(cols, rows)

def average(data, steps_trace_back):
    step = len(data)
    start = step - steps_trace_back
    data_len = steps_trace_back + 1
    
    recent_average = sum(data[start:]) / data_len
    
    return recent_average
    

def check_stable():
    
    min_step = 200
    stable_threshold = 0.005
    
    stable = False
    
    if average_speed[step] < 0.1:
        return True
    
    if step > min_step:
        if abs((average_speed[step] - average_speed[step-min_step])/average_speed[step]) < stable_threshold:
            stable = True
    return stable

def get_new_record():
    min_step = 30
    new_record = [average(density, min_step), \
                  average(average_speed, min_step), \
                  average(time_passing, min_step), \
                  average(flow, min_step)]
    
    return new_record

def export_data(output):
    # Write data to data frame, then to CSV file.
    file_name = "auto_%.2f_lane_%d_%s_%s.csv" % (ratio_auto, lane_number, str(T.strftime("%Y-%m-%d")), 
                               str(T.strftime("%H%M%S")))
    columns = ["density", "average_speed", "time_passing", "flow"]
    pd.DataFrame(output, columns = columns).drop_duplicates().to_csv(
        'output/' + file_name, index = False, encoding = "UTF-8")
    
    
# Draw cars and road
def draw(canvas, grid):
    # update window
    canvas.delete("all")
    
    # draw road grid
    y_offset = 20
    dx = 0.2 * w_cell
    
    lane_oneside = int(lane_number / 2)

    for i in range(0, cols):
        for j in range(0, rows):
            if grid[i][j].exist == 1:
                x = i * w_cell + w_cell
                
                if grid[i][j].auto == 1: # use different pics to distinguish auto and non-auto cars
                    if i < lane_oneside:
                        y = j * h_cell + h_cell + y_offset
                        car_image = dict_car_images['auto_L']
                    else:
                        y = (rows - j) * h_cell + h_cell + y_offset
                        car_image = dict_car_images['auto_R']
                else:
                    if i < lane_oneside:
                        y = j * h_cell + h_cell + y_offset
                        car_image = dict_car_images['human_L']
                    else:
                        y = (rows - j) * h_cell + h_cell + y_offset
                        car_image = dict_car_images['human_R']               
                

                canvas.create_image(x + dx, y - 2.5 * h_cell, image=car_image, anchor = 'nw')
    
    currentTime = 'Time: ' + str(round(step * dt, 2)) + ' seconds' # record time step
    canvas.create_text(w_cell, y_offset/2, text=currentTime, anchor = 'w')
    
    xc = w_cell * cols/2 + w_cell 
    yl_1 = h_cell + y_offset
    yl_2 = h_cell * (rows + 1) + y_offset
    delta = w_cell / 20
    canvas.create_line(xc + delta, yl_1, xc + delta, yl_2, fill = 'orange', width = w_cell / 20) # draw mid-line of the road
    canvas.create_line(xc - delta, yl_1, xc - delta, yl_2, fill = 'orange', width = w_cell / 20) 
    canvas.create_line(xc, yl_1, xc, yl_2, fill = 'white', width = w_cell / 20) 
    
    len_dash1 = max(int(w_cell / 6), 1) # draw dash lines
    len_dash2 = max(int(w_cell / 7), 1)
    for i in range(0, cols + 1): 
        xi = i * w_cell + w_cell
        if i == 0 or i == cols:
            canvas.create_line(xi, yl_1, xi, yl_2, fill = 'black')
        elif xi != xc:
            canvas.create_line(xi, yl_1, xi, yl_2, fill = 'black', dash=(len_dash1, len_dash2))
    
    
    # draw real-time monitoring board
    y = [average_speed, time_passing, flow, density]
    y_label = ['mph', 's', 'veh/h', 'veh/mile']
    titles = ['Average Speed', 'Passing Time', 'Flow', 'Density']
    for i in range(len(y)):
        # note: use set_xdata and set_ydata to avoid regenerate figures, and hereby avoid running out of memory
        lines[i].set_xdata(time) 
        lines[i].set_ydata(y[i])
        axs[i].set_xlim(min(time), max(time))
        axs[i].set_ylim(min(y[i]), max(y[i]) * 1.1)
        title_new = titles[i] + ': ' + str(round(y[i][-1],2)) + ' ' + y_label[i]
        axs[i].set_title(title_new)
        axs[i].relim()
        axs[i].autoscale_view(True, True, True)
        
    fig.canvas.draw()
    
    canvas.pack()

# recursively display
def draw_after():
    
    global step, grid, flow_count, average_flow, average_speed, time, update
    
    step += 1
    new_time = step * dt
    time.append(new_time)
    update = generateUpdateArray(cols, rows) # set all cells to be not updated
    
    # generate cars
    if step == 1:
        grid = rule_generate(grid)
    
    # apply lane changing rule
    grid = rule_changeLane(grid)
    
    # apply forward rule
    grid, total_speed, total_traffic = rule_forward(grid)
    
    # get statistics
    average_speed.append(total_speed / total_traffic * 3.6 / mile_adj) # mph
    
    if average_speed[-1] <= 0.1:
        time_passing.append(road_length * 10) # s
    else:
        time_passing.append(road_length / (1000 * mile_adj) / average_speed[-1] * 3600) # s
        
    density.append((total_traffic / 2) / road_length * 1000 * mile_adj) # veh/mile
    
    flow.append(density[-1] *  average_speed[-1]) # veh/h
    
    
    draw(canvas_road, grid)
    
    canvas_road.after(int(dt * 1000), draw_after)  # the unit for canvas.after is mili second<jupyter_output><empty_output><jupyter_text>3. Info<jupyter_code>def place_info():
    labels = ['Lane Number', 'Road Length (mile)', 'Density (veh/mile)', 'Auto Ratio']
    values = [str(lane_number), str(road_length / (1000 * mile_adj)), str(road_d), str(ratio_auto)]
    tk_labels = list()
    tk_values = list()
    tk_breaks = list()
    
    for i in range(len(labels)):
        tk_labels.append(tk.Label(canvas_control))
        tk_values.append(tk.Label(canvas_control))
        tk_breaks.append(tk.Label(canvas_control))
    try:
        for i in range(len(tk_labels)):
            tk_labels[i].config(text = labels[i])
            tk_values[i].config(text = values[i])
            tk_breaks[i].config(text = '---------')
            tk_labels[i].pack()
            tk_values[i].pack()
            tk_breaks[i].pack()
        
        button1 = tk.Button(master=canvas_control, text="Quit", command=_quit)
        button1.pack()
        
    except:
        a = 1

# quit the GUI
def _quit():
    root.quit()     # stops mainloop
    root.destroy() 
    
def _run():
    return True<jupyter_output><empty_output><jupyter_text>4. Main function<jupyter_code>if __name__ == "__main__":
    global step, grid, average_speed, time_passing, density, flow, time, update, output, road_density, count_loop
    global draw_handle
    step = 0 # simulation steps
    
# The following parameters are adjustable for research purpose
##########################################################
    lane_number = 12 # for both directions
    road_l = 1 # road length，unit mile
    road_d = 500 # road density, veh/mile
    ratio_auto = 0.6 # ratio of autonomous cars
    
    





    
    
    
    
    
    
    
    

    # 1.1 road
    #lane_number = 4 # for both directions
    mile_adj = 1.60934
    road_length = road_l * 1000 * mile_adj # road length，unit m
    road_density = max(int(road_d / mile_adj), 5) # road density, veh/km
    
    # 1.2 car
    vmax = 29.05 # max speed，unit m/s，~108km/h ~ 67.5 mph
    vmin = 0
    vm = 0 # initial mean speed，unit m/s
    dv = 25 # std for initial speed，unit m/s
    car_size = 5 # length of a car， 5m
    
    # 1.3 car following
    th = 2 # reaction time of a human driver, in seconds
    ta = 1 # reaction time of an autonomous car, in seconds
    max_forward = 100 # the car will be impacted by a car as much as 100m in front of it 
    alpha = 1 # parameter for updating acceleration
    beta = 1 # parameter for updating acceleration
    gamma = 1 # parameter for updating acceleration
    safe_margin = 3 # minimum distance between cars
    
    # 1.4 lane change
    pc = 0.5 # when safe condition is met, the probability for lane changing
    
    # 1.5 time step
    dt = 0.2 # real waiting time for each time step, in seconds
#########################################################   
    

    # parameters for displaying
    h_cell = 6 # height of the Cell，1m
    w_cell = 4 * h_cell # width of the cell, 4m (a real lane has a width of 3.7m)
    road_w_gui = (lane_number + 2) * w_cell  # display width
    road_h_gui = 600 # display height
    road_h_gui_scroll = road_length * h_cell + 100
    data_w_gui = 500
    
    # 2. definition of objects
    # 2.1 road grid
    cols = lane_number # lane numbers
    rows = int(road_length) # road length
    grid = make2DArray(cols, rows) # road grid
    update = generateUpdateArray(cols, rows) # update checking matrix
               
    # 2.2 interface
    root = tk.Tk()
    canvas_r = tk.Canvas(root, width=road_w_gui, height = road_h_gui)
    canvas_r.pack(side = 'left')
    
    scrollbar = ttk.Scrollbar(root, orient="vertical", command=canvas_r.yview)
    scrollable_frame = ttk.Frame(canvas_r)

    scrollable_frame.bind(
        "<Configure>",
        lambda e: canvas_r.configure(
            scrollregion=canvas_r.bbox("all")
        )
    )

    canvas_r.create_window((0, 0), window=scrollable_frame, anchor="nw")
    canvas_r.configure(yscrollcommand=scrollbar.set)
    canvas_road = tk.Canvas(scrollable_frame, width=road_w_gui, height=road_h_gui_scroll)
    canvas_road.pack()
    scrollbar.place(x = road_w_gui - 15, y = 0, relheight=1)
    
    car_w = int(2 / 3 * w_cell) # car width，pixels
    car_h = int(car_size * h_cell) # car length for，pixels
    images = ['auto_L', 'auto_R', 'human_L','human_R']
    dict_car_images = dict()
    for image in images:
        car_image = ImageTk.PhotoImage(Image.open('image/' + image + '.png').resize((car_w, car_h))) # load car images
        dict_car_images[image] =  car_image
    
    # 3. Statistics
    # 3.1 initialization of parameters
    time = [0]
    average_speed = [0]
    time_passing = [0]
    flow = [0]
    density = [0]
    
    # 3.2 real-time displaying
    fig = Figure(figsize=(6, 8), dpi=80)
    gs1 = gridspec.GridSpec(4, 1)
    gs1.update(wspace=0.1, hspace=0.5)
    axs = list()
    lines = list()
    txts = list()
    for i in range(4):
        axs.append(fig.add_subplot(gs1[i]))
        line, = axs[i].plot([0], [0])
        lines.append(line)
        
    axs[3].set_xlabel('Time (s)')
    
    canvas_data = FigureCanvasTkAgg(fig, master=root)  # A tk.DrawingArea.
    canvas_data.draw()
    
    canvas_control = tk.Canvas(root, width=30, height = road_h_gui)
    canvas_control.pack(side = 'right')
    place_info()
    
    canvas_data.get_tk_widget().pack()
    
    # 4. run
    output = list()
    draw_after()
       
    tk.mainloop()<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup