InstanceSpace Python v0.4.1

instancespace.stage_runner

A runner to run a list of stages, and a builder to attach extra stages to one.

build_stage_runner() is the entry point: given a caller-supplied, already -ordered base schedule plus any extra/plugin stages to attach via RunBefore/RunAfter, it produces a ready-to-use StageRunner.

StageScheduleElement = list[type[instancespace.stages.stage.Stage[typing.Any, typing.Any]]]
class StageResolutionError(builtins.Exception):

An error attaching an extra stage to a base schedule.

class StageRunningError(builtins.Exception):

An error during stage running.

class AnnotatedStageOutput(typing.NamedTuple):

The yielded output of running a stage.

AnnotatedStageOutput( stage: type[instancespace.stages.stage.Stage[typing.Any, typing.Any]], output: <function NamedTuple>)

Create new instance of AnnotatedStageOutput(stage, output)

stage: type[instancespace.stages.stage.Stage[typing.Any, typing.Any]]

Alias for field number 0

output: typing.NamedTuple

Alias for field number 1

class StageRunner:

A runner to run a list of stages.

def run_iter( self, additional_arguments: <function NamedTuple>) -> Generator[AnnotatedStageOutput, None, dict[str, typing.Any]]:

Run all stages, yielding after every run.

Parameters
  • additional_arguments (NamedTuple): The initial inputs for the first stages.
Yields
  • AnnotatedStageOutput: The stage that just ran and its output.
Returns
  • dict[str, Any]: All the available inputs and outputs after the last stage.
def run_stage( self, stage: type[instancespace.stages.stage.Stage[typing.Any, ~OUT]], **additional_arguments: Any) -> ~OUT:

Run a single stage.

Raises an error if the prerequisite stages did not run.

Parameters
  • stage (type[Stage[Any, OUT]]): The stage to run.
  • **additional_arguments (Any): Inputs for the stage. If an input is missing, the runner gets it from the outputs of the stages that ran before. If the input is still missing, the stage raises an error.
Returns
  • OUT: The output of the stage.
def run_all( self, additional_arguments: <function NamedTuple>) -> dict[str, typing.Any]:

Run all stages from start to finish.

Return the entire outputs data object when finished.

Parameters
  • additional_arguments (NamedTuple): The initial inputs for the first stages.
Returns
  • dict[str, Any]: All the available inputs and outputs after the last stage.
def run_until_stage( self, stop_at_stage: type[instancespace.stages.stage.Stage[typing.Any, typing.Any]], initial_arguments: <function NamedTuple>, **additional_arguments: Any) -> dict[str, typing.Any]:

Run every wave through the wave containing stop_at_stage.

Stages in one wave are unordered siblings. Running only the named sibling would leave the runner in an ambiguous half-wave state, so the complete target wave is always executed.

Parameters
  • stop_at_stage (StageClass): The runner stops after the wave that contains this stage.
  • initial_arguments (NamedTuple): The initial inputs for the first stages.
  • **additional_arguments (Any): More inputs. The runner adds them to the initial inputs.
Returns
  • dict[str, Any]: The available inputs and outputs after the target wave.
def named_tuple_to_stage_arguments( named_tuple: type[typing.NamedTuple]) -> set[instancespace.stage_runner._StageArgument]:

Extract a NamedTuple's fields as a set of _StageArguments.

def build_stage_runner( base_order: list[list[type[instancespace.stages.stage.Stage[typing.Any, typing.Any]]]], extra_stages: list[type[instancespace.stages.stage.Stage[typing.Any, typing.Any]]], initial_input_arguments: type[typing.NamedTuple] | set[instancespace.stage_runner._StageArgument]) -> StageRunner:

Attach extra stages to a caller-supplied base order and build a StageRunner.

base_order is a caller-supplied list[StageScheduleElement] (a list of "waves" - stages within the same wave are order-independent of each other). It is taken as given, not inferred: the caller owns their own pipeline's shape. This lets InstanceSpace hardcode its own known 7-stage order rather than have it re-derived from scratch on every construction.

Each stage in extra_stages must declare where it attaches relative to a stage already in base_order, via a RunBefore[X]/RunAfter[X] field in its _inputs() NamedTuple - it is not placed by matching its input/output types against the rest of the pipeline. It is inserted as its own new wave immediately before/after X's wave; multiple extras resolving to the same attachment point share that new wave.

Example:

runner = build_stage_runner(
    base_order=[[PrelimStage], [SiftedStage]],
    extra_stages=[MyPlugin],  # MyPlugin's inputs declare RunAfter[SiftedStage]
    initial_input_arguments=initial_input_arguments,
)