Source code for otx.hpo.hpo_base

# Copyright (C) 2024 Intel Corporation
# SPDX-License-Identifier: Apache-2.0
"""HPO algorithm abstract class."""

from __future__ import annotations

import json
import logging
import tempfile
from abc import ABC, abstractmethod
from enum import IntEnum
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal

from otx.hpo.search_space import SearchSpace
from otx.hpo.utils import check_mode_input, check_positive

    from import Hashable

logger = logging.getLogger(__name__)

class HpoBase(ABC):
    """Abstract class for HPO algorithms.

    This implements class for HPO algorithm base.
    Only common methods are implemented but not core algorithm of HPO.

        search_space (dict[str, dict[str, Any]]): hyper parameter search space to find.
        save_path (str | None, optional): path where result of HPO is saved.
        mode ("max" | "min", optional): One of {min, max}. Determines whether objective is
                                        minimizing or maximizing the score.
        num_trials (int | None, optional): How many training to conduct for HPO.
        num_workers (int, optional): How many trains are executed in parallel.
        num_full_iterations (int, optional): epoch for traninig after HPO.
        full_dataset_size (int, optional): train dataset size
        expected_time_ratio (int | float | None, optional): Time to use for HPO.
                                                            If HPO is configured automatically,
                                                            HPO use time about exepected_time_ratio *
                                                            train time after HPO times.
        maximum_resource (int | float | None, optional): Maximum resource to use for training each trial.
        resume (bool, optional): resume flag decide to use previous HPO results.
                                 If HPO completed, you can just use optimized hyper parameters.
                                 If HPO stopped in middle, you can resume in middle.
        prior_hyper_parameters (dict | list[dict] | None, optional) = Hyper parameters to try first.
        acceptable_additional_time_ratio (float | int, optional) = Decide how much additional time can be acceptable.

    # pylint: disable=too-many-instance-attributes

    def __init__(
        search_space: dict[str, dict[str, Any]],
        save_path: str | None = None,
        mode: Literal["max", "min"] = "max",
        num_trials: int | None = None,
        num_workers: int = 1,
        num_full_iterations: int | float = 1,
        full_dataset_size: int = 0,
        expected_time_ratio: int | float | None = None,
        maximum_resource: int | float | None = None,
        resume: bool = False,
        prior_hyper_parameters: dict | list[dict] | None = None,
        acceptable_additional_time_ratio: float | int = 1.0,
    ) -> None:
        # pylint: disable=too-many-arguments, too-many-locals
        check_positive(full_dataset_size, "full_dataset_size")
        check_positive(num_full_iterations, "num_full_iterations")
        if maximum_resource is not None:
            check_positive(maximum_resource, "maximum_resource")
        if num_trials is not None:
            check_positive(num_trials, "num_trials")
        check_positive(num_workers, "num_workers")

        if save_path is None:
            save_path = tempfile.mkdtemp(prefix="OTX-hpo-")
        self.save_path = save_path
        self.search_space = SearchSpace(search_space)
        self.mode = mode
        self.num_trials = num_trials
        self.num_workers = num_workers
        self.num_full_iterations = num_full_iterations
        self.full_dataset_size = full_dataset_size
        self.expected_time_ratio = expected_time_ratio
        self.maximum_resource: int | float | None = maximum_resource
        self.resume = resume
        self.hpo_status: dict = {}
        self.acceptable_additional_time_ratio = acceptable_additional_time_ratio
        if prior_hyper_parameters is None:
            prior_hyper_parameters = []
        elif isinstance(prior_hyper_parameters, dict):
            prior_hyper_parameters = [prior_hyper_parameters]
        self.prior_hyper_parameters = prior_hyper_parameters

    def print_result(self) -> None:
        """Print a HPO algorithm result."""
        raise NotImplementedError

    def save_results(self) -> None:
        """Save a HPO algorithm result."""
        raise NotImplementedError

    def is_done(self) -> bool:
        """Check whether HPO algorithm is done."""
        raise NotImplementedError

    def get_next_sample(self) -> Trial | None:
        """Get next sample to train."""
        raise NotImplementedError

    def auto_config(self):  # noqa: ANN201
        """Configure HPO algorithm automatically."""
        raise NotImplementedError

    def get_progress(self) -> int | float:
        """Get current progress of HPO algorithm."""
        raise NotImplementedError

    def report_score(self, score: float | int, resource: float | int, trial_id: Hashable, done: bool) -> TrialStatus:
        """Report a score to HPO algorithm."""
        raise NotImplementedError

    def get_best_config(self) -> dict[str, Any] | None:
        """Get best config of HPO algorithm."""
        raise NotImplementedError

class Trial:
    """Trial to train with given hyper parameters.

        trial_id (Any): Trial id.
        configuration (dict): Configuration to train with.
        train_environment (dict | None, optional): Train environment for the trial. Defaults to None.

    def __init__(self, trial_id: Hashable, configuration: dict, train_environment: dict | None = None) -> None:
        self._id = trial_id
        self._configuration = configuration
        self.score: dict[float | int, float | int] = {}
        self._train_environment = train_environment
        self._iteration: int | float | None = None
        self.status: TrialStatus = TrialStatus.READY
        self._done = False

    def id(self) -> Hashable:  # noqa: A003
        """Trial id."""
        return self._id

    def configuration(self) -> dict:
        """Configuration to train with."""
        return self._configuration

    def iteration(self) -> int | float | None:
        """Iteration to use for training."""
        return self._iteration

    def iteration(self, val: int | float) -> None:
        """Setter for iteration."""
        check_positive(val, "iteration")
        self._iteration = val
        if self.get_progress() < val:
            self._done = False

    def train_environment(self) -> dict | None:
        """Train environment for the trial."""
        return self._train_environment

    def get_train_configuration(self) -> dict[str, Any]:
        """Get configurations needed to trian."""
        self._configuration["iterations"] = self.iteration
        return {"id":, "configuration": self.configuration, "train_environment": self.train_environment}

    def register_score(self, score: int | float, resource: int | float) -> None:
        """Register score to the trial.

            score (int | float): Score to register.
            resource (int | float): Resource used to get score. It should be positive.
        check_positive(resource, "resource")
        self.score[resource] = score

    def get_best_score(
        mode: Literal["max", "min"] = "max",
        resource_limit: float | int | None = None,
    ) -> float | int | None:
        """Get best score of the trial.

            mode ("max" | "min", optional):
                Decide which is better between highest score or lowest score. Defaults to "max".
            resource_limit (float | int | None, optional): Find a best score among the score at resource
                                                           lower than this value. Defaults to None.

            float | int | None: Best score. If there is no score, return None.

        if resource_limit is None:
            scores = self.score.values()
            scores = [val for key, val in self.score.items() if key <= resource_limit]  # type: ignore[assignment, index]

        if len(scores) == 0:
            return None

        if mode == "max":
            return max(scores)
        return min(scores)

    def get_progress(self) -> float | int:
        """Get a progress of the trial.

            float | int: How many resource is used for the trial.
        if len(self.score) == 0:
            return 0
        return max(self.score.keys())

    def save_results(self, save_path: str) -> None:
        """Save a result in the 'save_path'.

            save_path (str): Path where to save a result.
        results = {
            "configuration": self.configuration,
            "train_environment": self.train_environment,
            "score": self.score,

        with Path(save_path).open("w", encoding="utf-8") as f:
            json.dump(results, f)

    def finalize(self) -> None:
        """Set done as True."""
        if not self.score:
            error_msg = f"Trial{} didn't report any score but tries to be done."
            raise RuntimeError(error_msg)
        self._done = True

    def is_done(self) -> bool:
        """Check the trial is done."""
        if self.iteration is None:
            error_msg = "iteration isn't set yet."
            raise ValueError(error_msg)
        return self._done or self.get_progress() >= self.iteration

[docs] class TrialStatus(IntEnum): """Enum class for trial status.""" UNKNOWN = -1 READY = 0 RUNNING = 1 STOP = 2 CUDAOOM = 3