(self, dataQue: Queue, timeStep: int)
| 52 | return edges, juncs |
| 53 | |
| 54 | def updateScene(self, dataQue: Queue, timeStep: int): |
| 55 | NowTLs = {} |
| 56 | for jid in self.junctions: |
| 57 | junc = self.netInfo.getJunction(jid) |
| 58 | for jlid in junc.JunctionLanes: |
| 59 | jl = self.netInfo.getJunctionLane(jlid) |
| 60 | tlid = jl.tlLogic |
| 61 | if tlid: |
| 62 | if tlid not in NowTLs.keys(): |
| 63 | currPhaseIndex = traci.trafficlight.getPhase(tlid) |
| 64 | tlLogic = self.netInfo.getTlLogic(tlid) |
| 65 | currPhase = tlLogic.currPhase(currPhaseIndex) |
| 66 | nextPhase = tlLogic.nextPhase(currPhaseIndex) |
| 67 | switchTime = round(traci.trafficlight.getNextSwitch( |
| 68 | tlid) - traci.simulation.getTime(), 1) |
| 69 | NowTLs[tlid] = (currPhase, nextPhase, switchTime) |
| 70 | dataQue.put(( |
| 71 | 'trafficLightStates', |
| 72 | (timeStep, tlid, currPhase, nextPhase, switchTime) |
| 73 | )) |
| 74 | else: |
| 75 | currPhase, nextPhase, switchTime = NowTLs[tlid] |
| 76 | jl.currTlState = currPhase[jl.tlsIndex] |
| 77 | jl.nexttTlState = nextPhase[jl.tlsIndex] |
| 78 | jl.switchTime = switchTime |
| 79 | |
| 80 | def addVeh(self, vdict: dict, vid: str) -> None: |
| 81 | if vdict and vid in vdict.keys(): |
no test coverage detected