testing_simulation.py

import traci
import numpy as np
import random
import timeit
import os

# phase codes based on environment.net.xml
PHASE_NS_GREEN = 0  # action 0 code 00
PHASE_NS_YELLOW = 1
PHASE_NSL_GREEN = 2  # action 1 code 01
PHASE_NSL_YELLOW = 3
PHASE_EW_GREEN = 4  # action 2 code 10
PHASE_EW_YELLOW = 5
PHASE_EWL_GREEN = 6  # action 3 code 11
PHASE_EWL_YELLOW = 7


class Simulation:
    def __init__(self, Model, TrafficGen, sumo_cmd, max_steps, green_duration, yellow_duration, num_states, num_actions):
        self._Model = Model
        self._TrafficGen = TrafficGen
        self._step = 0
        self._sumo_cmd = sumo_cmd
        self._max_steps = max_steps
        self._green_duration = green_duration
        self._yellow_duration = yellow_duration
        self._num_states = num_states
        self._num_actions = num_actions
        self._reward_episode = []
        self._queue_length_episode = []
        self._co_episode = []


    def run(self, episode):
        """
        Runs the testing simulation
        """
        start_time = timeit.default_timer()

        # first, generate the route file for this simulation and set up sumo
        self._TrafficGen.generate_routefile(seed=episode)
        traci.start(self._sumo_cmd)
        print("Simulating...")

        # inits
        self._step = 0
        self._waiting_times = {}
        self._wait_car_number = 0
        self._co = {}
        self._hc = {}
        self._nox = {}
        self._co2 = {}
        self._sum_co = 0
        self._sum_waiting_time = 0
        old_total_wait = 0
        old_total_co = 0
        old_total_hc = 0
        old_total_nox = 0
        old_total_co2 = 0
        old_action = -1  # dummy init

        while self._step < self._max_steps:

            # get current state of the intersection
            current_state = self._get_state()
            car = np.sum(current_state)

            # calculate reward of previous action: (change in cumulative waiting time between actions)
            # waiting time = seconds waited by a car since the spawn in the environment, cumulated for every car in incoming lanes
            # current_total_wait = self._collect_waiting_times()
            # current_total_wait, current_total_co, current_hc, current_nox, current_co2 = self._collect_waiting_times()
            # reward_wait = old_total_wait - current_total_wait
            # a = old_total_co - current_total_co
            # b = old_total_hc - current_hc
            # c = old_total_nox - current_nox
            # d = old_total_co2 - current_co2
            # reward_co = round((a/10), 1)
            # reward_hc = round(b, 1)
            # reward_nox = round(c, 1)
            # reward_co2 = round((d/1000), 1)
            # reward_em = reward_nox + reward_hc + reward_co + reward_co2

            current_total_wait = self._collect_waiting_car()
            reward = old_total_wait - current_total_wait

            # choose the light phase to activate, based on the current state of the intersection
            action = self._choose_action(current_state)

            # if the chosen phase is different from the last phase, activate the yellow phase
            if self._step != 0 and old_action != action:
                self._set_yellow_phase(old_action)
                self._simulate(self._yellow_duration)

            # execute the phase selected before
            self._set_green_phase(action)
            self._simulate(self._green_duration)

            # saving variables for later & accumulate reward
            old_action = action
            old_total_wait = current_total_wait
            # old_total_co = current_total_co
            # old_total_nox = current_nox
            # old_total_hc = current_hc
            # old_total_co2 = current_co2

            self._reward_episode.append(reward)

            co = self._sum_co
            wait = self._sum_waiting_time

        #print("Total reward:", np.sum(self._reward_episode))
        traci.close()
        simulation_time = round(timeit.default_timer() - start_time, 1)

        return simulation_time, wait, co


    def _simulate(self, steps_todo):
        """
        Proceed with the simulation in sumo
        """
        a = open("/3实验/6/Deep-QLearning-Agent-for-Traffic-Signal-Control-master/test-6e-co.txt", 'a')
        b = open("/3实验/6/Deep-QLearning-Agent-for-Traffic-Signal-Control-master/test-6e-wait.txt", 'a')


        if (self._step + steps_todo) >= self._max_steps:  # do not do more steps than the maximum allowed number of steps
            steps_todo = self._max_steps - self._step

        while steps_todo > 0:
            traci.simulationStep()  # simulate 1 step in sumo
            self._step += 1 # update the step counter
            steps_todo -= 1
            queue_length, co = self._get_queue_length()
            self._queue_length_episode.append(queue_length)
            self._sum_waiting_time += queue_length
            self._sum_co += co
            self._co_episode.append(co)

            a.write("%s\n" % co)
            b.write("%s\n" % queue_length)

    def _collect_waiting_car(self):
        """检索在各个车道上等待的车的数量"""
        emssion_class = ['Zero/default', "HBEFA3/LDV_G_EU6", 'HBEFA3/PC_G_EU4', 'HBEFA3/Bus', 'HBEFA3/HDV']
        incoming_roads = ["E2TL", "N2TL", "W2TL", "S2TL"]
        w_car = 0
        c_number = 0
        car_list = traci.vehicle.getIDList()
        for car_id in car_list:
            road_id = traci.vehicle.getRoadID(car_id)
            if road_id in incoming_roads:
                ve = traci.vehicle.getSpeed(car_id)
                v_class = traci.vehicle.getEmissionClass(car_id)
                if ve <= 0.1:
                    if v_class == emssion_class[0]:
                        c_number = 1
                    elif v_class == emssion_class[1]:
                        c_number = 2
                    elif v_class == emssion_class[2]:
                        c_number = 3
                    elif v_class == emssion_class[3]:
                        c_number = 4
                    elif v_class == emssion_class[4]:
                        c_number = 5
                    else:
                        c_number = 0
                w_car += c_number
            self._wait_car_number = w_car

        return self._wait_car_number
    def _collect_waiting_times(self):
        """
        Retrieve the waiting time of every car in the incoming roads
        """
        incoming_roads = ["E2TL", "N2TL", "W2TL", "S2TL"]
        car_list = traci.vehicle.getIDList()
        for car_id in car_list:
            wait_time = traci.vehicle.getAccumulatedWaitingTime(car_id)
            co = traci.vehicle.getCOEmission(car_id)
            hc = traci.vehicle.getHCEmission(car_id)
            nox = traci.vehicle.getNOxEmission(car_id)
            co2 = traci.vehicle.getCO2Emission(car_id)
            road_id = traci.vehicle.getRoadID(car_id)  # get the road id where the car is located
            if road_id in incoming_roads:  # consider only the waiting times of cars in incoming roads
                self._waiting_times[car_id] = wait_time
                self._co[car_id] = co
                self._hc[car_id] = hc
                self._nox[car_id] = nox
                self._co2[car_id] = co2
            else:
                if car_id in self._waiting_times: # a car that was tracked has cleared the intersection
                    del self._waiting_times[car_id]
                    del self._co[car_id]
                    del self._hc[car_id]
                    del self._nox[car_id]
                    del self._co2[car_id]
        total_waiting_time = sum(self._waiting_times.values())
        total_co = round(sum(self._co.values()), 1)
        total_hc = round(sum(self._hc.values()), 1)
        total_nox = round(sum(self._nox.values()), 1)
        total_co2 = round(sum(self._co2.values()), 1)
        return total_waiting_time, total_co, total_hc, total_nox, total_co2


    def _choose_action(self, state):
        """
        Pick the best action known based on the current state of the env
        """
        return np.argmax(self._Model.predict_one(state))


    def _set_yellow_phase(self, old_action):
        """
        Activate the correct yellow light combination in sumo
        """
        yellow_phase_code = old_action * 2 + 1 # obtain the yellow phase code, based on the old action (ref on environment.net.xml)
        traci.trafficlight.setPhase("TL", yellow_phase_code)


    def _set_green_phase(self, action_number):
        """
        Activate the correct green light combination in sumo
        """


        if action_number == 0:
            traci.trafficlight.setPhase("TL", PHASE_NS_GREEN)
        elif action_number == 1:
            traci.trafficlight.setPhase("TL", PHASE_NSL_GREEN)
        elif action_number == 2:
            traci.trafficlight.setPhase("TL", PHASE_EW_GREEN)
        elif action_number == 3:
            traci.trafficlight.setPhase("TL", PHASE_EWL_GREEN)


    def _get_queue_length(self):
        car = traci.edge.getCOEmission("N2TL")
        car = traci.edge.getCOEmission("S2TL")
        car = traci.edge.getCOEmission("E2TL")
        car = traci.edge.getCOEmission("W2TL")

        halt_N = traci.edge.getLastStepHaltingNumber("N2TL")
        halt_S = traci.edge.getLastStepHaltingNumber("S2TL")
        halt_E = traci.edge.getLastStepHaltingNumber("E2TL")
        halt_W = traci.edge.getLastStepHaltingNumber("W2TL")
        co_N = traci.edge.getCOEmission("N2TL")
        co_S = traci.edge.getCOEmission("S2TL")
        co_E = traci.edge.getCOEmission("E2TL")
        co_W = traci.edge.getCOEmission("W2TL")

        co2_N = traci.edge.getCO2Emission("N2TL")
        co2_S = traci.edge.getCO2Emission("S2TL")
        co2_E = traci.edge.getCO2Emission("E2TL")
        co2_W = traci.edge.getCO2Emission("W2TL")

        hc_N = traci.edge.getHCEmission("N2TL")
        hc_S = traci.edge.getHCEmission("S2TL")
        hc_E = traci.edge.getHCEmission("E2TL")
        hc_W = traci.edge.getHCEmission("W2TL")

        nox_N = traci.edge.getNOxEmission("N2TL")
        nox_S = traci.edge.getNOxEmission("S2TL")
        nox_E = traci.edge.getNOxEmission("E2TL")
        nox_W = traci.edge.getNOxEmission("W2TL")

        sum_co2 = co2_N + co2_S + co2_E + co2_W
        sum_co = co_S + co_N + co_W + co_E
        sum_hc = hc_E + hc_N + hc_S + hc_W
        sum_nox = nox_E + nox_N + nox_W +nox_S
        queue_length = halt_N + halt_S + halt_E + halt_W
        return queue_length, sum_co, sum_co2, sum_hc, sum_nox


    def _get_state(self):
        """
        Retrieve the state of the intersection from sumo, in the form of cell occupancy
        """
        state = np.zeros(self._num_states)
        car_list = traci.vehicle.getIDList()

        for car_id in car_list:
            lane_pos = traci.vehicle.getLanePosition(car_id)
            lane_id = traci.vehicle.getLaneID(car_id)
            lane_pos = 750 - lane_pos  # inversion of lane pos, so if the car is close to the traffic light -> lane_pos = 0 --- 750 = max len of a road

            # distance in meters from the traffic light -> mapping into cells
            if lane_pos < 7:
                lane_cell = 0
            elif lane_pos < 14:
                lane_cell = 1
            elif lane_pos < 21:
                lane_cell = 2
            elif lane_pos < 28:
                lane_cell = 3
            elif lane_pos < 40:
                lane_cell = 4
            elif lane_pos < 60:
                lane_cell = 5
            elif lane_pos < 100:
                lane_cell = 6
            elif lane_pos < 160:
                lane_cell = 7
            elif lane_pos < 400:
                lane_cell = 8
            elif lane_pos <= 750:
                lane_cell = 9

            # finding the lane where the car is located 
            # x2TL_3 are the "turn left only" lanes
            if lane_id == "W2TL_0" or lane_id == "W2TL_1" or lane_id == "W2TL_2":
                lane_group = 0
            elif lane_id == "W2TL_3":
                lane_group = 1
            elif lane_id == "N2TL_0" or lane_id == "N2TL_1" or lane_id == "N2TL_2":
                lane_group = 2
            elif lane_id == "N2TL_3":
                lane_group = 3
            elif lane_id == "E2TL_0" or lane_id == "E2TL_1" or lane_id == "E2TL_2":
                lane_group = 4
            elif lane_id == "E2TL_3":
                lane_group = 5
            elif lane_id == "S2TL_0" or lane_id == "S2TL_1" or lane_id == "S2TL_2":
                lane_group = 6
            elif lane_id == "S2TL_3":
                lane_group = 7
            else:
                lane_group = -1

            if lane_group >= 1 and lane_group <= 7:
                car_position = int(str(lane_group) + str(lane_cell))  # composition of the two postion ID to create a number in interval 0-79
                valid_car = True
            elif lane_group == 0:
                car_position = lane_cell
                valid_car = True
            else:
                valid_car = False  # flag for not detecting cars crossing the intersection or driving away from it

            if valid_car:
                state[car_position] = 1  # write the position of the car car_id in the state array in the form of "cell occupied"

        return state


    @property
    def queue_length_episode(self):
        return self._queue_length_episode


    @property
    def reward_episode(self):
        return self._reward_episode