def replay()

in aepsych/server.py [0:0]

• 38 lines of code
• 11 McCabe index (conditional complexity)

    def replay(self, uuid_to_replay, skip_computations=False):
        """
        Run a replay against the server. The UUID will be looked up in the database.
        if skip_computations is true, skip all the asks and queries, which should make the replay much faster.
        """
        if uuid_to_replay is None:
            raise RuntimeError("UUID is a required parameter to perform a replay")

        if self.db is None:
            raise RuntimeError("A database is required to perform a replay")

        if skip_computations is True:
            logger.info(
                "skip_computations=True, make sure to refit the final strat before doing anything!"
            )

        master_record = self.db.get_master_record(uuid_to_replay)

        if master_record is None:
            raise RuntimeError(
                f"The UUID {uuid_to_replay} isn't in the database. Unable to perform replay."
            )

        # this prevents writing back to the DB and creating a circular firing squad
        self.is_performing_replay = True

        # if there is a config in the DB we'll use it.
        if 0 < len(master_record.children_config):
            self._configure(master_record.children_config[0].config)

        for result in master_record.children_replay:
            request = result.message_contents
            logger.debug(f"replay - type = {result.message_type} request = {request}")
            if (
                request["type"] == "ask" or request["type"] == "query"
            ) and skip_computations is True:
                logger.info(
                    "Request type is ask or query and skip_computations==True, skipping!"
                )
                # HACK increment strat's count and manually move to next strat as needed, since
                # strats count based on `gen` calls not `add_data calls`.
                # TODO this should probably be the other way around when we refactor
                # the whole Modelbridge/Strategy axis.
                self.strat._strat._count += 1
                if (
                    isinstance(self.strat, SequentialStrategy)
                    and self.strat._count >= self.strat._strat.n_trials
                ):
                    self.strat._make_next_strat()
                continue
            if "version" in request.keys():
                result = self.versioned_handler(request)
            else:
                result = self.unversioned_handler(request)
        self.is_performing_replay = False