Spaces:
Sleeping
Sleeping
File size: 4,659 Bytes
86d5d3a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | import statistics
import random
class TrafficSimulator:
def __init__(self, difficulty: str = "easy"):
self.difficulty = difficulty
self.lanes = ["N", "S", "E", "W"]
self.queues = {lane: 0 for lane in self.lanes}
self.wait_times = {lane: 0.0 for lane in self.lanes}
self.signal_phase = "NS_GREEN"
self.time_since_last_change = 0
self.emergency = {lane: False for lane in self.lanes}
self.pedestrians = {lane: False for lane in self.lanes}
self.step_count = 0
if difficulty == "easy":
self.arrival_rates = {"N": 2, "S": 2, "E": 0, "W": 0}
elif difficulty == "medium":
self.arrival_rates = {"N": 2, "S": 2, "E": 2, "W": 2}
else: # hard
self.arrival_rates = {"N": 3, "S": 3, "E": 3, "W": 3}
def step(self, action_type: str) -> float:
"""Executes a simulation step and returns the reward."""
phase_switches = 0
if action_type == "SWITCH_PHASE":
self.signal_phase = "EW_GREEN" if self.signal_phase == "NS_GREEN" else "NS_GREEN"
self.time_since_last_change = 0
phase_switches = 1
else:
self.time_since_last_change += 1
self.step_count += 1
# Deterministic arrivals
arrivals = self._get_arrivals()
for lane, count in arrivals.items():
self.queues[lane] += count
# Hard mode logic: emergencies and pedestrians
if self.difficulty == "hard":
# Using step_count to ensure purely deterministic behavior
if self.step_count % 5 == 0:
self.emergency["N"] = True
if self.step_count % 7 == 0:
self.pedestrians["E"] = True
if self.step_count % 11 == 0:
self.emergency["E"] = True
if self.step_count % 3 == 0:
self.pedestrians["N"] = True
passed = {lane: 0 for lane in self.lanes}
green_lanes = ["N", "S"] if self.signal_phase == "NS_GREEN" else ["E", "W"]
red_lanes = ["E", "W"] if self.signal_phase == "NS_GREEN" else ["N", "S"]
emergency_cleared = 0
pedestrians_served = 0
# Process green lanes (cars move)
for lane in green_lanes:
if self.pedestrians[lane]:
pass # Cars must wait for pedestrians
else:
cars_can_pass = 3
passed_cars = min(self.queues[lane], cars_can_pass)
self.queues[lane] -= passed_cars
passed[lane] = passed_cars
if self.emergency[lane]:
emergency_cleared += 1
self.emergency[lane] = False
# Pedestrians waiting on red lanes can cross safely
for lane in red_lanes:
if self.pedestrians[lane]:
pedestrians_served += 1
self.pedestrians[lane] = False
# Accumulate wait times based on queue (proxy for Little's law)
for lane in self.lanes:
self.wait_times[lane] += self.queues[lane]
# Reward Calculation
# Weights from problem statement
alpha = 1.0
beta = 0.5
gamma = 0.2
delta = 10.0
eta = 3.0
lam = 0.1
mu = 0.05
N_passed = sum(passed.values())
sum_q = sum(self.queues.values())
sum_w = sum(self.wait_times.values())
N_emergency = emergency_cleared
N_ped = pedestrians_served
wait_values = list(self.wait_times.values())
var_wait = statistics.variance(wait_values) if len(wait_values) > 1 else 0.0
reward = (alpha * N_passed
- beta * sum_q
- gamma * sum_w
+ delta * N_emergency
+ eta * N_ped
- lam * var_wait
- mu * phase_switches)
return reward
def _get_arrivals(self):
"""Deterministic arrivals."""
arrivals = {}
for lane, base_rate in self.arrival_rates.items():
if base_rate > 0:
# Add simple cyclical variation to arrivals
offset = self.step_count % 3
arrivals[lane] = max(0, base_rate - 1 + offset)
else:
arrivals[lane] = 0
return arrivals
|