Coverage for python / lsst / pipe / base / pipeline.py: 21%
285 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-07 01:26 -0800
1# This file is part of pipe_base.
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/>.
28"""Module defining Pipeline class and related methods."""
30from __future__ import annotations
32__all__ = ["LabelSpecifier", "Pipeline", "TaskDef"]
34import copy
35import logging
36import re
37import urllib.parse
39# -------------------------------
40# Imports of standard modules --
41# -------------------------------
42from collections.abc import Callable, Set
43from dataclasses import dataclass
44from types import MappingProxyType
45from typing import TYPE_CHECKING, cast
47# -----------------------------
48# Imports for other modules --
49# -----------------------------
50from lsst.daf.butler import DataCoordinate, DimensionUniverse, Registry
51from lsst.resources import ResourcePath, ResourcePathExpression
52from lsst.utils import doImportType
53from lsst.utils.introspection import get_full_type_name
55from . import automatic_connection_constants as acc
56from . import pipeline_graph, pipelineIR
57from ._instrument import Instrument as PipeBaseInstrument
58from .config import PipelineTaskConfig
59from .connections import PipelineTaskConnections
60from .pipelineTask import PipelineTask
62if TYPE_CHECKING: # Imports needed only for type annotations; may be circular.
63 from lsst.obs.base import Instrument
64 from lsst.pex.config import Config
66# ----------------------------------
67# Local non-exported definitions --
68# ----------------------------------
70_LOG = logging.getLogger(__name__)
72# ------------------------
73# Exported definitions --
74# ------------------------
77@dataclass
78class LabelSpecifier:
79 """A structure to specify a subset of labels to load.
81 This structure may contain a set of labels to be used in subsetting a
82 pipeline, or a beginning and end point. Beginning or end may be empty, in
83 which case the range will be a half open interval. Unlike python iteration
84 bounds, end bounds are *INCLUDED*.
86 There are multiple potential definitions of range-based slicing for graphs
87 that are not a simple linear sequence. The definition used here is the
88 intersection of the tasks downstream of ``begin`` and the tasks upstream of
89 ``end``, i.e. tasks with no dependency relationship to a bounding task are
90 not included.
91 """
93 labels: set[str] | None = None
94 begin: str | None = None
95 end: str | None = None
97 def __post_init__(self) -> None:
98 if self.labels is not None and (self.begin or self.end):
99 raise ValueError(
100 "This struct can only be initialized with a labels set or a begin (and/or) end specifier"
101 )
104class TaskDef:
105 """TaskDef is a collection of information about task needed by Pipeline.
107 The information includes task name, configuration object and optional
108 task class. This class is just a collection of attributes and it exposes
109 all of them so that attributes could potentially be modified in place
110 (e.g. if configuration needs extra overrides).
112 Parameters
113 ----------
114 taskName : `str`, optional
115 The fully-qualified `PipelineTask` class name. If not provided,
116 ``taskClass`` must be.
117 config : `lsst.pipe.base.config.PipelineTaskConfig`, optional
118 Instance of the configuration class corresponding to this task class,
119 usually with all overrides applied. This config will be frozen. If
120 not provided, ``taskClass`` must be provided and
121 ``taskClass.ConfigClass()`` will be used.
122 taskClass : `type`, optional
123 `PipelineTask` class object; if provided and ``taskName`` is as well,
124 the caller guarantees that they are consistent. If not provided,
125 ``taskName`` is used to import the type.
126 label : `str`, optional
127 Task label, usually a short string unique in a pipeline. If not
128 provided, ``taskClass`` must be, and ``taskClass._DefaultName`` will
129 be used.
130 connections : `PipelineTaskConnections`, optional
131 Object that describes the dataset types used by the task. If not
132 provided, one will be constructed from the given configuration. If
133 provided, it is assumed that ``config`` has already been validated
134 and frozen.
135 """
137 def __init__(
138 self,
139 taskName: str | None = None,
140 config: PipelineTaskConfig | None = None,
141 taskClass: type[PipelineTask] | None = None,
142 label: str | None = None,
143 connections: PipelineTaskConnections | None = None,
144 ):
145 if taskName is None:
146 if taskClass is None:
147 raise ValueError("At least one of `taskName` and `taskClass` must be provided.")
148 taskName = get_full_type_name(taskClass)
149 elif taskClass is None:
150 taskClass = doImportType(taskName)
151 if config is None:
152 if taskClass is None:
153 raise ValueError("`taskClass` must be provided if `config` is not.")
154 config = taskClass.ConfigClass()
155 if label is None:
156 if taskClass is None:
157 raise ValueError("`taskClass` must be provided if `label` is not.")
158 label = taskClass._DefaultName
159 self.taskName = taskName
160 if connections is None:
161 # If we don't have connections yet, assume the config hasn't been
162 # validated yet.
163 try:
164 config.validate()
165 except Exception:
166 _LOG.error("Configuration validation failed for task %s (%s)", label, taskName)
167 raise
168 config.freeze()
169 connections = config.connections.ConnectionsClass(config=config)
170 self.config = config
171 self.taskClass = taskClass
172 self.label = label
173 self.connections = connections
175 @property
176 def configDatasetName(self) -> str:
177 """Name of a dataset type for configuration of this task (`str`)."""
178 return acc.CONFIG_INIT_OUTPUT_TEMPLATE.format(label=self.label)
180 @property
181 def metadataDatasetName(self) -> str:
182 """Name of a dataset type for metadata of this task (`str`)."""
183 return self.makeMetadataDatasetName(self.label)
185 @classmethod
186 def makeMetadataDatasetName(cls, label: str) -> str:
187 """Construct the name of the dataset type for metadata for a task.
189 Parameters
190 ----------
191 label : `str`
192 Label for the task within its pipeline.
194 Returns
195 -------
196 name : `str`
197 Name of the task's metadata dataset type.
198 """
199 return acc.METADATA_OUTPUT_TEMPLATE.format(label=label)
201 @property
202 def logOutputDatasetName(self) -> str | None:
203 """Name of a dataset type for log output from this task, `None` if
204 logs are not to be saved (`str`).
205 """
206 if self.config.saveLogOutput:
207 return acc.LOG_OUTPUT_TEMPLATE.format(label=self.label)
208 else:
209 return None
211 def __str__(self) -> str:
212 rep = "TaskDef(" + self.taskName
213 if self.label:
214 rep += ", label=" + self.label
215 rep += ")"
216 return rep
218 def __eq__(self, other: object) -> bool:
219 if not isinstance(other, TaskDef):
220 return False
221 # This does not consider equality of configs when determining equality
222 # as config equality is a difficult thing to define. Should be updated
223 # after DM-27847
224 return self.taskClass == other.taskClass and self.label == other.label
226 def __hash__(self) -> int:
227 return hash((self.taskClass, self.label))
229 @classmethod
230 def _unreduce(cls, taskName: str, config: PipelineTaskConfig, label: str) -> TaskDef:
231 """Unpickle pickle. Custom callable for unpickling.
233 All arguments are forwarded directly to the constructor; this
234 trampoline is only needed because ``__reduce__`` callables can't be
235 called with keyword arguments.
236 """
237 return cls(taskName=taskName, config=config, label=label)
239 def __reduce__(self) -> tuple[Callable[[str, PipelineTaskConfig, str], TaskDef], tuple[str, Config, str]]:
240 return (self._unreduce, (self.taskName, self.config, self.label))
243class Pipeline:
244 """A `Pipeline` is a representation of a series of tasks to run, and the
245 configuration for those tasks.
247 Parameters
248 ----------
249 description : `str`
250 A description of that this pipeline does.
251 """
253 PipelineSubsetCtrl = pipelineIR.PipelineSubsetCtrl
255 def __init__(self, description: str):
256 pipeline_dict = {"description": description, "tasks": {}}
257 self._pipelineIR = pipelineIR.PipelineIR(pipeline_dict)
259 @classmethod
260 def fromFile(cls, filename: str) -> Pipeline:
261 """Load a pipeline defined in a pipeline yaml file.
263 Parameters
264 ----------
265 filename : `str`
266 A path that points to a pipeline defined in yaml format. This
267 filename may also supply additional labels to be used in
268 subsetting the loaded Pipeline. These labels are separated from
269 the path by a ``#``, and may be specified as a comma separated
270 list, or a range denoted as beginning..end. Beginning or end may
271 be empty, in which case the range will be a half open interval.
272 Unlike python iteration bounds, end bounds are *INCLUDED*. Note
273 that range based selection is not well defined for pipelines that
274 are not linear in nature, and correct behavior is not guaranteed,
275 or may vary from run to run.
277 Returns
278 -------
279 pipeline: `Pipeline`
280 The pipeline loaded from specified location with appropriate (if
281 any) subsetting.
283 Notes
284 -----
285 This method attempts to prune any contracts that contain labels which
286 are not in the declared subset of labels. This pruning is done using a
287 string based matching due to the nature of contracts and may prune more
288 than it should.
289 """
290 return cls.from_uri(filename)
292 @classmethod
293 def from_uri(cls, uri: ResourcePathExpression) -> Pipeline:
294 """Load a pipeline defined in a pipeline yaml file at a location
295 specified by a URI.
297 Parameters
298 ----------
299 uri : convertible to `~lsst.resources.ResourcePath`
300 If a string is supplied this should be a URI path that points to a
301 pipeline defined in yaml format, either as a direct path to the
302 yaml file, or as a directory containing a ``pipeline.yaml`` file
303 the form used by `write_to_uri` with ``expand=True``). This uri may
304 also supply additional labels to be used in subsetting the loaded
305 `Pipeline`. These labels are separated from the path by a ``#``,
306 and may be specified as a comma separated list, or a range denoted
307 as beginning..end. Beginning or end may be empty, in which case the
308 range will be a half open interval. Unlike python iteration bounds,
309 end bounds are *INCLUDED*. Note that range based selection is not
310 well defined for pipelines that are not linear in nature, and
311 correct behavior is not guaranteed, or may vary from run to run.
312 The same specifiers can be used with a
313 `~lsst.resources.ResourcePath` object, by being the sole contents
314 in the fragments attribute.
316 Returns
317 -------
318 pipeline : `Pipeline`
319 The pipeline loaded from specified location with appropriate (if
320 any) subsetting.
322 Notes
323 -----
324 This method attempts to prune any contracts that contain labels which
325 are not in the declared subset of labels. This pruning is done using a
326 string based matching due to the nature of contracts and may prune more
327 than it should.
328 """
329 # Split up the uri and any labels that were supplied
330 uri, label_specifier = cls._parse_file_specifier(uri)
331 pipeline: Pipeline = cls.fromIR(pipelineIR.PipelineIR.from_uri(uri))
333 # If there are labels supplied, only keep those
334 if label_specifier is not None:
335 pipeline = pipeline.subsetFromLabels(label_specifier)
336 return pipeline
338 def subsetFromLabels(
339 self,
340 labelSpecifier: LabelSpecifier,
341 subsetCtrl: pipelineIR.PipelineSubsetCtrl = PipelineSubsetCtrl.DROP,
342 ) -> Pipeline:
343 """Subset a pipeline to contain only labels specified in
344 ``labelSpecifier``.
346 Parameters
347 ----------
348 labelSpecifier : `labelSpecifier`
349 Object containing labels that describes how to subset a pipeline.
350 subsetCtrl : `PipelineSubsetCtrl`
351 Control object which decides how subsets with missing labels are
352 handled. Setting to `PipelineSubsetCtrl.DROP` (the default) will
353 cause any subsets that have labels which are not in the set of all
354 task labels to be dropped. Setting to `PipelineSubsetCtrl.EDIT`
355 will cause the subset to instead be edited to remove the
356 nonexistent label.
358 Returns
359 -------
360 pipeline : `Pipeline`
361 A new pipeline object that is a subset of the old pipeline.
363 Raises
364 ------
365 ValueError
366 Raised if there is an issue with specified labels
368 Notes
369 -----
370 This method attempts to prune any contracts that contain labels which
371 are not in the declared subset of labels. This pruning is done using a
372 string based matching due to the nature of contracts and may prune more
373 than it should.
374 """
375 # Labels supplied as a set
376 if labelSpecifier.labels:
377 labelSet = labelSpecifier.labels
378 # Labels supplied as a range, first create a list of all the labels
379 # in the pipeline sorted according to task dependency. Then only
380 # keep labels that lie between the supplied bounds
381 else:
382 # Create a copy of the pipeline to use when assessing the label
383 # ordering. Use a dict for fast searching while preserving order.
384 # Remove contracts so they do not fail in the expansion step. This
385 # is needed because a user may only configure the tasks they intend
386 # to run, which may cause some contracts to fail if they will later
387 # be dropped
388 pipeline = copy.deepcopy(self)
389 pipeline._pipelineIR.contracts = []
390 graph = pipeline.to_graph()
392 # Verify the bounds are in the labels
393 if labelSpecifier.begin is not None and labelSpecifier.begin not in graph.tasks:
394 raise ValueError(
395 f"Beginning of range subset, {labelSpecifier.begin}, not found in pipeline definition"
396 )
397 if labelSpecifier.end is not None and labelSpecifier.end not in graph.tasks:
398 raise ValueError(
399 f"End of range subset, {labelSpecifier.end}, not found in pipeline definition"
400 )
402 labelSet = set(graph.tasks.between(labelSpecifier.begin, labelSpecifier.end))
403 return Pipeline.fromIR(self._pipelineIR.subset_from_labels(labelSet, subsetCtrl))
405 @staticmethod
406 def _parse_file_specifier(uri: ResourcePathExpression) -> tuple[ResourcePath, LabelSpecifier | None]:
407 """Split appart a uri and any possible label subsets"""
408 if isinstance(uri, str):
409 # This is to support legacy pipelines during transition
410 uri, num_replace = re.subn("[:](?!\\/\\/)", "#", uri)
411 if num_replace:
412 raise ValueError(
413 f"The pipeline file {uri} seems to use the legacy :"
414 " to separate labels, please use # instead."
415 )
416 if uri.count("#") > 1:
417 raise ValueError("Only one set of labels is allowed when specifying a pipeline to load")
418 # Everything else can be converted directly to ResourcePath.
419 uri = ResourcePath(uri)
420 label_subset = uri.fragment or None
422 specifier: LabelSpecifier | None
423 if label_subset is not None:
424 label_subset = urllib.parse.unquote(label_subset)
425 args: dict[str, set[str] | str | None]
426 # labels supplied as a list
427 if "," in label_subset:
428 if ".." in label_subset:
429 raise ValueError(
430 "Can only specify a list of labels or a range when loading a Pipeline, not both."
431 )
432 args = {"labels": set(label_subset.split(","))}
433 # labels supplied as a range
434 elif ".." in label_subset:
435 # Try to de-structure the labelSubset, this will fail if more
436 # than one range is specified
437 begin, end, *rest = label_subset.split("..")
438 if rest:
439 raise ValueError("Only one range can be specified when loading a pipeline")
440 args = {"begin": begin if begin else None, "end": end if end else None}
441 # Assume anything else is a single label
442 else:
443 args = {"labels": {label_subset}}
445 # MyPy doesn't like how cavalier kwarg construction is with types.
446 specifier = LabelSpecifier(**args) # type: ignore
447 else:
448 specifier = None
450 return uri, specifier
452 @classmethod
453 def fromString(cls, pipeline_string: str) -> Pipeline:
454 """Create a pipeline from string formatted as a pipeline document.
456 Parameters
457 ----------
458 pipeline_string : `str`
459 A string that is formatted according like a pipeline document.
461 Returns
462 -------
463 pipeline: `Pipeline`
464 The new pipeline.
465 """
466 pipeline = cls.fromIR(pipelineIR.PipelineIR.from_string(pipeline_string))
467 return pipeline
469 @classmethod
470 def fromIR(cls, deserialized_pipeline: pipelineIR.PipelineIR) -> Pipeline:
471 """Create a pipeline from an already created `PipelineIR` object.
473 Parameters
474 ----------
475 deserialized_pipeline : `PipelineIR`
476 An already created pipeline intermediate representation object.
478 Returns
479 -------
480 pipeline: `Pipeline`
481 The new pipeline.
482 """
483 pipeline = cls.__new__(cls)
484 pipeline._pipelineIR = deserialized_pipeline
485 return pipeline
487 @classmethod
488 def fromPipeline(cls, pipeline: Pipeline) -> Pipeline:
489 """Create a new pipeline by copying an already existing `Pipeline`.
491 Parameters
492 ----------
493 pipeline : `Pipeline`
494 An already created pipeline intermediate representation object.
496 Returns
497 -------
498 pipeline: `Pipeline`
499 The new pipeline.
500 """
501 return cls.fromIR(copy.deepcopy(pipeline._pipelineIR))
503 def __str__(self) -> str:
504 return str(self._pipelineIR)
506 def mergePipeline(self, pipeline: Pipeline) -> None:
507 """Merge another in-memory `Pipeline` object into this one.
509 This merges another pipeline into this object, as if it were declared
510 in the import block of the yaml definition of this pipeline. This
511 modifies this pipeline in place.
513 Parameters
514 ----------
515 pipeline : `Pipeline`
516 The `Pipeline` object that is to be merged into this object.
517 """
518 self._pipelineIR.merge_pipelines((pipeline._pipelineIR,))
520 def addLabelToSubset(self, subset: str, label: str) -> None:
521 """Add a task label from the specified subset.
523 Parameters
524 ----------
525 subset : `str`
526 The labeled subset to modify.
527 label : `str`
528 The task label to add to the specified subset.
530 Raises
531 ------
532 ValueError
533 Raised if the specified subset does not exist within the pipeline.
534 Raised if the specified label does not exist within the pipeline.
535 """
536 if label not in self._pipelineIR.tasks:
537 raise ValueError(f"Label {label} does not appear within the pipeline")
538 if subset not in self._pipelineIR.labeled_subsets:
539 raise ValueError(f"Subset {subset} does not appear within the pipeline")
540 self._pipelineIR.labeled_subsets[subset].subset.add(label)
542 def removeLabelFromSubset(self, subset: str, label: str) -> None:
543 """Remove a task label from the specified subset.
545 Parameters
546 ----------
547 subset : `str`
548 The labeled subset to modify.
549 label : `str`
550 The task label to remove from the specified subset.
552 Raises
553 ------
554 ValueError
555 Raised if the specified subset does not exist in the pipeline.
556 Raised if the specified label does not exist within the specified
557 subset.
558 """
559 if subset not in self._pipelineIR.labeled_subsets:
560 raise ValueError(f"Subset {subset} does not appear within the pipeline")
561 if label not in self._pipelineIR.labeled_subsets[subset].subset:
562 raise ValueError(f"Label {label} does not appear within the pipeline")
563 self._pipelineIR.labeled_subsets[subset].subset.remove(label)
565 def findSubsetsWithLabel(self, label: str) -> set[str]:
566 """Find any subsets which may contain the specified label.
568 This function returns the name of subsets which return the specified
569 label. May return an empty set if there are no subsets, or no subsets
570 containing the specified label.
572 Parameters
573 ----------
574 label : `str`
575 The task label to use in membership check.
577 Returns
578 -------
579 subsets : `set` of `str`
580 Returns a set (possibly empty) of subsets names which contain the
581 specified label.
583 Raises
584 ------
585 ValueError
586 Raised if the specified label does not exist within this pipeline.
587 """
588 results = set()
589 if label not in self._pipelineIR.tasks:
590 raise ValueError(f"Label {label} does not appear within the pipeline")
591 for subset in self._pipelineIR.labeled_subsets.values():
592 if label in subset.subset:
593 results.add(subset.label)
594 return results
596 @property
597 def task_labels(self) -> Set[str]:
598 """Labels of all tasks in the pipelines.
600 For simple pipelines with no imports, iteration over this set will
601 match the order in which tasks are defined in the pipeline file. In
602 all other cases the order is unspecified but deterministic. It is not
603 dependency-ordered (use ``to_graph().tasks.keys()`` for that).
604 """
605 return self._pipelineIR.tasks.keys()
607 @property
608 def subsets(self) -> MappingProxyType[str, set]:
609 """Returns a `MappingProxyType` where the keys are the labels of
610 labeled subsets in the `Pipeline` and the values are the set of task
611 labels contained within that subset.
612 """
613 return MappingProxyType(
614 {label: subsetIr.subset for label, subsetIr in self._pipelineIR.labeled_subsets.items()}
615 )
617 def addLabeledSubset(self, label: str, description: str, taskLabels: set[str]) -> None:
618 """Add a new labeled subset to the `Pipeline`.
620 Parameters
621 ----------
622 label : `str`
623 The label to assign to the subset.
624 description : `str`
625 A description of what the subset is for.
626 taskLabels : `set` [`str`]
627 The set of task labels to be associated with the labeled subset.
629 Raises
630 ------
631 ValueError
632 Raised if label already exists in the `Pipeline`.
633 Raised if a task label is not found within the `Pipeline`.
634 """
635 if label in self._pipelineIR.labeled_subsets.keys():
636 raise ValueError(f"Subset label {label} is already found within the Pipeline")
637 if extra := (taskLabels - self._pipelineIR.tasks.keys()):
638 raise ValueError(f"Task labels {extra} were not found within the Pipeline")
639 self._pipelineIR.labeled_subsets[label] = pipelineIR.LabeledSubset(label, taskLabels, description)
641 def removeLabeledSubset(self, label: str) -> None:
642 """Remove a labeled subset from the `Pipeline`.
644 Parameters
645 ----------
646 label : `str`
647 The label of the subset to remove from the `Pipeline`.
649 Raises
650 ------
651 ValueError
652 Raised if the label is not found within the `Pipeline`.
653 """
654 if label not in self._pipelineIR.labeled_subsets.keys():
655 raise ValueError(f"Subset label {label} was not found in the pipeline")
656 self._pipelineIR.labeled_subsets.pop(label)
658 def addInstrument(self, instrument: Instrument | str) -> None:
659 """Add an instrument to the pipeline, or replace an instrument that is
660 already defined.
662 Parameters
663 ----------
664 instrument : `~lsst.daf.butler.instrument.Instrument` or `str`
665 Either a derived class object of a `lsst.daf.butler.instrument` or
666 a string corresponding to a fully qualified
667 `lsst.daf.butler.instrument` name.
668 """
669 if isinstance(instrument, str):
670 pass
671 else:
672 # TODO: assume that this is a subclass of Instrument, no type
673 # checking
674 instrument = get_full_type_name(instrument)
675 self._pipelineIR.instrument = instrument
677 def getInstrument(self) -> str | None:
678 """Get the instrument from the pipeline.
680 Returns
681 -------
682 instrument : `str`, or None
683 The fully qualified name of a `lsst.obs.base.Instrument` subclass,
684 name, or None if the pipeline does not have an instrument.
685 """
686 return self._pipelineIR.instrument
688 def get_data_id(self, universe: DimensionUniverse) -> DataCoordinate:
689 """Return a data ID with all dimension constraints embedded in the
690 pipeline.
692 Parameters
693 ----------
694 universe : `lsst.daf.butler.DimensionUniverse`
695 Object that defines all dimensions.
697 Returns
698 -------
699 data_id : `lsst.daf.butler.DataCoordinate`
700 Data ID with all dimension constraints embedded in the
701 pipeline.
702 """
703 instrument_class_name = self._pipelineIR.instrument
704 if instrument_class_name is not None:
705 instrument_class = cast(PipeBaseInstrument, doImportType(instrument_class_name))
706 if instrument_class is not None:
707 return DataCoordinate.standardize(instrument=instrument_class.getName(), universe=universe)
708 return DataCoordinate.make_empty(universe)
710 def addTask(self, task: type[PipelineTask] | str, label: str) -> None:
711 """Add a new task to the pipeline, or replace a task that is already
712 associated with the supplied label.
714 Parameters
715 ----------
716 task : `PipelineTask` or `str`
717 Either a derived class object of a `PipelineTask` or a string
718 corresponding to a fully qualified `PipelineTask` name.
719 label : `str`
720 A label that is used to identify the `PipelineTask` being added.
721 """
722 if isinstance(task, str):
723 taskName = task
724 elif issubclass(task, PipelineTask):
725 taskName = get_full_type_name(task)
726 else:
727 raise ValueError(
728 "task must be either a child class of PipelineTask or a string containing"
729 " a fully qualified name to one"
730 )
731 if not label:
732 # in some cases (with command line-generated pipeline) tasks can
733 # be defined without label which is not acceptable, use task
734 # _DefaultName in that case
735 if isinstance(task, str):
736 task_class = cast(PipelineTask, doImportType(task))
737 label = task_class._DefaultName
738 self._pipelineIR.tasks[label] = pipelineIR.TaskIR(label, taskName)
740 def removeTask(self, label: str) -> None:
741 """Remove a task from the pipeline.
743 Parameters
744 ----------
745 label : `str`
746 The label used to identify the task that is to be removed.
748 Raises
749 ------
750 KeyError
751 If no task with that label exists in the pipeline.
752 """
753 self._pipelineIR.tasks.pop(label)
755 def addConfigOverride(self, label: str, key: str, value: object) -> None:
756 """Apply single config override.
758 Parameters
759 ----------
760 label : `str`
761 Label of the task.
762 key : `str`
763 Fully-qualified field name.
764 value : object
765 Value to be given to a field.
766 """
767 self._addConfigImpl(label, pipelineIR.ConfigIR(rest={key: value}))
769 def addConfigFile(self, label: str, filename: str) -> None:
770 """Add overrides from a specified file.
772 Parameters
773 ----------
774 label : `str`
775 The label used to identify the task associated with config to
776 modify.
777 filename : `str`
778 Path to the override file.
779 """
780 self._addConfigImpl(label, pipelineIR.ConfigIR(file=[filename]))
782 def addConfigPython(self, label: str, pythonString: str) -> None:
783 """Add Overrides by running a snippet of python code against a config.
785 Parameters
786 ----------
787 label : `str`
788 The label used to identity the task associated with config to
789 modify.
790 pythonString : `str`
791 A string which is valid python code to be executed. This is done
792 with config as the only local accessible value.
793 """
794 self._addConfigImpl(label, pipelineIR.ConfigIR(python=pythonString))
796 def _addConfigImpl(self, label: str, newConfig: pipelineIR.ConfigIR) -> None:
797 if label == "parameters":
798 self._pipelineIR.parameters.mapping.update(newConfig.rest)
799 if newConfig.file:
800 raise ValueError("Setting parameters section with config file is not supported")
801 if newConfig.python:
802 raise ValueError("Setting parameters section using python block in unsupported")
803 return
804 if label not in self._pipelineIR.tasks:
805 raise LookupError(f"There are no tasks labeled '{label}' in the pipeline")
806 self._pipelineIR.tasks[label].add_or_update_config(newConfig)
808 def write_to_uri(self, uri: ResourcePathExpression) -> None:
809 """Write the pipeline to a file or directory.
811 Parameters
812 ----------
813 uri : convertible to `~lsst.resources.ResourcePath`
814 URI to write to; may have any scheme with
815 `~lsst.resources.ResourcePath` write support or no scheme for a
816 local file/directory. Should have a ``.yaml`` extension.
817 """
818 self._pipelineIR.write_to_uri(uri)
820 def to_graph(
821 self, registry: Registry | None = None, visualization_only: bool = False
822 ) -> pipeline_graph.PipelineGraph:
823 """Construct a pipeline graph from this pipeline.
825 Constructing a graph applies all configuration overrides, freezes all
826 configuration, checks all contracts, and checks for dataset type
827 consistency between tasks (as much as possible without access to a data
828 repository). It cannot be reversed.
830 Parameters
831 ----------
832 registry : `lsst.daf.butler.Registry`, optional
833 Data repository client. If provided, the graph's dataset types
834 and dimensions will be resolved (see `PipelineGraph.resolve`).
835 visualization_only : `bool`, optional
836 Resolve the graph as well as possible even when dimensions and
837 storage classes cannot really be determined. This can include
838 using the ``universe.commonSkyPix`` as the assumed dimensions of
839 connections that use the "skypix" placeholder and using "<UNKNOWN>"
840 as a storage class name (which will fail if the storage class
841 itself is ever actually loaded).
843 Returns
844 -------
845 graph : `pipeline_graph.PipelineGraph`
846 Representation of the pipeline as a graph.
847 """
848 instrument_class_name = self._pipelineIR.instrument
849 data_id = {}
850 if instrument_class_name is not None:
851 instrument_class: type[Instrument] = doImportType(instrument_class_name)
852 if instrument_class is not None:
853 data_id["instrument"] = instrument_class.getName()
854 graph = pipeline_graph.PipelineGraph(data_id=data_id)
855 graph.description = self._pipelineIR.description
856 for label in self._pipelineIR.tasks:
857 self._add_task_to_graph(label, graph)
858 if self._pipelineIR.contracts is not None:
859 label_to_config = {x.label: x.config for x in graph.tasks.values()}
860 for contract in self._pipelineIR.contracts:
861 # execute this in its own line so it can raise a good error
862 # message if there was problems with the eval
863 success = eval(contract.contract, None, label_to_config)
864 if not success:
865 extra_info = f": {contract.msg}" if contract.msg is not None else ""
866 raise pipelineIR.ContractError(
867 f"Contract(s) '{contract.contract}' were not satisfied{extra_info}"
868 )
869 for label, subset in self._pipelineIR.labeled_subsets.items():
870 graph.add_task_subset(
871 label, subset.subset, subset.description if subset.description is not None else ""
872 )
873 for step_ir in self._pipelineIR.steps:
874 graph.steps.append(step_ir.label, step_ir.dimensions)
875 if registry is not None or visualization_only:
876 graph.resolve(registry=registry, visualization_only=visualization_only)
877 else:
878 graph.sort()
879 return graph
881 def _add_task_to_graph(self, label: str, graph: pipeline_graph.PipelineGraph) -> None:
882 """Add a single task from this pipeline to a pipeline graph that is
883 under construction.
885 Parameters
886 ----------
887 label : `str`
888 Label for the task to be added.
889 graph : `pipeline_graph.PipelineGraph`
890 Graph to add the task to.
891 """
892 if (taskIR := self._pipelineIR.tasks.get(label)) is None:
893 raise NameError(f"Label {label} does not appear in this pipeline")
894 taskClass: type[PipelineTask] = doImportType(taskIR.klass)
895 config = taskClass.ConfigClass()
896 instrument: PipeBaseInstrument | None = None
897 if (instrumentName := self._pipelineIR.instrument) is not None:
898 instrument_cls: type = doImportType(instrumentName)
899 instrument = instrument_cls()
900 config.applyConfigOverrides(
901 instrument,
902 getattr(taskClass, "_DefaultName", ""),
903 taskIR.config,
904 self._pipelineIR.parameters,
905 label,
906 )
907 graph.add_task(label, taskClass, config)
909 def __len__(self) -> int:
910 return len(self._pipelineIR.tasks)
912 def __eq__(self, other: object) -> bool:
913 if not isinstance(other, Pipeline):
914 return False
915 elif self._pipelineIR == other._pipelineIR:
916 # Shortcut: if the IR is the same, the expanded pipeline must be
917 # the same as well. But the converse is not true.
918 return True
919 else:
920 # Compare as much as we can (task classes and their edges).
921 if self.to_graph().diff_tasks(other.to_graph()):
922 return False
923 # After DM-27847, we should compare configuration here.
924 raise NotImplementedError(
925 "Pipelines cannot be compared because config instances cannot be compared; see DM-27847."
926 )