(self, vehid: str, currFrame: int)
| 158 | self.gui.drawMainWindowWhiteBG((x1, y1), (x2, y2)) |
| 159 | |
| 160 | def dbTrajectory(self, vehid: str, currFrame: int) -> Trajectory: |
| 161 | conn = sqlite3.connect(self.dataBase) |
| 162 | cur = conn.cursor() |
| 163 | cur.execute( |
| 164 | """SELECT frame, x, y, yaw, speed, accel, laneID, lanePos, routeIdx FROM frameINFO |
| 165 | WHERE vid = "{}" AND frame >= {} AND frame < {};""".format( |
| 166 | vehid, currFrame, currFrame + 50 |
| 167 | ) |
| 168 | ) |
| 169 | frameData = cur.fetchall() |
| 170 | if frameData: |
| 171 | # if the trajectory is segmented in time, only the |
| 172 | # data of the first segment will be taken. |
| 173 | validSeq = [frameData[0]] |
| 174 | for i in range(len(frameData) - 1): |
| 175 | if frameData[i + 1][0] - frameData[i][0] == 1: |
| 176 | validSeq.append(frameData[i + 1]) |
| 177 | |
| 178 | tState = [] |
| 179 | for vs in validSeq: |
| 180 | state = State( |
| 181 | x=vs[1], |
| 182 | y=vs[2], |
| 183 | yaw=vs[3], |
| 184 | vel=vs[4], |
| 185 | acc=vs[5], |
| 186 | laneID=vs[6], |
| 187 | s=vs[7], |
| 188 | routeIdx=vs[8], |
| 189 | ) |
| 190 | tState.append(state) |
| 191 | dbTrajectory = Trajectory(states=tState) |
| 192 | else: |
| 193 | self.sr.outOfRange.add(vehid) |
| 194 | return |
| 195 | |
| 196 | cur.close() |
| 197 | conn.close() |
| 198 | return dbTrajectory |
| 199 | |
| 200 | def dbVType(self, vid: str): |
| 201 | conn = sqlite3.connect(self.dataBase) |
no test coverage detected