Merge pull request #394 from lsst/tickets/DM-41606

DM-41606: Option to output pipetask report info to command line

doc/changes/DM-41606.feature.md

@@ -0,0 +1 @@
+Add human-readable option to report summary dictionaries.

python/lsst/pipe/base/execution_reports.py

@@ -197,7 +197,9 @@ def inspect_quantum(
             else:
                 dataset_type_report.n_produced += 1
 
-    def to_summary_dict(self, butler: Butler, do_store_logs: bool = True) -> dict[str, Any]:
+    def to_summary_dict(
+        self, butler: Butler, do_store_logs: bool = True, human_readable: bool = False
+    ) -> dict[str, Any]:
         """Summarize the results of the TaskExecutionReport in a dictionary.
 
         Parameters
@@ -206,6 +208,9 @@ def to_summary_dict(self, butler: Butler, do_store_logs: bool = True) -> dict[st
             The Butler used for this report.
         do_store_logs : `bool`
             Store the logs in the summary dictionary.
+        human_readable : `bool`
+            Store more human-readable information to be printed out to the
+            command-line.
 
         Returns
         -------
@@ -220,10 +225,18 @@ def to_summary_dict(self, butler: Butler, do_store_logs: bool = True) -> dict[st
             - n_quanta_blocked: The number of quanta which failed due to
               upstream failures.
             - n_succeded: The number of quanta which succeeded.
+
+            And possibly, if human-readable is passed:
+
+            - errors: A dictionary of data ids associated with each error
+              message. If `human-readable` and `do_store_logs`, this is stored
+              here. Otherwise, if `do_store_logs`, it is stored in
+              `failed_quanta` keyed by the quantum graph node id.
         """
         failed_quanta = {}
         for node_id, log_ref in self.failed.items():
-            quantum_info: dict[str, Any] = {"data_id": dict(log_ref.dataId.required)}
+            data_ids = dict(log_ref.dataId.required)
+            quantum_info: dict[str, Any] = {"data_id": data_ids}
             if do_store_logs:
                 try:
                     log = butler.get(log_ref)
@@ -235,7 +248,17 @@ def to_summary_dict(self, butler: Butler, do_store_logs: bool = True) -> dict[st
                     quantum_info["error"] = [
                         record.message for record in log if record.levelno >= logging.ERROR
                     ]
-            failed_quanta[str(node_id)] = quantum_info
+            if human_readable:
+                failed_quanta["data_id"] = data_ids
+                return {
+                    "outputs": {name: r.to_summary_dict() for name, r in self.output_datasets.items()},
+                    "failed_quanta": failed_quanta,
+                    "n_quanta_blocked": len(self.blocked),
+                    "n_succeeded": self.n_succeeded,
+                    "errors": quantum_info,
+                }
+            else:
+                failed_quanta[str(node_id)] = quantum_info
         return {
             "outputs": {name: r.to_summary_dict() for name, r in self.output_datasets.items()},
             "failed_quanta": failed_quanta,
@@ -274,7 +297,9 @@ class QuantumGraphExecutionReport:
     tasks: dict[str, TaskExecutionReport] = dataclasses.field(default_factory=dict)
     """A dictionary of TaskExecutionReports by task label (`dict`)."""
 
-    def to_summary_dict(self, butler: Butler, do_store_logs: bool = True) -> dict[str, Any]:
+    def to_summary_dict(
+        self, butler: Butler, do_store_logs: bool = True, human_readable: bool = False
+    ) -> dict[str, Any]:
         """Summarize the results of the `QuantumGraphExecutionReport` in a
         dictionary.
 
@@ -284,6 +309,9 @@ def to_summary_dict(self, butler: Butler, do_store_logs: bool = True) -> dict[st
             The Butler used for this report.
         do_store_logs : `bool`
             Store the logs in the summary dictionary.
+        human_readable : `bool`
+            Store more human-readable information to be printed out to the
+            command-line.
 
         Returns
         -------
@@ -292,7 +320,7 @@ def to_summary_dict(self, butler: Butler, do_store_logs: bool = True) -> dict[st
             each task in the quantum graph.
         """
         return {
-            task: report.to_summary_dict(butler, do_store_logs=do_store_logs)
+            task: report.to_summary_dict(butler, do_store_logs=do_store_logs, human_readable=human_readable)
             for task, report in self.tasks.items()
         }
 

tests/test_execution_reports.py

@@ -47,9 +47,20 @@ def test_make_reports(self) -> None:
             # make a simple qgraph to make an execution report on
             butler, qgraph = simpleQGraph.makeSimpleQGraph(root=root)
             report = QuantumGraphExecutionReport.make_reports(butler, qgraph)
-            dict_of_expected_failures = report.to_summary_dict(butler, do_store_logs=False)
-            self.assertIsNotNone(dict_of_expected_failures["task0"]["failed_quanta"])
-            self.assertEqual(dict_of_expected_failures["task1"]["outputs"]["add_dataset2"]["blocked"], 1)
-            self.assertDictEqual(dict_of_expected_failures["task2"]["failed_quanta"], {})
-            self.assertEqual(dict_of_expected_failures["task3"]["outputs"]["add_dataset4"]["produced"], 0)
-            self.assertEqual(dict_of_expected_failures["task4"]["n_quanta_blocked"], 1)
+            # make a summary dictionary with a certain amount of
+            # expected failures and check that they are there
+            exp_failures = report.to_summary_dict(butler, do_store_logs=False)
+            self.assertIsNotNone(exp_failures["task0"]["failed_quanta"])
+            self.assertEqual(exp_failures["task1"]["outputs"]["add_dataset2"]["blocked"], 1)
+            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)
+
+            # 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)
+            self.assertIsNotNone(hr_exp_failures["task0"]["failed_quanta"])
+            self.assertEqual(hr_exp_failures["task1"]["outputs"]["add_dataset2"]["blocked"], 1)
+            self.assertDictEqual(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)