(self, dataQue: Queue, timeStep: int)
| 29 | # else, judge if the upstream intersection or downstream intersection |
| 30 | # is in the range of the vehicle's deArea. |
| 31 | def updateScene(self, dataQue: Queue, timeStep: int): |
| 32 | ex, ey = traci.vehicle.getPosition(self.ego.id) |
| 33 | currGeox = int(ex // 100) |
| 34 | currGeoy = int(ey // 100) |
| 35 | |
| 36 | sceGeohashIDs = ( |
| 37 | (currGeox-1, currGeoy-1), |
| 38 | (currGeox, currGeoy-1), |
| 39 | (currGeox+1, currGeoy-1), |
| 40 | (currGeox-1, currGeoy), |
| 41 | (currGeox, currGeoy), |
| 42 | (currGeox+1, currGeoy), |
| 43 | (currGeox-1, currGeoy+1), |
| 44 | (currGeox, currGeoy+1), |
| 45 | (currGeox+1, currGeoy+1), |
| 46 | ) |
| 47 | |
| 48 | NowEdges: set = set() |
| 49 | NowJuncs: set = set() |
| 50 | |
| 51 | for sgh in sceGeohashIDs: |
| 52 | try: |
| 53 | geohash = self.netInfo.geoHashes[sgh] |
| 54 | except KeyError: |
| 55 | continue |
| 56 | NowEdges = NowEdges | geohash.edges |
| 57 | NowJuncs = NowJuncs | geohash.junctions |
| 58 | |
| 59 | self.edges = NowEdges |
| 60 | self.junctions = NowJuncs |
| 61 | |
| 62 | NowTLs = {} |
| 63 | for jid in NowJuncs: |
| 64 | junc = self.netInfo.getJunction(jid) |
| 65 | for jlid in junc.JunctionLanes: |
| 66 | jl = self.netInfo.getJunctionLane(jlid) |
| 67 | tlid = jl.tlLogic |
| 68 | if tlid: |
| 69 | if tlid not in NowTLs.keys(): |
| 70 | currPhaseIndex = traci.trafficlight.getPhase(tlid) |
| 71 | tlLogic = self.netInfo.getTlLogic(tlid) |
| 72 | currPhase = tlLogic.currPhase(currPhaseIndex) |
| 73 | nextPhase = tlLogic.nextPhase(currPhaseIndex) |
| 74 | switchTime = round(traci.trafficlight.getNextSwitch( |
| 75 | tlid) - traci.simulation.getTime(), 1) |
| 76 | NowTLs[tlid] = (currPhase, nextPhase, switchTime) |
| 77 | dataQue.put(( |
| 78 | 'trafficLightStates', |
| 79 | (timeStep, tlid, currPhase, nextPhase, switchTime) |
| 80 | )) |
| 81 | else: |
| 82 | currPhase, nextPhase, switchTime = NowTLs[tlid] |
| 83 | jl.currTlState = currPhase[jl.tlsIndex] |
| 84 | jl.nexttTlState = nextPhase[jl.tlsIndex] |
| 85 | jl.switchTime = switchTime |
| 86 | |
| 87 | def addVeh(self, vdict: dict, vid: str) -> None: |
| 88 | if vdict and vid in vdict.keys(): |
no test coverage detected