Skip to content

Target function script runner

smac.runner.target_function_script_runner #

TargetFunctionScriptRunner #

TargetFunctionScriptRunner(
    target_function: str,
    scenario: Scenario,
    required_arguments: list[str] = None,
)

Bases: AbstractSerialRunner

Class to execute target functions from scripts. Uses Popen to execute the script in a subprocess.

The following example shows how the script is called: target_function --instance=test --instance_features=test --seed=0 --hyperparameter1=5323

The script must return an echo in the following form (white-spaces are removed): cost=0.5; runtime=0.01; status=SUCCESS; additional_info=test (single-objective) cost=0.5, 0.4; runtime=0.01; status=SUCCESS; additional_info=test (multi-objective)

The status must be a string and must be one of the StatusType values. However, runtime, status and additional_info are optional.

Note

Everytime an instance is passed, also an instance feature in form of a comma-separated list (no spaces) of floats is passed. If no instance feature for the instance is given, an empty list is passed.

PARAMETER DESCRIPTION
target_function

The target function.

TYPE: Callable

scenario

TYPE: Scenario

required_arguments

A list of required arguments, which are passed to the target function.

TYPE: list[str] DEFAULT: None

Source code in smac/runner/target_function_script_runner.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def __init__(
    self,
    target_function: str,
    scenario: Scenario,
    required_arguments: list[str] = None,
):
    if required_arguments is None:
        required_arguments = []
    super().__init__(scenario=scenario, required_arguments=required_arguments)
    self._target_function = target_function

    # Check if target function is callable
    if not isinstance(self._target_function, str):
        raise TypeError(
            "Argument `target_function` must be a string but is type" f"`{type(self._target_function)}`."
        )

    if self._scenario.trial_memory_limit is not None:
        logger.warning("Trial memory limit is not supported for script target functions.")

    if self._scenario.trial_walltime_limit is not None:
        logger.warning("Trial walltime limit is not supported for script target functions.")

__call__ #

__call__(
    algorithm_kwargs: dict[str, Any],
) -> tuple[str, str]

Calls the algorithm, which is processed in the run method.

Source code in smac/runner/target_function_script_runner.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def __call__(
    self,
    algorithm_kwargs: dict[str, Any],
) -> tuple[str, str]:
    """Calls the algorithm, which is processed in the ``run`` method."""
    cmd = [self._target_function]
    for k, v in algorithm_kwargs.items():
        v = str(v)
        k = str(k)

        # Let's remove some spaces
        v = v.replace(" ", "")

        cmd += [f"--{k}={v}"]

    logger.debug(f"Calling: {' '.join(cmd)}")
    p = Popen(cmd, shell=False, stdout=PIPE, stderr=PIPE, universal_newlines=True)
    output, error = p.communicate()

    logger.debug("Stdout: %s" % output)
    logger.debug("Stderr: %s" % error)

    return output, error

count_available_workers #

count_available_workers() -> int

Returns the number of available workers. Serial workers only have one worker.

Source code in smac/runner/abstract_serial_runner.py 
41
42
43
def count_available_workers(self) -> int:
    """Returns the number of available workers. Serial workers only have one worker."""
    return 1

run #

run(
    config: Configuration,
    instance: Optional[str] = None,
    budget: Optional[float] = None,
    seed: Optional[int] = None,
    additional_info: Optional[dict[str, Any]] = None,
) -> tuple[
    StatusType,
    Union[float, list[float]],
    float,
    float,
    dict[str, Any],
]

Calls the target function.

PARAMETER DESCRIPTION
config

Configuration to be passed to the target function.

TYPE: Configuration

instance

The Problem instance.

TYPE: str | None, defaults to None DEFAULT: None

budget

A positive, real-valued number representing an arbitrary limit to the target function handled by the target function internally.

TYPE: float | None, defaults to None DEFAULT: None

seed

TYPE: int, defaults to None DEFAULT: None

additional_info

All further additional trial information.

TYPE: dict DEFAULT: None

RETURNS DESCRIPTION
status

Status of the trial.

TYPE: StatusType

cost

Resulting cost(s) of the trial.

TYPE: float | list[float]

runtime

The time the target function took to run.

TYPE: float

cpu_time

The time the target function took on the hardware to run.

TYPE: float

val_additional_info

All further additional trial information.

TYPE: dict

Source code in smac/runner/target_function_script_runner.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def run(
    self,
    config: Configuration,
    instance: Optional[str] = None,
    budget: Optional[float] = None,
    seed: Optional[int] = None,
    additional_info: Optional[dict[str, Any]] = None,
) -> tuple[StatusType, Union[float, list[float]], float, float, dict[str, Any]]:
    """Calls the target function.

    Parameters
    ----------
    config : Configuration
        Configuration to be passed to the target function.
    instance : str | None, defaults to None
        The Problem instance.
    budget : float | None, defaults to None
        A positive, real-valued number representing an arbitrary limit to the target function
        handled by the target function internally.
    seed : int, defaults to None
    additional_info : dict
        All further additional trial information.

    Returns
    -------
    status : StatusType
        Status of the trial.
    cost : float | list[float]
        Resulting cost(s) of the trial.
    runtime : float
        The time the target function took to run.
    cpu_time : float
        The time the target function took on the hardware to run.
    val_additional_info : dict
        All further additional trial information.
    """
    # The kwargs are passed to the target function.
    kwargs: dict[str, Any] = {}
    if "seed" in self._required_arguments:
        kwargs["seed"] = seed

    if "instance" in self._required_arguments:
        kwargs["instance"] = instance

        # In contrast to the normal target function runner, we also add the instance features here.
        if self._scenario.instance_features is not None and instance in self._scenario.instance_features:
            kwargs["instance_features"] = self._scenario.instance_features[instance]
        else:
            kwargs["instance_features"] = []

    if "budget" in self._required_arguments:
        kwargs["budget"] = budget

    if "cutoff" in self._required_arguments and additional_info is not None:
        kwargs["cutoff"] = additional_info["cutoff"]

    # Presetting
    cost: float | list[float] = self._crash_cost
    runtime = 0.0
    cpu_time = runtime
    val_additional_info = {}
    status = StatusType.SUCCESS

    # Add config arguments to the kwargs
    for k, v in dict(config).items():
        if k in kwargs:
            raise RuntimeError(f"The key {k} is already in use. Please use a different one.")
        kwargs[k] = v

    # Call target function
    start_time = time.time()
    cpu_time = time.process_time()
    output, error = self(kwargs)
    cpu_time = time.process_time() - cpu_time
    runtime = time.time() - start_time

    # Now we have to parse the std output
    # First remove white-spaces
    output = output.replace(" ", "")

    outputs = {}
    for pair in output.split(";"):
        try:
            kv = pair.split("=")
            k, v = kv[0], kv[1]

            # Get rid of the trailing newline
            v = v.strip()

            outputs[k] = v
        except Exception:
            pass

    # Parse status
    if "status" in outputs:
        status = getattr(StatusType, outputs["status"])

    # Parse costs (depends on the number of objectives)
    if "cost" in outputs:
        if self._n_objectives == 1:
            cost = float(outputs["cost"])
        else:
            costs = outputs["cost"].split(",")
            costs = [float(c) for c in costs]
            cost = costs

            if len(costs) != self._n_objectives:
                raise RuntimeError("The number of costs does not match the number of objectives.")
    else:
        status = StatusType.CRASHED

    # Overwrite runtime
    if "runtime" in outputs:
        runtime = float(outputs["runtime"])

    # Overwrite CPU time
    if "cpu_time" in outputs:
        cpu_time = float(outputs["cpu_time"])

    # Add additional info
    if "additional_info" in outputs:
        val_additional_info["additional_info"] = outputs["additional_info"]

    if status != StatusType.SUCCESS:
        val_additional_info["error"] = error

        if cost != self._crash_cost:
            cost = self._crash_cost
            logger.info(
                "The target function crashed but returned a cost. The cost is ignored and replaced by crash cost."
            )

    return status, cost, runtime, cpu_time, val_additional_info

run_wrapper #

run_wrapper(
    trial_info: TrialInfo,
    **dask_data_to_scatter: dict[str, Any]
) -> tuple[TrialInfo, TrialValue]

Wrapper around run() to execute and check the execution of a given config. This function encapsulates common handling/processing, so that run() implementation is simplified.

PARAMETER DESCRIPTION
trial_info

Object that contains enough information to execute a configuration run in isolation.

TYPE: RunInfo

dask_data_to_scatter

When a user scatters data from their local process to the distributed network, this data is distributed in a round-robin fashion grouping by number of cores. Roughly speaking, we can keep this data in memory and then we do not have to (de-)serialize the data every time we would like to execute a target function with a big dataset. For example, when your target function has a big dataset shared across all the target function, this argument is very useful.

TYPE: dict[str, Any] DEFAULT: {}

RETURNS DESCRIPTION
info

An object containing the configuration launched.

TYPE: TrialInfo

value

Contains information about the status/performance of config.

TYPE: TrialValue

Source code in smac/runner/abstract_runner.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def run_wrapper(
    self, trial_info: TrialInfo, **dask_data_to_scatter: dict[str, Any]
) -> tuple[TrialInfo, TrialValue]:
    """Wrapper around run() to execute and check the execution of a given config.
    This function encapsulates common
    handling/processing, so that run() implementation is simplified.

    Parameters
    ----------
    trial_info : RunInfo
        Object that contains enough information to execute a configuration run in isolation.
    dask_data_to_scatter: dict[str, Any]
        When a user scatters data from their local process to the distributed network,
        this data is distributed in a round-robin fashion grouping by number of cores.
        Roughly speaking, we can keep this data in memory and then we do not have to (de-)serialize the data
        every time we would like to execute a target function with a big dataset.
        For example, when your target function has a big dataset shared across all the target function,
        this argument is very useful.

    Returns
    -------
    info : TrialInfo
        An object containing the configuration launched.
    value : TrialValue
        Contains information about the status/performance of config.
    """
    start = time.time()
    cpu_time = time.process_time()
    try:
        status, cost, runtime, cpu_time, additional_info = self.run(
            config=trial_info.config,
            instance=trial_info.instance,
            budget=trial_info.budget,
            seed=trial_info.seed,
            additional_info=trial_info.additional_info,
            **dask_data_to_scatter,
        )
    except Exception as e:
        status = StatusType.CRASHED
        cost = self._crash_cost
        cpu_time = time.process_time() - cpu_time
        runtime = time.time() - start

        # Add context information to the error message
        exception_traceback = traceback.format_exc()
        error_message = repr(e)
        additional_info = {
            "traceback": exception_traceback,
            "error": error_message,
        }

    end = time.time()

    # Catch NaN or inf
    if not np.all(np.isfinite(cost)):
        logger.warning(
            "Target function returned infinity or nothing at all. Result is treated as CRASHED"
            f" and cost is set to {self._crash_cost}."
        )

        if "traceback" in additional_info:
            logger.warning(f"Traceback: {additional_info['traceback']}\n")

        status = StatusType.CRASHED

    if status == StatusType.CRASHED:
        cost = self._crash_cost

    trial_value = TrialValue(
        status=status,
        cost=cost,
        time=runtime,
        cpu_time=cpu_time,
        additional_info=additional_info,
        starttime=start,
        endtime=end,
    )

    return trial_info, trial_value

submit_trial #

submit_trial(trial_info: TrialInfo) -> None

This function submits a trial_info object in a serial fashion. As there is a single worker for this task, this interface can be considered a wrapper over the run method.

Both result/exceptions can be completely determined in this step so both lists are properly filled.

PARAMETER DESCRIPTION
trial_info

An object containing the configuration launched.

TYPE: TrialInfo

Source code in smac/runner/abstract_serial_runner.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
def submit_trial(self, trial_info: TrialInfo) -> None:
    """This function submits a trial_info object in a serial fashion. As there is a single
     worker for this task, this interface can be considered a wrapper over the `run` method.

    Both result/exceptions can be completely determined in this step so both lists
    are properly filled.

    Parameters
    ----------
    trial_info : TrialInfo
        An object containing the configuration launched.
    """
    self._results_queue.append(self.run_wrapper(trial_info))

wait #

wait() -> None

The SMBO/intensifier might need to wait for trials to finish before making a decision. For serial runners, no wait is needed as the result is immediately available.

Source code in smac/runner/abstract_serial_runner.py 
28
29
30
31
32
33
34
35
36
def wait(self) -> None:
    """The SMBO/intensifier might need to wait for trials to finish before making a decision.
    For serial runners, no wait is needed as the result is immediately available.
    """
    # There is no need to wait in serial runners. When launching a trial via submit, as
    # the serial trial uses the same process to run, the result is always available
    # immediately after. This method implements is just an implementation of the
    # abstract method via a simple return, again, because there is no need to wait
    return