Coverage for python/lsst/pipe/base/pipelineIR.py: 19%
450 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-09 03:38 -0700
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/>.
27from __future__ import annotations
29__all__ = (
30 "ConfigIR",
31 "ContractError",
32 "ContractIR",
33 "ImportIR",
34 "LabeledSubset",
35 "ParametersIR",
36 "PipelineIR",
37 "TaskIR",
38)
40import copy
41import enum
42import os
43import re
44import warnings
45from collections import Counter
46from collections.abc import Generator, Hashable, Iterable, MutableMapping
47from dataclasses import dataclass, field
48from typing import Any, Literal
50import yaml
52from lsst.resources import ResourcePath, ResourcePathExpression
53from lsst.utils.introspection import find_outside_stacklevel
56class PipelineSubsetCtrl(enum.Enum):
57 """An Enumeration of the various ways a pipeline subsetting operation will
58 handle labeled subsets when task labels they defined are missing.
59 """
61 DROP = enum.auto()
62 """Drop any subsets that contain labels which are no longer in the set of
63 task labels when subsetting an entire pipeline
64 """
65 EDIT = enum.auto()
66 """Edit any subsets that contain labels which are no longer in the set of
67 task labels to remove the missing label, but leave the subset when
68 subsetting a pipeline.
69 """
72class _Tags(enum.Enum):
73 KeepInstrument = enum.auto()
76class PipelineYamlLoader(yaml.SafeLoader):
77 """Specialized version of yaml's SafeLoader.
79 It checks and raises an exception if it finds that there are multiple
80 instances of the same key found inside a pipeline file at a given scope.
81 """
83 def construct_mapping(self, node: yaml.MappingNode, deep: bool = False) -> dict[Hashable, Any]:
84 # do the call to super first so that it can do all the other forms of
85 # checking on this node. If you check the uniqueness of keys first
86 # it would save the work that super does in the case of a failure, but
87 # it might fail in the case that the node was the incorrect node due
88 # to a parsing error, and the resulting exception would be difficult to
89 # understand.
90 mapping = super().construct_mapping(node, deep)
91 # Check if there are any duplicate keys
92 all_keys = Counter(key_node.value for key_node, _ in node.value)
93 duplicates = {k for k, i in all_keys.items() if i != 1}
94 if duplicates:
95 raise KeyError(
96 f"Pipeline files must not have duplicated keys, {duplicates} appeared multiple times"
97 )
98 return mapping
101class MultilineStringDumper(yaml.Dumper):
102 """Custom YAML dumper that makes multi-line strings use the '|'
103 continuation style instead of unreadable newlines and tons of quotes.
105 Basic approach is taken from
106 https://stackoverflow.com/questions/8640959/how-can-i-control-what-scalar-form-pyyaml-uses-for-my-data,
107 but is written as a Dumper subclass to make its effects non-global (vs
108 `yaml.add_representer`).
109 """
111 def represent_scalar(self, tag: str, value: Any, style: str | None = None) -> yaml.ScalarNode:
112 if style is None and tag == "tag:yaml.org,2002:str" and len(value.splitlines()) > 1:
113 style = "|"
114 return super().represent_scalar(tag, value, style)
117class ContractError(Exception):
118 """An exception that is raised when a pipeline contract is not
119 satisfied.
120 """
122 pass
125@dataclass
126class ContractIR:
127 """Intermediate representation of configuration contracts read from a
128 pipeline yaml file.
129 """
131 contract: str
132 """A string of python code representing one or more conditions on configs
133 in a pipeline. This code-as-string should, once evaluated, should be True
134 if the configs are fine, and False otherwise.
135 """
136 msg: str | None = None
137 """An optional message to be shown to the user if a contract fails
138 """
140 def to_primitives(self) -> dict[str, str]:
141 """Convert to a representation used in yaml serialization."""
142 accumulate = {"contract": self.contract}
143 if self.msg is not None:
144 accumulate["msg"] = self.msg
145 return accumulate
147 def __eq__(self, other: object) -> bool:
148 if not isinstance(other, ContractIR):
149 return False
150 return self.contract == other.contract and self.msg == other.msg
153@dataclass
154class LabeledSubset:
155 """Intermediate representation of named subset of task labels read from
156 a pipeline yaml file.
157 """
159 label: str
160 """The label used to identify the subset of task labels.
161 """
162 subset: set[str]
163 """A set of task labels contained in this subset.
164 """
165 description: str | None
166 """A description of what this subset of tasks is intended to do
167 """
169 @staticmethod
170 def from_primitives(label: str, value: list[str] | dict) -> LabeledSubset:
171 """Generate `LabeledSubset` objects given a properly formatted object
172 that as been created by a yaml loader.
174 Parameters
175 ----------
176 label : `str`
177 The label that will be used to identify this labeled subset.
178 value : `list` of `str` or `dict`
179 Object returned from loading a labeled subset section from a yaml
180 document.
182 Returns
183 -------
184 labeledSubset : `LabeledSubset`
185 A `LabeledSubset` object build from the inputs.
187 Raises
188 ------
189 ValueError
190 Raised if the value input is not properly formatted for parsing
191 """
192 if isinstance(value, MutableMapping):
193 subset = value.pop("subset", None)
194 if subset is None:
195 raise ValueError(
196 "If a labeled subset is specified as a mapping, it must contain the key 'subset'"
197 )
198 description = value.pop("description", None)
199 elif isinstance(value, Iterable):
200 subset = value
201 description = None
202 else:
203 raise ValueError(
204 f"There was a problem parsing the labeled subset {label}, make sure the "
205 "definition is either a valid yaml list, or a mapping with keys "
206 "(subset, description) where subset points to a yaml list, and description is "
207 "associated with a string"
208 )
209 return LabeledSubset(label, set(subset), description)
211 def to_primitives(self) -> dict[str, list[str] | str]:
212 """Convert to a representation used in yaml serialization."""
213 accumulate: dict[str, list[str] | str] = {"subset": list(self.subset)}
214 if self.description is not None:
215 accumulate["description"] = self.description
216 return accumulate
219@dataclass
220class ParametersIR:
221 """Intermediate representation of parameters that are global to a pipeline.
223 Attributes
224 ----------
225 mapping : `dict` [`str`, `str`]
226 A mutable mapping of identifiers as keys, and shared configuration
227 as values.
229 Notes
230 -----
231 These parameters are specified under a top level key named ``parameters``
232 and are declared as a yaml mapping. These entries can then be used inside
233 task configuration blocks to specify configuration values. They may not be
234 used in the special ``file`` or ``python`` blocks.
236 Examples
237 --------
238 .. code-block:: yaml
240 \u200bparameters:
241 shared_value: 14
242 tasks:
243 taskA:
244 class: modA
245 config:
246 field1: parameters.shared_value
247 taskB:
248 class: modB
249 config:
250 field2: parameters.shared_value
251 """
253 mapping: MutableMapping[str, Any]
254 """A mutable mapping of identifiers as keys, and shared configuration
255 as values.
256 """
258 def update(self, other: ParametersIR | None) -> None:
259 if other is not None:
260 self.mapping.update(other.mapping)
262 def to_primitives(self) -> MutableMapping[str, str]:
263 """Convert to a representation used in yaml serialization."""
264 return self.mapping
266 def __contains__(self, value: str) -> bool:
267 return value in self.mapping
269 def __getitem__(self, item: str) -> Any:
270 return self.mapping[item]
272 def __bool__(self) -> bool:
273 return bool(self.mapping)
276@dataclass
277class ConfigIR:
278 """Intermediate representation of configurations read from a pipeline yaml
279 file.
280 """
282 python: str | None = None
283 """A string of python code that is used to modify a configuration. This can
284 also be None if there are no modifications to do.
285 """
286 dataId: dict | None = None
287 """A dataId that is used to constrain these config overrides to only quanta
288 with matching dataIds. This field can be None if there is no constraint.
289 This is currently an unimplemented feature, and is placed here for future
290 use.
291 """
292 file: list[str] = field(default_factory=list)
293 """A list of paths which points to a file containing config overrides to be
294 applied. This value may be an empty list if there are no overrides to
295 apply.
296 """
297 rest: dict = field(default_factory=dict)
298 """This is a dictionary of key value pairs, where the keys are strings
299 corresponding to qualified fields on a config to override, and the values
300 are strings representing the values to apply.
301 """
303 def to_primitives(self) -> dict[str, str | dict | list[str]]:
304 """Convert to a representation used in yaml serialization."""
305 accumulate = {}
306 for name in ("python", "dataId", "file"):
307 # if this attribute is truthy add it to the accumulation
308 # dictionary
309 if getattr(self, name):
310 accumulate[name] = getattr(self, name)
311 # Add the dictionary containing the rest of the config keys to the
312 # # accumulated dictionary
313 accumulate.update(self.rest)
314 return accumulate
316 def formatted(self, parameters: ParametersIR) -> ConfigIR:
317 """Return a new ConfigIR object that is formatted according to the
318 specified parameters.
320 Parameters
321 ----------
322 parameters : `ParametersIR`
323 Object that contains variable mappings used in substitution.
325 Returns
326 -------
327 config : `ConfigIR`
328 A new ConfigIR object formatted with the input parameters.
329 """
330 new_config = copy.deepcopy(self)
331 for key, value in new_config.rest.items():
332 if not isinstance(value, str):
333 continue
334 match = re.match("parameters[.](.*)", value)
335 if match and match.group(1) in parameters:
336 new_config.rest[key] = parameters[match.group(1)]
337 if match and match.group(1) not in parameters:
338 warnings.warn(
339 f"config {key} contains value {match.group(0)} which is formatted like a "
340 "Pipeline parameter but was not found within the Pipeline, if this was not "
341 "intentional, check for a typo",
342 stacklevel=find_outside_stacklevel("lsst.pipe.base"),
343 )
344 return new_config
346 def maybe_merge(self, other_config: ConfigIR) -> Generator[ConfigIR, None, None]:
347 """Merge another instance of a `ConfigIR` into this instance if
348 possible. This function returns a generator that is either self
349 if the configs were merged, or self, and other_config if that could
350 not be merged.
352 Parameters
353 ----------
354 other_config : `ConfigIR`
355 An instance of `ConfigIR` to merge into this instance.
357 Yields
358 ------
359 Generator : `ConfigIR`
360 A generator containing either self, or self and other_config if
361 the configs could be merged or not respectively.
362 """
363 # Verify that the config blocks can be merged
364 if (
365 self.dataId != other_config.dataId
366 or self.python
367 or other_config.python
368 or self.file
369 or other_config.file
370 ):
371 yield from (self, other_config)
372 return
374 # create a set of all keys, and verify two keys do not have different
375 # values
376 key_union = self.rest.keys() & other_config.rest.keys()
377 for key in key_union:
378 if self.rest[key] != other_config.rest[key]:
379 yield from (self, other_config)
380 return
381 self.rest.update(other_config.rest)
383 # Combine the lists of override files to load
384 self_file_set = set(self.file)
385 other_file_set = set(other_config.file)
386 self.file = list(self_file_set.union(other_file_set))
388 yield self
390 def __eq__(self, other: object) -> bool:
391 if not isinstance(other, ConfigIR):
392 return False
393 return all(
394 getattr(self, attr) == getattr(other, attr) for attr in ("python", "dataId", "file", "rest")
395 )
398@dataclass
399class TaskIR:
400 """Intermediate representation of tasks read from a pipeline yaml file."""
402 label: str
403 """An identifier used to refer to a task.
404 """
405 klass: str
406 """A string containing a fully qualified python class to be run in a
407 pipeline.
408 """
409 config: list[ConfigIR] | None = None
410 """list of all configs overrides associated with this task, and may be
411 `None` if there are no config overrides.
412 """
414 def to_primitives(self) -> dict[str, str | list[dict]]:
415 """Convert to a representation used in yaml serialization."""
416 accumulate: dict[str, str | list[dict]] = {"class": self.klass}
417 if self.config:
418 accumulate["config"] = [c.to_primitives() for c in self.config]
419 return accumulate
421 def add_or_update_config(self, other_config: ConfigIR) -> None:
422 """Add a `ConfigIR` to this task if one is not present. Merges configs
423 if there is a `ConfigIR` present and the dataId keys of both configs
424 match, otherwise adds a new entry to the config list. The exception to
425 the above is that if either the last config or other_config has a
426 python block, then other_config is always added, as python blocks can
427 modify configs in ways that cannot be predicted.
429 Parameters
430 ----------
431 other_config : `ConfigIR`
432 A `ConfigIR` instance to add or merge into the config attribute of
433 this task.
434 """
435 if not self.config:
436 self.config = [other_config]
437 return
438 self.config.extend(self.config.pop().maybe_merge(other_config))
440 def __eq__(self, other: object) -> bool:
441 if not isinstance(other, TaskIR):
442 return False
443 return all(getattr(self, attr) == getattr(other, attr) for attr in ("label", "klass", "config"))
446@dataclass
447class ImportIR:
448 """An intermediate representation of imported pipelines."""
450 location: str
451 """This is the location of the pipeline to inherit. The path should be
452 specified as an absolute path. Environment variables may be used in the
453 path and should be specified as a python string template, with the name of
454 the environment variable inside braces.
455 """
456 include: list[str] | None = None
457 """list of tasks that should be included when inheriting this pipeline.
458 Either the include or exclude attributes may be specified, but not both.
459 """
460 exclude: list[str] | None = None
461 """list of tasks that should be excluded when inheriting this pipeline.
462 Either the include or exclude attributes may be specified, but not both.
463 """
464 importContracts: bool = True
465 """Boolean attribute to dictate if contracts should be inherited with the
466 pipeline or not.
467 """
468 importSteps: bool = True
469 """Boolean attribute to dictate if steps should be inherited with the
470 pipeline or not.
471 """
472 labeledSubsetModifyMode: PipelineSubsetCtrl = PipelineSubsetCtrl.DROP
473 """Controls how labeled subsets are handled when an import ends up not
474 including (either through an include or exclusion list) a task label that
475 is defined in the `Pipeline` being imported. DROP will remove any
476 subsets which contain a missing label. EDIT will change any subsets to not
477 include the missing label.
478 """
479 instrument: Literal[_Tags.KeepInstrument] | str | None = _Tags.KeepInstrument
480 """Instrument to assign to the Pipeline at import. The default value of
481 `_Tags.KeepInstrument`` indicates that whatever instrument the pipeline is
482 declared with will not be modified. setting this value to None will drop
483 any declared instrument prior to import.
484 """
486 def toPipelineIR(self) -> PipelineIR:
487 """Load in the Pipeline specified by this object, and turn it into a
488 PipelineIR instance.
490 Returns
491 -------
492 pipeline : `PipelineIR`
493 A pipeline generated from the imported pipeline file.
494 """
495 if self.include and self.exclude:
496 raise ValueError(
497 "An include list and an exclude list cannot both be specified"
498 " when declaring a pipeline import."
499 )
500 tmp_pipeline = PipelineIR.from_uri(os.path.expandvars(self.location))
501 if self.instrument is not _Tags.KeepInstrument:
502 tmp_pipeline.instrument = self.instrument
504 included_labels = set()
505 for label in tmp_pipeline.tasks:
506 if (
507 (self.include and label in self.include)
508 or (self.exclude and label not in self.exclude)
509 or (self.include is None and self.exclude is None)
510 ):
511 included_labels.add(label)
513 # Handle labeled subsets being specified in the include or exclude
514 # list, adding or removing labels.
515 if self.include is not None:
516 subsets_in_include = tmp_pipeline.labeled_subsets.keys() & self.include
517 for label in subsets_in_include:
518 included_labels.update(tmp_pipeline.labeled_subsets[label].subset)
520 elif self.exclude is not None:
521 subsets_in_exclude = tmp_pipeline.labeled_subsets.keys() & self.exclude
522 for label in subsets_in_exclude:
523 included_labels.difference_update(tmp_pipeline.labeled_subsets[label].subset)
525 if not self.importSteps:
526 tmp_pipeline.steps = []
528 tmp_pipeline = tmp_pipeline.subset_from_labels(included_labels, self.labeledSubsetModifyMode)
530 if not self.importContracts:
531 tmp_pipeline.contracts = []
533 return tmp_pipeline
535 def __eq__(self, other: object) -> bool:
536 if not isinstance(other, ImportIR):
537 return False
538 return all(
539 getattr(self, attr) == getattr(other, attr)
540 for attr in ("location", "include", "exclude", "importContracts")
541 )
544@dataclass
545class StepIR:
546 """Intermediate representation of a step definition."""
548 label: str
549 """The label associated with this step."""
551 dimensions: list[str]
552 """The dimensions to use when sharding this step."""
555class PipelineIR:
556 """Intermediate representation of a pipeline definition.
558 Parameters
559 ----------
560 loaded_yaml : `dict`
561 A dictionary which matches the structure that would be produced by a
562 yaml reader which parses a pipeline definition document.
564 Raises
565 ------
566 ValueError
567 Raised if:
569 - a pipeline is declared without a description;
570 - no tasks are declared in a pipeline, and no pipelines are to be
571 inherited;
572 - more than one instrument is specified;
573 - more than one inherited pipeline share a label.
574 """
576 def __init__(self, loaded_yaml: dict[str, Any]):
577 # Check required fields are present
578 if "description" not in loaded_yaml:
579 raise ValueError("A pipeline must be declared with a description")
580 if "tasks" not in loaded_yaml and len({"imports", "inherits"} - loaded_yaml.keys()) == 2:
581 raise ValueError("A pipeline must be declared with one or more tasks")
583 # These steps below must happen in this call order
585 # Process pipeline description
586 self.description = loaded_yaml.pop("description")
588 # Process tasks
589 self._read_tasks(loaded_yaml)
591 # Process instrument keys
592 inst = loaded_yaml.pop("instrument", None)
593 if isinstance(inst, list):
594 raise ValueError("Only one top level instrument can be defined in a pipeline")
595 self.instrument: str | None = inst
597 # Process any contracts
598 self._read_contracts(loaded_yaml)
600 # Process any defined parameters
601 self._read_parameters(loaded_yaml)
603 # Process any named label subsets
604 self._read_labeled_subsets(loaded_yaml)
606 # Process defined sets
607 self._read_step_declaration(loaded_yaml)
609 # Process any inherited pipelines
610 self._read_imports(loaded_yaml)
612 # verify named subsets, must be done after inheriting
613 self._verify_labeled_subsets()
615 # verify steps, must be done after inheriting
616 self._verify_steps()
618 def _read_contracts(self, loaded_yaml: dict[str, Any]) -> None:
619 """Process the contracts portion of the loaded yaml document
621 Parameters
622 ----------
623 loaded_yaml : `dict`
624 A dictionary which matches the structure that would be produced by
625 a yaml reader which parses a pipeline definition document
626 """
627 loaded_contracts = loaded_yaml.pop("contracts", [])
628 if isinstance(loaded_contracts, str):
629 loaded_contracts = [loaded_contracts]
630 self.contracts: list[ContractIR] = []
631 for contract in loaded_contracts:
632 if isinstance(contract, dict):
633 self.contracts.append(ContractIR(**contract))
634 if isinstance(contract, str):
635 self.contracts.append(ContractIR(contract=contract))
637 def _read_parameters(self, loaded_yaml: dict[str, Any]) -> None:
638 """Process the parameters portion of the loaded yaml document
640 Parameters
641 ----------
642 loaded_yaml : `dict`
643 A dictionary which matches the structure that would be produced by
644 a yaml reader which parses a pipeline definition document
645 """
646 loaded_parameters = loaded_yaml.pop("parameters", {})
647 if not isinstance(loaded_parameters, dict):
648 raise ValueError("The parameters section must be a yaml mapping")
649 self.parameters = ParametersIR(loaded_parameters)
651 def _read_labeled_subsets(self, loaded_yaml: dict[str, Any]) -> None:
652 """Process the subsets portion of the loaded yaml document
654 Parameters
655 ----------
656 loaded_yaml : `MutableMapping`
657 A dictionary which matches the structure that would be produced
658 by a yaml reader which parses a pipeline definition document
659 """
660 loaded_subsets = loaded_yaml.pop("subsets", {})
661 self.labeled_subsets: dict[str, LabeledSubset] = {}
662 if not loaded_subsets and "subset" in loaded_yaml:
663 raise ValueError("Top level key should be subsets and not subset, add an s")
664 for key, value in loaded_subsets.items():
665 self.labeled_subsets[key] = LabeledSubset.from_primitives(key, value)
667 def _read_step_declaration(self, loaded_yaml: dict[str, Any]) -> None:
668 """Process the steps portion of the loaded yaml document
670 Steps are subsets that are declared to be normal parts of the overall
671 processing of the pipeline. Not all subsets need to be a step, as they
672 can exist for certain targeted processing, such as debugging.
674 Parameters
675 ----------
676 loaded_yaml: `dict`
677 A dictionary which matches the structure that would be produced
678 by a yaml reader which parses a pipeline definition document
679 """
680 loaded_steps = loaded_yaml.pop("steps", [])
681 temp_steps: dict[str, StepIR] = {}
682 for declaration in loaded_steps:
683 new_step = StepIR(**declaration)
684 existing = temp_steps.setdefault(new_step.label, new_step)
685 if existing is not new_step:
686 raise ValueError(f"Step {existing.label} was declared twice.")
687 self.steps = [step for step in temp_steps.values()]
689 def _verify_labeled_subsets(self) -> None:
690 """Verify that all the labels in each named subset exist within the
691 pipeline.
692 """
693 # Verify that all labels defined in a labeled subset are in the
694 # Pipeline
695 for labeled_subset in self.labeled_subsets.values():
696 if not labeled_subset.subset.issubset(self.tasks.keys()):
697 raise ValueError(
698 f"Labels {labeled_subset.subset - self.tasks.keys()} were not found in the "
699 "declared pipeline"
700 )
701 # Verify subset labels are not already task labels
702 label_intersection = self.labeled_subsets.keys() & self.tasks.keys()
703 if label_intersection:
704 raise ValueError(f"Labeled subsets can not use the same label as a task: {label_intersection}")
706 def _verify_steps(self) -> None:
707 """Verify that all step definitions have a corresponding labeled
708 subset.
709 """
710 for step in self.steps:
711 if step.label not in self.labeled_subsets:
712 raise ValueError(
713 f"{step.label} was declared to be a step, but was not declared to be a labeled subset"
714 )
716 def _read_imports(self, loaded_yaml: dict[str, Any]) -> None:
717 """Process the inherits portion of the loaded yaml document
719 Parameters
720 ----------
721 loaded_yaml : `dict`
722 A dictionary which matches the structure that would be produced by
723 a yaml reader which parses a pipeline definition document
724 """
726 def process_args(argument: str | dict) -> dict:
727 if isinstance(argument, str):
728 return {"location": argument}
729 elif isinstance(argument, dict):
730 if "exclude" in argument and isinstance(argument["exclude"], str):
731 argument["exclude"] = [argument["exclude"]]
732 if "include" in argument and isinstance(argument["include"], str):
733 argument["include"] = [argument["include"]]
734 if "instrument" in argument and argument["instrument"] == "None":
735 argument["instrument"] = None
736 if "labeledSubsetModifyMode" in argument:
737 match argument["labeledSubsetModifyMode"]:
738 case "DROP":
739 argument["labeledSubsetModifyMode"] = PipelineSubsetCtrl.DROP
740 case "EDIT":
741 argument["labeledSubsetModifyMode"] = PipelineSubsetCtrl.EDIT
742 case unknown:
743 raise ValueError(f"{unknown} is not a valid mode for labeledSubsetModifyMode")
744 return argument
746 if not {"inherits", "imports"} - loaded_yaml.keys():
747 raise ValueError("Cannot define both inherits and imports sections, use imports")
748 tmp_import = loaded_yaml.pop("inherits", None)
749 if tmp_import is None:
750 tmp_import = loaded_yaml.pop("imports", None)
751 else:
752 raise ValueError("The 'inherits' key is not supported. Please use the key 'imports' instead")
753 if tmp_import is None:
754 self.imports: list[ImportIR] = []
755 elif isinstance(tmp_import, list):
756 self.imports = [ImportIR(**process_args(args)) for args in tmp_import]
757 else:
758 self.imports = [ImportIR(**process_args(tmp_import))]
760 self.merge_pipelines([fragment.toPipelineIR() for fragment in self.imports])
762 def merge_pipelines(self, pipelines: Iterable[PipelineIR]) -> None:
763 """Merge one or more other `PipelineIR` objects into this object.
765 Parameters
766 ----------
767 pipelines : `~collections.abc.Iterable` of `PipelineIR` objects
768 An `~collections.abc.Iterable` that contains one or more
769 `PipelineIR` objects to merge into this object.
771 Raises
772 ------
773 ValueError
774 Raised if there is a conflict in instrument specifications.
775 Raised if a task label appears in more than one of the input
776 `PipelineIR` objects which are to be merged.
777 Raised if a labeled subset appears in more than one of the input
778 `PipelineIR` objects which are to be merged, and with any subset
779 existing in this object.
780 """
781 # integrate any imported pipelines
782 accumulate_tasks: dict[str, TaskIR] = {}
783 accumulate_labeled_subsets: dict[str, LabeledSubset] = {}
784 accumulated_parameters = ParametersIR({})
785 accumulated_steps: dict[str, StepIR] = {}
787 for tmp_IR in pipelines:
788 if self.instrument is None:
789 self.instrument = tmp_IR.instrument
790 elif self.instrument != tmp_IR.instrument and tmp_IR.instrument is not None:
791 msg = (
792 "Only one instrument can be declared in a pipeline or its imports. "
793 f"Top level pipeline defines {self.instrument} but pipeline to merge "
794 f"defines {tmp_IR.instrument}."
795 )
796 raise ValueError(msg)
797 if duplicate_labels := accumulate_tasks.keys() & tmp_IR.tasks.keys():
798 msg = (
799 "Task labels in the imported pipelines must be unique. "
800 f"These labels appear multiple times: {duplicate_labels}"
801 )
802 raise ValueError(msg)
803 accumulate_tasks.update(tmp_IR.tasks)
804 self.contracts.extend(tmp_IR.contracts)
805 # verify that tmp_IR has unique labels for named subset among
806 # existing labeled subsets, and with existing task labels.
807 overlapping_subsets = accumulate_labeled_subsets.keys() & tmp_IR.labeled_subsets.keys()
808 task_subset_overlap = (
809 accumulate_labeled_subsets.keys() | tmp_IR.labeled_subsets.keys()
810 ) & accumulate_tasks.keys()
811 if overlapping_subsets or task_subset_overlap:
812 raise ValueError(
813 "Labeled subset names must be unique amongst imports in both labels and "
814 f" named Subsets. Duplicate: {overlapping_subsets | task_subset_overlap}"
815 )
816 accumulate_labeled_subsets.update(tmp_IR.labeled_subsets)
817 accumulated_parameters.update(tmp_IR.parameters)
818 for tmp_step in tmp_IR.steps:
819 existing = accumulated_steps.setdefault(tmp_step.label, tmp_step)
820 if existing != tmp_step:
821 raise ValueError(
822 f"There were conflicting step definitions in import {tmp_step}, {existing}"
823 )
825 for tmp_step in self.steps:
826 existing = accumulated_steps.setdefault(tmp_step.label, tmp_step)
827 if existing != tmp_step:
828 raise ValueError(f"There were conflicting step definitions in import {tmp_step}, {existing}")
830 # verify that any accumulated labeled subsets dont clash with a label
831 # from this pipeline
832 if accumulate_labeled_subsets.keys() & self.tasks.keys():
833 raise ValueError(
834 "Labeled subset names must be unique amongst imports in both labels and named Subsets"
835 )
836 # merge in the named subsets for self so this document can override any
837 # that have been declared
838 accumulate_labeled_subsets.update(self.labeled_subsets)
839 self.labeled_subsets = accumulate_labeled_subsets
841 # merge the dict of label:TaskIR objects, preserving any configs in the
842 # imported pipeline if the labels point to the same class
843 for label, task in self.tasks.items():
844 if label not in accumulate_tasks:
845 accumulate_tasks[label] = task
846 elif accumulate_tasks[label].klass == task.klass:
847 if task.config is not None:
848 for config in task.config:
849 accumulate_tasks[label].add_or_update_config(config)
850 else:
851 accumulate_tasks[label] = task
852 self.tasks: dict[str, TaskIR] = accumulate_tasks
853 accumulated_parameters.update(self.parameters)
854 self.parameters = accumulated_parameters
855 self.steps = list(accumulated_steps.values())
857 def _read_tasks(self, loaded_yaml: dict[str, Any]) -> None:
858 """Process the tasks portion of the loaded yaml document
860 Parameters
861 ----------
862 loaded_yaml : `dict`
863 A dictionary which matches the structure that would be produced by
864 a yaml reader which parses a pipeline definition document
865 """
866 self.tasks = {}
867 tmp_tasks = loaded_yaml.pop("tasks", None)
868 if tmp_tasks is None:
869 tmp_tasks = {}
871 if "parameters" in tmp_tasks:
872 raise ValueError("parameters is a reserved word and cannot be used as a task label")
874 for label, definition in tmp_tasks.items():
875 if isinstance(definition, str):
876 definition = {"class": definition}
877 config = definition.get("config", None)
878 if config is None:
879 task_config_ir = None
880 else:
881 if isinstance(config, dict):
882 config = [config]
883 task_config_ir = []
884 for c in config:
885 file = c.pop("file", None)
886 if file is None:
887 file = []
888 elif not isinstance(file, list):
889 file = [file]
890 task_config_ir.append(
891 ConfigIR(
892 python=c.pop("python", None), dataId=c.pop("dataId", None), file=file, rest=c
893 )
894 )
895 self.tasks[label] = TaskIR(label, definition["class"], task_config_ir)
897 def _remove_contracts(self, label: str) -> None:
898 """Remove any contracts that contain the given label
900 String comparison used in this way is not the most elegant and may
901 have issues, but it is the only feasible way when users can specify
902 contracts with generic strings.
903 """
904 new_contracts = []
905 for contract in self.contracts:
906 # match a label that is not preceded by an ASCII identifier, or
907 # is the start of a line and is followed by a dot
908 if re.match(f".*([^A-Za-z0-9_]|^){label}[.]", contract.contract):
909 continue
910 new_contracts.append(contract)
911 self.contracts = new_contracts
913 def subset_from_labels(
914 self, labelSpecifier: set[str], subsetCtrl: PipelineSubsetCtrl = PipelineSubsetCtrl.DROP
915 ) -> PipelineIR:
916 """Subset a pipelineIR to contain only labels specified in
917 labelSpecifier.
919 Parameters
920 ----------
921 labelSpecifier : `set` of `str`
922 Set containing labels that describes how to subset a pipeline.
923 subsetCtrl : `PipelineSubsetCtrl`
924 Control object which decides how subsets with missing labels are
925 handled. Setting to `PipelineSubsetCtrl.DROP` (the default) will
926 cause any subsets that have labels which are not in the set of all
927 task labels to be dropped. Setting to `PipelineSubsetCtrl.EDIT`
928 will cause the subset to instead be edited to remove the
929 nonexistent label.
931 Returns
932 -------
933 pipeline : `PipelineIR`
934 A new pipelineIR object that is a subset of the old pipelineIR.
936 Raises
937 ------
938 ValueError
939 Raised if there is an issue with specified labels.
941 Notes
942 -----
943 This method attempts to prune any contracts that contain labels which
944 are not in the declared subset of labels. This pruning is done using a
945 string based matching due to the nature of contracts and may prune more
946 than it should.
947 """
948 pipeline = copy.deepcopy(self)
950 # update the label specifier to expand any named subsets
951 toRemove = set()
952 toAdd = set()
953 for label in labelSpecifier:
954 if label in pipeline.labeled_subsets:
955 toRemove.add(label)
956 toAdd.update(pipeline.labeled_subsets[label].subset)
957 labelSpecifier.difference_update(toRemove)
958 labelSpecifier.update(toAdd)
959 # verify all the labels are in the pipeline
960 if not labelSpecifier.issubset(pipeline.tasks.keys() | pipeline.labeled_subsets):
961 difference = labelSpecifier.difference(pipeline.tasks.keys())
962 raise ValueError(
963 "Not all supplied labels (specified or named subsets) are in the pipeline "
964 f"definition, extra labels: {difference}"
965 )
966 # copy needed so as to not modify while iterating
967 pipeline_labels = set(pipeline.tasks.keys())
968 # Remove the labels from the pipelineIR, and any contracts that contain
969 # those labels (see docstring on _remove_contracts for why this may
970 # cause issues)
971 for label in pipeline_labels:
972 if label not in labelSpecifier:
973 pipeline.tasks.pop(label)
974 pipeline._remove_contracts(label)
976 # create a copy of the object to iterate over
977 labeled_subsets = copy.copy(pipeline.labeled_subsets)
978 # remove or edit any labeled subsets that no longer have a complete set
979 for label, labeled_subset in labeled_subsets.items():
980 if extraTaskLabels := (labeled_subset.subset - pipeline.tasks.keys()):
981 match subsetCtrl:
982 case PipelineSubsetCtrl.DROP:
983 del pipeline.labeled_subsets[label]
984 case PipelineSubsetCtrl.EDIT:
985 for extra in extraTaskLabels:
986 labeled_subset.subset.discard(extra)
987 elif subsetCtrl is PipelineSubsetCtrl.DROP and not labeled_subset.subset:
988 # When mode is DROP, also drop any subsets that were already
989 # empty. This ensures we drop steps that were emptied-out by
990 # (earlier) imports with exclude in EDIT mode. Note that we
991 # don't want to drop those steps when they're first excluded
992 # down to nothing, because the pipeline might be about to add
993 # new tasks back into them, and then we'd want to preserve the
994 # step definitions.
995 del pipeline.labeled_subsets[label]
997 # remove any steps that correspond to removed subsets
998 new_steps = []
999 for step in pipeline.steps:
1000 if step.label not in pipeline.labeled_subsets:
1001 continue
1002 new_steps.append(step)
1003 pipeline.steps = new_steps
1005 return pipeline
1007 @classmethod
1008 def from_string(cls, pipeline_string: str) -> PipelineIR:
1009 """Create a `PipelineIR` object from a string formatted like a pipeline
1010 document.
1012 Parameters
1013 ----------
1014 pipeline_string : `str`
1015 A string that is formatted according like a pipeline document.
1016 """
1017 loaded_yaml = yaml.load(pipeline_string, Loader=PipelineYamlLoader)
1018 return cls(loaded_yaml)
1020 @classmethod
1021 def from_uri(cls, uri: ResourcePathExpression) -> PipelineIR:
1022 """Create a `PipelineIR` object from the document specified by the
1023 input uri.
1025 Parameters
1026 ----------
1027 uri : convertible to `~lsst.resources.ResourcePath`
1028 Location of document to use in creating a `PipelineIR` object.
1030 Returns
1031 -------
1032 pipelineIR : `PipelineIR`
1033 The loaded pipeline.
1034 """
1035 loaded_uri = ResourcePath(uri)
1036 with loaded_uri.open("r") as buffer:
1037 loaded_yaml = yaml.load(buffer, Loader=PipelineYamlLoader)
1038 return cls(loaded_yaml)
1040 def write_to_uri(self, uri: ResourcePathExpression) -> None:
1041 """Serialize this `PipelineIR` object into a yaml formatted string and
1042 write the output to a file at the specified uri.
1044 Parameters
1045 ----------
1046 uri : convertible to `~lsst.resources.ResourcePath`
1047 Location of document to write a `PipelineIR` object.
1048 """
1049 with ResourcePath(uri).open("w") as buffer:
1050 yaml.dump(self.to_primitives(), buffer, sort_keys=False, Dumper=MultilineStringDumper)
1052 def to_primitives(self) -> dict[str, Any]:
1053 """Convert to a representation used in yaml serialization.
1055 Returns
1056 -------
1057 primitives : `dict`
1058 Dictionary that maps directly to the serialized YAML form.
1059 """
1060 accumulate = {"description": self.description}
1061 if self.instrument is not None:
1062 accumulate["instrument"] = self.instrument
1063 if self.parameters:
1064 accumulate["parameters"] = self.parameters.to_primitives()
1065 accumulate["tasks"] = {m: t.to_primitives() for m, t in self.tasks.items()}
1066 if len(self.contracts) > 0:
1067 # sort contracts lexicographical order by the contract string in
1068 # absence of any other ordering principle
1069 contracts_list = [c.to_primitives() for c in self.contracts]
1070 contracts_list.sort(key=lambda x: x["contract"])
1071 accumulate["contracts"] = contracts_list
1072 if self.labeled_subsets:
1073 accumulate["subsets"] = {k: v.to_primitives() for k, v in self.labeled_subsets.items()}
1074 return accumulate
1076 def __str__(self) -> str:
1077 """Instance formatting as how it would look in yaml representation"""
1078 return yaml.dump(self.to_primitives(), sort_keys=False, Dumper=MultilineStringDumper)
1080 def __repr__(self) -> str:
1081 """Instance formatting as how it would look in yaml representation"""
1082 return str(self)
1084 def __eq__(self, other: object) -> bool:
1085 if not isinstance(other, PipelineIR):
1086 return False
1087 # special case contracts because it is a list, but order is not
1088 # important
1089 return (
1090 all(
1091 getattr(self, attr) == getattr(other, attr)
1092 for attr in ("tasks", "instrument", "labeled_subsets", "parameters")
1093 )
1094 and len(self.contracts) == len(other.contracts)
1095 and all(c in self.contracts for c in other.contracts)
1096 )