SingleQuantumExecutor

class lsst.ctrl.mpexec.SingleQuantumExecutor(butler: Butler | None, taskFactory: TaskFactory, skipExistingIn: list[str] | None = None, clobberOutputs: bool = False, enableLsstDebug: bool = False, exitOnKnownError: bool = False, mock: bool = False, mock_configs: list[lsst.ctrl.mpexec.cli.utils._PipelineAction] | None = None, limited_butler_factory: Callable[[Quantum], LimitedButler] | None = None)

Bases: QuantumExecutor

Executor class which runs one Quantum at a time.

Parameters:
butlerButler or None

Data butler, None means that Quantum-backed butler should be used instead.

taskFactoryTaskFactory

Instance of a task factory.

skipExistingInlist [ str ], optional

Accepts list of collections, if all Quantum outputs already exist in the specified list of collections then that Quantum will not be rerun. If None then butler output RUN collection is searched for existing datasets. If empty list then there no check for existing outputs (which could result in conflicts when datasets are stored).

clobberOutputsbool, optional

If True, then existing qauntum outputs in output run collection will be removed prior to executing a quantum. If skipExistingIn is defined, only partial outputs from failed quanta will be overwritten (see notes). Only used when butler is not None.

enableLsstDebugbool, optional

Enable debugging with lsstDebug facility for a task.

exitOnKnownErrorbool, optional

If True, call sys.exit with the appropriate exit code for special known exceptions, after printing a traceback, instead of letting the exception propagate up to calling. This is always the behavior for InvalidQuantumError.

mockbool, optional

If True then mock task execution.

mock_configslist [ _PipelineAction ], optional

Optional config overrides for mock tasks.

limited_butler_factoryCallable, optional

A method that creates a LimitedButler instance for a given Quantum. This parameter must be defined if butler is None. If butler is not None then this parameter is ignored.

Notes

There is a non-trivial interaction between skipExistingIn and clobberOutputs areguments. Here is how they work together:

  • If skipExistingIn is specified (or None) then those collections are searched for quantum output datasets. If all outputs are found, then quantum is not executed and run completes successfully.

  • Otherwise if clobberOutputs is True then butler output RUN collection is checked for existing quantum outputs. If full or partial outputs are found, they are are pruned and quantum is executed.

  • Otherwise if clobberOutputs is False then butler output RUN collection is checked for existing quantum outputs. If any output dataset is found an exception is raised.

This leaves the case when partial quantum outputs may be found in skipExistingIn but that list does not include butler RUN collection. Those partial outputs are not prunned.

Methods Summary

checkExistingOutputs(quantum, taskDef, ...)

Decide whether this quantum needs to be executed.

execute(taskDef, quantum)

Execute single quantum.

getReport()

Return execution report from last call to execute.

initGlobals(quantum)

Initialize global state needed for task execution.

runQuantum(task, quantum, taskDef, ...)

Execute task on a single quantum.

updatedQuantumInputs(quantum, taskDef, ...)

Update quantum with extra information, returns a new updated Quantum.

writeMetadata(quantum, metadata, taskDef, ...)

Methods Documentation

checkExistingOutputs(quantum: Quantum, taskDef: TaskDef, limited_butler: LimitedButler) → bool

Decide whether this quantum needs to be executed.

If only partial outputs exist then they are removed if clobberOutputs is True, otherwise an exception is raised.

Parameters:
quantumQuantum

Quantum to check for existing outputs

taskDefTaskDef

Task definition structure.

Returns:
existbool

True if self.skipExistingIn is defined, and a previous execution of this quanta appears to have completed successfully (either because metadata was written or all datasets were written). False otherwise.

Raises:
RuntimeError

Raised if some outputs exist and some not.

execute(taskDef: TaskDef, quantum: Quantum) → Quantum

Execute single quantum.

Parameters:
taskDefTaskDef

Task definition structure.

quantumQuantum

Quantum for this execution.

Returns:
quantumQuantum

The quantum actually executed. At present this quantum will contain only unresolved DatasetRef instances for output datasets, reflecting the state of the quantum just before it was run (but after any adjustments for predicted but now missing inputs). This may change in the future to include resolved output DatasetRef objects.

Notes

Any exception raised by the task or code that wraps task execution is propagated to the caller of this method.

getReport() → QuantumReport | None

Return execution report from last call to execute.

Returns:
reportQuantumReport

Structure describing the status of the execution of a quantum. None is returned if implementation does not support this feature.

Raises:
RuntimeError

Raised if this method is called before execute.

initGlobals(quantum: Quantum) → None

Initialize global state needed for task execution.

Parameters:
quantumQuantum

Single Quantum instance.

Notes

There is an issue with initializing filters singleton which is done by instrument, to avoid requiring tasks to do it in runQuantum() we do it here when any dataId has an instrument dimension. Also for now we only allow single instrument, verify that all instrument names in all dataIds are identical.

This will need revision when filter singleton disappears.

runQuantum(task: PipelineTask, quantum: Quantum, taskDef: TaskDef, limited_butler: LimitedButler) → None

Execute task on a single quantum.

Parameters:
taskPipelineTask

Task object.

quantumQuantum

Single Quantum instance.

taskDefTaskDef

Task definition structure.

updatedQuantumInputs(quantum: Quantum, taskDef: TaskDef, limited_butler: LimitedButler) → Quantum

Update quantum with extra information, returns a new updated Quantum.

Some methods may require input DatasetRefs to have non-None dataset_id, but in case of intermediate dataset it may not be filled during QuantumGraph construction. This method will retrieve missing info from registry.

Parameters:
quantumQuantum

Single Quantum instance.

taskDefTaskDef

Task definition structure.

Returns:
updateQuantum

Updated Quantum instance

writeMetadata(quantum: Quantum, metadata: Any, taskDef: TaskDef, limited_butler: LimitedButler) → None