mirror of
https://github.com/Ahp06/SUMO_Emissions.git
synced 2024-11-21 19:16:30 +00:00
Axel HUYNH-PHUC
GitHub
commit
b012c4c4a8
@ -1,44 +1,78 @@
|
||||
"""
|
||||
Created on 17 oct. 2018
|
||||
|
||||
@author: Axel Huynh-Phuc, Thibaud Gasser
|
||||
"""
|
||||
from typing import Iterable
|
||||
|
||||
import traci
|
||||
from shapely.geometry.linestring import LineString
|
||||
|
||||
from model import Area, Vehicle
|
||||
|
||||
|
||||
def remove_vehicle(veh_id):
|
||||
traci.vehicle.remove(veh_id, traci.constants.REMOVE_PARKING)
|
||||
|
||||
|
||||
def lanes_in_area(area):
|
||||
for lane_id in traci.lane.getIDList():
|
||||
polygon_lane = LineString(traci.lane.getShape(lane_id))
|
||||
if area.rectangle.intersects(polygon_lane):
|
||||
yield lane_id
|
||||
|
||||
|
||||
def compute_edge_weight(edge_id):
|
||||
return (traci.edge.getCOEmission(edge_id)
|
||||
+ traci.edge.getNOxEmission(edge_id)
|
||||
+ traci.edge.getHCEmission(edge_id)
|
||||
+ traci.edge.getPMxEmission(edge_id)
|
||||
+ traci.edge.getCO2Emission(edge_id))
|
||||
|
||||
|
||||
def adjust_edges_weights():
|
||||
for edge_id in traci.edge.getIDList():
|
||||
weight = compute_edge_weight(edge_id) # by default edges weight = length/mean speed
|
||||
traci.edge.adaptTraveltime(edge_id, weight)
|
||||
|
||||
|
||||
def lock_area(area: Area):
|
||||
max_speed = 30
|
||||
print(f'Setting max speed into {area.name} to {max_speed} km/h')
|
||||
area.locked = True
|
||||
for lane in area._lanes:
|
||||
traci.lane.setMaxSpeed(lane.lane_id, max_speed / 3.6)
|
||||
"""
|
||||
Created on 17 oct. 2018
|
||||
|
||||
@author: Axel Huynh-Phuc, Thibaud Gasser
|
||||
"""
|
||||
from typing import Iterable
|
||||
|
||||
import traci
|
||||
from shapely.geometry.linestring import LineString
|
||||
|
||||
from model import Area, Vehicle
|
||||
from traci._trafficlight import Logic
|
||||
|
||||
def compute_edge_weight(edge_id):
|
||||
return (traci.edge.getCOEmission(edge_id)
|
||||
+ traci.edge.getNOxEmission(edge_id)
|
||||
+ traci.edge.getHCEmission(edge_id)
|
||||
+ traci.edge.getPMxEmission(edge_id)
|
||||
+ traci.edge.getCO2Emission(edge_id))/(traci.edge.getLaneNumber(edge_id))
|
||||
|
||||
def adjust_edges_weights():
|
||||
for edge_id in traci.edge.getIDList():
|
||||
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 not area.limited_speed:
|
||||
area.limited_speed = False
|
||||
for lane in area.lanes:
|
||||
traci.lane.setMaxSpeed(lane.lane_id, lane.initial_max_speed / 3.6)
|
||||
|
||||
#Reset traffic lights initial duration
|
||||
if not area.tls_adjusted:
|
||||
area.tls_adjusted = False
|
||||
for initial_logic in tl._logics:
|
||||
traci.trafficlights.setCompleteRedYellowGreenDefinition(tl.tl_id, initial_logic)
|
||||
|
||||
#Unlock the area
|
||||
if not area.locked:
|
||||
area.locked = False
|
||||
for lane in area._lanes:
|
||||
traci.lane.setAllowed(lane.lane_id, '') #empty means all classes are allowed
|
||||
|
||||
|
||||
@ -1,26 +1,117 @@
|
||||
"""
|
||||
Global configuration for the simulation
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
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'")
|
||||
|
||||
_SUMOCMD = 'sumo' # use 'sumo-gui' cmd for UI
|
||||
_SUMOCFG = "mulhouse_simulation/osm.sumocfg"
|
||||
CELLS_NUMBER = 10
|
||||
EMISSIONS_THRESHOLD = 500000
|
||||
n_steps = 200
|
||||
|
||||
lock_mode = True
|
||||
routing_mode = False
|
||||
|
||||
sumo_binary = os.path.join(os.environ['SUMO_HOME'], 'bin', _SUMOCMD)
|
||||
sumo_cmd = [sumo_binary, "-c", _SUMOCFG]
|
||||
|
||||
|
||||
"""
|
||||
Global configuration for the simulation
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import datetime
|
||||
import logging
|
||||
|
||||
###############################################################################
|
||||
############################# SIMULATION FILE #################################
|
||||
###############################################################################
|
||||
|
||||
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'")
|
||||
|
||||
_SUMOCMD = 'sumo' # use 'sumo-gui' cmd for UI
|
||||
_SUMOCFG = "mulhouse_simulation/osm.sumocfg"
|
||||
sumo_binary = os.path.join(os.environ['SUMO_HOME'], 'bin', _SUMOCMD)
|
||||
sumo_cmd = [sumo_binary, "-c", _SUMOCFG]
|
||||
|
||||
###############################################################################
|
||||
################################## LOGS #######################################
|
||||
###############################################################################
|
||||
|
||||
now = datetime.datetime.now()
|
||||
current_date = now.strftime("%Y_%m_%d_%H_%M_%S")
|
||||
LOG_FILENAME = f'sumo_logs_{current_date}.log'
|
||||
|
||||
# create logger
|
||||
logger = logging.getLogger("sumo_logger")
|
||||
logger.setLevel(logging.INFO)
|
||||
# create console handler and set level to info
|
||||
handler = logging.FileHandler(LOG_FILENAME)
|
||||
# create formatter
|
||||
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
||||
# add formatter to handler
|
||||
handler.setFormatter(formatter)
|
||||
# add handler to logger
|
||||
logger.addHandler(handler)
|
||||
|
||||
###############################################################################
|
||||
########################## SIMULATION CONFIGURATION ###########################
|
||||
###############################################################################
|
||||
|
||||
CELLS_NUMBER = 10
|
||||
EMISSIONS_THRESHOLD = 500000
|
||||
n_steps = 200
|
||||
|
||||
###############################################################################
|
||||
########################## ACTIONS CONFIGURATION ##############################
|
||||
###############################################################################
|
||||
|
||||
#Set this mode to True if you want running a basic simulation without actions
|
||||
without_actions_mode = False
|
||||
|
||||
#Limit the speed into areas when the threshold is exceeded
|
||||
speed_rf = 0.1
|
||||
limit_speed_mode = True
|
||||
|
||||
#Decrease all traffic lights duration into the area when the threshold is exceeded
|
||||
trafficLights_duration_rf = 0.2
|
||||
adjust_traffic_light_mode = True
|
||||
|
||||
#Vehicles are routed according to the less polluted route (HEAVY)
|
||||
weight_routing_mode = False
|
||||
|
||||
#Lock the area when the threshold is exceeded (NOT FIXED)
|
||||
lock_area_mode = False
|
||||
|
||||
#Weight routing mode cannot be combinated with other actions
|
||||
if weight_routing_mode:
|
||||
limit_speed_mode = False
|
||||
adjust_traffic_light_mode = False
|
||||
|
||||
#If without_actions_mode is choosen
|
||||
if without_actions_mode:
|
||||
limit_speed_mode = False
|
||||
adjust_traffic_light_mode = False
|
||||
weight_routing_mode = False
|
||||
lock_area_mode = False
|
||||
|
||||
###############################################################################
|
||||
########################## SIMULATION REFERENCES ##############################
|
||||
###############################################################################
|
||||
|
||||
# Total of emissions of all pollutants in mg for n steps of simulation without locking areas
|
||||
total_emissions200 = 43970763.15084749
|
||||
total_emissions300 = 87382632.08217141
|
||||
total_emissions400 = 140757491.8489904
|
||||
total_emissions500 = 202817535.43856794
|
||||
|
||||
###############################################################################
|
||||
########################## CONFIGURATION METHODS ##############################
|
||||
###############################################################################
|
||||
|
||||
def get_basics_emissions():
|
||||
if n_steps == 200:
|
||||
return total_emissions200
|
||||
if n_steps == 300:
|
||||
return total_emissions300
|
||||
if n_steps == 400:
|
||||
return total_emissions400
|
||||
if n_steps == 500:
|
||||
return total_emissions500
|
||||
|
||||
def show_config():
|
||||
return (str(f'Grid : {CELLS_NUMBER}x{CELLS_NUMBER}\n')
|
||||
+ str(f'step number = {n_steps}\n')
|
||||
+ str(f'weight routing mode = {weight_routing_mode}\n')
|
||||
+ str(f'lock area mode = {lock_area_mode}\n')
|
||||
+ str(f'limit speed mode = {limit_speed_mode}, RF = {speed_rf*100}%\n')
|
||||
+ str(f'adjust traffic light mode = {adjust_traffic_light_mode} , RF = {trafficLights_duration_rf*100}%\n'))
|
||||
|
||||
|
||||
@ -1,112 +1,154 @@
|
||||
from typing import List
|
||||
|
||||
import traci
|
||||
from shapely.geometry import LineString
|
||||
|
||||
import actions
|
||||
import config
|
||||
import sys
|
||||
from model import Area, Vehicle, Lane
|
||||
|
||||
|
||||
def init_grid(simulation_bounds, cells_number):
|
||||
grid = list()
|
||||
width = simulation_bounds[1][0] / cells_number
|
||||
height = simulation_bounds[1][1] / cells_number
|
||||
for i in range(cells_number):
|
||||
for j in range(cells_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))
|
||||
area = Area(ar_bounds)
|
||||
area.name = 'area ({},{})'.format(i, j)
|
||||
grid.append(area)
|
||||
traci.polygon.add(area.name, ar_bounds, (0, 255, 0))
|
||||
return grid
|
||||
|
||||
|
||||
def compute_vehicle_emissions(veh_id):
|
||||
return (traci.vehicle.getCOEmission(veh_id)
|
||||
+ traci.vehicle.getNOxEmission(veh_id)
|
||||
+ traci.vehicle.getHCEmission(veh_id)
|
||||
+ traci.vehicle.getPMxEmission(veh_id)
|
||||
+ traci.vehicle.getCO2Emission(veh_id))
|
||||
|
||||
|
||||
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)
|
||||
traci.vehicle.setRoutingMode(veh_id, traci.constants.ROUTING_MODE_AGGREGATED)
|
||||
vehicles.append(vehicle)
|
||||
return vehicles
|
||||
|
||||
|
||||
def get_all_lanes() -> List[Lane]:
|
||||
lanes = []
|
||||
for lane_id in traci.lane.getIDList():
|
||||
polygon_lane = LineString(traci.lane.getShape(lane_id))
|
||||
lanes.append(Lane(lane_id, polygon_lane))
|
||||
return lanes
|
||||
|
||||
|
||||
def get_emissions(grid: List[Area], vehicles: List[Vehicle]):
|
||||
for area in grid:
|
||||
for vehicle in vehicles:
|
||||
if vehicle.pos in area:
|
||||
area.emissions += vehicle.emissions
|
||||
if config.lock_mode and area.emissions > config.EMISSIONS_THRESHOLD and not area.locked:
|
||||
actions.lock_area(area)
|
||||
traci.polygon.setColor(area.name, (255, 0, 0))
|
||||
traci.polygon.setFilled(area.name, True)
|
||||
|
||||
|
||||
def add_lanes_to_areas(areas: List[Area]):
|
||||
lanes = get_all_lanes()
|
||||
for area in areas:
|
||||
for lane in lanes:
|
||||
if area.rectangle.intersects(lane.polygon):
|
||||
area.add_lane(lane)
|
||||
|
||||
|
||||
def main():
|
||||
grid = list()
|
||||
try:
|
||||
traci.start(config.sumo_cmd)
|
||||
grid = init_grid(traci.simulation.getNetBoundary(), config.CELLS_NUMBER)
|
||||
add_lanes_to_areas(grid)
|
||||
|
||||
step = 0
|
||||
while step < config.n_steps: # traci.simulation.getMinExpectedNumber() > 0:
|
||||
traci.simulationStep()
|
||||
|
||||
vehicles = get_all_vehicles()
|
||||
get_emissions(grid, vehicles)
|
||||
|
||||
if config.routing_mode:
|
||||
actions.adjust_edges_weights()
|
||||
# actions.rerouteAllVehicles()
|
||||
|
||||
step += 1
|
||||
sys.stdout.write(f'Simulation step = {step}/{config.n_steps}' + '\r')
|
||||
sys.stdout.flush()
|
||||
|
||||
finally:
|
||||
traci.close(False)
|
||||
|
||||
total_emissions = 0
|
||||
for area in grid:
|
||||
total_emissions += area.emissions
|
||||
|
||||
# Total of emissions of all pollutants in mg for 200 steps of simulation without locking areas
|
||||
total_emissions200 = 43970763.15084749
|
||||
|
||||
print(f'\n**** Total emissions = {total_emissions} mg ****')
|
||||
diff_with_lock = (total_emissions200 - total_emissions) / total_emissions200
|
||||
print(f'**** Reduction percentage of emissions = {diff_with_lock*100} % ****\n')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
from typing import List
|
||||
|
||||
import traci
|
||||
import time
|
||||
|
||||
from shapely.geometry import LineString
|
||||
from parse import search
|
||||
|
||||
import actions
|
||||
import config
|
||||
import sys
|
||||
from model import Area, Vehicle, Lane , TrafficLight , Phase , Logic
|
||||
from traci import trafficlight
|
||||
|
||||
logger = config.logger
|
||||
|
||||
def init_grid(simulation_bounds, cells_number):
|
||||
grid = list()
|
||||
width = simulation_bounds[1][0] / cells_number
|
||||
height = simulation_bounds[1][1] / cells_number
|
||||
for i in range(cells_number):
|
||||
for j in range(cells_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))
|
||||
area = Area(ar_bounds)
|
||||
area.name = 'Area ({},{})'.format(i, j)
|
||||
grid.append(area)
|
||||
traci.polygon.add(area.name, ar_bounds, (0, 255, 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)
|
||||
minDuration = search('minDuration: {:f}', phase_repr)
|
||||
maxDuration = search('maxDuration: {:f}', phase_repr)
|
||||
phaseDef = search('phaseDef: {}\n', phase_repr)
|
||||
|
||||
if phaseDef is None: phaseDef = ''
|
||||
else : phaseDef = phaseDef[0]
|
||||
|
||||
return Phase(duration[0], minDuration[0], maxDuration[0], phaseDef)
|
||||
|
||||
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):
|
||||
return (traci.vehicle.getCOEmission(veh_id)
|
||||
+traci.vehicle.getNOxEmission(veh_id)
|
||||
+traci.vehicle.getHCEmission(veh_id)
|
||||
+traci.vehicle.getPMxEmission(veh_id)
|
||||
+traci.vehicle.getCO2Emission(veh_id))
|
||||
|
||||
|
||||
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]):
|
||||
for area in grid:
|
||||
for vehicle in vehicles:
|
||||
if vehicle.pos in area:
|
||||
area.emissions += vehicle.emissions
|
||||
if area.emissions > 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)
|
||||
traci.polygon.setColor(area.name, (255, 0, 0))
|
||||
traci.polygon.setFilled(area.name, True)
|
||||
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)
|
||||
|
||||
def main():
|
||||
grid = list()
|
||||
try:
|
||||
traci.start(config.sumo_cmd)
|
||||
|
||||
logger.info('Loading data for the simulation')
|
||||
start = time.perf_counter()
|
||||
|
||||
grid = init_grid(traci.simulation.getNetBoundary(), config.CELLS_NUMBER)
|
||||
add_data_to_areas(grid)
|
||||
|
||||
loading_time = round(time.perf_counter() - start,2)
|
||||
logger.info(f'Data loaded ({loading_time}s)')
|
||||
|
||||
logger.info('Start of the simulation')
|
||||
step = 0
|
||||
while step < config.n_steps : #traci.simulation.getMinExpectedNumber() > 0:
|
||||
traci.simulationStep()
|
||||
|
||||
vehicles = get_all_vehicles()
|
||||
get_emissions(grid, vehicles)
|
||||
|
||||
if config.weight_routing_mode:
|
||||
logger.info('Action - Lane weights adjusted')
|
||||
actions.adjust_edges_weights()
|
||||
|
||||
step += 1
|
||||
|
||||
finally:
|
||||
traci.close(False)
|
||||
simulation_time = round(time.perf_counter() - start,2)
|
||||
logger.info(f'End of the simulation ({simulation_time}s)')
|
||||
|
||||
total_emissions = 0
|
||||
for area in grid:
|
||||
total_emissions += area.emissions
|
||||
|
||||
logger.info(f'Total emissions = {total_emissions} mg')
|
||||
|
||||
if not config.without_actions_mode:
|
||||
ref = config.get_basics_emissions()
|
||||
diff_with_actions = (ref - total_emissions)/ref
|
||||
logger.info(f'Reduction percentage of emissions = {diff_with_actions*100} %')
|
||||
|
||||
logger.info('With the configuration : \n' + str(config.show_config()))
|
||||
logger.info('Logs END')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
@ -1,64 +1,96 @@
|
||||
from typing import Tuple, Set
|
||||
|
||||
from shapely.geometry import Point, LineString
|
||||
from shapely.geometry import Polygon
|
||||
from shapely.geometry.base import BaseGeometry
|
||||
|
||||
|
||||
class Lane:
|
||||
|
||||
def __init__(self, lane_id: str, polygon: LineString):
|
||||
self.polygon = polygon
|
||||
self.lane_id = lane_id
|
||||
|
||||
def __hash__(self):
|
||||
"""Overrides the default implementation"""
|
||||
return hash(self.lane_id)
|
||||
|
||||
|
||||
class Area:
|
||||
|
||||
def __init__(self, coords, name=''):
|
||||
self.locked = False
|
||||
self.rectangle = Polygon(coords)
|
||||
self.name = name
|
||||
self.emissions = 0.0
|
||||
self._lanes: Set[Lane] = 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 remove_lane(self, lane: Lane):
|
||||
self._lanes.remove(lane)
|
||||
|
||||
@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: float = 0.0
|
||||
self.veh_id = veh_id
|
||||
self.pos = Point(pos)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return str(self.__dict__)
|
||||
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 Area:
|
||||
|
||||
def __init__(self, coords, name=''):
|
||||
self.limited_speed = False
|
||||
self.locked = False
|
||||
self.tls_adjusted = False
|
||||
self.rectangle = Polygon(coords)
|
||||
self.name = name
|
||||
self.emissions = 0.0
|
||||
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)
|
||||
|
||||
@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: float = 0.0
|
||||
self.veh_id = veh_id
|
||||
self.pos = Point(pos)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return str(self.__dict__)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user