Class ReplayModel

limsim/simModel/egoTracking/replay.py:21–451

dataBase: Replay database, please note that the database files for ego tracking and fixed scene are not common;

class ReplayModel:
    '''
        dataBase: Replay database, please note that the database files for ego tracking and fixed scene are not common;
    '''

    def __init__(self, dataBase: str, startFrame: int = None) -> None:
        print(
            '[green bold]Model initialized at {}.[/green bold]'.format(
                datetime.now().strftime('%H:%M:%S.%f')[:-3]
            )
        )
        self.dataBase = dataBase
        conn = sqlite3.connect(self.dataBase)
        cur = conn.cursor()

        # minTimeStep
        cur.execute("""SELECT MAX(frame) FROM frameINFO;""")
        maxTimeStep = cur.fetchone()[0] - 200
        if maxTimeStep < 0:
            maxTimeStep = 0
        cur.execute("""SELECT MIN(frame) FROM frameINFO;""")
        minTimeStep = cur.fetchone()[0]
        if startFrame:
            if startFrame > maxTimeStep:
                print(
                    '[yellow]The start frame is too large, and is reassigned to[/yellow] %i.'
                    % maxTimeStep
                )
                self.timeStep = maxTimeStep
            elif startFrame < minTimeStep:
                print(
                    '[yellow]The start frame is too small, and is reassigned to[/yellow] %i.'
                    % minTimeStep
                )
                self.timeStep = minTimeStep
            else:
                self.timeStep = startFrame
        else:
            self.timeStep = minTimeStep

        self.rb = Rebuild(dataBase)
        self.rb.getData()
        self.rb.buildTopology()

        cur.execute("""SELECT * FROM simINFO;""")
        simINFO = cur.fetchone()
        _, localPosx, localPosy, radius, egoID, strBoundary, _, _ = simINFO
        if egoID:
            self.egoID = egoID
            self.ego = self.initVeh(egoID, self.timeStep)
            netBoundaryList = strBoundary.split(' ')
            self.netBoundary: list[list[float]] = [
                list(map(float, p.split(','))) for p in netBoundaryList
            ]
        else:
            raise TypeError('Please select the appropriate database file.')

        cur.close()