Spaces:
Running on Zero
Running on Zero
File size: 4,365 Bytes
f234112 | 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 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 | class PhaseGenerator:
"""
Generates valid traffic signal phases from a ConflictGraph.
A phase is a group of movements that can receive green
simultaneously because none of them conflict.
"""
def __init__(self, conflict_graph):
self.graph = conflict_graph
def is_compatible(self, movements):
"""
Check whether all movements can operate together.
"""
movements = list(movements)
for i in range(len(movements)):
for j in range(i + 1, len(movements)):
if self.graph.are_conflicting(
movements[i],
movements[j]
):
return False
return True
def generate_phases(self):
"""
Generate all non-empty compatible movement groups.
Only allowed movements are considered.
"""
allowed_movements = [
movement.id
for movement in self.graph.topology.get_allowed_movements()
]
phases = []
def build_phase(start_index, current_phase):
if current_phase:
phases.append(tuple(current_phase))
for i in range(
start_index,
len(allowed_movements)
):
movement = allowed_movements[i]
if self.is_compatible(
current_phase + [movement]
):
current_phase.append(movement)
build_phase(
i + 1,
current_phase
)
current_phase.pop()
build_phase(0, [])
return phases
def generate_maximal_phases(self):
"""
Return only phases that cannot accept another
compatible movement.
These are maximal compatible movement groups.
"""
all_phases = self.generate_phases()
maximal_phases = []
for phase in all_phases:
phase_set = set(phase)
is_maximal = True
for other_phase in all_phases:
other_set = set(other_phase)
if (
phase_set < other_set
and phase_set.issubset(other_set)
):
is_maximal = False
break
if is_maximal:
maximal_phases.append(phase)
return maximal_phases
if __name__ == "__main__":
from topology import JunctionTopology
from conflict_graph import ConflictGraph
# -------------------------------------------------
# Create topology
# -------------------------------------------------
topology = JunctionTopology("J1")
topology.add_road("A", lanes=2)
topology.add_road("B", lanes=2)
topology.add_road("C", lanes=1)
topology.add_movement(
"A_to_B",
"A",
"B",
lanes=2
)
topology.add_movement(
"A_to_C",
"A",
"C",
lanes=1
)
topology.add_movement(
"B_to_A",
"B",
"A",
lanes=2
)
topology.add_movement(
"C_to_A",
"C",
"A",
lanes=1
)
# -------------------------------------------------
# Create conflict graph
# -------------------------------------------------
graph = ConflictGraph(topology)
graph.add_conflict(
"A_to_B",
"C_to_A"
)
graph.add_conflict(
"A_to_C",
"B_to_A"
)
# -------------------------------------------------
# Generate phases
# -------------------------------------------------
generator = PhaseGenerator(graph)
phases = generator.generate_phases()
maximal_phases = generator.generate_maximal_phases()
# -------------------------------------------------
# Display results
# -------------------------------------------------
print("Topology is valid.")
print("Conflict graph is valid.")
print()
print("All valid phases:")
for number, phase in enumerate(phases, start=1):
print(f"Phase {number}: {phase}")
print()
print("Maximal phases:")
for number, phase in enumerate(
maximal_phases,
start=1
):
print(f"Phase {number}: {phase}") |