PyHopper

https://pyhopper.readthedocs.io/en/latest/
Admin
PyHopper is a black-box optimizer made for high-dimensional hyperparameter optimization problems in...
Tokens:52,567
Snippets:319
Trust Score:9.6
Update:4 months ago
Show doc for...
Tokens:
Raw
### Basic Hyperparameter Search Setup

Source: https://pyhopper.readthedocs.io/en/latest/quickstart.html

Set up a hyperparameter search with various parameter types including integers, floats, and choices. This is a typical starting point for using Pyhopper.

```python
import pyhopper

def my_objective(param: dict) -> float:
    # Add code here
    return val_acc

search = pyhopper.Search(
    {
        "epochs": 20,
        "num_layers": pyhopper.int(1, 8, init=4),
        "batch_size": pyhopper.int(32, 512, multiple_of=32),
        "dropout": pyhopper.float(0, 0.5, precision=1),
        "lr": pyhopper.float(1e-5, 1e-2, log=True, precision=1),
        "opt": pyhopper.choice(["adam", "rmsprop", "sgd"], init="adam"),
        "weight_decay": pyhopper.choice([0, 1e-5, 1e-4, 1e-3], is_ordinal=True),
    }
)
search.run(my_objective, "max", "4h", n_jobs="per-gpu")

```

--------------------------------

### Install tensorboardX

Source: https://pyhopper.readthedocs.io/en/latest/recipes/tensorboard.html

Install the tensorboardX package, which is required for the TensorboardCallback.

```bash
pip3 install -U tensorboardx

```

--------------------------------

### Parallel Execution Output Example

Source: https://pyhopper.readthedocs.io/en/latest/quickstart.html

Observe the performance difference when running a search with multiple jobs.

```bash
> n_jobs=1 took 20.19 seconds
> n_jobs=4 took 5.08 seconds
```

--------------------------------

### Install PyHopper

Source: https://pyhopper.readthedocs.io/en/latest/index.html

Install PyHopper using pip. Ensure you are using pip3 for the installation.

```bash
pip3 install -U pyhopper

```

--------------------------------

### Quantile Pruner State Management

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/pruners/pruners.html

Demonstrates how to get and load the state for a quantile pruner, essential for resuming pruning processes.

```python
    def state_dict(self):
        return {"n": self.n, "intermediates": self.intermediates}

    def load_state_dict(self, state_dict):
        self.n = state_dict["n"]
        self.intermediates = state_dict["intermediates"]
```

--------------------------------

### Example Usage of Time Formatting

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/utils.html

Demonstrates the output of the `time_to_pretty_str` function with various time inputs, from milliseconds to large durations.

```python
print(time_to_pretty_str(0.0004))
print(time_to_pretty_str(0.004))
print(time_to_pretty_str(0.04))
print(time_to_pretty_str(0.4))
print(time_to_pretty_str(4.4))
print(time_to_pretty_str(47.4))
print(time_to_pretty_str(474.4))
print(time_to_pretty_str(4746.4))
print(time_to_pretty_str(47467.4))
print(time_to_pretty_str(474678.4))
print(time_to_pretty_str(4746788.4))
print(time_to_pretty_str(47467888.4))
print(time_to_pretty_str(474678888.4))
```

--------------------------------

### Search.run

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Starts the hyperparameter tuning process. This method orchestrates the search for optimal hyperparameters based on a given objective function and search space.

```APIDOC
## Search.run

### Description
Starts the hyperparameter tuning process. This method orchestrates the search for optimal hyperparameters based on a given objective function and search space.

### Method
`run`

### Parameters
- **objective_function**: callable - The function to optimize.
- **direction**: str - The direction of optimization ('maximize' or 'minimize'). Defaults to 'maximize'.
- **runtime**: Union[int, float, str, None] - The maximum runtime for the search. Can be specified in seconds (int/float) or with a unit suffix (e.g., '10s', '5m', '1h').
- **steps**: Union[int, str, None] - The maximum number of steps (evaluations) for the search.
- **endless_mode**: bool - If True, the search will continue indefinitely until manually stopped.
- **seeding_steps**: Optional[int] - The number of initial steps to use for seeding the search process.
- **seeding_runtime**: Union[int, float, str, None] - The runtime allocated for seeding steps.
- **seeding_ratio**: Optional[float] - The ratio of seeding steps to total steps. Defaults to 0.25.
- **pruner**: any - A pruner object to use for early stopping of unpromising trials.
- **n_jobs**: int - The number of parallel jobs to run. Defaults to 1.
- **quiet**: bool - If True, suppresses output during the search.
- **ignore_nans**: bool - If True, NaNs returned by the objective function will be ignored.
- **mp_backend**: str - The multiprocessing backend to use ('auto', 'loky', 'multiprocessing', 'ray'). Defaults to 'auto'.
- **enable_rejection_cache**: bool - Whether to enable caching for rejected trials.
- **callbacks**: Union[callable, list, None] - A callable or list of callables to be executed during the search.
- **start_temperature**: float - The initial temperature for simulated annealing. Defaults to 1.
- **end_temperature**: float - The final temperature for simulated annealing. Defaults to 0.2.
- **kwargs**: any - Additional keyword arguments to pass to the objective function.
- **checkpoint_path**: any - Path to save/load checkpoints.
- **overwrite_checkpoint**: bool - Whether to overwrite existing checkpoints.
- **keep_history**: bool - Whether to keep the history of trials.

### Request Example
```python
def obj_func(param):
    return -(param["x"]-3)**2

search = pyhopper.Search(
    x = pyhopper.float(-5,5),
)
search.run(obj_func,"max","10s")
```

### Response
This method does not return a value directly; it initiates and manages the hyperparameter tuning process.
```

--------------------------------

### Run PyHopper Search with TensorboardCallback

Source: https://pyhopper.readthedocs.io/en/latest/recipes/tensorboard.html

This example demonstrates how to set up and run a hyperparameter search using PyHopper, integrating the TensorboardCallback to log experiment progress. Ensure numpy is imported as np.

```python
import time

import pyhopper
from pyhopper.callbacks.tensorboard import TensorboardCallback

def of(param):
    time.sleep(1)
    return np.random.default_rng().random()

search = pyhopper.Search(
    {
        "a": pyhopper.float(0, 1),
        "b": pyhopper.int(50, 100),
        "c": pyhopper.choice([0, 1, 2]),
    },
)
search.run(
    of,
    "max",
    "3min",
    callbacks=[
        TensorboardCallback("tb_example",comment="My experiment"),
    ],
)


```

--------------------------------

### Dask Backend Initialization

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/parallel.html

Initializes the backend using `dask.distributed`. This backend is experimental and requires `dask` to be installed. It sets up a `LocalCluster` and `Client` for distributed task execution.

```python
from dask.distributed import Client, LocalCluster, wait

self._backend_FIRST_COMPLETED = "FIRST_COMPLETED"
self._backend_ALL_COMPLETED = "ALL_COMPLETED"
cluster = LocalCluster(n_workers=n_jobs)
self._backend_task_executor = Client(cluster)
self._backend_wait_func = wait
```

--------------------------------

### PyTorch CIFAR-10 Training Setup

Source: https://pyhopper.readthedocs.io/en/latest/examples/torch_cifar10.html

Initializes a ResNet34 model, loss function, optimizer (SGD with momentum), and learning rate scheduler (CosineAnnealingLR) for training on CIFAR-10. It also sets up the data loaders.

```python
def train_cifar10(params, for_validation=True):
    model = torchvision.models.resnet34(pretrained=False, num_classes=10)
    model.conv1 = torch.nn.Conv2d(
        3, 64, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1), bias=False
    )
    model.maxpool = torch.nn.Identity()
    model = model.cuda()

    loss_fn = torch.nn.CrossEntropyLoss()

    # optimizer
    optimizer = torch.optim.SGD(
        model.parameters(),
        lr=params["lr"],
        momentum=0.9,
        weight_decay=params["weight_decay"],
        nesterov=True,
    )

    train_loader, val_loader = get_cifar_loader(
        128, params["erasing_prob"], for_validation
    )
    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
        optimizer, len(train_loader) * 100, eta_min=params["eta_min"]
    )
```

--------------------------------

### Log Evaluation Times with Custom Callback

Source: https://pyhopper.readthedocs.io/en/latest/recipes/custom_callbacks.html

This example demonstrates using `on_evaluate_start` and `on_evaluate_end` to log the sampling and finishing times of parameters, as well as the duration of the evaluation. Ensure `datetime` and `time` are imported.

```python
import datetime as dt
import time

class MyCallback(pyhopper.callbacks.Callback):
    def on_evaluate_start(self, param: dict, info: pyhopper.ParamInfo):
        print(
            f"{param['x']} sampled at {dt.datetime.fromtimestamp(info.sampled_at):%Y %B %d - %H:%m}"
        )

    def on_evaluate_end(self, param: dict, f: float, info: pyhopper.ParamInfo):
        print(
            f"{param['x']} finished at {dt.datetime.fromtimestamp(info.finished_at):%Y %B %d - %H:%m}"
        )
        print(f"Took {info.finished_at-info.sampled_at:0.1f} seconds")

def of(param):
    time.sleep(1)
    return np.random.default_rng().random()


search = pyhopper.Search(
    {
        "x": pyhopper.int(0, 10),
    },
)
r1 = search.run(
    of,
    "max",
    steps=3,
    quiet=True,
    callbacks=[MyCallback()],
)


```

--------------------------------

### Basic PyHopper Search Example

Source: https://pyhopper.readthedocs.io/en/latest/index.html

Demonstrates how to define an objective function and set up a PyHopper search for hyperparameter optimization. Supports various parameter types including integers, floats with formatting, NumPy arrays, and choices.

```python
import pyhopper

def my_objective(params: dict) -> float:
    model = build_model(params["hidden_size"],...)
    # .... train and evaluate the model
    return val_accuracy

search = pyhopper.Search(
    units   = pyhopper.int(100,500),
    dropout = pyhopper.float(0,0.4,"0.1f"), # 1 decimal digit
    lr      = pyhopper.float(1e-5,1e-2,"0.1g"), # loguniform, 1 significant
    matrix  = pyhopper.float(-1,1,shape=(20,20)), # numpy array
    opt     = pyhopper.choice(["adam","rmsprop","sgd"]),
)
best_params = search.run(my_objective, "maximize", "8h", n_jobs="per-gpu")

```

--------------------------------

### CIFAR-10 Data Loader Setup

Source: https://pyhopper.readthedocs.io/en/latest/examples/torch_cifar10.html

Configures data transformations, including random cropping, horizontal flipping, random erasing, and normalization. It also sets up data loaders for training and validation/testing with specified batch sizes and samplers.

```python
import numpy as np
import torchvision
import torch
import torch.utils.data
from torch.utils.data import SubsetRandomSampler
from tqdm.auto import tqdm

import pyhopper


def get_cifar_loader(batch_size, erasing_prob, for_validation):
    mean = np.array([125.30691805, 122.95039414, 113.86538318]) / 255.0
    std = np.array([62.99321928, 62.08870764, 66.70489964]) / 255.0

    train_transform = torchvision.transforms.Compose(
        [
            torchvision.transforms.RandomCrop(32, padding=4),
            torchvision.transforms.RandomHorizontalFlip(),
            torchvision.transforms.ToTensor(),
            torchvision.transforms.RandomErasing(p=erasing_prob),
            torchvision.transforms.Normalize(mean, std),
        ]
    )
    test_transform = torchvision.transforms.Compose(
        [
            torchvision.transforms.ToTensor(),
            torchvision.transforms.Normalize(mean, std),
        ]
    )

    dataset_dir = "~/.torchvision/datasets/CIFAR10"
    train_dataset = torchvision.datasets.CIFAR10(
        dataset_dir, train=True, transform=train_transform, download=True
    )
    train_sampler, test_sampler = torch.utils.data.RandomSampler(train_dataset), None
    if for_validation:
        test_dataset = torchvision.datasets.CIFAR10(
            dataset_dir, train=True, transform=test_transform, download=True
        )
        indices = np.random.default_rng(12345).permutation(len(train_dataset))
        valid_size = int(0.05 * len(train_dataset))
        train_idx, valid_idx = indices[valid_size:], indices[:valid_size]
        train_sampler = SubsetRandomSampler(train_idx)
        test_sampler = SubsetRandomSampler(valid_idx)
    else:
        test_dataset = torchvision.datasets.CIFAR10(
            dataset_dir, train=False, transform=test_transform, download=True
        )

    train_loader = torch.utils.data.DataLoader(
        train_dataset,
        batch_size=batch_size,
        sampler=train_sampler,
        num_workers=4,
        pin_memory=True,
        drop_last=True,
    )
    test_loader = torch.utils.data.DataLoader(
        test_dataset,
        batch_size=256,
        sampler=test_sampler,
        num_workers=4,
        pin_memory=True,
        drop_last=False,
    )
    return train_loader, test_loader
```

--------------------------------

### Implementing Pruners for Efficiency

Source: https://pyhopper.readthedocs.io/en/latest/quickstart.html

Use pruners to discontinue unpromising candidates early in the search process, saving computational resources. This example uses a QuantilePruner to remove candidates in the bottom 80%.

```python
search = pyhopper.Search(...)

search.run(
    pyhopper.wrap_n_times(noisy_objective,3),
    "maximize",
    steps=50,
    pruner = pyhopper.pruners.QuantilePruner(0.8) # discontinue candidates if they are in the bottom 80%
)

```

--------------------------------

### Pyhopper Search Initialization

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Initializes the search process with various configuration options. This snippet shows the core setup for a search run, including defining the objective function, search parameters, and execution context.

```python
if kwargs is None:
            kwargs = {}

        self._caught_exception = False
        schedule = ScheduledRun(
            steps,
            runtime,
            endless_mode,
            seeding_steps=seeding_steps,
            seeding_runtime=seeding_runtime,
            seeding_ratio=seeding_ratio,
            start_temperature=start_temperature,
            end_temperature=end_temperature,
        )
        task_executor = None
        if n_jobs != 1:
            task_executor = TaskManager(n_jobs, mp_backend)
            if task_executor.n_jobs == 1:
                task_executor = None  # '1x per-gpu' on single GPU machines -> No need for multiprocess overhead

        callbacks = convert_to_list(callbacks)
        if keep_history:
            callbacks.append(self._history)
        if checkpoint_path is not None:
            checkpoint_path = convert_to_checkpoint_path(checkpoint_path)
            self._checkpoint_path = checkpoint_path
            callbacks.append(CheckpointCallback(checkpoint_path))

        self._pruner = pruner
        self._run_context = RunContext(
            direction,
            pruner,
            ignore_nans,
            schedule,
            callbacks,
            task_executor,
            quiet,
        )

        self._f_cache.set_enable(enable_rejection_cache)
        self._signal_listener.register_signal(
            schedule.signal_gradually_quit, self._force_termination
        )
```

--------------------------------

### Get Initial Solution for Parameters

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Recursively determines the initial value for a given parameter, handling nested structures like dictionaries, lists, and wrapped samples.

```python
def _get_initial_solution(self, param):
        if isinstance(param, Parameter):
            init = param.initial_value
            if isinstance(init, WrappedSample):
                init.value = self._get_initial_solution(init.value)
            return init
        elif isinstance(param, dict):
            return {k: self._get_initial_solution(v) for k, v in param.items()}
        elif isinstance(param, list):
            return [self._get_initial_solution(v) for v in param]
        else:
            return param
```

--------------------------------

### Get GPU List and Multi-Instance Support

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/parallel.html

Retrieves a list of available GPU IDs and determines if multi-instance mode is supported. It checks the CUDA_VISIBLE_DEVICES environment variable if set, otherwise uses all available GPUs. Raises an error if nvidia-smi output cannot be parsed and no GPUs are specified.

```python
def get_gpu_list():
    gpu_ids, gpu_names, gpu_default_modes = parse_nvidia_smi()
    if "CUDA_VISIBLE_DEVICES" in os.environ.keys():
        # User specified which GPUs to use (could be a subset of the GPUs on the machine)
        if gpu_ids is None:
            print(
                "Warning: Could not parse output of 'nvidia-smi'. This probably means you need to upgrade the nvidia "
                "device drivers or reboot the machine. "
            )
        gpus_used = [s for s in os.environ["CUDA_VISIBLE_DEVICES"].split(",")]
    else:
        if gpu_ids is None:
            raise ValueError(
                "Error: Could not parse output of 'nvidia-smi'. Either specify the GPU ids via the "
                "'CUDA_VISIBLE_DEVICES' environment variable, or upgrade the nvidia device drivers and reboot the "
                "machine. "
            )
        # No explicit GPU defined -> let's use them all
        gpus_used = gpu_ids
    supports_multi_instance = True
    if gpu_ids is not None:
        for gpu in gpus_used:
            if not gpu_default_modes[gpu]:
                supports_multi_instance = False
    return gpus_used, supports_multi_instance
```

--------------------------------

### Defining Conditional Hyperparameter Choices for Optimizers

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Shows an example of using `pyhopper.cases` to define conditional hyperparameters for different optimization algorithms (sgd, nesterov, adam). Each optimizer has a specific set of hyperparameters (e.g., learning rate, momentum, beta) defined within its case.

```python
search = pyhopper.Search(
  optimizer = pyhopper.cases(
    sgd = {"lr": pyhopper.float(1e-5,1e-2)},
    nesterov = {"lr": pyhopper.float(1e-5,1e-2), "momentum": pyhopper.float(0.01,0.99)},
    adam = {"lr": pyhopper.float(1e-5,1e-2), "beta": pyhopper.float(0.1,0.2)},
)

```

--------------------------------

### Format Float Hyperparameters with fmt Argument

Source: https://pyhopper.readthedocs.io/en/latest/quickstart.html

Use the `fmt` argument as a shortcut for `log` and `precision`. For example, `"0.2f"` sets linear sampling with 2 digits after the comma, and `"0.1g"` sets logarithmic sampling with 1 significant digit.

```python
search = pyhopper.Search(
    {
        "dropout": pyhopper.float(0, 0.5, "0.2f"), # 2 digits after the comma (linear)
        "lr": pyhopper.float(1e-5, 1e-1, "0.1g"), # 1 significant digit (logarithmic)
    }
)

```

--------------------------------

### Defining a Conditional Parameter with PyHopper

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Illustrates how to create a conditional parameter using `pyhopper.cases`. This is useful for defining nested configuration spaces where choices depend on other outcomes. The example shows different types of cases, including simple values, integer ranges, and lists containing strings and integers.

```python
search = pyhopper.Search(
  cond_param = pyhopper.cases(
    case1="abc",
    other_case=pyhopper.int(0,10),
    third_case=["xyz",pyhopper.int(-10,0)])
)
# Generates samples
# {'cond_param': ('case1', 'abc')}
# {'cond_param': ('other_case', 0)}
# {'cond_param': ('other_case', 3)}
# {'cond_param': ('case1', 'abc')}
# {'cond_param': ('third_case', ['xyz', -2])}
# {'cond_param': ('third_case', ['xyz', -9])}
```

--------------------------------

### Sample Solution

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Initiates the sampling process for a complete solution based on the defined parameters.

```python
def sample_solution(self):
        return self._sample_solution_rec(self._params)
```

--------------------------------

### Quantile Pruner Initialization

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/pruners/pruners.html

Initializes the QuantilePruner. It takes a quantile `q` and an optional `warmup` period during which no pruning occurs.

```python
class QuantilePruner(Pruner):
    def __init__(self, q, warmup=5):
        super().__init__()
        if 1 <= q < 100:
            q /= 100
        if not 0 < q < 1:
            raise ValueError(f"q must be between 0 and 1 (got {q})")
        self.q = q
        self.n = None
        self.warmup = warmup
        self.intermediates = None
```

--------------------------------

### Set Global Pruner

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/pruners/pruners.html

Sets the global pruner object and resets the list of intermediate results. This function is typically called once before starting a search.

```python
def set_global_pruner(pruner):
    global _global_pruner_obj
    global _global_intermediate_results_list
    _global_pruner_obj = pruner
    _global_intermediate_results_list = []
```

--------------------------------

### CheckpointCallback

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/callbacks/callbacks.html

The CheckpointCallback automatically saves the search state to a specified checkpoint path during the search process. It restores the search from the checkpoint if one exists at the start.

```APIDOC
## CheckpointCallback

### Description
Saves and restores the search state to/from a checkpoint file.

### Methods
- **__init__(self, checkpoint_path)**: Initializes the callback with the path to the checkpoint file.
- **on_search_start(self, search: "pyhopper.Search")**: Restores the search from the checkpoint if it exists.
- **on_evaluate_end(self, candidate: dict, f: float, info: ParamInfo)**: Saves the search state after evaluation.
- **on_evaluate_pruned(self, candidate: dict, info: ParamInfo)**: Saves the search state if a candidate is pruned.
- **on_evaluate_nan(self, candidate: dict, info: ParamInfo)**: Saves the search state if a candidate evaluates to NaN.
- **on_new_best(self, new_best: dict, f: float, info: ParamInfo)**: Saves the search state when a new best is found.
- **on_search_end(self)**: Saves the final search state.
```

--------------------------------

### Run Hyperparameter Tuning

Source: https://pyhopper.readthedocs.io/en/latest/api/search.html

Initiate the hyperparameter tuning process by providing an objective function, direction, and runtime constraints. This method explores the search space to find optimal parameters.

```python
>>> def obj_func(param):
>>>     return -(param["x"]-3)**2
>>>
>>> search = pyhopper.Search(
>>>     x = pyhopper.float(-5,5),
>>> )
>>> search.run(obj_func,"max","10s")
```

--------------------------------

### Recursive Mutation from Best Solution

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Recursively mutates a solution structure starting from the best known solution. It uses a temperature parameter to control mutation probability and a bitmask to determine which parameters to mutate.

```python
def _mutate_from_best_rec(
        self,
        temperature,
        node=None,
        best_node=None,
        bitmask=None,
    ):
        if isinstance(node, Parameter):
            p = True if bitmask is None else bitmask.pop()
            # consume one bit -> tells us if we should mutate or not
            sample = node.mutate(best_node, temperature=temperature) if p else best_node
            if isinstance(sample, WrappedSample):
                if best_node.aux != sample.aux:
                    # switched to a new case
                    sample.value = self._sample_solution_rec(sample.value)
                else:
                    sample.value = self._mutate_from_best_rec(
                        temperature,
                        node=sample.value,
                        best_node=best_node.value,
                        bitmask=None,  # no further masking after level 1
                    )
            return sample
        elif isinstance(node, dict):
            return {
                k: self._mutate_from_best_rec(
                    temperature, node=node[k], best_node=best_node[k], bitmask=bitmask
                )
                for k, v in node.items()
            }
        elif isinstance(node, list):
            return [
                self._mutate_from_best_rec(
                    temperature, node=node[i], best_node=best_node[i], bitmask=bitmask
                )
                for i in range(len(node))
            ]
        else:
            return node
```

--------------------------------

### TopK Pruner Initialization

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/pruners/pruners.html

Initializes the TopKPruner, setting up internal lists to track the top k intermediate results and their corresponding values.

```python
[docs] class TopKPruner(Pruner):
    def __init__(self, k):
        super().__init__()
        self.k = k
        self.n = None
        self.top_k_intermediates = []
        self.top_k_of = []
```

--------------------------------

### Checkpoint Callback Implementation

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/callbacks/callbacks.html

The CheckpointCallback saves the search state to a specified file path after each evaluation or pruning step. It also handles restoring the search from a checkpoint if one exists when the search starts.

```python
class CheckpointCallback(Callback):
    def __init__(self, checkpoint_path):
        self._checkpoint_path = convert_to_checkpoint_path(checkpoint_path)
        self._search_obj = None

    @property
    def checkpoint_path(self):
        return self._checkpoint_path

    def on_search_start(self, search: "pyhopper.Search"):
        self._search_obj = search
        if os.path.isfile(self._checkpoint_path):
            self._search_obj.load(self._checkpoint_path)
            self._search_obj._run_context.pbar.write(
                f"Restored search from checkpoint '{self._checkpoint_path}'"
            )

    def on_evaluate_end(self, candidate: dict, f: float, info: ParamInfo):
        self._search_obj.save(self._checkpoint_path)

    def on_evaluate_pruned(self, candidate: dict, info: ParamInfo):
        self._search_obj.save(self._checkpoint_path)

    def on_evaluate_nan(self, candidate: dict, info: ParamInfo):
        self._search_obj.save(self._checkpoint_path)

    def on_new_best(self, new_best: dict, f: float, info: ParamInfo):
        self._search_obj.save(self._checkpoint_path)

    def on_search_end(self):
        self._search_obj.save(
            self._checkpoint_path
        )  # Search is done -> forget run context

```

--------------------------------

### SaveBestOnDisk on_new_best Method

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/callbacks/callbacks.html

Saves the new best result to the specified file using pickle when a new best is found during optimization.

```python
    def on_new_best(self, new_best, f):
        with open(self._filename, "wb") as f:
            pickle.dump(new_best, f)
```

--------------------------------

### PowerOfIntParameter Initialization and Sampling

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/parameters.html

Initializes a parameter constrained to powers of two and defines its sampling logic. It works by sampling from a logarithmic scale and then exponentiating the result.

```python
class PowerOfIntParameter(IntParameter):
    def __init__(
        self,
        shape,
        lb,
        ub,
        init,
        power_of,
        multiple_of,
        mutation_strategy,
        sampling_strategy,
    ):
        super().__init__(
            shape, lb, ub, init, multiple_of, mutation_strategy, sampling_strategy
        )
        self._power_of = power_of
        self._log_param = IntParameter(
            shape, int(np.log2(lb)), int(np.log2(ub)), None, None, None, None
        )
        if init is None:
            init = self._log_param.initial_value
            init = 2**init
            init = self._round_to_multiple_of(init)
            init = self._cast_if_scalar(init)
            self.initial_value = init

    def sample(self):
        if self._sampling_strategy is not None:
            # Custom sampling
            new_value = self._sampling_strategy()
        else:
            new_value = self._log_param.sample()
            new_value = 2**new_value
        new_value = self._round_to_multiple_of(new_value)

        if self._lb is not None or self._ub is not None:
            new_value = np.clip(new_value, self._lb, self._ub)
        return self._cast_if_scalar(new_value)
```

--------------------------------

### Example Output of Cross-validation

Source: https://pyhopper.readthedocs.io/en/latest/recipes/custom_validation.html

This output shows the print statements from the `noisy_objective` function during the PyHopper search. Each candidate is evaluated multiple times (n=3 in this case), and the `eval_index` indicates which cross-validation step is currently being performed for a given candidate.

```text
> x=-0.2101340164769267, eval_index=0
> x=-0.2101340164769267, eval_index=1
> x=-0.2101340164769267, eval_index=2
> x=-0.0846185419082141, eval_index=0
> x=-0.0846185419082141, eval_index=1
> x=-0.0846185419082141, eval_index=2


```

--------------------------------

### SaveBestOnDisk Callback Initialization

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/callbacks/callbacks.html

Initializes the SaveBestOnDisk callback, setting up the filename for saving the best result. It prevents specifying both filename and directory.

```python
class SaveBestOnDisk(Callback):
    def __init__(self, filename=None, dir=None):
        if dir is not None and filename is not None:
            raise ValueError("Cannot specify filename and dir at the same time.")
        if dir is not None:
            os.makedirs(dir, exist_ok=True)
            for i in range(10000):
                filename = os.path.join(dir, f"run_{i:05d}.pkl")
                if not os.path.isfile(filename):
                    break

        self._filename = filename
```

--------------------------------

### Base Callback Class

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/callbacks/callbacks.html

The base Callback class defines methods that can be overridden to perform actions at different stages of the hyperparameter search. These include starting and ending the search, evaluating candidates, and handling new best results.

```python
import pickle
import os

import pyhopper
from pyhopper.utils import ParamInfo, convert_to_checkpoint_path
import time



class Callback:
    def on_search_start(self, search: "pyhopper.Search"):
        """Called at the beginning of the search

        :param search: `pyhopper.Search` object handling the search
        """
        pass



    def on_evaluate_start(self, candidate: dict, info: ParamInfo):
        """Called after `candidate` was sampled and scheduled for evaluation

        :param candidate: Parameter value of the candidate to be evaluated
        """
        pass



    def on_evaluate_end(self, candidate: dict, f: float, info: ParamInfo):
        """Called after `candidate` was successfully evaluated

        :param candidate: Parameter value of the evaluated candidate
        :param f: Value of the objective function corresponding to the candidate
        """
        pass



    def on_duplicate_sampled(self, candidate: dict, info: ParamInfo):
        """Called if `candidate` was sampled twice

        :param candidate: Parameter value of the sampled candidate
        """
        pass



    def on_evaluate_pruned(self, candidate: dict, info: ParamInfo):
        """Called if `candidate` was pruned (by an :meth:`pyhopper.pruners.Pruner`)

        :param candidate: Parameter value of the pruned candidate
        """
        pass



    def on_evaluate_nan(self, candidate: dict, info: ParamInfo):
        """Called if `candidate` is evaluated to NaN and the `ignore_nans` argument of `run` was set to True

        :param candidate: Parameter value that evaluate to NaN
        """
        pass



    def on_new_best(self, new_best: dict, f: float, info: ParamInfo):
        """Called when a new best parameter is found

        :param new_best: Value of the new best parameter
        :param f: Value of the objective function corresponding to the new best parameter
        """
        pass



    def on_search_end(self):
        """Called at the end of the search process"""
        pass



    def state_dict(self):
        """Called when a checkpoint of the hyperparameter search is created to backup the state
        :return A dict containing the internal (runtime) state of the callback. None if the callback has no state"""
        return None



    def load_state_dict(self, state_dict):
        """
        Restores the internal state of the callback.
        :param state_dict: A dict created by the self.state_dict method
        """
        pass

```

--------------------------------

### Format Steps to Pretty String

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/utils.html

Converts a number of steps into a human-readable string with suffixes like 'k' for thousands and 'M' for millions. Handles None input by returning '-'.

```python
def steps_to_pretty_str(steps):
    if steps is None:
        return "-"
    if steps > 1e6:
        return f"{steps//1e6:0.0f}M"
    if steps > 1e3:
        return f"{steps//1e3:0.0f}k"
    return str(steps)
```

--------------------------------

### Create Search Object with Dictionary

Source: https://pyhopper.readthedocs.io/en/latest/api/search.html

Instantiate a Search object by passing a dictionary that defines the search space. This is equivalent to using named arguments.

```python
>>> search = pyhopper.Search({
>>>     "x": pyhopper.float(0,1),
>>>     "y": pyhopper.int(0,10),
>>>     "z": 2,
>>> })
```

--------------------------------

### Search Object Initialization

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Initializes the Search object, setting up the parameter space, best solution, and internal caches. It merges provided arguments (dictionaries or keyword arguments) to define the search space.

```python
parameters = {}
        if len(args) > 0:
            parameters = merge_dicts(*args)

        parameters = merge_dicts(parameters, kwargs)

        self._params = parameters
        self._best_solution = Candidate(self._get_initial_solution(self._params))

        self._free_param_count = self._count_free_parameters()
        self._best_f = None
        self._f_cache = EvaluationCache()
        self._run_context = None
        self._manually_queued_candidates = []

        self._signal_listener = SignalListener()
        self._history = History()
        self._checkpoint_path = None
        self._caught_exception = False
```

--------------------------------

### Submit Candidate Evaluation

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Submits a candidate solution for evaluation, checking for cache hits and invoking callbacks before evaluation.

```python
def _submit_candidate(self, objective_function, candidate_type, candidate, kwargs):
        if candidate.unwrapped_value in self._f_cache:
            return False

        param_info = ParamInfo(candidate_type, sampled_at=time.time())
        for c in self._run_context.callbacks:
            c.on_evaluate_start(candidate.unwrapped_value, param_info)
```

--------------------------------

### Create Search Object with Named Arguments

Source: https://pyhopper.readthedocs.io/en/latest/api/search.html

Instantiate a Search object by passing search space parameters as named arguments. This is a convenient way to define the optimization problem.

```python
>>> search = pyhopper.Search(
>>>     x = pyhopper.float(0,1),
>>>     y = pyhopper.int(0,10),
>>>     z = 2,
>>> )
```

--------------------------------

### Define and Run Hyperparameter Search with PyHopper

Source: https://pyhopper.readthedocs.io/en/latest/quickstart.html

This snippet shows how to initialize the `pyhopper.Search` class with a dictionary defining the search space, including constants, integers, floats, and choices. It then runs the search using a provided objective function, specifying the optimization direction and number of steps. Remove `quiet=True` to see a progress bar and summary.

```python
import pyhopper

def dummy_objective(param :dict) -> float:
    print(param)
    return 0

search = pyhopper.Search(
    {
        "my_const": "cifar10",
        "my_int": pyhopper.int(100,500),
        "my_float": pyhopper.float(0,0.4),
        "my_choice": pyhopper.choice(["adam","rmsprop","sgd"]),
    }
)
search.run(dummy_objective,"maximize",steps=5,quiet=True)

```

```text
> {'my_const': 'cifar10', 'my_int': 364, 'my_float': 0.3404501564, 'my_choice': 'sgd'}
> {'my_const': 'cifar10', 'my_int': 408, 'my_float': 0.1056544366, 'my_choice': 'adam'}
> {'my_const': 'cifar10', 'my_int': 206, 'my_float': 0.2091606387, 'my_choice': 'adam'}
> {'my_const': 'cifar10', 'my_int': 438, 'my_float': 0.1960442887, 'my_choice': 'sgd'}
> {'my_const': 'cifar10', 'my_int': 156, 'my_float': 0.2719664365, 'my_choice': 'sgd'}

```

--------------------------------

### Mutate from Best Solution

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Initiates the mutation process from the best solution found so far. It calculates the number of parameters to mutate based on temperature and shuffles a bitmask to randomize mutation.

```python
def mutate_from_best(self, temperature):

        temperature = float(np.clip(temperature, 0, 1))
        node = self._params
        best_node = self._best_solution.value
        # With decreasing temperature we resample/mutate fewer parameters
        amount_to_mutate = int(
            max(round(temperature * self.free_param_count), 1) # at least 1, at most all
        )
        bitmask = [i < amount_to_mutate for i in range(self.free_param_count)]
        np.random.default_rng().shuffle(bitmask)

        return self._mutate_from_best_rec(temperature, node, best_node, bitmask)
```

--------------------------------

### ChoiceParameter Initialization and Sampling

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/parameters.html

Initializes a parameter that selects from a list of options. It uses an internal integer parameter to manage the selection index and provides sampling logic.

```python
class ChoiceParameter(Parameter):
    def __init__(self,
        options,
        init_index,
        is_ordinal,
        mutation_strategy,
        sampling_strategy,
    ):
        super().__init__()
        self._options = options
        self._is_ordinal = is_ordinal
        self._mutation_strategy = mutation_strategy
        self._sampling_strategy = sampling_strategy
        self._int_param = IntParameter(
            None, 0, len(options) - 1, None, None, None, None
        )
        if init_index is None:
            if is_ordinal:
                init_index = (len(options) - 1) // 2
            else:
                init_index = 0
        self.initial_value = WrappedSample(options[init_index], init_index)

    def sample(self):
        if self._sampling_strategy is not None:
            new_index = self._sampling_strategy()
        else:
            new_index = self._int_param.sample()

        return WrappedSample(value=self._options[new_index], aux=new_index)
```

--------------------------------

### load

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Loads the internal state of the search from a checkpoint file. This includes the run context, cache, history, best found solution, and optionally the pruner state.

```APIDOC
## load

### Description
Loads the internal state of the search from a checkpoint file. This includes the run context, cache, history, best found solution, and optionally the pruner state.

### Parameters
#### Path Parameters
- **checkpoint_path** (str) - Required - File from which to load the checkpoint
- **pruner** (Pruner) - Optional - Pruner object whose internal state should also be loaded from the checkpoint

### Raises
- **ValueError**: If the checkpoint file cannot be parsed or if there's a conflict with the pruner object.
```

--------------------------------

### Concurrent Futures Backend Initialization

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/parallel.html

Initializes the backend using `concurrent.futures.ProcessPoolExecutor`. This is the default backend for parallel processing.

```python
self._backend_wait_func = concurrent.futures.wait
self._backend_FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED
self._backend_ALL_COMPLETED = concurrent.futures.ALL_COMPLETED
self._backend_task_executor = concurrent.futures.ProcessPoolExecutor(
    max_workers=n_jobs
)
```

--------------------------------

### TaskManager Initialization with Multiprocessing Backends

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/parallel.html

Initializes the TaskManager, validating the multiprocessing backend and configuring job distribution. It supports 'auto', 'multiprocessing', 'dask', and 'dask-cuda', with specific handling for 'per_gpu' job specifications.

```python
class TaskManager:
    def __init__(self, n_jobs, mp_backend):
        admissible_mp_backends = [
            "auto",
            "multiprocessing",
            "dask",
            "dask-cuda",
        ]
        if mp_backend not in admissible_mp_backends:
            raise ValueError(
                f"Unknown multiprocessing backend '{mp_backend}'. Valid options are {str(admissible_mp_backends)}"
            )
        self._gpu_allocator = None
        self._pending_candidates = []
        self._pending_futures = []
        if isinstance(n_jobs, str) and (
            n_jobs.endswith("per_gpu") or n_jobs.endswith("per-gpu")
        ):
            if mp_backend == "dask-cuda":
                mp_backend = "dask-cuda"
            elif mp_backend in ["auto", "multiprocessing"]:
                mp_backend = "multiprocessing"
                # workaround if dask-cuda does not work
                factor = parse_factor(n_jobs)
                self._gpu_allocator = GPUAllocator(factor)
                n_jobs = self._gpu_allocator.num_gpus
                self._queue_max = n_jobs
            else:
                raise ValueError(
                    f"Cannot use mp_backend ```{mp_backend}``` when using per_gpu mode. "
                    "Valid options are 'dask-cuda' and 'multiprocessing' ('auto' defaults to 'multiprocessing')"
                )
        else:
            n_jobs = int(n_jobs)
            if mp_backend == "auto":
                mp_backend = "multiprocessing"
            if n_jobs <= 0:
                n_jobs = len(os.sched_getaffinity(0))
            self._queue_max = n_jobs

        if mp_backend == "dask-cuda":
            # Experimental
            try:
                from dask_cuda import LocalCUDACluster
                from dask.distributed import Client, wait
            except ImportError:
                raise ValueError(
                    "Could not import cuda-dask. Make sure dask is installed ```pip3 install -U cuda-dask```. "
                    + str(sys.exc_info()[0])
                )
            self._backend_FIRST_COMPLETED = "FIRST_COMPLETED"
            self._backend_ALL_COMPLETED = "ALL_COMPLETED"
            cluster = LocalCUDACluster()
            # print("CUDA_VISIBLE_DEVICES:", cluster.cuda_visible_devices)
            self._queue_max = len(cluster.cuda_visible_devices)
            self._backend_task_executor = Client(cluster)
            self._backend_wait_func = wait
            n_jobs = -1
        elif mp_backend == "multiprocessing":
            import concurrent.futures
```

--------------------------------

### Create a Custom Parameter

Source: https://pyhopper.readthedocs.io/en/latest/api/parameters.html

Define a parameter with custom seeding and mutation functions. An initial value can also be provided.

```python
>>> pyhopper.custom(_seeding_fn =None_, _mutation_fn =None_, _init =None_)
```

--------------------------------

### best

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/search.html

Returns a dictionary containing the best found parameter set so far. Returns None if no candidate has been evaluated yet.

```APIDOC
## best

### Description
A dict object containing the best found parameter so far. None if no candidate has been evaluated yet.

### Returns
- **Optional[dict]**: A dictionary of the best parameters found, or None.
```

--------------------------------

### Create a Float Parameter

Source: https://pyhopper.readthedocs.io/en/latest/api/parameters.html

Defines a floating-point parameter. Use bounds to set the range, and 'log=True' for log-uniform distribution. Format strings like '0.1f' or '0.1g' can simplify setting distribution and precision.

```python
>>> pyhopper.float(1) # uniform distribution in range [0,1]
>>> pyhopper.float(-1,1) # uniform distribution in range [-1,1]
>>> pyhopper.float(1e-5,1e-2, log=True) # loguniform distrubution
>>> pyhopper.float(0,0.5, "0.1f") # uniform distribution, quantized to 0.1 increments (1 decimal digit)
>>> pyhopper.float(1e-5,1e-2, "0.1g") # loguniform distribution, logquantized to to 1 signficiant digit
>>> pyhopper.float(-1,1, shape=(3,3)) # multidimensional parameter
```

```python
import pyhopper

pyhopper.float(0,1) #  Bounded by 0 and 1
pyhopper.float(1)   #  Implicitly bounded by 0 and 1
pyhopper.float()    #  Unbounded
```

--------------------------------

### FloatParameter Initialization

Source: https://pyhopper.readthedocs.io/en/latest/_modules/pyhopper/parameters.html

Initializes a float parameter with bounds, precision, and strategies for mutation and sampling. Handles default initialization if not provided.

```python
class FloatParameter(Parameter):
    def __init__(
        self,
        shape,
        lb,
        ub,
        init,
        precision,
        mutation_strategy,
        sampling_strategy,
    ):
        super().__init__()
        self._lb = lb
        self._ub = ub
        self._precision = precision
        self._mutation_strategy = mutation_strategy
        self._sampling_strategy = sampling_strategy

        self._shape = shape

        if init is None:
            if lb is None and ub is None:
                init = 0
            elif lb is None:
                init = ub
            elif ub is None:
                init = lb
            else:
                init = (lb + ub) / 2
            init = self._round_and_clip(init)
            init = self._cast_if_scalar(init)
        self.initial_value = init
```

--------------------------------

### Saving and Loading Pruner States with Checkpoints

Source: https://pyhopper.readthedocs.io/en/latest/recipes/checkpoints.html

Shows how to include the internal state of a pruner object when saving and loading checkpoints. This ensures that the pruner's progress is also restored when resuming a search.

```python
search = pyhopper.Search( ... )
pruner = pyhopper.pruners.TopKPruner(10)

search.run(..., pruner=pruner, checkpoint_path="my_checkpoint.ckpt")

# ....

other_search = pyhopper.Search( ... )
other_pruner = pyhopper.pruners.TopKPruner(10)
other_search.load("my_checkpoint.ckpt", pruner=other_pruner)
# Restores the search history and the internal state of the pruner
```
Tokens:
Raw