Spaces:
Running on Zero
Running on Zero
| class ConflictGraph: | |
| """ | |
| Represents conflicts between traffic movements. | |
| Each movement is a node. | |
| An edge between two movements means: | |
| those two movements cannot receive green | |
| at the same time. | |
| """ | |
| def __init__(self, topology): | |
| self.topology = topology | |
| # movement_id -> set of conflicting movement_ids | |
| self.conflicts = {} | |
| # Initialize every movement | |
| for movement_id in topology.movements: | |
| self.conflicts[movement_id] = set() | |
| def add_conflict(self, movement_a, movement_b): | |
| """ | |
| Add a two-way conflict between two movements. | |
| """ | |
| if movement_a not in self.topology.movements: | |
| raise ValueError( | |
| f"Unknown movement: {movement_a}" | |
| ) | |
| if movement_b not in self.topology.movements: | |
| raise ValueError( | |
| f"Unknown movement: {movement_b}" | |
| ) | |
| if movement_a == movement_b: | |
| raise ValueError( | |
| "A movement cannot conflict with itself" | |
| ) | |
| self.conflicts[movement_a].add(movement_b) | |
| self.conflicts[movement_b].add(movement_a) | |
| def get_conflicts(self, movement_id): | |
| """ | |
| Return all movements that conflict | |
| with the given movement. | |
| """ | |
| if movement_id not in self.conflicts: | |
| raise ValueError( | |
| f"Unknown movement: {movement_id}" | |
| ) | |
| return self.conflicts[movement_id] | |
| def are_conflicting(self, movement_a, movement_b): | |
| """ | |
| Check whether two movements conflict. | |
| """ | |
| if movement_a not in self.conflicts: | |
| raise ValueError( | |
| f"Unknown movement: {movement_a}" | |
| ) | |
| if movement_b not in self.conflicts: | |
| raise ValueError( | |
| f"Unknown movement: {movement_b}" | |
| ) | |
| return movement_b in self.conflicts[movement_a] | |
| def validate(self): | |
| """ | |
| Validate the conflict graph. | |
| """ | |
| for movement_id, conflicts in self.conflicts.items(): | |
| for conflicting_movement in conflicts: | |
| if conflicting_movement not in self.conflicts: | |
| raise ValueError( | |
| f"Conflict references unknown movement: " | |
| f"{conflicting_movement}" | |
| ) | |
| # Conflict must be symmetric | |
| if movement_id not in self.conflicts[ | |
| conflicting_movement | |
| ]: | |
| raise ValueError( | |
| f"Conflict is not symmetric: " | |
| f"{movement_id} <-> " | |
| f"{conflicting_movement}" | |
| ) | |
| return True | |
| def to_dict(self): | |
| """ | |
| Convert conflict graph into a dictionary. | |
| """ | |
| return { | |
| movement_id: sorted(list(conflicts)) | |
| for movement_id, conflicts in self.conflicts.items() | |
| } | |
| if __name__ == "__main__": | |
| # Import our topology module | |
| from topology import JunctionTopology | |
| # ------------------------------------------------- | |
| # Create example 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) | |
| # Example conflict relationships. | |
| # | |
| # These are explicitly supplied for the test. | |
| # We are NOT assuming that road names determine | |
| # physical conflicts. | |
| graph.add_conflict( | |
| "A_to_B", | |
| "C_to_A" | |
| ) | |
| graph.add_conflict( | |
| "A_to_C", | |
| "B_to_A" | |
| ) | |
| # ------------------------------------------------- | |
| # Validate | |
| # ------------------------------------------------- | |
| topology.validate() | |
| graph.validate() | |
| print("Topology is valid.") | |
| print("Conflict graph is valid.") | |
| print() | |
| print(graph.to_dict()) |