6930

pending completion

Build # 6930

Build Type

Pull #994

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/flow-project/flow/commit/<a class=hub.com/flow-project/flow/commit/<a class="double-link" href="https://git"><a class=hub.com/flow-project/flow/commit/<a class="double-link" href="https://git"><a class=hub.com/flow-project/flow/commit/90756572c1fd6bfc51aaf34bcba5b4022fa2a587">90756572c<a href="https://github.com/flow-project/flow/commit/90756572c1fd6bfc51aaf34bcba5b4022fa2a587">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/flow-project/flow/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/flow-project/flow/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/flow-project/flow/commit/90756572c1fd6bfc51aaf34bcba5b4022fa2a587&quot;&gt;90756572c&lt;/a&gt;&lt;a href=&quot;https://github.com/flow-project/flow/commit/90756572c1fd6bfc51aaf34bcba5b4022fa2a587&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/flow-project/flow/commit/90756572c1fd6bfc51aaf34bcba5b402&lt;/a&gt;2fa2a587">&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/flow-project/flow/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/flow-project/flow/commit/<a class="double-link" href="https://github.com/flow-project/flow/commit/aad7183675be01bacd3e947431e1102235adbcfa">aad718367</a><a href="https://github.com/flow-project/flow/commit/90756572c1fd6bfc51aaf34bcba5b4022fa2a587">&amp;quot;&amp;gt;aad718367&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;aad718367&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/flow-project/flow/commit/90756572c1fd6bfc51aaf34bcba5b4022fa2a&lt;/a&gt;587&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/flow-project/flow/commit/<a class="double-link" href="https://github.com/flow-project/flow/commit/f1ded5430">f1ded5430">f1ded5430</a>

Pull Request Pull Request #994: Ak/i210 master merge

Run Details

259 of 259 new or added lines in 19 files covered. (100.0%)

9974 of 11911 relevant lines covered (83.74%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.89

/flow/core/experiment.py

"""Contains an experiment class for running simulations."""
from flow.utils.registry import make_create_env
from flow.data_pipeline.data_pipeline import upload_to_s3
from flow.data_pipeline.data_pipeline import get_configuration
from flow.data_pipeline.leaderboard_utils import network_name_translate
from flow.visualize.time_space_diagram import tsd_main
from collections import defaultdict
from datetime import timezone
from datetime import datetime
import logging
import time
import numpy as np
import uuid
import os


class Experiment:
    """
    Class for systematically running simulations in any supported simulator.

    This class acts as a runner for a network and environment. In order to use
    it to run an network and environment in the absence of a method specifying
    the actions of RL agents in the network, type the following:

        >>> from flow.envs import Env
        >>> flow_params = dict(...)  # see the examples in exp_config
        >>> exp = Experiment(flow_params)  # for some experiment configuration
        >>> exp.run(num_runs=1)

    If you wish to specify the actions of RL agents in the network, this may be
    done as follows:

        >>> rl_actions = lambda state: 0  # replace with something appropriate
        >>> exp.run(num_runs=1, rl_actions=rl_actions)

    Finally, if you would like to like to plot and visualize your results, this
    class can generate csv files from emission files produced by sumo. These
    files will contain the speeds, positions, edges, etc... of every vehicle
    in the network at every time step.

    In order to ensure that the simulator constructs an emission file, set the
    ``emission_path`` attribute in ``SimParams`` to some path.

        >>> from flow.core.params import SimParams
        >>> flow_params['sim'] = SimParams(emission_path="./data")

    Once you have included this in your environment, run your Experiment object
    as follows:

        >>> exp.run(num_runs=1, convert_to_csv=True)

    After the experiment is complete, look at the "./data" directory. There
    will be two files, one with the suffix .xml and another with the suffix
    .csv. The latter should be easily interpretable from any csv reader (e.g.
    Excel), and can be parsed using tools such as numpy and pandas.

    Attributes
    ----------
    custom_callables : dict < str, lambda >
        strings and lambda functions corresponding to some information we want
        to extract from the environment. The lambda will be called at each step
        to extract information from the env and it will be stored in a dict
        keyed by the str.
    env : flow.envs.Env
        the environment object the simulator will run
    """

    def __init__(self, flow_params, custom_callables=None):
        """Instantiate the Experiment class.

        Parameters
        ----------
        flow_params : dict
            flow-specific parameters
        custom_callables : dict < str, lambda >
            strings and lambda functions corresponding to some information we
            want to extract from the environment. The lambda will be called at
            each step to extract information from the env and it will be stored
            in a dict keyed by the str.
        """
        self.custom_callables = custom_callables or {}

        # Get the env name and a creator for the environment.
        create_env, _ = make_create_env(flow_params)

        # Create the environment.
        self.env = create_env()

        logging.info(" Starting experiment {} at {}".format(
            self.env.network.name, str(datetime.utcnow())))

        logging.info("Initializing environment.")

    def run(self,
            num_runs,
            rl_actions=None,
            convert_to_csv=False,
            to_aws=None,
            only_query="",
            is_baseline=False):
        """Run the given network for a set number of runs.

        Parameters
        ----------
        num_runs : int
            number of runs the experiment should perform
        rl_actions : method, optional
            maps states to actions to be performed by the RL agents (if
            there are any)
        convert_to_csv : bool
            Specifies whether to convert the emission file created by sumo
            into a csv file
        to_aws: str
            Specifies the S3 partition you want to store the output file,
            will be used to later for query. If NONE, won't upload output
            to S3.
        only_query: str
            Specifies which queries should be automatically run when the
            simulation data gets uploaded to S3. If an empty str is passed in,
            then it implies no queries should be run on this.
        is_baseline: bool
            Specifies whether this is a baseline run.

        Returns
        -------
        info_dict : dict < str, Any >
            contains returns, average speed per step
        """
        num_steps = self.env.env_params.horizon

        # raise an error if convert_to_csv is set to True but no emission
        # file will be generated, to avoid getting an error at the end of the
        # simulation
        if convert_to_csv and self.env.sim_params.emission_path is None:
            raise ValueError(
                'The experiment was run with convert_to_csv set '
                'to True, but no emission file will be generated. If you wish '
                'to generate an emission file, you should set the parameter '
                'emission_path in the simulation parameters (SumoParams or '
                'AimsunParams) to the path of the folder where emissions '
                'output should be generated. If you do not wish to generate '
                'emissions, set the convert_to_csv parameter to False.')

        # used to store
        info_dict = {
            "returns": [],
            "velocities": [],
            "outflows": [],
        }
        info_dict.update({
            key: [] for key in self.custom_callables.keys()
        })

        if rl_actions is None:
            def rl_actions(*_):
                return None

        # time profiling information
        t = time.time()
        times = []

        if convert_to_csv and self.env.simulator == "traci":
            # data pipeline
            source_id = 'flow_{}'.format(uuid.uuid4().hex)
            metadata = defaultdict(lambda: [])

            # collect current time
            cur_datetime = datetime.now(timezone.utc)
            cur_date = cur_datetime.date().isoformat()
            cur_time = cur_datetime.time().isoformat()

            # collecting information for metadata table
            metadata['source_id'].append(source_id)
            metadata['submission_time'].append(cur_time)
            metadata['network'].append(
                network_name_translate(self.env.network.name.split('_20')[0]))
            metadata['is_baseline'].append(str(is_baseline))
            if to_aws:
                name, strategy = get_configuration()
                metadata['submitter_name'].append(name)
                metadata['strategy'].append(strategy)

            # emission-specific parameters
            dir_path = self.env.sim_params.emission_path
            trajectory_table_path = os.path.join(
                dir_path, '{}.csv'.format(source_id))
            metadata_table_path = os.path.join(
                dir_path, '{}_METADATA.csv'.format(source_id))
        else:
            source_id = None
            trajectory_table_path = None
            metadata_table_path = None
            metadata = None
            cur_date = None

        for i in range(num_runs):
            ret = 0
            vel = []
            custom_vals = {key: [] for key in self.custom_callables.keys()}
            state = self.env.reset()
            for j in range(num_steps):
                t0 = time.time()
                state, reward, done, _ = self.env.step(rl_actions(state))
                t1 = time.time()
                times.append(1 / (t1 - t0))

                # Compute the velocity speeds and cumulative returns.
                veh_ids = self.env.k.vehicle.get_ids()
                vel.append(np.mean(self.env.k.vehicle.get_speed(veh_ids)))
                ret += reward

                # Compute the results for the custom callables.
                for (key, lambda_func) in self.custom_callables.items():
                    custom_vals[key].append(lambda_func(self.env))

                if type(done) is dict and done['__all__'] or done is True:
                    break

            # Store the information from the run in info_dict.
            outflow = self.env.k.vehicle.get_outflow_rate(int(500))
            info_dict["returns"].append(ret)
            info_dict["velocities"].append(np.mean(vel))
            info_dict["outflows"].append(outflow)
            for key in custom_vals.keys():
                info_dict[key].append(np.mean(custom_vals[key]))

            print("Round {0}, return: {1}".format(i, ret))

            # Save emission data at the end of every rollout. This is skipped
            # by the internal method if no emission path was specified.
            if self.env.simulator == "traci":
                self.env.k.simulation.save_emission(run_id=i)

        # Print the averages/std for all variables in the info_dict.
        for key in info_dict.keys():
            print("Average, std {}: {}, {}".format(
                key, np.mean(info_dict[key]), np.std(info_dict[key])))

        print("Total time:", time.time() - t)
        print("steps/second:", np.mean(times))
        self.env.terminate()

        if to_aws:
            tsd_main(
                trajectory_table_path,
                {'network': self.env.network.__class__},
                min_speed=0,
                max_speed=10,
                start=self.env.env_params.warmup_steps
            )
            upload_to_s3(
                'circles.data.pipeline',
                'metadata_table/date={0}/partition_name={1}_METADATA/'
                '{1}_METADATA.csv'.format(cur_date, source_id),
                metadata_table_path
            )
            upload_to_s3(
                'circles.data.pipeline',
                'fact_vehicle_trace/date={0}/partition_name={1}/'
                '{1}.csv'.format(cur_date, source_id),
                trajectory_table_path,
                {'network': metadata['network'][0],
                 'is_baseline': metadata['is_baseline'][0]}
            )
            upload_to_s3(
                'circles.data.pipeline',
                'time_space_diagram/date={0}/partition_name={1}/'
                '{1}.png'.format(cur_date, source_id),
                trajectory_table_path.replace('csv', 'png')
            )

        return info_dict

1	"""Contains an experiment class for running simulations."""
2	from flow.utils.registry import make_create_env	1✔
3	from flow.data_pipeline.data_pipeline import upload_to_s3	1✔
4	from flow.data_pipeline.data_pipeline import get_configuration	1✔
5	from flow.data_pipeline.leaderboard_utils import network_name_translate	1✔
6	from flow.visualize.time_space_diagram import tsd_main	1✔
7	from collections import defaultdict	1✔
8	from datetime import timezone	1✔
9	from datetime import datetime	1✔
10	import logging	1✔
11	import time	1✔
12	import numpy as np	1✔
13	import uuid	1✔
14	import os	1✔
15
16
17	class Experiment:	1✔
18	"""
19	Class for systematically running simulations in any supported simulator.
20
21	This class acts as a runner for a network and environment. In order to use
22	it to run an network and environment in the absence of a method specifying
23	the actions of RL agents in the network, type the following:
24
25	>>> from flow.envs import Env
26	>>> flow_params = dict(...) # see the examples in exp_config
27	>>> exp = Experiment(flow_params) # for some experiment configuration
28	>>> exp.run(num_runs=1)
29
30	If you wish to specify the actions of RL agents in the network, this may be
31	done as follows:
32
33	>>> rl_actions = lambda state: 0 # replace with something appropriate
34	>>> exp.run(num_runs=1, rl_actions=rl_actions)
35
36	Finally, if you would like to like to plot and visualize your results, this
37	class can generate csv files from emission files produced by sumo. These
38	files will contain the speeds, positions, edges, etc... of every vehicle
39	in the network at every time step.
40
41	In order to ensure that the simulator constructs an emission file, set the
42	``emission_path`` attribute in ``SimParams`` to some path.
43
44	>>> from flow.core.params import SimParams
45	>>> flow_params['sim'] = SimParams(emission_path="./data")
46
47	Once you have included this in your environment, run your Experiment object
48	as follows:
49
50	>>> exp.run(num_runs=1, convert_to_csv=True)
51
52	After the experiment is complete, look at the "./data" directory. There
53	will be two files, one with the suffix .xml and another with the suffix
54	.csv. The latter should be easily interpretable from any csv reader (e.g.
55	Excel), and can be parsed using tools such as numpy and pandas.
56
57	Attributes
58	----------
59	custom_callables : dict < str, lambda >
60	strings and lambda functions corresponding to some information we want
61	to extract from the environment. The lambda will be called at each step
62	to extract information from the env and it will be stored in a dict
63	keyed by the str.
64	env : flow.envs.Env
65	the environment object the simulator will run
66	"""
67
68	def __init__(self, flow_params, custom_callables=None):	1✔
69	"""Instantiate the Experiment class.
70
71	Parameters
72	----------
73	flow_params : dict
74	flow-specific parameters
75	custom_callables : dict < str, lambda >
76	strings and lambda functions corresponding to some information we
77	want to extract from the environment. The lambda will be called at
78	each step to extract information from the env and it will be stored
79	in a dict keyed by the str.
80	"""
81	self.custom_callables = custom_callables or {}	1✔
82
83	# Get the env name and a creator for the environment.
84	create_env, _ = make_create_env(flow_params)	1✔
85
86	# Create the environment.
87	self.env = create_env()	1✔
88
89	logging.info(" Starting experiment {} at {}".format(	1✔
90	self.env.network.name, str(datetime.utcnow())))
91
92	logging.info("Initializing environment.")	1✔
93
94	def run(self,	1✔
95	num_runs,
96	rl_actions=None,
97	convert_to_csv=False,
98	to_aws=None,
99	only_query="",
100	is_baseline=False):
101	"""Run the given network for a set number of runs.
102
103	Parameters
104	----------
105	num_runs : int
106	number of runs the experiment should perform
107	rl_actions : method, optional
108	maps states to actions to be performed by the RL agents (if
109	there are any)
110	convert_to_csv : bool
111	Specifies whether to convert the emission file created by sumo
112	into a csv file
113	to_aws: str
114	Specifies the S3 partition you want to store the output file,
115	will be used to later for query. If NONE, won't upload output
116	to S3.
117	only_query: str
118	Specifies which queries should be automatically run when the
119	simulation data gets uploaded to S3. If an empty str is passed in,
120	then it implies no queries should be run on this.
121	is_baseline: bool
122	Specifies whether this is a baseline run.
123
124	Returns
125	-------
126	info_dict : dict < str, Any >
127	contains returns, average speed per step
128	"""
129	num_steps = self.env.env_params.horizon	1✔
130
131	# raise an error if convert_to_csv is set to True but no emission
132	# file will be generated, to avoid getting an error at the end of the
133	# simulation
134	if convert_to_csv and self.env.sim_params.emission_path is None:	1✔
135	raise ValueError(	×
136	'The experiment was run with convert_to_csv set '
137	'to True, but no emission file will be generated. If you wish '
138	'to generate an emission file, you should set the parameter '
139	'emission_path in the simulation parameters (SumoParams or '
140	'AimsunParams) to the path of the folder where emissions '
141	'output should be generated. If you do not wish to generate '
142	'emissions, set the convert_to_csv parameter to False.')
143
144	# used to store
145	info_dict = {	1✔
146	"returns": [],
147	"velocities": [],
148	"outflows": [],
149	}
150	info_dict.update({	1✔
151	key: [] for key in self.custom_callables.keys()
152	})
153
154	if rl_actions is None:	1✔
155	def rl_actions(*_):	1✔
156	return None	1✔
157
158	# time profiling information
159	t = time.time()	1✔
160	times = []	1✔
161
162	if convert_to_csv and self.env.simulator == "traci":	1✔
163	# data pipeline
164	source_id = 'flow_{}'.format(uuid.uuid4().hex)	1✔
165	metadata = defaultdict(lambda: [])	1✔
166
167	# collect current time
168	cur_datetime = datetime.now(timezone.utc)	1✔
169	cur_date = cur_datetime.date().isoformat()	1✔
170	cur_time = cur_datetime.time().isoformat()	1✔
171
172	# collecting information for metadata table
173	metadata['source_id'].append(source_id)	1✔
174	metadata['submission_time'].append(cur_time)	1✔
175	metadata['network'].append(	1✔
176	network_name_translate(self.env.network.name.split('_20')[0]))
177	metadata['is_baseline'].append(str(is_baseline))	1✔
178	if to_aws:	1✔
179	name, strategy = get_configuration()	×
180	metadata['submitter_name'].append(name)	×
181	metadata['strategy'].append(strategy)	×
182
183	# emission-specific parameters
184	dir_path = self.env.sim_params.emission_path	1✔
185	trajectory_table_path = os.path.join(	1✔
186	dir_path, '{}.csv'.format(source_id))
187	metadata_table_path = os.path.join(	1✔
188	dir_path, '{}_METADATA.csv'.format(source_id))
189	else:
190	source_id = None	1✔
191	trajectory_table_path = None	1✔
192	metadata_table_path = None	1✔
193	metadata = None	1✔
194	cur_date = None	1✔
195
196	for i in range(num_runs):	1✔
197	ret = 0	1✔
198	vel = []	1✔
199	custom_vals = {key: [] for key in self.custom_callables.keys()}	1✔
200	state = self.env.reset()	1✔
201	for j in range(num_steps):	1✔
202	t0 = time.time()	1✔
203	state, reward, done, _ = self.env.step(rl_actions(state))	1✔
204	t1 = time.time()	1✔
205	times.append(1 / (t1 - t0))	1✔
206
207	# Compute the velocity speeds and cumulative returns.
208	veh_ids = self.env.k.vehicle.get_ids()	1✔
209	vel.append(np.mean(self.env.k.vehicle.get_speed(veh_ids)))	1✔
210	ret += reward	1✔
211
212	# Compute the results for the custom callables.
213	for (key, lambda_func) in self.custom_callables.items():	1✔
214	custom_vals[key].append(lambda_func(self.env))	×
215
216	if type(done) is dict and done['__all__'] or done is True:	1✔
217	break	1✔
218
219	# Store the information from the run in info_dict.
220	outflow = self.env.k.vehicle.get_outflow_rate(int(500))	1✔
221	info_dict["returns"].append(ret)	1✔
222	info_dict["velocities"].append(np.mean(vel))	1✔
223	info_dict["outflows"].append(outflow)	1✔
224	for key in custom_vals.keys():	1✔
225	info_dict[key].append(np.mean(custom_vals[key]))	×
226
227	print("Round {0}, return: {1}".format(i, ret))	1✔
228
229	# Save emission data at the end of every rollout. This is skipped
230	# by the internal method if no emission path was specified.
231	if self.env.simulator == "traci":	1✔
232	self.env.k.simulation.save_emission(run_id=i)	1✔
233
234	# Print the averages/std for all variables in the info_dict.
235	for key in info_dict.keys():	1✔
236	print("Average, std {}: {}, {}".format(	1✔
237	key, np.mean(info_dict[key]), np.std(info_dict[key])))
238
239	print("Total time:", time.time() - t)	1✔
240	print("steps/second:", np.mean(times))	1✔
241	self.env.terminate()	1✔
242
243	if to_aws:	1✔
244	tsd_main(	×
245	trajectory_table_path,
246	{'network': self.env.network.__class__},
247	min_speed=0,
248	max_speed=10,
249	start=self.env.env_params.warmup_steps
250	)
251	upload_to_s3(	×
252	'circles.data.pipeline',
253	'metadata_table/date={0}/partition_name={1}_METADATA/'
254	'{1}_METADATA.csv'.format(cur_date, source_id),
255	metadata_table_path
256	)
257	upload_to_s3(	×
258	'circles.data.pipeline',
259	'fact_vehicle_trace/date={0}/partition_name={1}/'
260	'{1}.csv'.format(cur_date, source_id),
261	trajectory_table_path,
262	{'network': metadata['network'][0],
263	'is_baseline': metadata['is_baseline'][0]}
264	)
265	upload_to_s3(	×
266	'circles.data.pipeline',
267	'time_space_diagram/date={0}/partition_name={1}/'
268	'{1}.png'.format(cur_date, source_id),
269	trajectory_table_path.replace('csv', 'png')
270	)
271
272	return info_dict	1✔

flow-project / flow / 6930

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous