CheckpointRunner

class negmas.checkpoints.CheckpointRunner(folder: str | ~pathlib.Path, id: str | None = None, callback: ~typing.Callable[[~negmas.types.named.NamedObject, int], None] | None = None, watch: bool = False, object_type: type[~negmas.types.named.NamedObject] = <class 'negmas.types.named.NamedObject'>)

Bases: object

Runs an object based on its checkpoints saved in an earlier run

Attributes Summary

current_step	Gets the current step number
first_step	Get the first stored step number (None if it does not exist)
last_step	Get the last stored step number (None if it does not exist)
loaded_object	The object stored in the current checkpoint
next_step	Get the next stored step number (None if it does not exist)
previous_step	Get the previous stored step number (None if it does not exist)
steps	A list of all stored steps

Methods Summary

fork([copy_past_checkpoints, every, folder, ...])	Creates a copy of the internal object that can be run safely.
goto(step[, exact])	Goes to the nearest step for the given one returning the exact step number.
register_callback(callback)	Registers a callback to be called whenever a new step is loaded
reset()	Goes before the first step
run()	Run all steps.
step()	Go one step forward in the stored steps.

Attributes Documentation

current_step

Gets the current step number

first_step

Get the first stored step number (None if it does not exist)

last_step

Get the last stored step number (None if it does not exist)

loaded_object

The object stored in the current checkpoint

next_step

Get the next stored step number (None if it does not exist)

previous_step

Get the previous stored step number (None if it does not exist)

steps

A list of all stored steps

Methods Documentation

fork(copy_past_checkpoints: bool = False, every: int = 1, folder: str | Path | None = None, filename: str | None = None, info: dict[str, Any] | None = None, exist_ok: bool = True, single: bool = True) → NamedObject | None

Creates a copy of the internal object that can be run safely.

Parameters:

• copy_past_checkpoints – If true, all checkpoints upto and including current_step will be copied to the given folder

• every – Number of steps per checkpoint. If < 1 no checkpoints will be saved

• folder – The directory to store checkpoints under

• filename – Name of the file to save the checkpoint under. If None, a unique name will be chosen. If single_checkpoint was False, then multiple files will be used prefixed with the step number

• info – Any extra information to save in the json file associated with each checkpoint

• exist_ok – Override existing files if any

• single – If True, only the most recent checkpoint will be kept

Returns:

goto(step: int, exact=False) → int | None

Goes to the nearest step for the given one returning the exact step number.

Parameters:

• step – The step we want to goto

• exact – If True, must go to the exact step number, otherwise go to the nearest step stored in a checkpoint

Returns:

• None if the current step is the nearest to the given step. Otherwise the exact step we moved to

register_callback(callback: Callable[[NamedObject, int], None]) → None

Registers a callback to be called whenever a new step is loaded

Parameters:

callback – A callable that takes the named object (after it is loaded and an integer specifying the step number and returns None.

reset() → None

Goes before the first step

run()

Run all steps. Notice that if register_callback was used to register some callback functions, they will be called for every stored stepped during the run.

step() → int | None

Go one step forward in the stored steps.

Returns:

The number of the current step or None if we are already on the last step.