(self, vehid: str, currFrame: int)
| 223 | conn.close() |
| 224 | |
| 225 | def dbTrajectory(self, vehid: str, currFrame: int) -> dict: |
| 226 | conn = sqlite3.connect(self.dataBase) |
| 227 | cur = conn.cursor() |
| 228 | cur.execute( |
| 229 | """SELECT frame, x, y, yaw, speed, accel, laneID, lanePos, routeIdx FROM frameINFO |
| 230 | WHERE vid = '{}' AND frame >= {} AND frame < {};""".format( |
| 231 | vehid, currFrame, currFrame + 50 |
| 232 | ) |
| 233 | ) |
| 234 | frameData = cur.fetchall() |
| 235 | if frameData: |
| 236 | # if the trajectory is segmented in time, only the data of the first |
| 237 | # segment will be taken. |
| 238 | validSeq = [frameData[0]] |
| 239 | for i in range(len(frameData)-1): |
| 240 | if frameData[i+1][0] - frameData[i][0] == 1: |
| 241 | validSeq.append(frameData[i+1]) |
| 242 | |
| 243 | tState = [] |
| 244 | for vs in validSeq: |
| 245 | state = State( |
| 246 | x=vs[1], y=vs[2], yaw=vs[3], vel=vs[4], |
| 247 | acc=vs[5], laneID=vs[6], s=vs[7], routeIdx=vs[8] |
| 248 | ) |
| 249 | tState.append(state) |
| 250 | dbTrajectory = Trajectory(states=tState) |
| 251 | else: |
| 252 | self.lsr.outOfRange.add(vehid) |
| 253 | return |
| 254 | |
| 255 | cur.close() |
| 256 | conn.close() |
| 257 | return dbTrajectory |
| 258 | |
| 259 | def dbVType(self, vid: str): |
| 260 | conn = sqlite3.connect(self.dataBase) |
no test coverage detected