| 171 | |
| 172 | |
| 173 | class LocalSceneReplay: |
| 174 | def __init__(self, netInfo: Rebuild, localPos: DummyVehicle) -> None: |
| 175 | self.netInfo = netInfo |
| 176 | self.localPos = localPos |
| 177 | self.edges, self.junctions = self.getRoadGraph() |
| 178 | self.currVehicles: dict[str, Vehicle] = {} |
| 179 | self.vehINAoI: dict[str, Vehicle] = {} |
| 180 | self.outOfAoI: dict[str, Vehicle] = {} |
| 181 | self.outOfRange: set[str] = set() |
| 182 | |
| 183 | def getRoadGraph(self) -> tuple[set[str]]: |
| 184 | ex, ey = self.localPos.x, self.localPos.y |
| 185 | currGeox = int(ex // 100) |
| 186 | currGeoy = int(ey // 100) |
| 187 | |
| 188 | sceGeohashIDs = ( |
| 189 | (currGeox-1, currGeoy-1), |
| 190 | (currGeox, currGeoy-1), |
| 191 | (currGeox+1, currGeoy-1), |
| 192 | (currGeox-1, currGeoy), |
| 193 | (currGeox, currGeoy), |
| 194 | (currGeox+1, currGeoy), |
| 195 | (currGeox-1, currGeoy+1), |
| 196 | (currGeox, currGeoy+1), |
| 197 | (currGeox+1, currGeoy+1), |
| 198 | ) |
| 199 | |
| 200 | edges: set = set() |
| 201 | juncs: set = set() |
| 202 | |
| 203 | for sgh in sceGeohashIDs: |
| 204 | try: |
| 205 | geohash = self.netInfo.geoHashes[sgh] |
| 206 | except KeyError: |
| 207 | continue |
| 208 | edges = edges | geohash.edges |
| 209 | juncs = juncs | geohash.junctions |
| 210 | |
| 211 | return edges, juncs |
| 212 | |
| 213 | def updateScene(self, dataBase: str, timeStep: int): |
| 214 | NowTLs = {} |
| 215 | conn = sqlite3.connect(dataBase) |
| 216 | cur = conn.cursor() |
| 217 | cur.execute( |
| 218 | """SELECT * FROM trafficLightStates WHERE frame=%i;""" % timeStep |
| 219 | ) |
| 220 | tlsINFO = cur.fetchall() |
| 221 | if tlsINFO: |
| 222 | for tls in tlsINFO: |
| 223 | frame, tlid, currPhase, nextPhase, switchTime = tls |
| 224 | NowTLs[tlid] = (currPhase, nextPhase, switchTime) |
| 225 | |
| 226 | cur.close() |
| 227 | conn.close() |
| 228 | |
| 229 | if NowTLs: |
| 230 | for jid in self.junctions: |