(self, dataBase: str, timeStep: int)
| 208 | self.junctions: set = None |
| 209 | |
| 210 | def updateScene(self, dataBase: str, timeStep: int): |
| 211 | ex, ey = self.ego.x, self.ego.y |
| 212 | currGeox = int(ex // 100) |
| 213 | currGeoy = int(ey // 100) |
| 214 | |
| 215 | sceGeohashIDs = ( |
| 216 | (currGeox-1, currGeoy-1), |
| 217 | (currGeox, currGeoy-1), |
| 218 | (currGeox+1, currGeoy-1), |
| 219 | (currGeox-1, currGeoy), |
| 220 | (currGeox, currGeoy), |
| 221 | (currGeox+1, currGeoy), |
| 222 | (currGeox-1, currGeoy+1), |
| 223 | (currGeox, currGeoy+1), |
| 224 | (currGeox+1, currGeoy+1), |
| 225 | ) |
| 226 | |
| 227 | NowEdges: set = set() |
| 228 | NowJuncs: set = set() |
| 229 | |
| 230 | for sgh in sceGeohashIDs: |
| 231 | try: |
| 232 | geohash = self.netInfo.geoHashes[sgh] |
| 233 | except KeyError: |
| 234 | continue |
| 235 | NowEdges = NowEdges | geohash.edges |
| 236 | NowJuncs = NowJuncs | geohash.junctions |
| 237 | |
| 238 | self.edges = NowEdges |
| 239 | self.junctions = NowJuncs |
| 240 | |
| 241 | NowTLs = {} |
| 242 | conn = sqlite3.connect(dataBase) |
| 243 | cur = conn.cursor() |
| 244 | cur.execute( |
| 245 | '''SELECT * FROM trafficLightStates WHERE frame=%i;''' % timeStep) |
| 246 | tlsINFO = cur.fetchall() |
| 247 | if tlsINFO: |
| 248 | for tls in tlsINFO: |
| 249 | frame, tlid, currPhase, nextPhase, switchTime = tls |
| 250 | NowTLs[tlid] = (currPhase, nextPhase, switchTime) |
| 251 | |
| 252 | cur.close() |
| 253 | conn.close() |
| 254 | |
| 255 | if NowTLs: |
| 256 | for jid in NowJuncs: |
| 257 | junc = self.netInfo.getJunction(jid) |
| 258 | if junc: |
| 259 | for jlid in junc.JunctionLanes: |
| 260 | jl = self.netInfo.getJunctionLane(jlid) |
| 261 | tlid = jl.tlLogic |
| 262 | if tlid: |
| 263 | try: |
| 264 | currPhase, nextPhase, switchTime = NowTLs[tlid] |
| 265 | except KeyError: |
| 266 | continue |
| 267 | jl.currTlState = currPhase[jl.tlsIndex] |
nothing calls this directly
no test coverage detected