Add files via upload

envs/env.py

@@ -0,0 +1,417 @@
+"""
+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]

envs/functions.py

@@ -0,0 +1,48 @@
+from inspect import indentsize
+import numpy as np
+import math
+
+# prints formatted price
+
+
+def formatPrice(n):
+    return ("-$" if n < 0 else "$") + "{0:.2f}".format(abs(n))
+
+
+def formatPercent(n):
+    return "{0:.2f}".format(n*100) + "%"
+
+# returns the vector containing stock data from a fixed file
+
+
+def getStockDataVec(data_dir, key, index=1000):
+    vec = []
+    lines = open(data_dir + key + ".csv", "r").read().splitlines()
+
+    for line in lines[1:]:
+        l1 = line.split(",")[4]  # [1:7] --> not use by no have " "
+        # vec.append(float(line.split(",")[1])*volume)
+        vec.append(float(l1)*index)
+        # list1 = [float(x) for x in l1]
+        # print('num', float(l1)*volume)
+
+    return vec
+
+# returns the sigmoid
+
+
+def sigmoid(x):
+    return 1 / (1 + math.exp(-x))
+
+# returns an an n-day state representation ending at time t
+
+
+def getState(data, t, n):
+    d = t - n + 1
+    block = data[d:t + 1] if d >= 0 else -d * \
+        [data[0]] + data[0:t + 1]  # pad with t0
+    res = []
+    for i in range(n - 1):
+        res.append(sigmoid(block[i + 1] - block[i]))
+
+    return np.array(res)  # [res]

envs/real_net_env.py

@@ -0,0 +1,112 @@
+"""
+Particular class of real traffic network
+@author: Tianshu Chun 
+"""
+import numpy as np
+import matplotlib.pyplot as plt
+import os
+import seaborn as sns
+from envs.env import PhaseMap, PhaseSet, TrafficSimulator
+# from real_net.data.build_file import gen_rou_file
+
+import sys
+import os
+sys.path.append(os.path.abspath(
+    "/home/smartcube/tuannguyen/deeprl_signal_control/envs/"))
+sys.path.append(os.path.abspath(
+    "/home/smartcube/tuannguyen/deeprl_signal_control/real_net/data"))
+
+sns.set_color_codes()
+
+STATE_NAMES = ['wave']
+# node: (phase key, neighbor list)
+NODES = {'long': ('3.0', []),
+         'short': ('3.1', [])}
+
+PHASES = {
+    '3.0': ['hold', 'buy', 'sell'],
+    '3.1': ['hold', 'sell', 'buy']
+}
+
+
+class RealNetPhase(PhaseMap):
+    def __init__(self):
+        self.phases = {}
+        for key, val in PHASES.items():
+            self.phases[key] = PhaseSet(val)
+
+
+class RealNetController:
+    def __init__(self, node_names, nodes):
+        self.name = 'greedy'
+        self.node_names = node_names
+        self.nodes = nodes
+
+    def forward(self, obs):
+        actions = []
+        for ob, node_name in zip(obs, self.node_names):
+            actions.append(self.greedy(ob, node_name))
+        return actions
+
+    def greedy(self, ob, node_name):
+        # get the action space
+        phases = PHASES[NODES[node_name][0]]
+        flows = []
+        node = self.nodes[node_name]
+        # get the green waves
+        for phase in phases:
+            wave = 0
+            visited_ilds = set()
+            for i, signal in enumerate(phase):
+                if signal == 'G':
+                    # find controlled lane
+                    lane = node.lanes_in[i]
+                    # ild = 'ild:' + lane
+                    ild = lane
+                    # if it has not been counted, add the wave
+                    if ild not in visited_ilds:
+                        j = node.ilds_in.index(ild)
+                        wave += ob[j]
+                        visited_ilds.add(ild)
+            flows.append(wave)
+        return np.argmax(np.array(flows))
+
+
+class RealNetEnv(TrafficSimulator):
+    def __init__(self, config, port=0, output_path='', is_record=False, record_stat=False):
+        super().__init__(config, output_path, is_record, record_stat, port=port)
+
+    def _get_node_phase_id(self, node_name):
+        return self.phase_node_map[node_name]
+
+    def _init_neighbor_map(self):
+        return dict([(key, val[1]) for key, val in NODES.items()])
+
+    def _init_map(self):
+        self.neighbor_map = self._init_neighbor_map()
+        self.phase_map = RealNetPhase()
+        self.phase_node_map = dict([(key, val[0])
+                                   for key, val in NODES.items()])
+        self.state_names = STATE_NAMES
+
+    # def _init_sim_config(self, seed):
+    #     # comment out to call build_file.py
+    #     return gen_rou_file(self.data_path,
+    #                         self.flow_rate,
+    #                         seed=seed,
+    #                         thread=self.sim_thread)
+
+    def plot_stat(self, rewards):
+        self.state_stat['reward'] = rewards
+        for name, data in self.state_stat.items():
+            fig = plt.figure(figsize=(8, 6))
+            plot_cdf(data)
+            plt.ylabel(name)
+            fig.savefig(self.output_path + self.name + '_' + name + '.png')
+
+
+def plot_cdf(X, c='b', label=None):
+    sorted_data = np.sort(X)
+    yvals = np.arange(len(sorted_data))/float(len(sorted_data)-1)
+    # print('sorted_data', sorted_data)
+    plt.plot(sorted_data, yvals, color=c, label=label)