(self, vehid: str, currFrame: int)
| 78 | self.tpEnd = 0 |
| 79 | |
| 80 | def dbTrajectory(self, vehid: str, currFrame: int) -> Trajectory: |
| 81 | conn = sqlite3.connect(self.dataBase) |
| 82 | cur = conn.cursor() |
| 83 | cur.execute( |
| 84 | """SELECT frame, x, y, yaw, speed, accel, laneID, lanePos, routeIdx FROM frameINFO |
| 85 | WHERE vid = '{}' AND frame >= {} AND frame < {};""".format( |
| 86 | vehid, currFrame, currFrame + 50 |
| 87 | ) |
| 88 | ) |
| 89 | frameData = cur.fetchall() |
| 90 | if frameData: |
| 91 | # if the trajectory is segmented in time, only the |
| 92 | # data of the first segment will be taken. |
| 93 | validSeq = [frameData[0]] |
| 94 | for i in range(len(frameData) - 1): |
| 95 | if frameData[i + 1][0] - frameData[i][0] == 1: |
| 96 | validSeq.append(frameData[i + 1]) |
| 97 | |
| 98 | tState = [] |
| 99 | for vs in validSeq: |
| 100 | state = State( |
| 101 | x=vs[1], |
| 102 | y=vs[2], |
| 103 | yaw=vs[3], |
| 104 | vel=vs[4], |
| 105 | acc=vs[5], |
| 106 | laneID=vs[6], |
| 107 | s=vs[7], |
| 108 | routeIdx=vs[8], |
| 109 | ) |
| 110 | tState.append(state) |
| 111 | dbTrajectory = Trajectory(states=tState) |
| 112 | else: |
| 113 | self.lsr.outOfRange.add(vehid) |
| 114 | return |
| 115 | |
| 116 | cur.close() |
| 117 | conn.close() |
| 118 | return dbTrajectory |
| 119 | |
| 120 | def dbVType(self, vid: str): |
| 121 | conn = sqlite3.connect(self.dataBase) |
no test coverage detected