checkpoint

Pipeline checkpoint and resume system for VariantCentrifuge.

This module provides a robust checkpoint system that tracks pipeline execution state, allowing resumption from the last successful step in case of interruption or failure.

class variantcentrifuge.checkpoint.AtomicFileOperation(final_path, suffix=None)[source]

Bases: object

Context manager for atomic file operations with automatic temp file cleanup.

Ensures that output files are only visible in their final location after being completely written, using unique sibling temporary files and atomic os.replace.

Parameters:
__init__(final_path, suffix=None)[source]
Parameters:
class variantcentrifuge.checkpoint.CheckpointContext(pipeline_state, step_name, command_hash=None, parameters=None)[source]

Bases: object

Context manager for checkpoint operations within a code block.

Parameters:
__init__(pipeline_state, step_name, command_hash=None, parameters=None)[source]
Parameters:
add_input_file(filepath)[source]
Parameters:

filepath (str)

Return type:

None

add_output_file(filepath)[source]
Parameters:

filepath (str)

Return type:

None

class variantcentrifuge.checkpoint.FileInfo(path, size, mtime, checksum=None)[source]

Bases: object

Information about a file for validation.

Parameters:
path: str
size: int
mtime: float
checksum: str | None = None
classmethod from_file(filepath, calculate_checksum=False)[source]

Create FileInfo from an existing file.

Parameters:

  • filepath (str)

  • calculate_checksum (bool)

Return type:

FileInfo

validate(calculate_checksum=False)[source]

Validate that the file matches the stored info.

Parameters:

calculate_checksum (bool)

Return type:

bool

__init__(path, size, mtime, checksum=None)
Parameters:
class variantcentrifuge.checkpoint.PipelineState(output_dir, enable_checksum=False)[source]

Bases: object

Manages pipeline execution state for checkpoint/resume functionality.

Parameters:

  • output_dir (str)

  • enable_checksum (bool)

STATE_FILE_NAME = '.variantcentrifuge_state.json'
STATE_VERSION = '1.0'
__init__(output_dir, enable_checksum=False)[source]
Parameters:

  • output_dir (str)

  • enable_checksum (bool)

initialize(configuration, pipeline_version)[source]

Initialize a new pipeline run.

Parameters:
Return type:

None

load()[source]

Load existing state from file.

Return type:

bool

save()[source]

Save current state to file atomically (thread-safe).

Return type:

None

can_resume(configuration, pipeline_version, validate_outputs=True)[source]

Check if pipeline can be resumed with given configuration.

Parameters:
Return type:

bool

get_resume_point()[source]

Get the name of the last successfully completed step.

Return type:

str | None

should_skip_step(step_name)[source]

Check if a step should be skipped (already completed with valid outputs).

Parameters:

step_name (str)

Return type:

bool

start_step(step_name, command_hash=None, parameters=None)[source]

Mark a step as started.

Parameters:
Return type:

None

complete_step(step_name, input_files=None, output_files=None)[source]

Mark a step as completed.

Parameters:
Return type:

None

finalize_step(step_name)[source]

Mark a step as in finalizing phase.

Parameters:

step_name (str)

Return type:

None

fail_step(step_name, error)[source]

Mark a step as failed.

Parameters:
Return type:

None

clear_step_completion(step_name, save=True)[source]

Clear completion status for a step to force re-execution.

Parameters:
Return type:

None

get_summary()[source]

Get a human-readable summary of the pipeline state.

Return type:

str

get_completed_stages()[source]

Get all completed stages sorted by completion time.

Return type:

list[tuple]

get_available_resume_points()[source]

Get stage names that can be used as resume points.

Return type:

list[str]

validate_resume_from_stage(stage_name, available_stages)[source]

Validate that resuming from specific stage is possible.

Parameters:
Return type:

tuple

get_stages_to_execute(resume_from, all_stages)[source]

Get ordered list of stages to execute when resuming from specific stage.

Parameters:
Return type:

list[str]

get_detailed_status()[source]

Get detailed status information for enhanced display.

Return type:

dict[str, Any]

can_resume_from_stage(stage_name, available_stages)[source]

Check if resuming from a specific stage is possible.

Parameters:
Return type:

bool

cleanup_stale_stages()[source]

Clean up stages that were left in ‘running’ state from previous pipeline runs.

Return type:

None

class variantcentrifuge.checkpoint.StepInfo(name, status, start_time=None, end_time=None, command_hash=None, input_files=list(), output_files=list(), parameters=dict(), error=None)[source]

Bases: object

Information about a pipeline step.

Parameters:
name: str
status: str
start_time: float | None = None
end_time: float | None = None
command_hash: str | None = None
input_files: list[FileInfo]
output_files: list[FileInfo]
parameters: dict[str, Any]
error: str | None = None
property duration: float | None

Calculate step duration in seconds.

to_dict()[source]

Convert to dictionary for JSON serialization.

Return type:

dict[str, Any]

__init__(name, status, start_time=None, end_time=None, command_hash=None, input_files=<factory>, output_files=<factory>, parameters=<factory>, error=None)
Parameters:
classmethod from_dict(data)[source]

Create StepInfo from dictionary.

Parameters:

data (dict[str, Any])

Return type:

StepInfo

variantcentrifuge.checkpoint.checkpoint(step_name, input_files=None, output_files=None, parameters=None)[source]

Add checkpoint functionality to pipeline steps.

Parameters:
Return type:

Callable[..., Any]

variantcentrifuge.checkpoint.format_state_summary(state, state_file)[source]

Format a human-readable summary of the pipeline state.

Parameters:
Return type:

str

variantcentrifuge.checkpoint.hash_configuration(config)[source]

Create a hash of the configuration for change detection.

Parameters:

config (dict[str, Any])

Return type:

str

variantcentrifuge.checkpoint.hash_legacy_configuration(config)[source]

Create the pre-normalization checkpoint hash for compatibility.

Parameters:

config (dict[str, Any])

Return type:

str

variantcentrifuge.checkpoint.legacy_annotation_equivalent_configs(config)[source]

Build legacy-key configs that represent the normalized annotation behavior.

Parameters:

config (dict[str, Any])

Return type:

list[dict[str, Any]]

variantcentrifuge.checkpoint.resume_configuration_hashes(config)[source]

Return current and legacy-annotation-key-equivalent config hashes accepted for resume.

Parameters:

config (dict[str, Any])

Return type:

set[str]

Overview

The checkpoint module provides pipeline state tracking and resume functionality for VariantCentrifuge. It enables robust handling of long-running analyses by saving progress at key pipeline steps and allowing resumption after interruptions.

Key Classes

PipelineState

The main class for managing pipeline checkpoints. It tracks:

  • Pipeline steps and their completion status

  • Input/output files for each step

  • Configuration hash to ensure consistency

  • Pipeline version for compatibility checks

CheckpointContext

A context manager that automatically tracks step execution:

  • Records start and end times

  • Captures input and output files

  • Handles errors gracefully

  • Saves state on successful completion

FileInfo

Tracks file metadata for validation:

  • File path, size, and modification time

  • Optional checksum calculation

  • Validation to ensure files haven’t changed

StepInfo

Stores information about each pipeline step:

  • Step name and status (pending, running, completed, failed)

  • Execution times

  • Input/output files

  • Step-specific parameters

  • Error information if failed

Usage Example

from variantcentrifuge.checkpoint import PipelineState, CheckpointContext

# Initialize pipeline state
pipeline_state = PipelineState(output_dir, enable_checksum=False)
pipeline_state.initialize(config, pipeline_version)

# Check if we can resume from a previous run
if pipeline_state.load() and pipeline_state.can_resume(config, pipeline_version):
    logger.info("Resuming from checkpoint")

# Use checkpoint context for a pipeline step
if not pipeline_state.should_skip_step("variant_extraction"):
    with CheckpointContext(pipeline_state, "variant_extraction") as ctx:
        # Add input files
        ctx.add_input_file(input_vcf)

        # Perform the step
        output_file = extract_variants(input_vcf, output_path)

        # Add output files
        ctx.add_output_file(output_file)

        # State is automatically saved on successful completion

Parallel Processing Support

The checkpoint system is designed to work with parallel processing:

  • Thread-safe state updates

  • Tracks individual chunks in parallel runs

  • Proper ordering of parallel step completion

  • Handles worker failures gracefully

State File Format

The checkpoint state is stored as JSON in .variantcentrifuge_state.json:

{
  "version": "1.0",
  "pipeline_version": "0.5.0",
  "start_time": 1234567890.0,
  "configuration_hash": "abc123...",
  "steps": {
    "gene_bed_creation": {
      "name": "gene_bed_creation",
      "status": "completed",
      "start_time": 1234567890.0,
      "end_time": 1234567891.0,
      "input_files": [],
      "output_files": [
        {
          "path": "output/genes.bed",
          "size": 12345,
          "mtime": 1234567891.0,
          "checksum": null
        }
      ],
      "parameters": {},
      "error": null
    }
  }
}