(self)
| 501 | super().__init__(dataBase, networkFile, obsFile) |
| 502 | |
| 503 | def getData(self): |
| 504 | conn = sqlite3.connect(self.dataBase) |
| 505 | cur = conn.cursor() |
| 506 | |
| 507 | cur.execute('SELECT * FROM junctionINFO;') |
| 508 | junctionINFO = cur.fetchall() |
| 509 | if junctionINFO: |
| 510 | for ji in junctionINFO: |
| 511 | junctionID = ji[0] |
| 512 | jrawShape = ji[1] |
| 513 | juncShape = self.processRawShape(jrawShape) |
| 514 | # Add the first point to form a closed shape |
| 515 | juncShape.append(juncShape[0]) |
| 516 | junc = Junction(junctionID) |
| 517 | junc.shape = juncShape |
| 518 | self.junctions[junctionID] = junc |
| 519 | |
| 520 | cur.execute('SELECT * FROM edgeINFO;') |
| 521 | edgeInfo = cur.fetchall() |
| 522 | if edgeInfo: |
| 523 | for ed in edgeInfo: |
| 524 | eid, laneNumber, fromJunction, toJunction = ed |
| 525 | self.edges[eid] = Edge( |
| 526 | id=eid, lane_num=laneNumber, |
| 527 | from_junction=fromJunction, |
| 528 | to_junction=toJunction |
| 529 | ) |
| 530 | |
| 531 | cur.execute('SELECT * FROM laneINFO;') |
| 532 | laneINFO = cur.fetchall() |
| 533 | if laneINFO: |
| 534 | for la in laneINFO: |
| 535 | lid, rawShape, lwidth, lspeed, eid, llength = la |
| 536 | lshape = self.processRawShape(rawShape) |
| 537 | lane = NormalLane(lid, lwidth, lspeed, eid, llength) |
| 538 | lane = NormalLane( |
| 539 | id=lid, width=lwidth, speed_limit=lspeed, |
| 540 | affiliated_edge=self.getEdge(eid), sumo_length=llength |
| 541 | ) |
| 542 | shapeUnzip = list(zip(*lshape)) |
| 543 | # interpolate shape points for better represent shape |
| 544 | shapeUnzip = [ |
| 545 | np.interp( |
| 546 | np.linspace(0, len(shapeUnzip[0])-1, 50), |
| 547 | np.arange(0, len(shapeUnzip[0])), |
| 548 | shapeUnzip[i] |
| 549 | ) for i in range(2) |
| 550 | ] |
| 551 | lane.course_spline = Spline2D(shapeUnzip[0], shapeUnzip[1]) |
| 552 | lane.getPlotElem() |
| 553 | self.lanes[lid] = lane |
| 554 | self.getEdge(eid).lanes.add(lid) |
| 555 | |
| 556 | cur.execute('SELECT * FROM junctionLaneINFO;') |
| 557 | JunctionLaneINFO = cur.fetchall() |
| 558 | if JunctionLaneINFO: |
| 559 | for jl in JunctionLaneINFO: |
| 560 | jlid, jlwidth, jlspeed, jlLength, tlLogicID, tlsIndex = jl |
no test coverage detected