def end_trial()

in src/python/tensorflow_cloud/tuner/tuner.py [0:0]

• 35 lines of code
• 7 McCabe index (conditional complexity)

    def end_trial(self, trial_id: Text, status: Text = "COMPLETED"):
        """Record the measured objective for a set of parameter values."""
        keras_tuner_trial = None
        for tuner_id, ongoing_trial in self.ongoing_trials.items():
            if ongoing_trial.trial_id == trial_id:
                tf.get_logger().info(
                    "End trial requested by tuner ({})".format(tuner_id)
                )
                keras_tuner_trial = self.ongoing_trials.pop(tuner_id)
                break

        if not keras_tuner_trial:
            raise ValueError(
                "Ongoing trial with id: {} not found.".format(trial_id))

        keras_tuner_trial.status = status
        if status == trial_module.TrialStatus.COMPLETED:
            trial_infeasible = False
            infeasibility_reason = None
        elif status == trial_module.TrialStatus.INVALID:
            trial_infeasible = True
            infeasibility_reason = status
        else:
            raise ValueError(
                'Unexpected status passed. Expected "COMPLETED" or '
                '"INVALID", found {}'.format(status)
            )

        vizier_trial = self.service.complete_trial(
            trial_id, trial_infeasible, infeasibility_reason
        )

        if status == trial_module.TrialStatus.COMPLETED:
            final_measurement = vizier_trial["finalMeasurement"]
            # If epochs = 1, set the best_step = 0.
            keras_tuner_trial.best_step = int(
                final_measurement.get("stepCount", 0))
            keras_tuner_trial.score = final_measurement["metrics"][0].get(
                "value")
        self._save_trial(keras_tuner_trial)
        self.save()