thib8956/sumo-emissions
1
0
Fork 0
mirror of https://github.com/Ahp06/SUMO_Emissions.git synced 2024-11-22 03:26:30 +00:00
Code Issues Releases Wiki Activity

Reformat code 🚀

Browse Source
This commit is contained in:
Thibaud Gasser 2018-12-16 21:43:21 +01:00
parent 2a4ce7f935
commit 61da2d7303
3 changed files with 138 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
sumo_project/config.py
View File

@ -12,16 +12,16 @@ from model import Emission
class Config:
# Total of emissions of all pollutants in mg for n steps of simulation without acting on areas
# These constants are simulation dependant, you must change them according to your simulation
ref200 = Emission(co2=42816869.05436445,co=1128465.0343051048,nox=18389.648337283958,hc=6154.330914019103,pmx=885.0829265236318)
ref200 = Emission(co2=42816869.05436445, co=1128465.0343051048, nox=18389.648337283958, hc=6154.330914019103,
pmx=885.0829265236318)
def __init__(self):
'''Default constructor'''
"""Default constructor"""
def import_config_file(self,config_file):
with open(config_file,'r') as f:
def import_config_file(self, config_file):
with open(config_file, 'r') as f:
data = json.load(f)
self._SUMOCMD = data["_SUMOCMD"]
@ -43,29 +43,30 @@ class Config:
self.check_config()
def check_config(self):
#Weight routing mode cannot be combinated with other actions
# Weight routing mode cannot be combinated with other actions
if self.weight_routing_mode:
self.limit_speed_mode = False
self.adjust_traffic_light_mode = False
self.lock_area_mode = False
#If without_actions_mode is choosen
# If without_actions_mode is choosen
if self.without_actions_mode:
self.limit_speed_mode = False
self.adjust_traffic_light_mode = False
self.weight_routing_mode = False
self.lock_area_mode = False
def __repr__(self) -> str:
return (str(f'grid : {self.areas_number}x{self.areas_number}\n')
+ str(f'step number = {self.n_steps}\n')
+ str(f'window size = {self.window_size}\n')
+ str(f'weight routing mode = {self.weight_routing_mode}\n')
+ str(f'lock area mode = {self.lock_area_mode}\n')
+ str(f'limit speed mode = {self.limit_speed_mode}, RF = {self.speed_rf*100}%\n')
+ str(f'adjust traffic light mode = {self.adjust_traffic_light_mode} , RF = {self.trafficLights_duration_rf*100}%\n'))
return (
f'grid : {self.areas_number}x{self.areas_number}\n'
f'step number = {self.n_steps}\n'
f'window size = {self.window_size}\n'
f'weight routing mode = {self.weight_routing_mode}\n'
f'lock area mode = {self.lock_area_mode}\n'
f'limit speed mode = {self.limit_speed_mode}, RF = {self.speed_rf * 100}%\n'
f'adjust traffic light mode = {self.adjust_traffic_light_mode},'
'RF = {self.trafficLights_duration_rf * 100}%\n'
)
def init_traci(self):
if 'SUMO_HOME' in os.environ:
@ -77,7 +78,7 @@ class Config:
sumo_binary = os.path.join(os.environ['SUMO_HOME'], 'bin', self._SUMOCMD)
self.sumo_cmd = [sumo_binary, "-c", self._SUMOCFG]
def init_logger(self, save_logs = False):
def init_logger(self, save_logs=False):
now = datetime.datetime.now()
current_date = now.strftime("%Y_%m_%d_%H_%M_%S")
@ -90,7 +91,7 @@ class Config:
logger.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
if save_logs :
if save_logs:
file_handler = logging.FileHandler(log_filename)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
@ -104,5 +105,3 @@ class Config:
def get_ref_emissions(self):
if self.n_steps == 200:
return self.ref200

58
sumo_project/emissions.py
View File

@ -1,26 +1,25 @@
import argparse
import csv
import datetime
import itertools
import os
import sys
import time
from traci import trafficlight
import traci
from typing import List
import datetime
import os
import traci
from parse import search
from shapely.geometry import LineString
import actions
from config import Config
from model import Area, Vehicle, Lane , TrafficLight , Phase , Logic, Emission
from model import Area, Vehicle, Lane, TrafficLight, Phase, Logic, Emission
# Absolute path of the directory the script is in
SCRIPTDIR = os.path.dirname(__file__)
def init_grid(simulation_bounds, areas_number,window_size):
def init_grid(simulation_bounds, areas_number, window_size):
grid = list()
width = simulation_bounds[1][0] / areas_number
height = simulation_bounds[1][1] / areas_number
@ -47,14 +46,16 @@ def get_all_lanes() -> List[Lane]:
def parse_phase(phase_repr):
duration = search('duration: {:f}', phase_repr)
minDuration = search('minDuration: {:f}', phase_repr)
maxDuration = search('maxDuration: {:f}', phase_repr)
phaseDef = search('phaseDef: {}\n', phase_repr)
min_duration = search('min_duration: {:f}', phase_repr)
max_duration = search('max_duration: {:f}', phase_repr)
phase_def = search('phase_def: {}\n', phase_repr)
if phaseDef is None: phaseDef = ''
else : phaseDef = phaseDef[0]
if phase_def is None:
phase_def = ''
else:
phase_def = phase_def[0]
return Phase(duration[0], minDuration[0], maxDuration[0], phaseDef)
return Phase(duration[0], min_duration[0], max_duration[0], phase_def)
def add_data_to_areas(areas: List[Area]):
@ -75,14 +76,14 @@ def add_data_to_areas(areas: List[Area]):
def compute_vehicle_emissions(veh_id):
co2 = traci.vehicle.getCO2Emission(veh_id)
co = traci.vehicle.getCOEmission(veh_id)
nox = traci.vehicle.getNOxEmission(veh_id)
hc = traci.vehicle.getHCEmission(veh_id)
pmx = traci.vehicle.getPMxEmission(veh_id)
return Emission(co2,co,nox,hc,pmx)
return Emission(co2, co, nox, hc, pmx)
def get_all_vehicles() -> List[Vehicle]:
vehicles = list()
@ -106,10 +107,11 @@ def get_emissions(grid: List[Area], vehicles: List[Vehicle], current_step, confi
if area.sum_emissions_into_window(current_step, config.window_size) >= config.emissions_threshold:
if config.limit_speed_mode and not area.limited_speed:
logger.info(f'Action - Decreased max speed into {area.name} by {config.speed_rf*100}%')
logger.info(f'Action - Decreased max speed into {area.name} by {config.speed_rf * 100}%')
actions.limit_speed_into_area(area, vehicles, config.speed_rf)
if config.adjust_traffic_light_mode and not area.tls_adjusted:
logger.info(f'Action - Decreased traffic lights duration by {config.trafficLights_duration_rf*100}%')
logger.info(
f'Action - Decreased traffic lights duration by {config.trafficLights_duration_rf * 100}%')
actions.adjust_traffic_light_phase_duration(area, config.trafficLights_duration_rf)
if config.lock_area_mode and not area.locked:
@ -126,7 +128,8 @@ def get_emissions(grid: List[Area], vehicles: List[Vehicle], current_step, confi
actions.reverse_actions(area)
traci.polygon.setFilled(area.name, False)
def get_reduction_percentage(ref,total):
def get_reduction_percentage(ref, total):
return (ref - total) / ref * 100
@ -163,7 +166,7 @@ def run(config, logger):
logger.info('Simulation started...')
step = 0
while step < config.n_steps : # traci.simulation.getMinExpectedNumber() > 0:
while step < config.n_steps: # traci.simulation.getMinExpectedNumber() > 0:
traci.simulationStep()
vehicles = get_all_vehicles()
@ -174,11 +177,11 @@ def run(config, logger):
finally:
traci.close(False)
export_data_to_csv(config,grid)
export_data_to_csv(config, grid)
simulation_time = round(time.perf_counter() - start, 2)
logger.info(f'End of the simulation ({simulation_time}s)')
logger.info(f'Real-time factor : {config.n_steps/simulation_time}')
logger.info(f'Real-time factor : {config.n_steps / simulation_time}')
total_emissions = Emission()
for area in grid:
@ -186,12 +189,12 @@ def run(config, logger):
logger.info(f'Total emissions = {total_emissions.value()} mg')
if not config.without_actions_mode :
if not config.without_actions_mode:
ref = config.get_ref_emissions()
if not (ref is None):
global_diff = (ref.value() - total_emissions.value()) / ref.value()
logger.info(f'Global reduction percentage of emissions = {global_diff*100} %')
logger.info(f'Global reduction percentage of emissions = {global_diff * 100} %')
logger.info(f'-> CO2 emissions = {get_reduction_percentage(ref.co2, total_emissions.co2)} %')
logger.info(f'-> CO emissions = {get_reduction_percentage(ref.co, total_emissions.co)} %')
logger.info(f'-> Nox emissions = {get_reduction_percentage(ref.nox, total_emissions.nox)} %')
@ -203,16 +206,16 @@ def add_options(parser):
parser.add_argument("-f", "--configfile", type=str, default='configs/default_config.json', required=False,
help='Choose your configuration file from your working directory')
parser.add_argument("-save", "--save", action="store_true",
help = 'Save the logs into the logs folder')
parser.add_argument("-steps", "--steps", type= int, default= 200, required = False,
help='Save the logs into the logs folder')
parser.add_argument("-steps", "--steps", type=int, default=200, required=False,
help='Choose the simulated time (in seconds)')
parser.add_argument("-ref", "--ref", action="store_true",
help='Launch a reference simulation (without acting on areas)')
parser.add_argument("-gui", "--gui", action ="store_true",
help= "Set GUI mode")
parser.add_argument("-gui", "--gui", action="store_true",
help="Set GUI mode")
def main(args):
parser = argparse.ArgumentParser(description="")
add_options(parser)
args = parser.parse_args(args)
@ -238,5 +241,6 @@ def main(args):
logger.info(f'Simulated time : {args.steps}s')
run(config, logger)
if __name__ == '__main__':
main(sys.argv[1:])

20
sumo_project/model.py
View File

@ -1,10 +1,10 @@
import collections
from traci._trafficlight import Logic as SUMO_Logic
from typing import Tuple, Set
from shapely.geometry import Point, LineString
from shapely.geometry import Polygon
from shapely.geometry.base import BaseGeometry
from traci._trafficlight import Logic as SUMO_Logic
class Lane:
@ -18,8 +18,9 @@ class Lane:
"""Overrides the default implementation"""
return hash(self.lane_id)
class Phase:
def __init__(self, duration: float, minDuration: float, maxDuration : float, phaseDef: str):
def __init__(self, duration: float, minDuration: float, maxDuration: float, phaseDef: str):
self.duration = duration
self.minDuration = minDuration
self.maxDuration = maxDuration
@ -29,11 +30,13 @@ class Phase:
repr = f'Phase(duration:{self.duration},minDuration:{self.minDuration},maxDuration:{self.maxDuration},phaseDef:{self.phaseDef})'
return str(repr)
class Logic:
def __init__(self, logic: SUMO_Logic, phases: Set[Phase]):
self._logic = logic
self._phases: Set[Phase] = phases
class TrafficLight:
def __init__(self, tl_id: str, logics: Set[Logic]):
@ -44,17 +47,18 @@ class TrafficLight:
"""Overrides the default implementation"""
return hash(self.tl_id)
class Emission:
def __init__(self, co2 = 0, co = 0 , nox = 0, hc = 0, pmx = 0):
def __init__(self, co2=0, co=0, nox=0, hc=0, pmx=0):
self.co2 = co2
self.co = co
self.nox = nox
self.hc = hc
self.pmx = pmx
def __add__(self,other):
return Emission(self.co2 + other.co2, self.co + other.co, self.nox + other.nox, self.hc + other.hc, self.pmx + other.pmx)
def __add__(self, other):
return Emission(self.co2 + other.co2, self.co + other.co, self.nox + other.nox, self.hc + other.hc,
self.pmx + other.pmx)
def value(self):
return self.co2 + self.co + self.nox + self.hc + self.pmx
@ -63,6 +67,7 @@ class Emission:
repr = f'Emission(co2={self.co2},co={self.co},nox={self.nox},hc={self.hc},pmx={self.pmx})'
return str(repr)
class Area:
def __init__(self, coords, name, window_size):
@ -73,7 +78,7 @@ class Area:
self.rectangle = Polygon(coords)
self.name = name
self.emissions_by_step = []
self.window = collections.deque(maxlen = window_size)
self.window = collections.deque(maxlen=window_size)
self._lanes: Set[Lane] = set()
self._tls: Set[TrafficLight] = set()
@ -122,6 +127,7 @@ class Area:
(xmax, ymax),
(xmax, ymin)))
class Vehicle:
def __init__(self, veh_id: int, pos: Tuple[float, float]):