def experiment_to_sqa()

in ax/storage/sqa_store/encoder.py [0:0]

• 67 lines of code
• 12 McCabe index (conditional complexity)

    def experiment_to_sqa(self, experiment: Experiment) -> SQAExperiment:
        """Convert Ax Experiment to SQLAlchemy.

        In addition to creating and storing a new Experiment object, we need to
        create and store copies of the Trials, Metrics, Parameters,
        ParameterConstraints, and Runner owned by this Experiment.
        """
        optimization_metrics = self.optimization_config_to_sqa(
            experiment.optimization_config
        )

        tracking_metrics = []
        for metric in experiment._tracking_metrics.values():
            tracking_metrics.append(self.metric_to_sqa(metric))

        parameters, parameter_constraints = self.search_space_to_sqa(
            experiment.search_space
        )

        status_quo_name = None
        status_quo_parameters = None
        if experiment.status_quo is not None:
            status_quo_name = not_none(experiment.status_quo).name
            status_quo_parameters = not_none(experiment.status_quo).parameters

        trials = []
        for trial in experiment.trials.values():
            trial_sqa = self.trial_to_sqa(trial=trial)
            trials.append(trial_sqa)

        experiment_data = []
        for trial_index, data_by_timestamp in experiment.data_by_trial.items():
            for timestamp, data in data_by_timestamp.items():
                data = self.data_to_sqa(
                    data=data, trial_index=trial_index, timestamp=timestamp
                )
                experiment_data.append(data)

        experiment_type = self.get_enum_value(
            value=experiment.experiment_type, enum=self.config.experiment_type_enum
        )

        properties = experiment._properties
        runners = []
        if isinstance(experiment, MultiTypeExperiment):
            properties[Keys.SUBCLASS] = "MultiTypeExperiment"
            for trial_type, runner in experiment._trial_type_to_runner.items():
                runner_sqa = self.runner_to_sqa(runner, trial_type)
                runners.append(runner_sqa)

            for metric in tracking_metrics:
                metric.trial_type = experiment._metric_to_trial_type[metric.name]
                if metric.name in experiment._metric_to_canonical_name:
                    metric.canonical_name = experiment._metric_to_canonical_name[
                        metric.name
                    ]
        elif experiment.runner:
            runners.append(self.runner_to_sqa(not_none(experiment.runner)))

        # pyre-ignore[9]: Expected `Base` for 1st...yping.Type[Experiment]`.
        experiment_class: Type[SQAExperiment] = self.config.class_to_sqa_class[
            Experiment
        ]
        exp_sqa = experiment_class(
            id=experiment.db_id,  # pyre-ignore
            description=experiment.description,
            is_test=experiment.is_test,
            name=experiment.name,
            status_quo_name=status_quo_name,
            status_quo_parameters=status_quo_parameters,
            time_created=experiment.time_created,
            experiment_type=experiment_type,
            metrics=optimization_metrics + tracking_metrics,
            parameters=parameters,
            parameter_constraints=parameter_constraints,
            trials=trials,
            runners=runners,
            data=experiment_data,
            properties=properties,
            default_trial_type=experiment.default_trial_type,
            default_data_type=experiment.default_data_type,
        )
        return exp_sqa