(self, vehid: str, currFrame: int)
| 253 | conn.close() |
| 254 | |
| 255 | def dbTrajectory(self, vehid: str, currFrame: int) -> dict: |
| 256 | conn = sqlite3.connect(self.dataBase) |
| 257 | cur = conn.cursor() |
| 258 | cur.execute( |
| 259 | """SELECT frame, x, y, yaw, speed, accel, laneID, lanePos, routeIdx FROM frameINFO |
| 260 | WHERE vid = '{}' AND frame >= {} AND frame < {};""".format( |
| 261 | vehid, currFrame, currFrame + 50 |
| 262 | ) |
| 263 | ) |
| 264 | frameData = cur.fetchall() |
| 265 | if frameData: |
| 266 | # if the trajectory is segmented in time, only the data of the first |
| 267 | # segment will be taken. |
| 268 | validSeq = [frameData[0]] |
| 269 | for i in range(len(frameData)-1): |
| 270 | if frameData[i+1][0] - frameData[i][0] == 1: |
| 271 | validSeq.append(frameData[i+1]) |
| 272 | |
| 273 | tState = [] |
| 274 | for vs in validSeq: |
| 275 | state = State( |
| 276 | x=vs[1], y=vs[2], yaw=vs[3], vel=vs[4], |
| 277 | acc=vs[5], laneID=vs[6], s=vs[7], routeIdx=vs[8] |
| 278 | ) |
| 279 | tState.append(state) |
| 280 | dbTrajectory = Trajectory(states=tState) |
| 281 | else: |
| 282 | if vehid not in self.sr.vehINAoI.keys(): |
| 283 | self.sr.outOfRange.add(vehid) |
| 284 | return |
| 285 | |
| 286 | cur.close() |
| 287 | conn.close() |
| 288 | return dbTrajectory |
| 289 | |
| 290 | def dbVType(self, vid: str): |
| 291 | conn = sqlite3.connect(self.dataBase) |
no test coverage detected