langtest.utils.checkpoints.CheckpointManager#

class CheckpointManager(checkpoint_folder='checkpoints')#

Bases: object

__init__(checkpoint_folder='checkpoints')#

Initialize the CheckpointManager.

Parameters:

checkpoint_folder (str) – The directory to store checkpoints and batch information.

Methods

__init__([checkpoint_folder])	Initialize the CheckpointManager.
load_batches()	Load all remaining batches.
load_checkpoint()	Load complete checkpoints and concatenate results.
load_remaining_batch()	Load remaining batch checkpoints and concatenate results.
save_all_batches(batches)	Save all batches for potential recovery.
save_checkpoint(check_point_extension, ...)	Save a checkpoint with partial results.
update_status(batch_number)	Update the status by removing the checkpoint file associated with a specific batch number.

load_batches() → dict#

Load all remaining batches.

Returns:

Dictionary containing batch numbers as keys and batches as values.

Return type:

dict

load_checkpoint() → List[Sample]#

Load complete checkpoints and concatenate results.

Returns:

Concatenated list of results from all checkpoints.

Return type:

List[Sample]

load_remaining_batch() → List[Sample]#

Load remaining batch checkpoints and concatenate results.

Returns:

Concatenated list of results from all remaining batch checkpoints.

Return type:

List[Sample]

save_all_batches(batches: dict)#

Save all batches for potential recovery.

Parameters:

batches – Dictionary containing batches with batch numbers as keys and batches as values.

save_checkpoint(check_point_extension: str, results_so_far: List[Sample])#

Save a checkpoint with partial results.

Parameters:

• check_point_extension (str) – Extension or identifier for the checkpoint.

• results_so_far (List[Sample]) – Partial results to be saved.

update_status(batch_number: int)#

Update the status by removing the checkpoint file associated with a specific batch number.

Parameters:

batch_number (int) – The batch number to update the status for.