| 59 | |
| 60 | |
| 61 | class simulationWorker(QObject): |
| 62 | finished = Signal() |
| 63 | updateWindow = Signal() |
| 64 | planningError = Signal() |
| 65 | |
| 66 | def __init__(self, model: Model) -> None: |
| 67 | super().__init__() |
| 68 | self.model = model |
| 69 | self.planner = TrafficManager(model) |
| 70 | |
| 71 | def run(self): |
| 72 | self.model.start() |
| 73 | while not self.model.tpEnd and self.model.is_gui_running: |
| 74 | self.model.moveStep() |
| 75 | # if self.model.timeStep % 2 == 0: |
| 76 | self.updateWindow.emit() |
| 77 | time.sleep(0.01) |
| 78 | if self.model.timeStep % 5 == 0: |
| 79 | roadgraph, vehicles = self.model.exportSce() |
| 80 | if self.model.tpStart and roadgraph: |
| 81 | trajectories = self.planner.plan( |
| 82 | self.model.timeStep * 0.1, roadgraph, vehicles |
| 83 | ) |
| 84 | if len(trajectories) == 0: |
| 85 | self.planningError.emit() |
| 86 | # print('Sorry. Route planning failed. Please choose another route.') |
| 87 | break |
| 88 | self.model.setTrajectories(trajectories) |
| 89 | else: |
| 90 | self.model.ego.exitControlMode() |
| 91 | self.model.updateVeh() |
| 92 | |
| 93 | self.model.destroy() |
| 94 | self.finished.emit() |
| 95 | |
| 96 | |
| 97 | class myQGraphicsPolygonItem(QGraphicsPolygonItem): |
nothing calls this directly
no outgoing calls
no test coverage detected