""" Traffic network simulator w/ defined sumo files @author: Tianshu Chu """ from envs.functions import getState, getStockDataVec, formatPrice, formatPercent import xml.etree.cElementTree as ET import subprocess import pandas as pd import numpy as np import logging import sys import os os.environ['SUMO_HOME'] = "/usr/share/sumo" sys.path.append(os.path.join(os.environ.get("SUMO_HOME"), 'tools')) DEFAULT_PORT = 8000 SEC_IN_MS = 1000 # hard code real-net reward norm REALNET_REWARD_NORM = 20 class PhaseSet: def __init__(self, phases): self.num_phase = len(phases) self.num_lane = len(phases[0]) self.phases = phases # self._init_phase_set() @staticmethod def _get_phase_lanes(phase, signal='r'): phase_lanes = [] for i, l in enumerate(phase): if l == signal: phase_lanes.append(i) return phase_lanes def _init_phase_set(self): self.red_lanes = [] # self.green_lanes = [] for phase in self.phases: self.red_lanes.append(self._get_phase_lanes(phase)) # self.green_lanes.append(self._get_phase_lanes(phase, signal='G')) class PhaseMap: def __init__(self): self.phases = {} def get_phase(self, phase_id, action): # phase_type is either green or yellow return self.phases[phase_id].phases[int(action)] def get_phase_num(self, phase_id): return self.phases[phase_id].num_phase def get_lane_num(self, phase_id): # the lane number is link number return self.phases[phase_id].num_lane def get_red_lanes(self, phase_id, action): # the lane number is link number return self.phases[phase_id].red_lanes[int(action)] class Node: def __init__(self, name, neighbor=[], control=False): self.control = control # disabled self.lanes_in = [] self.ilds_in = [] # for state self.fingerprint = [] # local policy self.name = name self.neighbor = neighbor self.num_state = 0 # wave and wait should have the same dim self.num_fingerprint = 0 self.wave_state = [] # local state self.wait_state = [] # local state self.phase_id = -1 self.n_a = 0 # self.prev_action = -1 class TrafficSimulator: def __init__(self, config, output_path, is_record, record_stats, port=0): self.name = config.get('scenario') self.control_interval_sec = config.getint('control_interval_sec') self.yellow_interval_sec = config.getint('yellow_interval_sec') self.port = DEFAULT_PORT + port self.sim_thread = port self.obj = config.get('objective') self.data_path = config.get('data_path') self.price_data = config.get('price_data') self.key = config.get('key') self.window_size = config.getint('window_size') self.balance = config.getint('balance') self.agent = config.get('agent') self.coop_gamma = config.getfloat('coop_gamma') self.cur_episode = 0 self.norms = {'wave': config.getfloat('norm_wave'), 'wait': config.getfloat('norm_wait')} self.clips = {'wave': config.getfloat('clip_wave'), 'wait': config.getfloat('clip_wait')} self.coef_wait = config.getfloat('coef_wait') self.train_mode = True test_seeds = config.get('test_seeds').split(',') test_seeds = [int(s) for s in test_seeds] self.data = getStockDataVec(self.price_data, self.key) self.l = len(self.data) - 1 self.t = 0 self.n = self.window_size + 1 self.inventory = {} self.agent_type = {'long': 1, 'short': -1} self.total_profit = 0 # self.drawdown = [] self.balance_list = [] self.cur_balance = self.balance self._init_map() self.init_data(is_record, record_stats, output_path) self.init_test_seeds(test_seeds) self._init_nodes() def _get_node_phase_id(self, node_name): # needs to be overwriteen raise NotImplementedError() def _get_state(self): # hard code the state ordering as wave, wait, fp state = [] # measure the most recent state self._measure_state_step() # get the appropriate state vectors for node_name in self.node_names: # node_name 10026 node = self.nodes[node_name] # wave is required in state cur_state = [node.wave_state] # include wave states of neighbors for nnode_name in node.neighbor: # discount the neigboring states cur_state.append( self.nodes[nnode_name].wave_state * self.coop_gamma) # include wait state if 'wait' in self.state_names: cur_state.append(node.wait_state) # include fingerprints of neighbors for nnode_name in node.neighbor: cur_state.append(self.nodes[nnode_name].fingerprint) state.append(np.concatenate(cur_state)) return state def _init_nodes(self): nodes = {} trafficlight_id = ['long', 'short'] for node_name in trafficlight_id: if node_name in self.neighbor_map: neighbor = self.neighbor_map[node_name] else: logging.info('node %s can not be found!' % node_name) neighbor = [] nodes[node_name] = Node(node_name, neighbor=neighbor, control=True) # Init inventory self.inventory[node_name] = [] self.nodes = nodes self.node_names = sorted(list(nodes.keys())) s = 'Env: init %d node information:\n' % len(self.node_names) for node in self.nodes.values(): s += node.name + ':\n' s += '\tneigbor: %r\n' % node.neighbor # s += '\tlanes_in: %r\n' % node.lanes_in s += '\tilds_in: %r\n' % node.ilds_in # s += '\tedges_in: %r\n' % node.edges_in logging.info(s) self._init_action_space() self._init_state_space() def _init_action_space(self): # for local and neighbor coop level self.n_a_ls = [] for node_name in self.node_names: node = self.nodes[node_name] # phase_id 3.0 phase_id = self._get_node_phase_id(node_name) node.phase_id = phase_id node.n_a = self.phase_map.get_phase_num(phase_id) self.n_a_ls.append(node.n_a) # for global coop level self.n_a = np.prod(np.array(self.n_a_ls)) def _init_map(self): # needs to be overwriteen self.neighbor_map = None self.phase_map = None self.state_names = None raise NotImplementedError() def _init_policy(self): policy = [] for node_name in self.node_names: phase_num = self.nodes[node_name].n_a p = 1. / phase_num policy.append(np.array([p] * phase_num)) return policy def _init_state_space(self): self._reset_state() self.n_s_ls = [] self.n_w_ls = [] self.n_f_ls = [] for node_name in self.node_names: node = self.nodes[node_name] # num_wave = node.num_state num_wave = self.window_size num_fingerprint = 0 for nnode_name in node.neighbor: if self.agent not in ['a2c', 'greedy']: # all marl agents have neighborhood communication num_wave += self.nodes[nnode_name].num_state if self.agent == 'ma2c': # only ma2c uses neighbor's policy num_fingerprint += self.nodes[nnode_name].num_fingerprint num_wait = 0 if 'wait' not in self.state_names else node.num_state self.n_s_ls.append(num_wave + num_wait + num_fingerprint) self.n_f_ls.append(num_fingerprint) self.n_w_ls.append(num_wait) self.n_s = np.sum(np.array(self.n_s_ls)) def _measure_reward_step(self, action): rewards = [] for node_name, a in zip(self.node_names, list(action)): reward = 0 if a == 1: # buy self.inventory[node_name].append(self.data[self.t]) # print("Buy: " + formatPrice(self.data[self.t])) elif a == 2 and len(self.inventory[node_name]) > 0: # sell order_price = self.inventory[node_name].pop(0) profit = (self.data[self.t] - order_price) * \ self.agent_type[node_name] reward = max(profit, 0) # if node_name == 'short': self.total_profit += profit # if profit < 0: # self.drawdown.append(-profit/self.cur_balance) self.cur_balance += profit self.balance_list.append(round(self.cur_balance, 2)) # print("Sell: " + formatPrice(self.data[self.t]) + " | Profit: " + formatPrice( # self.data[self.t] - bought_price)) rewards.append(reward) return np.array(rewards) def _measure_state_step(self): for node_name in self.node_names: node = self.nodes[node_name] for state_name in self.state_names: cur_state = getState(self.data, self.t, self.n) if self.record_stats: self.state_stat[state_name] += list(cur_state) # normalization norm_cur_state = self._norm_clip_state(cur_state, self.norms[state_name], self.clips[state_name]) node.wave_state = norm_cur_state def _measure_traffic_step(self): cur_traffic = {'episode': self.cur_episode, 'time_sec': self.t } self.traffic_data.append(cur_traffic) @staticmethod def _norm_clip_state(x, norm, clip=-1): x = x / norm return x if clip < 0 else np.clip(x, 0, clip) def _reset_state(self): for node_name in self.node_names: node = self.nodes[node_name] # prev action for yellow phase before each switch node.prev_action = 0 # fingerprint is previous policy[:-1] node.num_fingerprint = node.n_a - 1 # node.num_state = self._get_node_state_num(node) def _simulate(self): self.t += 1 if self.is_record: self._measure_traffic_step() def _transfer_action(self, action): '''Transfer global action to a list of local actions''' phase_nums = [] for node in self.control_node_names: phase_nums.append(self.nodes[node].phase_num) action_ls = [] for i in range(len(phase_nums) - 1): action, cur_action = divmod(action, phase_nums[i]) action_ls.append(cur_action) action_ls.append(action) return action_ls def _update_waits(self, action): for node_name, a in zip(self.node_names, action): red_lanes = set() node = self.nodes[node_name] for i in self.phase_map.get_red_lanes(node.phase_id, a): red_lanes.add(node.lanes_in[i]) for i in range(len(node.waits)): lane = node.ilds_in[i] if lane in red_lanes: node.waits[i] += self.control_interval_sec else: node.waits[i] = 0 def init_data(self, is_record, record_stats, output_path): self.is_record = is_record self.record_stats = record_stats self.output_path = output_path if self.is_record: self.traffic_data = [] self.control_data = [] self.trip_data = [] if self.record_stats: self.state_stat = {} for state_name in self.state_names: self.state_stat[state_name] = [] def init_test_seeds(self, test_seeds): self.test_num = len(test_seeds) self.test_seeds = test_seeds def output_data(self): if not self.is_record: logging.error('Env: no record to output!') control_data = pd.DataFrame(self.control_data) control_data.to_csv(self.output_path + ('%s_%s_control.csv' % (self.name, self.agent))) traffic_data = pd.DataFrame(self.traffic_data) traffic_data.to_csv(self.output_path + ('%s_%s_traffic.csv' % (self.name, self.agent))) trip_data = pd.DataFrame(self.trip_data) trip_data.to_csv(self.output_path + ('%s_%s_trip.csv' % (self.name, self.agent))) def reset(self): self._reset_state() self.t = 0 self.cur_episode += 1 # initialize fingerprint self.update_fingerprint(self._init_policy()) # next environment random condition should be different self.inventory = dict.fromkeys(self.inventory, []) self.total_profit = 0 # self.drawdown = [] self.balance_list = [] self.cur_balance = self.balance return self._get_state() def step(self, action): state = self._get_state() reward = self._measure_reward_step(action) global_reward = np.sum(reward) # for fair comparison # New t for new price self.t += 1 # self._simulate() done = True if self.t == self.l - 1 else False if done: print("--------------------------------") print("Total Profit: " + formatPrice(self.total_profit)) # print('Max Drawdown: ' + formatPercent(max(self.drawdown))) print('Balance List', self.balance_list) print('Final Balance: ' + formatPrice(self.cur_balance)) print("--------------------------------") if self.is_record: action_r = ','.join(['%d' % a for a in action]) cur_control = {'episode': self.cur_episode, 'time_sec': self.t, 'total_profit': formatPrice(self.total_profit), 'action': action_r, 'reward': global_reward} self.control_data.append(cur_control) # use local rewards in test if not self.train_mode: return state, reward, done, global_reward # discounted global reward for ma2c new_reward = [] for node_name, r in zip(self.node_names, reward): cur_reward = r for nnode_name in self.nodes[node_name].neighbor: i = self.node_names.index(nnode_name) cur_reward += self.coop_gamma * reward[i] if self.name != 'real_net': new_reward.append(cur_reward) else: n_node = 1 + len(self.nodes[node_name].neighbor) new_reward.append( cur_reward / (n_node * REALNET_REWARD_NORM)) reward = np.array(new_reward) return state, reward, done, global_reward def update_fingerprint(self, policy): for node_name, pi in zip(self.node_names, policy): self.nodes[node_name].fingerprint = np.array(pi)[:-1]