mirror of
https://github.com/Ahp06/SUMO_Emissions.git
synced 2024-11-24 12:36:30 +00:00
283 lines
10 KiB
Python
283 lines
10 KiB
Python
'''
|
|
Created on 19 janv. 2019
|
|
|
|
@author: Axel Huynh-Phuc
|
|
'''
|
|
|
|
"""
|
|
This module defines the entry point of the application
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import datetime
|
|
import itertools
|
|
import logging
|
|
import multiprocessing
|
|
import os
|
|
import sys
|
|
import time
|
|
import traci
|
|
|
|
import jsonpickle
|
|
|
|
from config import Config
|
|
from data import Data
|
|
import emissions
|
|
from model import Emission
|
|
|
|
|
|
"""
|
|
Init the Traci API
|
|
"""
|
|
if 'SUMO_HOME' in os.environ:
|
|
tools = os.path.join(os.environ['SUMO_HOME'], 'tools')
|
|
sys.path.append(tools)
|
|
else:
|
|
sys.exit("please declare environment variable 'SUMO_HOME'")
|
|
|
|
class RunProcess(multiprocessing.Process):
|
|
"""
|
|
Run process inheriting from multiprocessing.Process
|
|
"""
|
|
|
|
def __init__(self, data: Data, config: Config, save_logs: bool, csv_export: bool):
|
|
"""
|
|
RunProcess constructor
|
|
:param data: The data instance
|
|
:param config: The config instance
|
|
:param save_logs: If save_logs == True, it will save the logs into the logs directory
|
|
:param csv_export: If csv_export == True, it will export all emissions data into a csv file
|
|
"""
|
|
multiprocessing.Process.__init__(self)
|
|
self.data = data
|
|
self.config = config
|
|
self.save_logs = save_logs
|
|
self.csv_export = csv_export
|
|
|
|
def init_logger(self):
|
|
"""
|
|
Init logger properties
|
|
"""
|
|
now = datetime.datetime.now()
|
|
current_date = now.strftime("%Y_%m_%d_%H_%M_%S")
|
|
|
|
logdir = f'{self.data.dir}/logs/'
|
|
logging.info(logdir)
|
|
if not os.path.exists(logdir):
|
|
os.mkdir(logdir)
|
|
|
|
conf_name = self.config.config_filename.replace('.json', '')
|
|
log_filename = f'{logdir}/{current_date}.log'
|
|
|
|
self.logger = logging.getLogger(f'sumo_logger')
|
|
self.logger.setLevel(logging.INFO)
|
|
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
|
|
|
if self.save_logs:
|
|
file_handler = logging.FileHandler(log_filename)
|
|
file_handler.setFormatter(formatter)
|
|
self.logger.addHandler(file_handler)
|
|
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(formatter)
|
|
self.logger.addHandler(handler)
|
|
|
|
def export_data_to_csv(self):
|
|
"""
|
|
Export all Emission objects as a CSV file into the csv directory
|
|
"""
|
|
csv_dir = f'{self.data.dir}/csv'
|
|
if not os.path.exists(csv_dir):
|
|
os.mkdir(csv_dir)
|
|
|
|
now = datetime.datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
|
|
conf_name = self.config.config_filename.replace('.json', '')
|
|
|
|
csvfile = os.path.join(csv_dir, f'{self.data.dump_name}_{conf_name}_{now}.csv')
|
|
with open(csvfile, 'w') as f:
|
|
writer = csv.writer(f)
|
|
# Write CSV headers
|
|
writer.writerow(itertools.chain(('Step',), (a.name for a in self.data.grid)))
|
|
# Write all areas emission value for each step
|
|
for step in range(self.config.n_steps):
|
|
em_for_step = (f'{a.emissions_by_step[step].value():.3f}' for a in self.data.grid)
|
|
writer.writerow(itertools.chain((step,), em_for_step))
|
|
|
|
def run(self):
|
|
"""
|
|
Launch a simulation, will be called when a RunProcess instance is started
|
|
"""
|
|
try:
|
|
self.init_logger()
|
|
self.logger.info(f'Running simulation dump "{self.data.dump_name}" with the config "{self.config.config_filename}" ...')
|
|
|
|
if self.config.without_actions_mode:
|
|
self.logger.info('Reference simulation')
|
|
|
|
traci.start(self.config.sumo_cmd)
|
|
|
|
for area in self.data.grid: # Set acquisition window size
|
|
area.set_window_size(self.config.window_size)
|
|
traci.polygon.add(area.name, area.rectangle.exterior.coords, (255, 0, 0)) # Add polygon for UI
|
|
|
|
self.logger.info(f'Loaded simulation file : {self.config._SUMOCFG}')
|
|
self.logger.info('Loading data for the simulation')
|
|
|
|
start = time.perf_counter()
|
|
self.logger.info('Simulation started...')
|
|
step = 0
|
|
while step < self.config.n_steps:
|
|
traci.simulationStep()
|
|
|
|
vehicles = emissions.get_all_vehicles()
|
|
emissions.get_emissions(self, vehicles, step)
|
|
step += 1
|
|
|
|
print(f'step = {step}/{self.config.n_steps}', end='\r')
|
|
|
|
finally:
|
|
traci.close(False)
|
|
|
|
total_emissions = Emission()
|
|
for area in self.data.grid:
|
|
total_emissions += area.sum_all_emissions()
|
|
|
|
self.logger.info(f'Total emissions = {total_emissions.value()} mg')
|
|
for pollutant in ['co2','co','nox','hc','pmx']:
|
|
value = total_emissions.__getattribute__(pollutant)
|
|
self.logger.info(f'{pollutant.upper()} = {value} mg')
|
|
|
|
simulation_time = round(time.perf_counter() - start, 2)
|
|
self.logger.info(f'End of the simulation ({simulation_time}s)')
|
|
|
|
# 1 step is equal to one second simulated
|
|
self.logger.info(f'Real-time factor : {self.config.n_steps / simulation_time}')
|
|
|
|
if self.csv_export:
|
|
self.export_data_to_csv()
|
|
self.logger.info(f'Exported data into the csv folder')
|
|
|
|
def create_dump(dump_name, simulation_dir, areas_number):
|
|
"""
|
|
Create a new dump with config file and dump_name chosen
|
|
:param dump_name: The name of the data dump
|
|
:param simulation_dir: The simulation directory
|
|
:param areas_number: The number of areas in grid
|
|
:return:
|
|
"""
|
|
|
|
#sumo_binary = os.path.join(os.environ['SUMO_HOME'], 'bin', 'sumo')
|
|
#sumo_cmd = [sumo_binary, "-c", f'files/simulations/{simulation_dir}/osm.sumocfg']
|
|
|
|
for f in os.listdir(simulation_dir):
|
|
if f.endswith('.sumocfg'):
|
|
_SUMOCFG = os.path.join(simulation_dir, f)
|
|
|
|
sumo_binary = os.path.join(os.environ['SUMO_HOME'], 'bin', 'sumo')
|
|
sumo_cmd = [sumo_binary, "-c", _SUMOCFG]
|
|
|
|
|
|
traci.start(sumo_cmd)
|
|
if not os.path.isfile(f'{simulation_dir}/dump/{dump_name}.json'):
|
|
start = time.perf_counter()
|
|
data = Data(dump_name, traci.simulation.getNetBoundary(), areas_number, simulation_dir)
|
|
data.init_grid()
|
|
data.add_data_to_areas()
|
|
data.save()
|
|
|
|
loading_time = round(time.perf_counter() - start, 2)
|
|
print(f'Data loaded ({loading_time}s)')
|
|
print(f'Dump {dump_name} created')
|
|
else:
|
|
print(f'Dump with name {dump_name} already exists')
|
|
|
|
traci.close(False)
|
|
|
|
def add_options(parser):
|
|
"""
|
|
Add command line options
|
|
:param parser: The command line parser
|
|
:return:
|
|
"""
|
|
|
|
parser.add_argument("-new_dump", "--new_dump", type=str,
|
|
help='Load and create a new data dump with the configuration file chosen')
|
|
parser.add_argument("-areas", "--areas", type=int,
|
|
help='Will create a grid with "areas x areas" areas')
|
|
parser.add_argument("-simulation_dir", "--simulation_dir", type=str,
|
|
help='Choose the simulation directory')
|
|
|
|
parser.add_argument("-run", "--run", type=str,
|
|
help='Run a simulation process with the dump chosen')
|
|
parser.add_argument("-c", "--c", metavar =('config1','config2'), nargs='+', type=str,
|
|
help='Choose your(s) configuration file(s) from your working directory')
|
|
parser.add_argument("-c_dir", "--c_dir", type=str,
|
|
help='Choose a directory which contains your(s) configuration file(s)')
|
|
parser.add_argument("-save", "--save", action="store_true",
|
|
help='Save the logs into the logs folder')
|
|
parser.add_argument("-csv", "--csv", action="store_true",
|
|
help="Export all data emissions into a CSV file")
|
|
|
|
def check_user_entry(args):
|
|
"""
|
|
Check the user entry consistency
|
|
"""
|
|
if (args.new_dump is not None):
|
|
if(args.areas is None or args.simulation_dir is None):
|
|
print('The -new_dump argument requires the -areas and -simulation_dir options')
|
|
return False
|
|
|
|
if (args.run is not None):
|
|
if(args.c is None and args.c_dir is None):
|
|
print('The -run argument requires the -c or -c_dir')
|
|
return False
|
|
|
|
return True
|
|
|
|
def main(args):
|
|
"""
|
|
The entry point of the application
|
|
:param args: Command line options
|
|
:return:
|
|
"""
|
|
parser = argparse.ArgumentParser()
|
|
add_options(parser)
|
|
args = parser.parse_args(args)
|
|
|
|
if(check_user_entry(args)):
|
|
|
|
if args.new_dump is not None:
|
|
if (args.simulation_dir is not None) and (args.areas is not None):
|
|
create_dump(args.new_dump, args.simulation_dir, args.areas)
|
|
|
|
if args.run is not None:
|
|
dump_path = f'{args.run}'
|
|
if os.path.isfile(dump_path):
|
|
with open(dump_path, 'r') as f:
|
|
data = jsonpickle.decode(f.read())
|
|
|
|
process = []
|
|
files = []
|
|
|
|
if args.c is not None:
|
|
for config in args.c:
|
|
files.append(f'{config}')
|
|
|
|
if args.c_dir is not None:
|
|
path = f'{args.c_dir}'
|
|
bundle_files = [f for f in os.listdir(path) if os.path.isfile(os.path.join(path, f))]
|
|
for config in bundle_files:
|
|
files.append(os.path.join(path, config))
|
|
|
|
for conf in files: # Initialize all process
|
|
config = Config(conf,data)
|
|
p = RunProcess(data, config, args.save, args.csv)
|
|
process.append(p)
|
|
p.start()
|
|
|
|
for p in process : p.join()
|
|
|
|
if __name__ == '__main__':
|
|
main(sys.argv[1:])
|