Coverage for python / lsst / ctrl / mpexec / showInfo.py: 11%
205 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-13 01:31 -0800
1# This file is part of ctrl_mpexec.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <http://www.gnu.org/licenses/>.
28from __future__ import annotations
30__all__ = ["ShowInfo"]
32import fnmatch
33import re
34import sys
35from collections import defaultdict
36from collections.abc import Mapping
37from typing import Any
39import lsst.pex.config as pexConfig
40import lsst.pex.config.history as pexConfigHistory
41from lsst.daf.butler import Butler, DatasetRef, DatasetType, NamedKeyMapping
42from lsst.daf.butler.datastore.record_data import DatastoreRecordData
43from lsst.pipe.base import PipelineGraph
44from lsst.pipe.base.pipeline_graph import visualization
45from lsst.pipe.base.quantum_graph import PredictedQuantumGraph
46from lsst.resources import ResourcePathExpression
48from . import util
49from ._pipeline_graph_factory import PipelineGraphFactory
52class _FilteredStream:
53 """A file-like object that filters some config fields.
55 Note
56 ----
57 This class depends on implementation details of ``Config.saveToStream``
58 methods, in particular that that method uses single call to write()
59 method to save information about single config field, and that call
60 combines comments string(s) for a field and field path and value.
61 This class will not work reliably on the "import" strings, so imports
62 should be disabled by passing ``skipImports=True`` to ``saveToStream()``.
63 """
65 def __init__(self, pattern: str, stream: Any = None) -> None:
66 if stream is None:
67 stream = sys.stdout
68 self.stream = stream
69 # obey case if pattern isn't lowercase or requests NOIGNORECASE
70 mat = re.search(r"(.*):NOIGNORECASE$", pattern)
72 if mat:
73 pattern = mat.group(1)
74 self._pattern = re.compile(fnmatch.translate(pattern))
75 else:
76 if pattern != pattern.lower():
77 print(
78 f'Matching "{pattern}" without regard to case (append :NOIGNORECASE to prevent this)',
79 file=self.stream,
80 )
81 self._pattern = re.compile(fnmatch.translate(pattern), re.IGNORECASE)
83 def write(self, showStr: str) -> None:
84 # Strip off doc string line(s) and cut off at "=" for string matching
85 matchStr = showStr.rstrip().split("\n")[-1].split("=")[0]
86 if self._pattern.search(matchStr):
87 self.stream.write(showStr)
90class ShowInfo:
91 """Show information about a pipeline or quantum graph.
93 Parameters
94 ----------
95 show : `list` [`str`]
96 A list of show commands, some of which may have additional parameters
97 specified using an ``=``.
98 stream : I/O stream or None
99 The output stream to use. `None` will be treated as `sys.stdout`.
101 Raises
102 ------
103 ValueError
104 Raised if some show commands are not recognized.
105 """
107 pipeline_commands = {
108 "pipeline",
109 "config",
110 "history",
111 "tasks",
112 "dump-config",
113 "pipeline-graph",
114 "task-graph",
115 "subsets",
116 }
117 graph_commands = {"graph", "workflow", "uri"}
119 def __init__(self, show: list[str], stream: Any = None) -> None:
120 if stream is None:
121 # Defer assigning sys.stdout to allow click to redefine it if
122 # it wants. Assigning the default at class definition leads
123 # to confusion on reassignment.
124 stream = sys.stdout
125 commands: dict[str, list[str]] = defaultdict(list)
126 for value in show:
127 command, _, args = value.partition("=")
128 commands[command].append(args)
129 self.commands = commands
130 self.stream = stream
131 self.handled: set[str] = set()
133 known = self.pipeline_commands | self.graph_commands
134 unknown = set(commands) - known
135 if unknown:
136 raise ValueError(
137 f"Unknown value(s) for show: {unknown} (choose from '{', '.join(sorted(known))}')"
138 )
139 self.needs_full_qg: bool = "graph" in self.commands.keys() or "uri" in self.commands.keys()
141 @property
142 def unhandled(self) -> frozenset[str]:
143 """Return the commands that have not yet been processed."""
144 return frozenset(set(self.commands) - self.handled)
146 def show_pipeline_info(self, pipeline_graph_factory: PipelineGraphFactory) -> None:
147 """Display useful information about the pipeline.
149 Parameters
150 ----------
151 pipeline_graph_factory : `PipelineGraphFactory`
152 Factory object that holds the pipeline and can produce a pipeline
153 graph.
154 """
155 for command in self.pipeline_commands:
156 if command not in self.commands:
157 continue
158 args = self.commands[command]
159 match command:
160 case "pipeline":
161 print(pipeline_graph_factory.pipeline, file=self.stream)
162 case "config":
163 for arg in args:
164 self._showConfig(pipeline_graph_factory(visualization_only=True), arg, False)
165 case "dump-config":
166 for arg in args:
167 self._showConfig(pipeline_graph_factory(visualization_only=True), arg, True)
168 case "history":
169 for arg in args:
170 self._showConfigHistory(pipeline_graph_factory(visualization_only=True), arg)
171 case "tasks":
172 self._showTaskHierarchy(pipeline_graph_factory(visualization_only=True))
173 case "subsets":
174 print(
175 "\n".join(
176 f"{subset}:\n" + "\n".join(f" - {s}" for s in sorted(tasks))
177 for subset, tasks in dict(pipeline_graph_factory.pipeline.subsets).items()
178 ),
179 file=self.stream,
180 )
181 case "pipeline-graph":
182 visualization.show(
183 pipeline_graph_factory(visualization_only=True), self.stream, dataset_types=True
184 )
185 case "task-graph":
186 visualization.show(
187 pipeline_graph_factory(visualization_only=True), self.stream, dataset_types=False
188 )
189 case _:
190 raise RuntimeError(f"Unexpectedly tried to process command {command!r}.")
191 self.handled.add(command)
193 def show_graph_info(
194 self,
195 qg: PredictedQuantumGraph,
196 butler_config: ResourcePathExpression | None = None,
197 ) -> None:
198 """Show information associated with this graph.
200 Parameters
201 ----------
202 qg : `lsst.pipe.base.quantum_graph.PredictedQuantumGraph`
203 Quantum graph.
204 butler_config : convertible to `lsst.resources.ResourcePath`, optional
205 Path to configuration for the butler.
206 """
207 for command in self.graph_commands:
208 if command not in self.commands:
209 continue
210 match command:
211 case "graph":
212 self._showGraph(qg)
213 case "uri":
214 if butler_config is None:
215 raise ValueError("Showing URIs requires the -b option")
216 self._showUri(qg, butler_config)
217 case "workflow":
218 self._showWorkflow(qg)
219 case _:
220 raise RuntimeError(f"Unexpectedly tried to process command {command!r}.")
221 self.handled.add(command)
223 def _showConfig(self, pipeline_graph: PipelineGraph, showArgs: str, dumpFullConfig: bool) -> None:
224 """Show task configuration
226 Parameters
227 ----------
228 pipeline : `lsst.pipe.base.pipeline_graph.Pipeline`
229 Pipeline definition as a graph.
230 showArgs : `str`
231 Defines what to show
232 dumpFullConfig : `bool`
233 If true then dump complete task configuration with all imports.
234 """
235 stream: Any = self.stream
236 if dumpFullConfig:
237 # Task label can be given with this option
238 taskName = showArgs
239 else:
240 # The argument can have form [TaskLabel::][pattern:NOIGNORECASE]
241 matConfig = re.search(r"^(?:(\w+)::)?(?:config.)?(.+)?", showArgs)
242 assert matConfig is not None, "regex always matches"
243 taskName = matConfig.group(1)
244 pattern = matConfig.group(2)
245 if pattern:
246 stream = _FilteredStream(pattern, stream=stream)
248 tasks = util.filterTaskNodes(pipeline_graph, taskName)
249 if not tasks:
250 raise ValueError(f"Pipeline has no tasks named {taskName}")
252 for task_node in tasks:
253 print(f"### Configuration for task `{task_node.label}'", file=self.stream)
254 task_node.config.saveToStream(stream, root="config", skipImports=not dumpFullConfig)
256 def _showConfigHistory(self, pipeline_graph: PipelineGraph, showArgs: str) -> None:
257 """Show history for task configuration.
259 Parameters
260 ----------
261 pipeline_graph : `lsst.pipe.base.pipeline_graph.PipelineGraph`
262 Pipeline definition as a graph.
263 showArgs : `str`
264 Defines what to show
265 """
266 taskName = None
267 pattern = None
268 matHistory = re.search(r"^(?:(\w+)::)?(?:config[.])?(.+)", showArgs)
269 if matHistory:
270 taskName = matHistory.group(1)
271 pattern = matHistory.group(2)
272 if not pattern:
273 raise ValueError("Please provide a value with --show history (e.g. history=Task::param)")
275 tasks = util.filterTaskNodes(pipeline_graph, taskName)
276 if not tasks:
277 raise ValueError(f"Pipeline has no tasks named {taskName}")
279 found = False
280 for task_node in tasks:
281 config = task_node.config
283 # Look for any matches in the config hierarchy for this name
284 for nmatch, thisName in enumerate(fnmatch.filter(config.names(), pattern)):
285 if nmatch > 0:
286 print("", file=self.stream)
288 cpath, _, cname = thisName.rpartition(".")
289 try:
290 if not cpath:
291 # looking for top-level field
292 hconfig = task_node.config
293 else:
294 hconfig = eval("config." + cpath, {}, {"config": config})
295 except AttributeError:
296 print(
297 f"Error: Unable to extract attribute {cpath} from task {task_node.label}",
298 file=sys.stderr,
299 )
300 hconfig = None
302 # Sometimes we end up with a non-Config so skip those
303 if isinstance(hconfig, pexConfig.Config | pexConfig.ConfigurableInstance) and hasattr(
304 hconfig, cname
305 ):
306 print(f"### Configuration field for task `{task_node.label}'", file=self.stream)
307 print(pexConfigHistory.format(hconfig, cname), file=self.stream)
308 found = True
310 if not found:
311 raise ValueError(f"None of the tasks has field matching {pattern}")
313 def _showTaskHierarchy(self, pipeline_graph: PipelineGraph) -> None:
314 """Print task hierarchy to stdout
316 Parameters
317 ----------
318 pipeline_graph : `lsst.pipe.base.pipeline_graph.PipelineGraph`
319 Pipeline definition as a graph.
320 """
321 for task_node in pipeline_graph.tasks.values():
322 print(f"### Subtasks for task `{task_node.task_class_name}'", file=self.stream)
324 for configName, taskName in util.subTaskIter(task_node.config):
325 print(f"{configName}: {taskName}", file=self.stream)
327 def _showGraph(self, qg: PredictedQuantumGraph) -> None:
328 """Print quanta information to stdout
330 Parameters
331 ----------
332 qg : `lsst.pipe.base.quantum_graph.PredictedQuantumGraph`
333 Quantum graph.
334 """
336 def _print_refs(
337 mapping: NamedKeyMapping[DatasetType, tuple[DatasetRef, ...]],
338 datastore_records: Mapping[str, DatastoreRecordData],
339 ) -> None:
340 """Print complete information on quantum input or output refs."""
341 for key, refs in mapping.items():
342 if refs:
343 print(f" {key}:", file=self.stream)
344 for ref in refs:
345 print(f" - {ref}", file=self.stream)
346 for datastore_name, record_data in datastore_records.items():
347 if record_map := record_data.records.get(ref.id):
348 print(f" records for {datastore_name}:", file=self.stream)
349 for table_name, records in record_map.items():
350 print(f" - {table_name}:", file=self.stream)
351 for record in records:
352 print(f" - {record}:", file=self.stream)
353 else:
354 print(f" {key}: []", file=self.stream)
356 for task_label, quanta_for_task in qg.quanta_by_task.items():
357 print(f"{task_label} ({qg.pipeline_graph.tasks[task_label].task_class_name})", file=self.stream)
358 execution_quanta = qg.build_execution_quanta(task_label=task_label)
359 for data_id, quantum_id in quanta_for_task.items():
360 quantum = execution_quanta[quantum_id]
361 print(f" Quantum {quantum_id} dataId={data_id}:", file=self.stream)
362 print(" inputs:", file=self.stream)
363 _print_refs(quantum.inputs, quantum.datastore_records)
364 print(" outputs:", file=self.stream)
365 _print_refs(quantum.outputs, quantum.datastore_records)
367 def _showWorkflow(self, qg: PredictedQuantumGraph) -> None:
368 """Print quanta information and dependency to stdout
370 Parameters
371 ----------
372 qg : `lsst.pipe.base.quantum_graph.PredictedQuantumGraph`
373 Quantum graph.
374 """
375 xgraph = qg.quantum_only_xgraph
376 for child_id, child_data in xgraph.nodes.items():
377 print(f"Quantum {child_id}: {child_data['pipeline_node'].task_class_name}", file=self.stream)
378 for parent_id in xgraph.predecessors(child_id):
379 print(f"Parent Quantum {parent_id} - Child Quantum {child_id}", file=self.stream)
381 def _showUri(self, qg: PredictedQuantumGraph, butler_config: ResourcePathExpression) -> None:
382 """Print input and predicted output URIs to stdout.
384 Parameters
385 ----------
386 qg : `lsst.pipe.base.quantum_graph.PredictedQuantumGraph`
387 Quantum graph.
388 butler_config : convertible to `lsst.resources.ResourcePath`
389 Path to configuration for the butler.
390 """
392 def dumpURIs(butler: Butler, thisRef: DatasetRef) -> None:
393 primary, components = butler.getURIs(thisRef, predict=True, run="TBD")
394 if primary:
395 print(f" {primary}", file=self.stream)
396 else:
397 print(" (disassembled artifact)", file=self.stream)
398 for compName, compUri in components.items():
399 print(f" {compName}: {compUri}", file=self.stream)
401 with Butler.from_config(butler_config) as butler:
402 xgraph = qg.quantum_only_xgraph
403 execution_quanta = qg.build_execution_quanta()
404 for quantum_id, quantum_data in xgraph.nodes.items():
405 print(
406 f"Quantum {quantum_id}: {quantum_data['pipeline_node'].task_class_name}", file=self.stream
407 )
408 print(" inputs:", file=self.stream)
409 execution_quantum = execution_quanta[quantum_id]
410 for refs in execution_quantum.inputs.values():
411 for ref in refs:
412 dumpURIs(butler, ref)
413 print(" outputs:", file=self.stream)
414 for refs in execution_quantum.outputs.values():
415 for ref in refs:
416 dumpURIs(butler, ref)