Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

DM-48536: improvements and testing for pipetask report #322

Merged
merged 4 commits into from
Feb 9, 2025
Merged

DM-48536: improvements and testing for pipetask report #322

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
74b2515
Update to reflect QPG summary change.
TallJimbo Jan 24, 2025
3473648
Rewrite pipetask report brief to delegate to pipe_base.
TallJimbo Jan 24, 2025
03ef1df
Add tests of execution status and exception reporting.
TallJimbo Feb 3, 2025
78a2d0d
Move QPG summary printing to pipe_base.
TallJimbo Feb 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
138 changes: 9 additions & 129 deletions python/lsst/ctrl/mpexec/cli/script/report.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
from astropy.table import Table

from lsst.daf.butler import Butler
from lsst.pipe.base import QuantumGraph, QuantumSuccessCaveats
from lsst.pipe.base import QuantumGraph
from lsst.pipe.base.execution_reports import QuantumGraphExecutionReport
from lsst.pipe.base.quantum_provenance_graph import QuantumProvenanceGraph, Summary

Expand Down Expand Up @@ -235,137 +235,17 @@ def print_summary(summary: Summary, full_output_filename: str | None, brief: boo
Parameters
----------
summary : `QuantumProvenanceGraph.Summary`
This `Pydantic` model contains all the information derived from the
`QuantumProvenanceGraph`.
This `Pydantic` model contains all the information derived from the
`QuantumProvenanceGraph`.
full_output_filename : `str | None`
Name of the JSON file in which to store summary information, if
passed.
Name of the JSON file in which to store summary information, if
passed.
brief : `bool`
Only display short (counts-only) summary on stdout. This includes
counts and not error messages or data_ids (similar to BPS report).
This option will still report all `cursed` datasets and `wonky`
quanta.
Only display short (counts-only) summary on stdout. This includes
counts and not error messages or data_ids (similar to BPS report).
Ignored (considered `False`) if ``full_output_filename`` is passed.
"""
quanta_table = []
failed_quanta_table = []
wonky_quanta_table = []
for label, task_summary in summary.tasks.items():
if task_summary.n_wonky > 0:
print(
f"{label} has produced wonky quanta. Recommend processing cease until the issue is resolved."
)
for quantum_summary in task_summary.wonky_quanta:
wonky_quanta_table.append(
{
"Task": label,
"Data ID": quantum_summary.data_id,
"Runs and Status": quantum_summary.runs,
"Messages": quantum_summary.messages,
}
)
if len(task_summary.caveats) > 1:
caveats = "(multiple)"
elif len(task_summary.caveats) == 1:
((code, data_ids),) = task_summary.caveats.items()
caveats = f"{code}({len(data_ids)})"
else:
caveats = ""
quanta_table.append(
{
"Task": label,
"Unknown": task_summary.n_unknown,
"Successful": task_summary.n_successful,
"Caveats": caveats,
"Blocked": task_summary.n_blocked,
"Failed": task_summary.n_failed,
"Wonky": task_summary.n_wonky,
"TOTAL": sum(
[
task_summary.n_successful,
task_summary.n_unknown,
task_summary.n_blocked,
task_summary.n_failed,
task_summary.n_wonky,
]
),
"EXPECTED": task_summary.n_expected,
}
)
if task_summary.failed_quanta:
for quantum_summary in task_summary.failed_quanta:
failed_quanta_table.append(
{
"Task": label,
"Data ID": quantum_summary.data_id,
"Runs and Status": quantum_summary.runs,
"Messages": quantum_summary.messages,
}
)
quanta = Table(quanta_table)
quanta.pprint_all()
print("")
print("Caveat codes:")
for k, v in QuantumSuccessCaveats.legend().items():
print(f"{k}: {v}")
print("")
# Dataset loop
dataset_table = []
cursed_datasets = []
unsuccessful_datasets = {}
for dataset_type_name, dataset_type_summary in summary.datasets.items():
dataset_table.append(
{
"Dataset": dataset_type_name,
"Visible": dataset_type_summary.n_visible,
"Shadowed": dataset_type_summary.n_shadowed,
"Predicted Only": dataset_type_summary.n_predicted_only,
"Unsuccessful": dataset_type_summary.n_unsuccessful,
"Cursed": dataset_type_summary.n_cursed,
"TOTAL": sum(
[
dataset_type_summary.n_visible,
dataset_type_summary.n_shadowed,
dataset_type_summary.n_predicted_only,
dataset_type_summary.n_unsuccessful,
dataset_type_summary.n_cursed,
]
),
"EXPECTED": dataset_type_summary.n_expected,
}
)
if dataset_type_summary.n_cursed > 0:
for cursed_dataset in dataset_type_summary.cursed_datasets:
print(
f"{dataset_type_name} has cursed quanta with message(s) {cursed_dataset.messages}. "
"Recommend processing cease until the issue is resolved."
)
cursed_datasets.append(
{
"Dataset Type": dataset_type_name,
"Producer Data Id": cursed_dataset.producer_data_id,
}
)
if dataset_type_summary.n_unsuccessful > 0:
unsuccessful_datasets[dataset_type_name] = dataset_type_summary.unsuccessful_datasets
datasets = Table(dataset_table)
datasets.pprint_all()
curse_table = Table(cursed_datasets)
# Display wonky quanta
if wonky_quanta_table:
print("Wonky Quanta")
pprint.pprint(wonky_quanta_table)
# Display cursed datasets
if cursed_datasets:
print("Cursed Datasets")
curse_table.pprint_all()
summary.pprint(brief=(brief or bool(full_output_filename)))
if full_output_filename:
with open(full_output_filename, "w") as stream:
stream.write(summary.model_dump_json(indent=2))
else:
if not brief:
if failed_quanta_table:
print("Failed Quanta")
pprint.pprint(failed_quanta_table)
if unsuccessful_datasets:
print("Unsuccessful Datasets")
pprint.pprint(unsuccessful_datasets)
3 changes: 0 additions & 3 deletions tests/test_cliCmdReport.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,9 +167,6 @@ def test_report(self):
self.assertIn("TOTAL", result_v2_terminal_out.stdout)
self.assertIn("EXPECTED", result_v2_terminal_out.stdout)

# Check that title from the error summary appears
self.assertIn("Unsuccessful Datasets", result_v2_terminal_out.stdout)

# Test cli for the QPG brief option
result_v2_brief = self.runner.invoke(
pipetask_cli,
Expand Down
152 changes: 148 additions & 4 deletions tests/test_simple_pipeline_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -261,15 +261,75 @@ def test_partial_outputs_success(self):
prov.assemble_quantum_provenance_graph(self.butler, [executor.quantum_graph])
(quantum_key_a,) = prov.quanta["a"]
quantum_info_a = prov.get_quantum_info(quantum_key_a)
_, quantum_run_a = qpg.QuantumRun.find_final(quantum_info_a)
self.assertEqual(
quantum_info_a["caveats"],
quantum_run_a.caveats,
QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
| QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
| QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR,
)
self.assertEqual(
quantum_run_a.exception.type_name,
"lsst.pipe.base.tests.mocks.MockAlgorithmError",
)
self.assertEqual(
quantum_run_a.exception.metadata,
{"badness": 12},
)
(quantum_key_b,) = prov.quanta["b"]
quantum_info_b = prov.get_quantum_info(quantum_key_b)
self.assertEqual(quantum_info_b["caveats"], QuantumSuccessCaveats.NO_CAVEATS)
_, quantum_run_b = qpg.QuantumRun.find_final(quantum_info_b)
self.assertEqual(quantum_run_b.caveats, QuantumSuccessCaveats.NO_CAVEATS)
prov_summary = prov.to_summary(self.butler)
# One partial-outputs case, with an empty data ID:
self.assertEqual(prov_summary.tasks["a"].caveats, {"*P": [{}]})
self.assertEqual(
prov_summary.tasks["a"].exceptions.keys(), {"lsst.pipe.base.tests.mocks.MockAlgorithmError"}
)
self.assertEqual(
prov_summary.tasks["a"]
.exceptions["lsst.pipe.base.tests.mocks.MockAlgorithmError"][0]
.exception.metadata,
{"badness": 12},
)
# No caveats for the second task, since it didn't need the first task's
# output anyway.
self.assertEqual(prov_summary.tasks["b"].caveats, {})
self.assertEqual(prov_summary.tasks["b"].exceptions, {})
# Check table forms for summaries of the same information.
quantum_table = prov_summary.make_quantum_table()
self.assertEqual(list(quantum_table["Task"]), ["a", "b"])
self.assertEqual(list(quantum_table["Unknown"]), [0, 0])
self.assertEqual(list(quantum_table["Successful"]), [1, 1])
self.assertEqual(list(quantum_table["Caveats"]), ["*P(1)", ""])
self.assertEqual(list(quantum_table["Blocked"]), [0, 0])
self.assertEqual(list(quantum_table["Failed"]), [0, 0])
self.assertEqual(list(quantum_table["Wonky"]), [0, 0])
self.assertEqual(list(quantum_table["TOTAL"]), [1, 1])
self.assertEqual(list(quantum_table["EXPECTED"]), [1, 1])
dataset_table = prov_summary.make_dataset_table()
self.assertEqual(
list(dataset_table["Dataset"]),
["intermediate", "a_metadata", "a_log", "output", "b_metadata", "b_log"],
)
self.assertEqual(list(dataset_table["Visible"]), [0, 1, 1, 1, 1, 1])
self.assertEqual(list(dataset_table["Shadowed"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["Predicted Only"]), [1, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["Unsuccessful"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["Cursed"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["TOTAL"]), [1, 1, 1, 1, 1, 1])
self.assertEqual(list(dataset_table["EXPECTED"]), [1, 1, 1, 1, 1, 1])
exception_table = prov_summary.make_exception_table()
self.assertEqual(list(exception_table["Task"]), ["a"])
self.assertEqual(
list(exception_table["Exception"]), ["lsst.pipe.base.tests.mocks.MockAlgorithmError"]
)
self.assertEqual(list(exception_table["Count"]), [1])
bad_quantum_tables = prov_summary.make_bad_quantum_tables()
self.assertEqual(bad_quantum_tables.keys(), {"a"})
self.assertEqual(list(bad_quantum_tables["a"]["Status(Caveats)"]), ["SUCCESSFUL(P)"])
self.assertEqual(list(bad_quantum_tables["a"]["Exception"]), ["MockAlgorithmError"])
self.assertFalse(prov_summary.make_bad_dataset_tables())

def test_no_work_found(self):
"""Test executing two quanta where the first raises
Expand Down Expand Up @@ -304,21 +364,56 @@ def test_no_work_found(self):
prov.assemble_quantum_provenance_graph(self.butler, [executor.quantum_graph])
(quantum_key_a,) = prov.quanta["a"]
quantum_info_a = prov.get_quantum_info(quantum_key_a)
_, quantum_run_a = qpg.QuantumRun.find_final(quantum_info_a)
self.assertEqual(
quantum_info_a["caveats"],
quantum_run_a.caveats,
QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
| QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
| QuantumSuccessCaveats.NO_WORK,
)
(quantum_key_b,) = prov.quanta["b"]
quantum_info_b = prov.get_quantum_info(quantum_key_b)
_, quantum_run_b = qpg.QuantumRun.find_final(quantum_info_b)
self.assertEqual(
quantum_info_b["caveats"],
quantum_run_b.caveats,
QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
| QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
| QuantumSuccessCaveats.ADJUST_QUANTUM_RAISED
| QuantumSuccessCaveats.NO_WORK,
)
prov_summary = prov.to_summary(self.butler)
# One NoWorkFound, raised by runQuantum, with an empty data ID:
self.assertEqual(prov_summary.tasks["a"].caveats, {"*N": [{}]})
self.assertEqual(prov_summary.tasks["a"].exceptions, {})
# One NoWorkFound, raised by adjustQuantum, with an empty data ID.
self.assertEqual(prov_summary.tasks["b"].caveats, {"*A": [{}]})
self.assertEqual(prov_summary.tasks["b"].exceptions, {})
# Check table forms for summaries of the same information.
quantum_table = prov_summary.make_quantum_table()
self.assertEqual(list(quantum_table["Task"]), ["a", "b"])
self.assertEqual(list(quantum_table["Unknown"]), [0, 0])
self.assertEqual(list(quantum_table["Successful"]), [1, 1])
self.assertEqual(list(quantum_table["Caveats"]), ["*N(1)", "*A(1)"])
self.assertEqual(list(quantum_table["Blocked"]), [0, 0])
self.assertEqual(list(quantum_table["Failed"]), [0, 0])
self.assertEqual(list(quantum_table["Wonky"]), [0, 0])
self.assertEqual(list(quantum_table["TOTAL"]), [1, 1])
self.assertEqual(list(quantum_table["EXPECTED"]), [1, 1])
dataset_table = prov_summary.make_dataset_table()
self.assertEqual(
list(dataset_table["Dataset"]),
["intermediate", "a_metadata", "a_log", "output", "b_metadata", "b_log"],
)
self.assertEqual(list(dataset_table["Visible"]), [0, 1, 1, 0, 1, 1])
self.assertEqual(list(dataset_table["Shadowed"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["Predicted Only"]), [1, 0, 0, 1, 0, 0])
self.assertEqual(list(dataset_table["Unsuccessful"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["Cursed"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["TOTAL"]), [1, 1, 1, 1, 1, 1])
self.assertEqual(list(dataset_table["EXPECTED"]), [1, 1, 1, 1, 1, 1])
self.assertFalse(prov_summary.make_exception_table())
self.assertFalse(prov_summary.make_bad_quantum_tables())
self.assertFalse(prov_summary.make_bad_dataset_tables())

def test_partial_outputs_failure(self):
"""Test executing two quanta where the first raises
Expand Down Expand Up @@ -357,6 +452,55 @@ def test_partial_outputs_failure(self):
executor.run(register_dataset_types=True)
self.assertFalse(self.butler.exists("intermediate"))
self.assertFalse(self.butler.exists("output"))
prov = qpg.QuantumProvenanceGraph()
prov.assemble_quantum_provenance_graph(self.butler, [executor.quantum_graph])
(quantum_key_a,) = prov.quanta["a"]
quantum_info_a = prov.get_quantum_info(quantum_key_a)
_, quantum_run_a = qpg.QuantumRun.find_final(quantum_info_a)
self.assertEqual(quantum_run_a.status, qpg.QuantumRunStatus.FAILED)
self.assertIsNone(quantum_run_a.caveats)
self.assertIsNone(quantum_run_a.exception)
(quantum_key_b,) = prov.quanta["b"]
quantum_info_b = prov.get_quantum_info(quantum_key_b)
self.assertEqual(quantum_info_b["status"], qpg.QuantumInfoStatus.BLOCKED)
prov_summary = prov.to_summary(self.butler)
# One partial-outputs failure case for the first task.
self.assertEqual(prov_summary.tasks["a"].n_failed, 1)
# No direct failures, but one blocked for the second
self.assertEqual(prov_summary.tasks["b"].n_failed, 0)
self.assertEqual(prov_summary.tasks["b"].n_blocked, 1)
# Check table forms for summaries of the same information.
quantum_table = prov_summary.make_quantum_table()
self.assertEqual(list(quantum_table["Task"]), ["a", "b"])
self.assertEqual(list(quantum_table["Unknown"]), [0, 0])
self.assertEqual(list(quantum_table["Successful"]), [0, 0])
self.assertEqual(list(quantum_table["Caveats"]), ["", ""])
self.assertEqual(list(quantum_table["Blocked"]), [0, 1])
self.assertEqual(list(quantum_table["Failed"]), [1, 0])
self.assertEqual(list(quantum_table["Wonky"]), [0, 0])
self.assertEqual(list(quantum_table["TOTAL"]), [1, 1])
self.assertEqual(list(quantum_table["EXPECTED"]), [1, 1])
dataset_table = prov_summary.make_dataset_table()
self.assertEqual(
list(dataset_table["Dataset"]),
["intermediate", "a_metadata", "a_log", "output", "b_metadata", "b_log"],
)
# Note that a_log is UNSUCCESSFUL, not VISIBLE, despite being present
# in butler because those categories are mutually exclusive and we
# don't want to consider any outputs of failed quanta to be successful.
self.assertEqual(list(dataset_table["Visible"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["Shadowed"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["Predicted Only"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["Unsuccessful"]), [1, 1, 1, 1, 1, 1])
self.assertEqual(list(dataset_table["Cursed"]), [0, 0, 0, 0, 0, 0])
self.assertEqual(list(dataset_table["TOTAL"]), [1, 1, 1, 1, 1, 1])
self.assertEqual(list(dataset_table["EXPECTED"]), [1, 1, 1, 1, 1, 1])
self.assertFalse(prov_summary.make_exception_table())
bad_quantum_tables = prov_summary.make_bad_quantum_tables()
self.assertEqual(bad_quantum_tables.keys(), {"a"})
self.assertEqual(list(bad_quantum_tables["a"]["Status(Caveats)"]), ["FAILED"])
self.assertEqual(list(bad_quantum_tables["a"]["Exception"]), [""])
self.assertFalse(prov_summary.make_bad_dataset_tables())

def test_existence_check_skips(self):
"""Test that pre-execution existence checks are not performed for
Expand Down
Loading