Merge pull request #424 from lsst/tickets/DM-44368

DM-44368: Include number of expected instances in pipetask report task-level summary

doc/changes/DM-44368.feature.md

@@ -0,0 +1 @@
+Include number of expected instances in `pipetask report` task-level summary for the `QuantumGraphExecutionReport`.

python/lsst/pipe/base/execution_reports.py

@@ -68,6 +68,10 @@ class DatasetTypeExecutionReport:
     """Count of datasets produced (`int`).
     """
 
+    n_expected: int = 0
+    """Count of datasets expected (`int`)
+    """
+
     def to_summary_dict(self) -> dict[str, Any]:
         r"""Summarize the DatasetTypeExecutionReport in a dictionary.
 
@@ -84,6 +88,7 @@ def to_summary_dict(self) -> dict[str, Any]:
             "failed": len(self.failed),
             "not_produced": len(self.not_produced),
             "blocked": len(self.blocked),
+            "expected": self.n_expected,
         }
 
 
@@ -113,6 +118,10 @@ class TaskExecutionReport:
     `NoWorkFound`.
     """
 
+    n_expected: int = 0
+    """A count of expected quanta.
+    """
+
     blocked: dict[uuid.UUID, DataCoordinate] = dataclasses.field(default_factory=dict)
     """A mapping of data IDs of quanta that were not attempted due to an
     upstream failure (`dict`).
@@ -197,6 +206,7 @@ def inspect_quantum(
                     dataset_type_report.not_produced.add(output_ref)
             else:
                 dataset_type_report.n_produced += 1
+            dataset_type_report.n_expected += 1
 
     def to_summary_dict(
         self, butler: Butler, do_store_logs: bool = True, human_readable: bool = False
@@ -262,6 +272,7 @@ def to_summary_dict(
             "outputs": {name: r.to_summary_dict() for name, r in self.output_datasets.items()},
             "n_quanta_blocked": len(self.blocked),
             "n_succeeded": self.n_succeeded,
+            "n_expected": self.n_expected,
         }
         if human_readable:
             result["failed_quanta"] = failed_data_ids
@@ -399,7 +410,6 @@ def make_reports(
                     status_graph.add_edge(quantum_id, ref.id)
                 for ref in itertools.chain.from_iterable(quantum.inputs.values()):
                     status_graph.add_edge(ref.id, quantum_id)
-
         for task_node in qg.pipeline_graph.tasks.values():
             task_report = TaskExecutionReport()
             if task_node.log_output is None:
@@ -413,6 +423,7 @@ def make_reports(
                     metadata_name=task_node.metadata_output.dataset_type_name,
                     log_name=task_node.log_output.dataset_type_name,
                 )
+                task_report.n_expected = len(qg.get_task_quanta(task_node.label).items())
             report.tasks[task_node.label] = task_report
         return report
 

tests/test_execution_reports.py

@@ -55,7 +55,7 @@ def test_make_reports(self) -> None:
             self.assertDictEqual(exp_failures["task2"]["failed_quanta"], {})
             self.assertEqual(exp_failures["task3"]["outputs"]["add_dataset4"]["produced"], 0)
             self.assertEqual(exp_failures["task4"]["n_quanta_blocked"], 1)
-
+            self.assertIsNotNone(exp_failures["task0"]["n_expected"])
             # now we'll make a human-readable summary dict and
             # repeat the tests:
             hr_exp_failures = report.to_summary_dict(butler, do_store_logs=False, human_readable=True)
@@ -64,3 +64,4 @@ def test_make_reports(self) -> None:
             self.assertListEqual(hr_exp_failures["task2"]["failed_quanta"], [])
             self.assertEqual(hr_exp_failures["task3"]["outputs"]["add_dataset4"]["produced"], 0)
             self.assertEqual(hr_exp_failures["task4"]["n_quanta_blocked"], 1)
+            self.assertIsNotNone(hr_exp_failures["task0"]["n_expected"])