negmas.checkpoints

Implements Checkpoint functionality for easy dumping and restoration of any NamedObject in negmas.

class negmas.checkpoints.CheckpointMixin

Bases: object

Adds the ability to save checkpoints to a NamedObject

checkpoint_final_step() → Path | None

Should be called at the end of the simulation to save the final state

Remarks:

• Should be called after all processing of the final step is conducted.

checkpoint_init(step_attrib: str = 'current_step', every: int = 1, folder: PathLike | Path | str | None = None, filename: str | None = None, info: dict[str, Any] | None = None, exist_ok: bool = True, single: bool = True)

Initializes the object to automatically save a checkpoint

Parameters:

• step_attrib – The attribute that defines the current step. If None, there is no step concept

• 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

Remarks:

• single_checkpoint implies exist_ok

checkpoint_on_step_started() → Path | None

Should be called on every step to save checkpoints as needed.

Returns:

The path on which the checkpoint is stored if one is stored. None otherwise.

Remarks:

• Should be called at the BEGINNING of every step before any processing takes place

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

property current_step: int

Gets the current step number

property first_step: int

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

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

property last_step: int

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

property loaded_object: NamedObject | None

The object stored in the current checkpoint

property next_step: int | None

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

property previous_step: int | None

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

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.

property steps: list[int]

A list of all stored steps