Fixed tls duration mode

sumo_project/actions.py

@@ -6,11 +6,10 @@ Created on 17 oct. 2018
 from typing import Iterable
 
 import traci
-import inspect
 from shapely.geometry.linestring import LineString
 
 from model import Area, Vehicle
-from traci._trafficlight import Logic, Phase
+from traci._trafficlight import Logic
 
 
 def remove_vehicle(veh_id):
@@ -34,16 +33,19 @@ def limit_speed_into_area(area: Area, vehicles: Iterable[Vehicle], max_speed):
     for lane in area._lanes:
         traci.lane.setMaxSpeed(lane.lane_id, max_speed/3.6)
 
-
+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)
         
-def adjust_traffic_light_phase_duration(area,reduction_factor):
-    #attributes = inspect.getmembers(Phase, lambda a:not(inspect.isroutine(a))) 
-    #print ([a[0] for a in attributes])
+    return traci.trafficlight.Logic("new-program", 0 , 0 , 0 , new_phases)    
+
+def adjust_traffic_light_phase_duration(area, reduction_factor):
+    print(f'Decrease of traffic lights duration by a factor of {reduction_factor}')
     for tl in area._tls:
         for logic in tl._logics:
-           phases = traci.trafficlight.Logic.getPhases(logic)
-           for phase in phases:
-               print(phase)
+            traci.trafficlights.setCompleteRedYellowGreenDefinition(tl.tl_id, modifyLogic(logic,reduction_factor))
     
     #phaseDuration = traci.trafficlight.getPhaseDuration(tl.tl_id)
     #traci.trafficlight.setPhaseDuration(tl.tl_id, phaseDuration*reduction_factor)

sumo_project/config.py

@@ -17,10 +17,19 @@ CELLS_NUMBER = 10
 EMISSIONS_THRESHOLD = 500000
 n_steps = 200 
 
+#Limit the speed into areas when the threshold is exceeded
 limit_speed_mode = True
+#Vehicles are routed according to the less polluted route
 weight_routing_mode = False
+#Decrease all traffic lights duration into the area when the threshold is exceeded
+rf_trafficLights_duration = 0.2
 adjust_traffic_light_mode = True
 
+#Weight routing mode cannot be combinated with other actions 
+if weight_routing_mode:
+    limit_speed_mode = False
+    adjust_traffic_light_mode = False
+
 sumo_binary = os.path.join(os.environ['SUMO_HOME'], 'bin', _SUMOCMD)
 sumo_cmd = [sumo_binary, "-c", _SUMOCFG]
 
@@ -29,5 +38,5 @@ def showConfig():
     + str(f'step number = {n_steps}\n')
     + str(f'limit speed mode = {limit_speed_mode}\n')
     + str(f'weight routing mode= {weight_routing_mode}\n')
-    + str(f'adjust traffic light mode = {adjust_traffic_light_mode}\n'))
+    + str(f'adjust traffic light mode = {adjust_traffic_light_mode} , RF = {rf_trafficLights_duration}\n'))
     

sumo_project/emissions.py

@@ -2,11 +2,12 @@ from typing import List
 
 import traci
 from shapely.geometry import LineString
+from parse import *
 
 import actions
 import config
 import sys
-from model import Area, Vehicle, Lane , TrafficLight
+from model import Area, Vehicle, Lane , TrafficLight , Phase , Logic
 from traci import trafficlight
 
 
@@ -28,10 +29,10 @@ def init_grid(simulation_bounds, cells_number):
 
 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))
+            +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]:
@@ -49,7 +50,7 @@ def get_all_lanes() -> List[Lane]:
     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))
+        lanes.append(Lane(lane_id, polygon_lane, initial_max_speed))
     return lanes
 
 
@@ -59,26 +60,42 @@ def get_emissions(grid: List[Area], vehicles: List[Vehicle]):
             if vehicle.pos in area:
                 area.emissions += vehicle.emissions
         if config.limit_speed_mode and area.emissions > config.EMISSIONS_THRESHOLD and not area.locked:
-            actions.limit_speed_into_area(area, vehicles,30)
+            actions.limit_speed_into_area(area, vehicles, 30)
             traci.polygon.setColor(area.name, (255, 0, 0))
             traci.polygon.setFilled(area.name, True)
             if config.adjust_traffic_light_mode:
-                actions.adjust_traffic_light_phase_duration(area, 0.75)
+                actions.adjust_traffic_light_phase_duration(area, config.rf_trafficLights_duration)
+
+                
+def parsePhase(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 
+        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 
-                    logics = traci.trafficlight.getCompleteRedYellowGreenDefinition(tl_id)
+                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(parsePhase(phase.__repr__()))
+                            logics.append(Logic(l,phases)) 
                         area.add_tl(TrafficLight(tl_id,logics))
+            
         
-
-
 def main():
     grid = list()
     try:
@@ -100,7 +117,7 @@ def main():
             progress = round(step/config.n_steps*100,2)
             sys.stdout.write(f'Progress :  {progress}%'+'\r')
             sys.stdout.flush()
-
+            
     finally:
         traci.close(False)
 
@@ -108,14 +125,16 @@ def main():
         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 of emissions of all pollutants in mg for n steps of simulation without locking areas
         total_emissions200 = 43970763.15084749  
+        total_emissions300 = 87382632.08217141
                 
         print("\n**** RESULTS ****")
         print(f'Total emissions = {total_emissions} mg')
-        diff_with_lock = (total_emissions200 - total_emissions)/total_emissions200
+        diff_with_lock = (total_emissions200 - total_emissions) / total_emissions200
         print(f'Reduction percentage of emissions = {diff_with_lock*100} %')
         print("With the configuration :\n" + str(config.showConfig()))
+
         
 if __name__ == '__main__':
     main()

sumo_project/model.py

@@ -3,7 +3,7 @@ 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
+from traci._trafficlight import Logic as SUMO_Logic
 
 
 class Lane:
@@ -17,17 +17,32 @@ class Lane:
         """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=''):
@@ -68,7 +83,6 @@ class Area:
             (xmax, ymax),
             (xmax, ymin)))
 
-
 class Vehicle:
 
     def __init__(self, veh_id: int, pos: Tuple[float, float]):