mirror of
https://github.com/Ahp06/SUMO_Emissions.git
synced 2024-11-21 19:16:30 +00:00
Added docstrings to all methods of the application
This commit is contained in:
Ahp06
parent
0ba52d2c9a
commit
5b812d2dc1
@ -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
|
||||
|
||||
|
||||
"""
|
||||
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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
}
|
||||
@ -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:])
|
||||
|
||||
@ -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__)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user