CheckpointRunner

class negmas.checkpoints.CheckpointRunner(folder: Union[str, pathlib.Path], id: Optional[str] = None, callback: Optional[Callable[[negmas.common.NamedObject, int], None]] = None, watch: bool = False, object_type: Type[negmas.common.NamedObject] = <class 'negmas.common.NamedObject'>)[source]

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: Optional[Union[str, pathlib.Path]] = None, filename: Optional[str] = None, info: Optional[Dict[str, Any]] = None, exist_ok: bool = True, single: bool = True)Optional[negmas.common.NamedObject][source]

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)Optional[negmas.checkpoints.CheckpointRunner.step][source]

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[[negmas.common.NamedObject, int], None])None[source]

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[source]

Goes before the first step

run()[source]

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()Optional[int][source]

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.