From 5b812d2dc17dedfd04db84ada5906b1bf5b78ac6 Mon Sep 17 00:00:00 2001 From: Ahp06 Date: Thu, 17 Jan 2019 18:02:24 +0100 Subject: [PATCH] Added docstrings to all methods of the application --- sumo_project/actions.py | 221 +++++---- sumo_project/config.py | 250 +++++----- sumo_project/configs/default_config.json | 42 +- sumo_project/emissions.py | 562 +++++++++++++---------- sumo_project/model.py | 419 +++++++++++------ 5 files changed, 900 insertions(+), 594 deletions(-) diff --git a/sumo_project/actions.py b/sumo_project/actions.py index 302a9ce..14de80d 100644 --- a/sumo_project/actions.py +++ b/sumo_project/actions.py @@ -1,85 +1,136 @@ -""" -Created on 17 oct. 2018 - -@author: Axel Huynh-Phuc, Thibaud Gasser -""" -import traci -from traci._trafficlight import Logic -from typing import Iterable - -from shapely.geometry.linestring import LineString - -from model import Area, Vehicle - - -def compute_edge_weight(edge_id): - - co2 = traci.edge.getCO2Emission(edge_id) - co = traci.edge.getCOEmission(edge_id) - nox = traci.edge.getNOxEmission(edge_id) - hc = traci.edge.getHCEmission(edge_id) - pmx = traci.edge.getPMxEmission(edge_id) - - return (co2 + co + nox + hc + pmx) - -def adjust_edges_weights(area): - area.weight_adjusted = True - for lane in area._lanes: - edge_id = traci.lane.getEdgeID(lane.lane_id) - weight = compute_edge_weight(edge_id) # by default edges weight = length/mean speed - traci.edge.setEffort(edge_id, weight) - - for veh_id in traci.vehicle.getIDList(): - traci.vehicle.rerouteEffort(veh_id) - -def limit_speed_into_area(area: Area, vehicles: Iterable[Vehicle], speed_rf): - area.limited_speed = True - for lane in area._lanes: - traci.lane.setMaxSpeed(lane.lane_id, speed_rf * lane.initial_max_speed) - -def modifyLogic(logic, rf): #rf for "reduction factor" - new_phases = [] - for phase in logic._phases: - new_phase = traci.trafficlight.Phase(phase.duration*rf,phase.minDuration*rf,phase.maxDuration*rf,phase.phaseDef) - new_phases.append(new_phase) - - return traci.trafficlight.Logic("new-program", 0 , 0 , 0 , new_phases) - -def adjust_traffic_light_phase_duration(area, reduction_factor): - area.tls_adjusted = True - for tl in area._tls: - for logic in tl._logics: - traci.trafficlights.setCompleteRedYellowGreenDefinition(tl.tl_id, modifyLogic(logic,reduction_factor)) - -def count_vehicles_in_area(area): - vehicles_in_area = 0 - for lane in area._lanes: - vehicles_in_area += traci.lane.getLastStepVehicleNumber(lane.lane_id) - return vehicles_in_area - -def lock_area(area): - area.locked = True - for lane in area._lanes: - traci.lane.setDisallowed(lane.lane_id, 'passenger') - -def reverse_actions(area): - #Reset max speed to original - if area.limited_speed: - area.limited_speed = False - for lane in area._lanes: - traci.lane.setMaxSpeed(lane.lane_id, lane.initial_max_speed) - - #Reset traffic lights initial duration - if area.tls_adjusted: - area.tls_adjusted = False - for tl in area._tls: - for initial_logic in tl._logics: - traci.trafficlights.setCompleteRedYellowGreenDefinition(tl.tl_id, initial_logic._logic) - - #Unlock the area - if area.locked: - area.locked = False - for lane in area._lanes: - traci.lane.setAllowed(lane.lane_id, '') #empty means all classes are allowed - - \ No newline at end of file +""" +Created on 17 oct. 2018 + +@author: Axel Huynh-Phuc, Thibaud Gasser +""" + +from typing import Iterable + +import traci +from model import Area, Vehicle + +""" +This module defines all possible actions on the simulation +""" + +def compute_edge_weight(edge_id): + """ + Sum the different pollutant emissions on the edge with the identifier edge_id + :param edge_id: The edge ID + :return: The sum (in mg) of all pollutant emissions + """ + co2 = traci.edge.getCO2Emission(edge_id) + co = traci.edge.getCOEmission(edge_id) + nox = traci.edge.getNOxEmission(edge_id) + hc = traci.edge.getHCEmission(edge_id) + pmx = traci.edge.getPMxEmission(edge_id) + + return co2 + co + nox + hc + pmx + + +def adjust_edges_weights(area): + """ + Changes the edge weight of all edges into the area + :param area: The Area object + :return: + """ + area.weight_adjusted = True + for lane in area._lanes: + edge_id = traci.lane.getEdgeID(lane.lane_id) + weight = compute_edge_weight(edge_id) # by default edges weight = length/mean speed + traci.edge.setEffort(edge_id, weight) + + for veh_id in traci.vehicle.getIDList(): + traci.vehicle.rerouteEffort(veh_id) + + +def limit_speed_into_area(area: Area, speed_rf): + """ + Limit the speed into the area by speed_rf factor + :param area: The Area object + :param speed_rf: The speed reduction factor (must be positive) + :return: + """ + area.limited_speed = True + for lane in area._lanes: + traci.lane.setMaxSpeed(lane.lane_id, speed_rf * lane.initial_max_speed) + + +def modifyLogic(logic, rf): + """ + Change the logic of a traffic light by decreasing the overall duration of the traffic light + :param logic: The Logic object + :param rf: The reduction factor (must be positive) + :return: A new Logic object with all phases modified + """ + new_phases = [] + for phase in logic._phases: + new_phase = traci.trafficlight.Phase(phase.duration * rf, phase.minDuration * rf, phase.maxDuration * rf, + phase.phaseDef) + new_phases.append(new_phase) + + return traci.trafficlight.Logic("new-program", 0, 0, 0, new_phases) + + +def adjust_traffic_light_phase_duration(area, reduction_factor): + """ + Set all logics modification on traffic lights into the area + :param area: The Area object + :param reduction_factor: The reduction factor (must be positive) + :return: + """ + area.tls_adjusted = True + for tl in area._tls: + for logic in tl._logics: + traci.trafficlights.setCompleteRedYellowGreenDefinition(tl.tl_id, modifyLogic(logic, reduction_factor)) + + +def count_vehicles_in_area(area): + """ + Count the vehicles number into the area + :param area: The Area object + :return: The number of vehicles into the area + """ + vehicles_in_area = 0 + for lane in area._lanes: + vehicles_in_area += traci.lane.getLastStepVehicleNumber(lane.lane_id) + return vehicles_in_area + + +def lock_area(area): + """ + Prohibits access to the area to a particular vehicle class + NOT FIXED : Some vehicles continue to go into the area if they can not turn around and stay there + :param area: The Area object + :return: + """ + area.locked = True + for lane in area._lanes: + # The passenger class is an example, you have to adapt this code + traci.lane.setDisallowed(lane.lane_id, 'passenger') + + +def reverse_actions(area): + """ + Reverse all actions made in an area + :param area: The Area object + :return: + """ + # Reset max speed to original + if area.limited_speed: + area.limited_speed = False + for lane in area._lanes: + traci.lane.setMaxSpeed(lane.lane_id, lane.initial_max_speed) + + # Reset traffic lights initial duration + if area.tls_adjusted: + area.tls_adjusted = False + for tl in area._tls: + for initial_logic in tl._logics: + traci.trafficlights.setCompleteRedYellowGreenDefinition(tl.tl_id, initial_logic._logic) + + # Unlock the area + if area.locked: + area.locked = False + for lane in area._lanes: + traci.lane.setAllowed(lane.lane_id, '') # empty means all classes are allowed diff --git a/sumo_project/config.py b/sumo_project/config.py index ef2fa0c..e943ba3 100644 --- a/sumo_project/config.py +++ b/sumo_project/config.py @@ -1,107 +1,143 @@ -""" -Global configuration for the simulation -""" - -import datetime -import json -import logging -import os -import sys - -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) - - def __init__(self): - """Default constructor""" - - def import_config_file(self, config_file): - with open(config_file, 'r') as f: - data = json.load(f) - - self._SUMOCMD = data["_SUMOCMD"] - self._SUMOCFG = data["_SUMOCFG"] - - self.areas_number = data["areas_number"] - self.emissions_threshold = data["emissions_threshold"] - self.n_steps = data["n_steps"] - self.window_size = data["window_size"] - - self.without_actions_mode = data["without_actions_mode"] - self.limit_speed_mode = data["limit_speed_mode"] - self.speed_rf = data["speed_rf"] - self.adjust_traffic_light_mode = data["adjust_traffic_light_mode"] - self.trafficLights_duration_rf = data["trafficLights_duration_rf"] - self.weight_routing_mode = data["weight_routing_mode"] - self.lock_area_mode = data["lock_area_mode"] - - self.check_config() - - def check_config(self): - # 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 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 ( - 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},' - f'RF = {self.trafficLights_duration_rf * 100}%\n' - ) - - def init_traci(self): - 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'") - - 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): - now = datetime.datetime.now() - current_date = now.strftime("%Y_%m_%d_%H_%M_%S") - - if not os.path.exists('logs'): - os.makedirs('logs') - - log_filename = f'logs/sumo_logs_{current_date}.log' - - logger = logging.getLogger("sumo_logger") - logger.setLevel(logging.INFO) - formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") - - if save_logs: - file_handler = logging.FileHandler(log_filename) - file_handler.setFormatter(formatter) - logger.addHandler(file_handler) - - handler = logging.StreamHandler() - handler.setFormatter(formatter) - logger.addHandler(handler) - - return logger - - def get_ref_emissions(self): - if self.n_steps == 200: - return self.ref200 +""" +Created on 17 oct. 2018 + +@author: Axel Huynh-Phuc, Thibaud Gasser +""" + +import datetime +import json +import logging +import os +import sys + +from model import Emission + +""" +This module defines the global configuration for the simulation +""" + + +class Config: + """ + The Config class defines all simulation properties that can be changed + """ + + # 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) + + def __init__(self): + """ + Default constructor + """ + + def import_config_file(self, config_file): + """ + Import your configuration file in JSON format + :param config_file: The path to your configuration file + :return: + """ + with open(config_file, 'r') as f: + data = json.load(f) + + self._SUMOCMD = data["_SUMOCMD"] + self._SUMOCFG = data["_SUMOCFG"] + + self.areas_number = data["areas_number"] + self.emissions_threshold = data["emissions_threshold"] + self.n_steps = data["n_steps"] + self.window_size = data["window_size"] + + self.without_actions_mode = data["without_actions_mode"] + self.limit_speed_mode = data["limit_speed_mode"] + self.speed_rf = data["speed_rf"] + self.adjust_traffic_light_mode = data["adjust_traffic_light_mode"] + self.trafficLights_duration_rf = data["trafficLights_duration_rf"] + self.weight_routing_mode = data["weight_routing_mode"] + self.lock_area_mode = data["lock_area_mode"] + + self.check_config() + + def check_config(self): + """ + Check the relevance of user configuration choices + :return: + """ + # 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 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: All properties chosen by the user + """ + 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},' + f'RF = {self.trafficLights_duration_rf * 100}%\n' + ) + + def init_traci(self): + """ + Init the Traci API + :return: + """ + 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'") + + 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): + """ + Init the application logger + :param save_logs: If save_logs is True, it will save the logs into the logs directory + :return: + """ + now = datetime.datetime.now() + current_date = now.strftime("%Y_%m_%d_%H_%M_%S") + + if not os.path.exists('logs'): + os.makedirs('logs') + + log_filename = f'logs/sumo_logs_{current_date}.log' + + logger = logging.getLogger("sumo_logger") + logger.setLevel(logging.INFO) + formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") + + if save_logs: + file_handler = logging.FileHandler(log_filename) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + handler = logging.StreamHandler() + handler.setFormatter(formatter) + logger.addHandler(handler) + + return logger + + def get_ref_emissions(self): + """ + :return: Return the sum of all emissions (in mg) from the simulation of reference + """ + if self.n_steps == 200: + return self.ref200 diff --git a/sumo_project/configs/default_config.json b/sumo_project/configs/default_config.json index 4bf6bf2..4565a2a 100644 --- a/sumo_project/configs/default_config.json +++ b/sumo_project/configs/default_config.json @@ -1,22 +1,22 @@ -{ - "_SUMOCMD": "sumo", - "_SUMOCFG": "simulations/mulhouse_simulation/osm.sumocfg", - - "areas_number": 10, - "emissions_threshold": 500000, - "n_steps": 200, - "window_size":100, - - "without_actions_mode": true, - - "limit_speed_mode": false, - "speed_rf": 0.1, - - "adjust_traffic_light_mode": false, - "trafficLights_duration_rf": 0.2, - - "weight_routing_mode": false, - - "lock_area_mode": false - +{ + "_SUMOCMD": "sumo", + "_SUMOCFG": "simulations/mulhouse_simulation/osm.sumocfg", + + "areas_number": 10, + "emissions_threshold": 500000, + "n_steps": 200, + "window_size":100, + + "without_actions_mode": false, + + "limit_speed_mode": true, + "speed_rf": 0.1, + + "adjust_traffic_light_mode": true, + "trafficLights_duration_rf": 0.2, + + "weight_routing_mode": false, + + "lock_area_mode": false + } \ No newline at end of file diff --git a/sumo_project/emissions.py b/sumo_project/emissions.py index 1293e80..e49d3ee 100644 --- a/sumo_project/emissions.py +++ b/sumo_project/emissions.py @@ -1,242 +1,320 @@ -import argparse -import csv -import datetime -import itertools -import os -import sys -import time -from typing import List - -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 - -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 - for i in range(areas_number): - for j in range(areas_number): - # bounds coordinates for the area : (xmin, ymin, xmax, ymax) - ar_bounds = ((i * width, j * height), (i * width, (j + 1) * height), - ((i + 1) * width, (j + 1) * height), ((i + 1) * width, j * height)) - name = 'Area ({},{})'.format(i, j) - area = Area(ar_bounds, name, window_size) - grid.append(area) - traci.polygon.add(area.name, ar_bounds, (255, 0, 0)) - return grid - - -def get_all_lanes() -> List[Lane]: - lanes = [] - for lane_id in traci.lane.getIDList(): - polygon_lane = LineString(traci.lane.getShape(lane_id)) - initial_max_speed = traci.lane.getMaxSpeed(lane_id) - lanes.append(Lane(lane_id, polygon_lane, initial_max_speed)) - return lanes - - -def parse_phase(phase_repr): - duration = search('duration: {:f}', phase_repr) - min_duration = search('minDuration: {:f}', phase_repr) - max_duration = search('maxDuration: {:f}', phase_repr) - phase_def = search('phaseDef: {}\n', phase_repr) - - if phase_def is None: - phase_def = '' - else: - phase_def = phase_def[0] - - return Phase(duration[0], min_duration[0], max_duration[0], phase_def) - - -def add_data_to_areas(areas: List[Area]): - lanes = get_all_lanes() - for area in areas: - for lane in lanes: # add lanes - if area.rectangle.intersects(lane.polygon): - area.add_lane(lane) - for tl_id in traci.trafficlight.getIDList(): # add traffic lights - if lane.lane_id in traci.trafficlight.getControlledLanes(tl_id): - logics = [] - for l in traci.trafficlight.getCompleteRedYellowGreenDefinition(tl_id): # add logics - phases = [] - for phase in traci.trafficlight.Logic.getPhases(l): # add phases to logics - phases.append(parse_phase(phase.__repr__())) - logics.append(Logic(l, phases)) - area.add_tl(TrafficLight(tl_id, logics)) - - -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) - - -def get_all_vehicles() -> List[Vehicle]: - vehicles = list() - for veh_id in traci.vehicle.getIDList(): - veh_pos = traci.vehicle.getPosition(veh_id) - vehicle = Vehicle(veh_id, veh_pos) - vehicle.emissions = compute_vehicle_emissions(veh_id) - vehicles.append(vehicle) - return vehicles - - -def get_emissions(grid: List[Area], vehicles: List[Vehicle], current_step, config, logger): - for area in grid: - total_emissions = Emission() - for vehicle in vehicles: - if vehicle.pos in area: - total_emissions += vehicle.emissions - - area.emissions_by_step.append(total_emissions) - - 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}%') - 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}%') - actions.adjust_traffic_light_phase_duration(area, config.trafficLights_duration_rf) - - if config.lock_area_mode and not area.locked: - if actions.count_vehicles_in_area(area): - logger.info(f'Action - {area.name} blocked') - actions.lock_area(area) - - if config.weight_routing_mode and not area.weight_adjusted: - actions.adjust_edges_weights(area) - - traci.polygon.setFilled(area.name, True) - - else: - actions.reverse_actions(area) - traci.polygon.setFilled(area.name, False) - - -def get_reduction_percentage(ref, total): - return (ref - total) / ref * 100 - - -def export_data_to_csv(config, grid): - csv_dir = os.path.join(SCRIPTDIR, 'csv') - if not os.path.exists(csv_dir): - os.mkdir(csv_dir) - - now = datetime.datetime.now().strftime("%Y_%m_%d_%H_%M_%S") - - with open(f'csv/{now}.csv', 'w') as f: - writer = csv.writer(f) - # Write CSV headers - writer.writerow(itertools.chain(('Step',), (a.name for a in grid))) - # Write all areas emission value for each step - for step in range(config.n_steps): - em_for_step = (f'{a.emissions_by_step[step].value():.3f}' for a in grid) - writer.writerow(itertools.chain((step,), em_for_step)) - - -def run(config, logger): - grid = list() - try: - traci.start(config.sumo_cmd) - logger.info(f'Loaded simulation file : {config._SUMOCFG}') - logger.info('Loading data for the simulation') - start = time.perf_counter() - - grid = init_grid(traci.simulation.getNetBoundary(), config.areas_number, config.window_size) - add_data_to_areas(grid) - - loading_time = round(time.perf_counter() - start, 2) - logger.info(f'Data loaded ({loading_time}s)') - - logger.info('Simulation started...') - step = 0 - while step < config.n_steps: # traci.simulation.getMinExpectedNumber() > 0: - traci.simulationStep() - - vehicles = get_all_vehicles() - get_emissions(grid, vehicles, step, config, logger) - step += 1 - - print(f'step = {step}/{config.n_steps}', end='\r') - - finally: - traci.close(False) - 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}') - - total_emissions = Emission() - for area in grid: - total_emissions += area.sum_all_emissions() - - logger.info(f'Total emissions = {total_emissions.value()} mg') - - 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'-> 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)} %') - logger.info(f'-> HC emissions = {get_reduction_percentage(ref.hc, total_emissions.hc)} %') - logger.info(f'-> PMx emissions = {get_reduction_percentage(ref.pmx, total_emissions.pmx)} %') - - -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='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") - - -def main(args): - parser = argparse.ArgumentParser(description="") - add_options(parser) - args = parser.parse_args(args) - - config = Config() - config.import_config_file(args.configfile) - config.init_traci() - logger = config.init_logger(save_logs=args.save) - - if args.ref: - config.without_actions_mode = True - logger.info(f'Reference simulation') - - if args.steps: - config.n_steps = args.steps - - if args.gui: - config._SUMOCMD = "sumo-gui" - - config.check_config() - - logger.info(f'Loaded configuration file : {args.configfile}') - logger.info(f'Simulated time : {args.steps}s') - run(config, logger) - - -if __name__ == '__main__': - main(sys.argv[1:]) +""" +Created on 17 oct. 2018 + +@author: Axel Huynh-Phuc, Thibaud Gasser +""" + +import argparse +import csv +import datetime +import itertools +import os +import sys +import time +from typing import List + +import actions +import traci +from config import Config +from model import Area, Vehicle, Lane, TrafficLight, Phase, Logic, Emission +from parse import search +from shapely.geometry import LineString + +""" +This module defines the entry point of the application +""" + +def init_grid(simulation_bounds, areas_number, window_size): + """ + Initialize the grid of the loaded map from the configuration + :param simulation_bounds: The map bounds + :param areas_number: The number of areas + :param window_size: The size of the acquisition window + :return: A list of areas + """ + grid = list() + width = simulation_bounds[1][0] / areas_number + height = simulation_bounds[1][1] / areas_number + for i in range(areas_number): + for j in range(areas_number): + # bounds coordinates for the area : (xmin, ymin, xmax, ymax) + ar_bounds = ((i * width, j * height), (i * width, (j + 1) * height), + ((i + 1) * width, (j + 1) * height), ((i + 1) * width, j * height)) + name = 'Area ({},{})'.format(i, j) + area = Area(ar_bounds, name, window_size) + grid.append(area) + traci.polygon.add(area.name, ar_bounds, (255, 0, 0)) + return grid + + +def get_all_lanes() -> List[Lane]: + """ + Recover and creates a list of Lane objects + :return: The lanes list + """ + lanes = [] + for lane_id in traci.lane.getIDList(): + polygon_lane = LineString(traci.lane.getShape(lane_id)) + initial_max_speed = traci.lane.getMaxSpeed(lane_id) + lanes.append(Lane(lane_id, polygon_lane, initial_max_speed)) + return lanes + + +def parse_phase(phase_repr): + """ + Because the SUMO object Phase does not contain accessors, + we parse the string representation to retrieve data members. + :param phase_repr: The Phase string representation + :return: An new Phase instance + """ + duration = search('duration: {:f}', phase_repr) + min_duration = search('minDuration: {:f}', phase_repr) + max_duration = search('maxDuration: {:f}', phase_repr) + phase_def = search('phaseDef: {}\n', phase_repr) + + if phase_def is None: + phase_def = '' + else: + phase_def = phase_def[0] + + return Phase(duration[0], min_duration[0], max_duration[0], phase_def) + + +def add_data_to_areas(areas: List[Area]): + """ + Adds all recovered data to different areas + :param areas: The list of areas + :return: + """ + lanes = get_all_lanes() + for area in areas: + for lane in lanes: # add lanes + if area.rectangle.intersects(lane.polygon): + area.add_lane(lane) + for tl_id in traci.trafficlight.getIDList(): # add traffic lights + if lane.lane_id in traci.trafficlight.getControlledLanes(tl_id): + logics = [] + for l in traci.trafficlight.getCompleteRedYellowGreenDefinition(tl_id): # add logics + phases = [] + for phase in traci.trafficlight.Logic.getPhases(l): # add phases to logics + phases.append(parse_phase(phase.__repr__())) + logics.append(Logic(l, phases)) + area.add_tl(TrafficLight(tl_id, logics)) + + +def compute_vehicle_emissions(veh_id): + """ + Recover the emissions of different pollutants from a vehicle and create an Emission instance + :param veh_id: + :return: A new Emission instance + """ + 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) + + +def get_all_vehicles() -> List[Vehicle]: + """ + Recover all useful information about vehicles and creates a vehicles list + :return: A list of vehicles instances + """ + vehicles = list() + for veh_id in traci.vehicle.getIDList(): + veh_pos = traci.vehicle.getPosition(veh_id) + vehicle = Vehicle(veh_id, veh_pos) + vehicle.emissions = compute_vehicle_emissions(veh_id) + vehicles.append(vehicle) + return vehicles + + +def get_emissions(grid: List[Area], vehicles: List[Vehicle], current_step, config, logger): + """ + For each area retrieves the acquired emissions in the window, + and acts according to the configuration chosen by the user + :param grid: The list of areas + :param vehicles: The list of vehicles + :param current_step: The simulation current step + :param config: The simulation configuration + :param logger: The simulation logger + :return: + """ + for area in grid: + total_emissions = Emission() + for vehicle in vehicles: + if vehicle.pos in area: + total_emissions += vehicle.emissions + + area.emissions_by_step.append(total_emissions) + + if area.sum_emissions_into_window(current_step) >= 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}%') + actions.limit_speed_into_area(area, 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}%') + actions.adjust_traffic_light_phase_duration(area, config.trafficLights_duration_rf) + + if config.lock_area_mode and not area.locked: + if actions.count_vehicles_in_area(area): + logger.info(f'Action - {area.name} blocked') + actions.lock_area(area) + + if config.weight_routing_mode and not area.weight_adjusted: + actions.adjust_edges_weights(area) + + traci.polygon.setFilled(area.name, True) + + else: + actions.reverse_actions(area) + traci.polygon.setFilled(area.name, False) + + +def get_reduction_percentage(ref, total): + """ + Return the reduction percentage of total emissions between reference and an other simulation + :param ref: + :param total: + :return: + """ + return (ref - total) / ref * 100 + + +def export_data_to_csv(config, grid): + """ + Export all Emission objects as a CSV file into the csv directory + :param config: The simulation configuration + :param grid: The list of areas + :return: + """ + csv_dir = 'csv' + if not os.path.exists(csv_dir): + os.mkdir(csv_dir) + + now = datetime.datetime.now().strftime("%Y_%m_%d_%H_%M_%S") + + with open(f'csv/{now}.csv', 'w') as f: + writer = csv.writer(f) + # Write CSV headers + writer.writerow(itertools.chain(('Step',), (a.name for a in grid))) + # Write all areas emission value for each step + for step in range(config.n_steps): + em_for_step = (f'{a.emissions_by_step[step].value():.3f}' for a in grid) + writer.writerow(itertools.chain((step,), em_for_step)) + + +def run(config, logger): + """ + Run the simulation with the configuration chosen + :param config: The simulation configuration + :param logger: The simulation logger + :return: + """ + grid = list() + try: + traci.start(config.sumo_cmd) + logger.info(f'Loaded simulation file : {config._SUMOCFG}') + logger.info('Loading data for the simulation') + start = time.perf_counter() + + grid = init_grid(traci.simulation.getNetBoundary(), config.areas_number, config.window_size) + add_data_to_areas(grid) + + loading_time = round(time.perf_counter() - start, 2) + logger.info(f'Data loaded ({loading_time}s)') + + logger.info('Simulation started...') + step = 0 + while step < config.n_steps: # traci.simulation.getMinExpectedNumber() > 0: + traci.simulationStep() + + vehicles = get_all_vehicles() + get_emissions(grid, vehicles, step, config, logger) + step += 1 + + print(f'step = {step}/{config.n_steps}', end='\r') + + finally: + traci.close(False) + 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}') + + total_emissions = Emission() + for area in grid: + total_emissions += area.sum_all_emissions() + + logger.info(f'Total emissions = {total_emissions.value()} mg') + + 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'-> 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)} %') + logger.info(f'-> HC emissions = {get_reduction_percentage(ref.hc, total_emissions.hc)} %') + logger.info(f'-> PMx emissions = {get_reduction_percentage(ref.pmx, total_emissions.pmx)} %') + + +def add_options(parser): + """ + Add command line options + :param parser: The command line parser + :return: + """ + 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='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") + + +def main(args): + """ + The entry point of the application + :param args: Command line options + :return: + """ + parser = argparse.ArgumentParser(description="") + add_options(parser) + args = parser.parse_args(args) + + config = Config() + config.import_config_file(args.configfile) + config.init_traci() + logger = config.init_logger(save_logs=args.save) + + if args.ref: + config.without_actions_mode = True + logger.info(f'Reference simulation') + + if args.steps: + config.n_steps = args.steps + + if args.gui: + config._SUMOCMD = "sumo-gui" + + config.check_config() + + logger.info(f'Loaded configuration file : {args.configfile}') + logger.info(f'Simulated time : {args.steps}s') + run(config, logger) + + +if __name__ == '__main__': + main(sys.argv[1:]) diff --git a/sumo_project/model.py b/sumo_project/model.py index 3b4bc82..cf5ae04 100644 --- a/sumo_project/model.py +++ b/sumo_project/model.py @@ -1,139 +1,280 @@ -import collections -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: - - def __init__(self, lane_id: str, polygon: LineString, initial_max_speed: float): - self.polygon = polygon - self.lane_id = lane_id - self.initial_max_speed = initial_max_speed - - def __hash__(self): - """Overrides the default implementation""" - return hash(self.lane_id) - - -class Phase: - def __init__(self, duration: float, minDuration: float, maxDuration: float, phaseDef: str): - self.duration = duration - self.minDuration = minDuration - self.maxDuration = maxDuration - self.phaseDef = phaseDef - - def __repr__(self) -> str: - 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]): - self.tl_id = tl_id - self._logics: Set[Logic] = logics - - def __hash__(self): - """Overrides the default implementation""" - return hash(self.tl_id) - - -class Emission: - 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 value(self): - return self.co2 + self.co + self.nox + self.hc + self.pmx - - def __repr__(self) -> str: - 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): - self.limited_speed = False - self.locked = False - self.tls_adjusted = False - self.weight_adjusted = False - self.rectangle = Polygon(coords) - self.name = name - self.emissions_by_step = [] - self.window = collections.deque(maxlen=window_size) - self._lanes: Set[Lane] = set() - self._tls: Set[TrafficLight] = set() - - def __eq__(self, other): - return self.rectangle.__eq__(other) - - def __contains__(self, item): - return self.rectangle.contains(item) - - @property - def bounds(self): - return self.rectangle.bounds - - def intersects(self, other: BaseGeometry) -> bool: - return self.rectangle.intersects(other) - - def add_lane(self, lane: Lane): - self._lanes.add(lane) - - def add_tl(self, tl: TrafficLight): - self._tls.add(tl) - - def remove_lane(self, lane: Lane): - self._lanes.remove(lane) - - def sum_all_emissions(self): - sum = Emission() - for emission in self.emissions_by_step: - sum += emission - return sum - - def sum_emissions_into_window(self, current_step, window_size): - - self.window.appendleft(self.emissions_by_step[current_step].value()) - - sum = 0 - for i in range(self.window.__len__()): - sum += self.window[i] - return sum - - @classmethod - def from_bounds(cls, xmin, ymin, xmax, ymax): - return cls(( - (xmin, ymin), - (xmin, ymax), - (xmax, ymax), - (xmax, ymin))) - - -class Vehicle: - - def __init__(self, veh_id: int, pos: Tuple[float, float]): - self.emissions: Emission = Emission() - self.veh_id = veh_id - self.pos = Point(pos) - - def __repr__(self) -> str: - return str(self.__dict__) +""" +Created on 17 oct. 2018 + +@author: Axel Huynh-Phuc, Thibaud Gasser +""" + +import collections +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 + +""" +This module defines the business model of our application +""" + + +class Lane: + """ + The Lane class includes the polygon defining the lane + and keep in memory the initial maximum speed on the lane + """ + + def __init__(self, lane_id: str, polygon: LineString, initial_max_speed: float): + """ + Lane constructor + + :param lane_id: The ID of the lane + :param polygon: The polygon defining the shape of the lane + :param initial_max_speed: The initial maximum speed + """ + self.polygon = polygon + self.lane_id = lane_id + self.initial_max_speed = initial_max_speed + + def __hash__(self): + """Overrides the default implementation""" + return hash(self.lane_id) + + +class Phase: + """ + The Phase class defines a phase of a traffic light + """ + + def __init__(self, duration: float, minDuration: float, maxDuration: float, phaseDef: str): + """ + Phase constructor + + :param duration: The duration of the phase (in seconds) + :param minDuration: The minimum duration of the phase + :param maxDuration: The maximum duration of the phase + :param phaseDef: The definition of the phase, following the definition rules of SUMO + (See : http://sumo.dlr.de/wiki/Simulation/Traffic_Lights#.3Cphase.3E_Attributes) + """ + + self.duration = duration + self.minDuration = minDuration + self.maxDuration = maxDuration + self.phaseDef = phaseDef + + def __repr__(self) -> str: + """ + :return: The Phase string representation + """ + repr = f'Phase(duration:{self.duration},minDuration:{self.minDuration},maxDuration:{self.maxDuration},phaseDef:{self.phaseDef})' + return str(repr) + + +class Logic: + """ + The Logic class defines the strategy of a traffic light. + This class includes the Logic instance of SUMO with all phases corresponding to it. + A Logic object contains multiple phases. + """ + + def __init__(self, logic: SUMO_Logic, phases: Set[Phase]): + """ + Logic constructor + :param logic: The SUMO Logic object + :param phases: The list of phases belonging to this logic + """ + self._logic = logic + self._phases: Set[Phase] = phases + + +class TrafficLight: + """ + This TrafficLight class defines a traffic light + """ + + def __init__(self, tl_id: str, logics: Set[Logic]): + """ + TrafficLight constructor + :param tl_id: The traffic light ID + :param logics: The list of logics belonging to the traffic light + """ + self.tl_id = tl_id + self._logics: Set[Logic] = logics + + def __hash__(self): + """Overrides the default implementation""" + return hash(self.tl_id) + + +class Emission: + """ + This class defines the different pollutant emissions + """ + + def __init__(self, co2=0, co=0, nox=0, hc=0, pmx=0): + """ + Emission constructor + :param co2: Quantity of CO2(in mg) + :param co: Quantity of C0(in mg) + :param nox: Quantity of Nox(in mg) + :param hc: Quantity of HC(in mg) + :param pmx: Quantity of PMx(in mg) + """ + self.co2 = co2 + self.co = co + self.nox = nox + self.hc = hc + self.pmx = pmx + + def __add__(self, other): + """ + Add two emission objects + :param other: The other Emission object to add + :return: A new object whose emission values ​​are the sum of both Emission object + """ + 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: The sum of all emissions + """ + return self.co2 + self.co + self.nox + self.hc + self.pmx + + def __repr__(self) -> str: + """ + :return: The Emission string representation + """ + repr = f'Emission(co2={self.co2},co={self.co},nox={self.nox},hc={self.hc},pmx={self.pmx})' + return str(repr) + + +class Area: + """ + The Area class defines a grid area of ​​the map + """ + + def __init__(self, coords, name, window_size): + """ + Area constructor + :param coords: The coordinates of the zone, + defined by the bounds coordinates of this area : (xmin, ymin, xmax, ymax) + :param name: The Area name + :param window_size: The size of the acquisition window + """ + self.limited_speed = False + self.locked = False + self.tls_adjusted = False + self.weight_adjusted = False + self.rectangle = Polygon(coords) + self.name = name + self.emissions_by_step = [] + self.window = collections.deque(maxlen=window_size) + self._lanes: Set[Lane] = set() + self._tls: Set[TrafficLight] = set() + + def __eq__(self, other): + """ + Overrides the equal definition + :param other: The other Area object + :return: True if the two rectangles are equals + """ + return self.rectangle.__eq__(other) + + def __contains__(self, item): + """ + :param item: A position on the map + :return: True if the area contains the item + """ + return self.rectangle.contains(item) + + @property + def bounds(self): + """ + Return the bounds rectangle of this area + :return: + """ + return self.rectangle.bounds + + def intersects(self, other: BaseGeometry) -> bool: + """ + :param other: A BaseGeometry object + :return: True if this area intersects with other + """ + return self.rectangle.intersects(other) + + def add_lane(self, lane: Lane): + """ + Add a new lane object into lanes list + :param lane: A Lane object + :return: + """ + self._lanes.add(lane) + + def add_tl(self, tl: TrafficLight): + """ + Add a new trafficLight object into lanes list + :param tl: A TrafficLight object + :return: + """ + self._tls.add(tl) + + def remove_lane(self, lane: Lane): + """ + Remove a lane from lanes list + :param lane: The Lane object to remove + :return: + """ + self._lanes.remove(lane) + + def sum_all_emissions(self): + """ + Sum all Emissions object from initial step to final step + :return: The sum Emission object + """ + sum = Emission() + for emission in self.emissions_by_step: + sum += emission + return sum + + def sum_emissions_into_window(self, current_step): + """ + Sum all Emissions object into the acquisition window + :param current_step: The current step of the simulation + :return: + """ + self.window.appendleft(self.emissions_by_step[current_step].value()) + + sum = 0 + for i in range(self.window.__len__()): + sum += self.window[i] + return sum + + @classmethod + def from_bounds(cls, xmin, ymin, xmax, ymax): + return cls(( + (xmin, ymin), + (xmin, ymax), + (xmax, ymax), + (xmax, ymin))) + + +class Vehicle: + """ + The Vehicle class defines a vehicle object + """ + + def __init__(self, veh_id: int, pos: Tuple[float, float]): + """ + Vehicle constructor + :param veh_id: The vehicle ID + :param pos: The position of the vehicle one the map + """ + self.emissions: Emission = Emission() + self.veh_id = veh_id + self.pos = Point(pos) + + def __repr__(self) -> str: + """ + :return: The Vehicle string representation + """ + return str(self.__dict__)