Skip to content

Commit

Permalink
Store output refs in quantum metadata
Browse files Browse the repository at this point in the history
  • Loading branch information
timj committed Feb 4, 2025
1 parent af99950 commit 6ee031a
Showing 1 changed file with 13 additions and 4 deletions.
17 changes: 13 additions & 4 deletions python/lsst/ctrl/mpexec/singleQuantumExecutor.py
Original file line number Diff line number Diff line change
Expand Up @@ -273,9 +273,9 @@ def _execute(
task = self.taskFactory.makeTask(task_node, limited_butler, init_input_refs)
logInfo(None, "start", metadata=quantumMetadata) # type: ignore[arg-type]
try:
quantumMetadata["caveats"] = self.runQuantum(
caveats, outputsPut = self.runQuantum(
task, quantum, task_node, limited_butler, quantum_id=quantum_id
).value
)
except Exception as e:
_LOG.error(
"Execution of task '%s' on quantum %s failed. Exception %s: %s",
Expand All @@ -285,6 +285,11 @@ def _execute(
str(e),
)
raise
else:
quantumMetadata["caveats"] = caveats.value
# Stringify the UUID for easier compatibility with
# PropertyList.
quantumMetadata["outputs"] = [str(output) for output in outputsPut]
logInfo(None, "end", metadata=quantumMetadata) # type: ignore[arg-type]
fullMetadata = task.getFullMetadata()
fullMetadata["quantum"] = quantumMetadata
Expand Down Expand Up @@ -483,7 +488,7 @@ def runQuantum(
/,
limited_butler: LimitedButler,
quantum_id: uuid.UUID | None = None,
) -> QuantumSuccessCaveats:
) -> tuple[QuantumSuccessCaveats, list[uuid.UUID]]:
"""Execute task on a single quantum.
Parameters
Expand All @@ -503,6 +508,9 @@ def runQuantum(
-------
flags : `QuantumSuccessCaveats`
Flags that describe qualified successes.
ids_put : list[ `uuid.UUID` ]
Record of all the dataset IDs that were written by this quantum
being executed.
"""
flags = QuantumSuccessCaveats.NO_CAVEATS

Expand Down Expand Up @@ -556,7 +564,8 @@ def runQuantum(
flags |= QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
if not butlerQC.outputsPut == butlerQC.allOutputs:
flags |= QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
return flags
ids_put = [output[2] for output in butlerQC.outputsPut]
return flags, ids_put

def writeMetadata(
self, quantum: Quantum, metadata: Any, task_node: TaskNode, /, limited_butler: LimitedButler
Expand Down

0 comments on commit 6ee031a

Please sign in to comment.