instancespace.progress_reporter

Progress reporting for instance space analysis.

Reports build()-time stage progress to external systems via HTTP callbacks or file output. This is what lets a pipeline that is split across multiple processes (e.g. one SLURM job per stage, triggered by a scheduler rather than a single long-lived Python process looping over every stage) tell an orchestrator what happened without that orchestrator polling for it.

class OutputDetail(builtins.str, enum.Enum):

Level of detail to include in progress reports.

NONE = <OutputDetail.NONE: 'none'>

METADATA = <OutputDetail.METADATA: 'metadata'>

FULL = <OutputDetail.FULL: 'full'>

class StageStatus(builtins.str, enum.Enum):

Status of a pipeline stage.

PENDING = <StageStatus.PENDING: 'pending'>

RUNNING = <StageStatus.RUNNING: 'running'>

COMPLETED = <StageStatus.COMPLETED: 'completed'>

FAILED = <StageStatus.FAILED: 'failed'>

def serialize_stage_output( instance_space: Any, stage_name: str, detail_level: OutputDetail = <OutputDetail.METADATA: 'metadata'>) -> dict[str, typing.Any]:

Serialize stage output for inclusion in progress reports.

Parameters

instance_space (Any): The InstanceSpace object after stage completion.
stage_name (str): Name of the completed stage.
detail_level (OutputDetail): How much detail to include.

Returns

dict[str, Any]: Dictionary with stage output information.

class ProgressReporter(abc.ABC):

Abstract base class for progress reporters.

Progress reporters are responsible for communicating stage execution status to external systems. They can include varying levels of detail about stage outputs in their reports.

output_detail: OutputDetail = <OutputDetail.METADATA: 'metadata'>

@abstractmethod

def report_stage_started(self, stage_name: str) -> None:

Report that a stage has started.

@abstractmethod

def report_stage_completed( self, stage_name: str, duration_seconds: float | None = None, instance_space: Any = None) -> None:

Report that a stage has completed successfully.

Parameters

stage_name (str): Name of the completed stage.
duration_seconds (float | None): How long the stage took.
instance_space (Any): The InstanceSpace object. The reporter uses it to save the intermediate state and to add the output data to the report.

@abstractmethod

def report_stage_failed(self, stage_name: str, error: str) -> None:

Report that a stage has failed.

@abstractmethod

def report_job_completed(self, instance_space: Any = None) -> None:

Report that the entire job has completed.

Parameters

instance_space (Any): The final InstanceSpace object. The reporter uses it to add the final output data to the report.

@abstractmethod

def report_job_failed(self, error: str) -> None:

Report that the job has failed.

class HttpProgressReporter(ProgressReporter):

Reports progress via HTTP POST to a callback URL.

Example callback URL: http://backend:8081/internal/jobs/123/stage-callback

The reporter can include different levels of detail in its callbacks:

NONE: Just status updates
METADATA: Status + output type/size information
FULL: Status + metadata + base64-encoded pickle of InstanceSpace

HttpProgressReporter( callback_url: str, job_id: int, auth_token: str | None = None, timeout_seconds: int = 10, output_detail: OutputDetail = <OutputDetail.METADATA: 'metadata'>, include_pickle_on_completion: bool = False)

Initialize the HTTP progress reporter.

Parameters

callback_url (str): URL to POST progress updates to.
job_id (int): Job ID for the callback payload.
auth_token (str | None): Optional bearer token for authentication.
timeout_seconds (int): HTTP request timeout.
output_detail (OutputDetail): Level of detail to include for stage outputs.
include_pickle_on_completion (bool): If True, the final callback includes the full pickle. For the final callback, this setting overrides output_detail.

callback_url

job_id

auth_token

timeout_seconds

output_detail = <OutputDetail.METADATA: 'metadata'>

include_pickle_on_completion

def report_stage_started(self, stage_name: str) -> None:

Report that a stage has started.

def report_stage_completed( self, stage_name: str, duration_seconds: float | None = None, instance_space: Any = None) -> None:

Report that a stage has completed successfully.

def report_stage_failed(self, stage_name: str, error: str) -> None:

Report that a stage has failed.

def report_job_completed(self, instance_space: Any = None) -> None:

Report that the entire job has completed.

def report_job_failed(self, error: str) -> None:

Report that the job has failed.

class FileProgressReporter(ProgressReporter):

Reports progress by writing to a JSON file.

Useful as a fallback when HTTP callbacks aren't available, or for debugging/testing. If stages_dir is given, this also writes an unsigned pickle snapshot of the InstanceSpace after every stage - a convenient debugging artifact, but not the supported checkpoint/resume mechanism (see InstanceSpace.save()/InstanceSpace.load(), which sign the file the same way Model.save()/Model.load() do).

FileProgressReporter(progress_file: pathlib.Path, stages_dir: pathlib.Path | None = None)

Initialize the file progress reporter.

Parameters

progress_file (Path): Path to the progress.json file.
stages_dir (Path | None): Optional directory to save intermediate stage pickles.

progress_file

stages_dir

def report_stage_started(self, stage_name: str) -> None:

Report that a stage has started.

def report_stage_completed( self, stage_name: str, duration_seconds: float | None = None, instance_space: Any = None) -> None:

Report that a stage has completed successfully.

def report_stage_failed(self, stage_name: str, error: str) -> None:

Report that a stage has failed.

def report_job_completed(self, instance_space: Any = None) -> None:

Report that the entire job has completed.

def report_job_failed(self, error: str) -> None:

Report that the job has failed.

Inherited Members

ProgressReporter: output_detail

class CompositeProgressReporter(ProgressReporter):

Combines multiple progress reporters.

Useful for reporting to both HTTP and file simultaneously.

CompositeProgressReporter(reporters: list[ProgressReporter])

Initialize with a list of reporters.

reporters

def report_stage_started(self, stage_name: str) -> None:

Report that a stage has started, to every configured reporter.

def report_stage_completed( self, stage_name: str, duration_seconds: float | None = None, instance_space: Any = None) -> None:

Report that a stage has completed, to every configured reporter.

def report_stage_failed(self, stage_name: str, error: str) -> None:

Report that a stage has failed, to every configured reporter.

def report_job_completed(self, instance_space: Any = None) -> None:

Report that the job has completed, to every configured reporter.

def report_job_failed(self, error: str) -> None:

Report that the job has failed, to every configured reporter.

Inherited Members

ProgressReporter: output_detail

class NullProgressReporter(ProgressReporter):

A no-op reporter for when progress reporting is disabled (the default).

def report_stage_started(self, stage_name: str) -> None:

Do nothing.

def report_stage_completed( self, stage_name: str, duration_seconds: float | None = None, instance_space: Any = None) -> None:

Do nothing.

def report_stage_failed(self, stage_name: str, error: str) -> None:

Do nothing.

def report_job_completed(self, instance_space: Any = None) -> None:

Do nothing.

def report_job_failed(self, error: str) -> None:

Do nothing.

Inherited Members

ProgressReporter: output_detail