mlos_bench.storage.sql.experiment

Storage.Experiment interface implementation for saving and restoring the benchmark experiment data using SQLAlchemy backend.

Classes

Experiment

Logic for retrieving and storing the results of a single experiment.

Module Contents

class mlos_bench.storage.sql.experiment.Experiment(*, engine: sqlalchemy.engine.Engine, schema: mlos_bench.storage.sql.schema.DbSchema, tunables: mlos_bench.tunables.tunable_groups.TunableGroups, experiment_id: str, trial_id: int, root_env_config: str, description: str, opt_targets: dict[str, Literal['min', 'max']])[source]

Bases: mlos_bench.storage.base_storage.Storage.Experiment

Logic for retrieving and storing the results of a single experiment.

Parameters:

engine (sqlalchemy.engine.Engine)
schema (mlos_bench.storage.sql.schema.DbSchema)
tunables (mlos_bench.tunables.tunable_groups.TunableGroups)
experiment_id (str)
trial_id (int)
root_env_config (str)
description (str)
opt_targets (dict[str, Literal['min', 'max']])

get_trial_by_id(trial_id: int) → mlos_bench.storage.base_storage.Storage.Trial | None[source]

Gets a Trial by its ID.

Parameters:: trial_id (int) – ID of the Trial to retrieve for this Experiment.
Returns:: trial – The Trial object, or None if it doesn’t exist.
Return type:: Storage.Trial | None

load(last_trial_id: int = -1) → tuple[list[int], list[dict], list[dict[str, Any] | None], list[mlos_bench.environments.status.Status]][source]

Load (tunable values, benchmark scores, status) to warm-up the optimizer.

If last_trial_id is present, load only the data from the (completed) trials that were scheduled after the given trial ID. Otherwise, return data from ALL merged-in experiments and attempt to impute the missing tunable values.

Parameters:: last_trial_id (int) – (Optional) Trial ID to start from.
Returns:: (trial_ids, configs, scores, status) – Trial ids, Tunable values, benchmark scores, and status of the trials.
Return type:: ([int], [dict], [dict] | None, [Status])

load_telemetry(trial_id: int) → list[tuple[datetime.datetime, str, Any]][source]

Retrieve the telemetry data for a given trial.

Parameters:: trial_id (int) – Trial ID.
Returns:: metrics – Telemetry data.
Return type:: list[tuple[datetime.datetime, str, Any]]

load_tunable_config(config_id: int) → dict[str, Any][source]

Load tunable values for a given config ID.

Parameters:: config_id (int)
Return type:: dict[str, Any]

merge(experiment_ids: list[str]) → None[source]

Merge in the results of other (compatible) experiments trials. Used to help warm up the optimizer for this experiment.

Parameters:: experiment_ids (list[str]) – List of IDs of the experiments to merge in.
Return type:: None

pending_trials(timestamp: datetime.datetime, *, running: bool = False, trial_runner_assigned: bool | None = None) → collections.abc.Iterator[mlos_bench.storage.base_storage.Storage.Trial][source]

Return an iterator over PENDING Trial instances that have a scheduled start time to run on or before the specified timestamp.

Parameters:

timestamp (datetime.datetime) – The time in UTC to check for scheduled Trials.
running (bool) – If True, include the Trials that are also RUNNING or READY. Otherwise, return only the scheduled trials.
trial_runner_assigned (bool | None) – If True, include the Trials that are assigned to a TrialRunner. If False, return only the trials that are not assigned to any TrialRunner. If None, return all trials regardless of their assignment.

Returns:

trials – An iterator over the scheduled (and maybe running) trials.

Return type:

Iterator[Storage.Trial]