Coverage for python / lsst / pipe / base / pipelineIR.py: 18%
474 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/>.
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, cast
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 rename: dict[str, str] = field(default_factory=dict)
465 """dict of tasks to rename, keyed by old name with new name value."""
466 importContracts: bool = True
467 """Boolean attribute to dictate if contracts should be inherited with the
468 pipeline or not.
469 """
470 importSteps: bool = True
471 """Boolean attribute to dictate if steps should be inherited with the
472 pipeline or not.
473 """
474 labeledSubsetModifyMode: PipelineSubsetCtrl = PipelineSubsetCtrl.DROP
475 """Controls how labeled subsets are handled when an import ends up not
476 including (either through an include or exclusion list) a task label that
477 is defined in the `Pipeline` being imported. DROP will remove any
478 subsets which contain a missing label. EDIT will change any subsets to not
479 include the missing label.
480 """
481 instrument: Literal[_Tags.KeepInstrument] | str | None = _Tags.KeepInstrument
482 """Instrument to assign to the Pipeline at import. The default value of
483 `_Tags.KeepInstrument`` indicates that whatever instrument the pipeline is
484 declared with will not be modified. setting this value to None will drop
485 any declared instrument prior to import.
486 """
488 def toPipelineIR(self) -> PipelineIR:
489 """Load in the Pipeline specified by this object, and turn it into a
490 PipelineIR instance.
492 Returns
493 -------
494 pipeline : `PipelineIR`
495 A pipeline generated from the imported pipeline file.
496 """
497 if self.include and self.exclude:
498 raise ValueError(
499 "An include list and an exclude list cannot both be specified"
500 " when declaring a pipeline import."
501 )
502 if rename_keys := self.rename.keys():
503 rename_values_set = set(self.rename.values())
504 if len(rename_values_set) != len(rename_keys):
505 raise ValueError(f"rename {rename_keys=} must not have duplicates")
506 if rename_values_set.intersection(rename_keys):
507 raise ValueError(
508 f"rename keys={rename_keys} must not intersect with values={self.rename.values()}"
509 )
511 tmp_pipeline = PipelineIR.from_uri(os.path.expandvars(self.location))
512 if self.instrument is not _Tags.KeepInstrument:
513 tmp_pipeline.instrument = self.instrument
515 included_labels = set()
516 renamed_tasks = {}
517 for label in tmp_pipeline.tasks:
518 is_included = self.include and label in self.include
519 if (
520 is_included
521 or (self.exclude and label not in self.exclude)
522 or (self.include is None and self.exclude is None)
523 ):
524 if (label_new := self.rename.get(label)) is not None:
525 renamed_tasks[label] = label_new
526 if is_included:
527 self.include = [
528 label_new if (x == label) else label for x in cast(list[str], self.include)
529 ]
530 else:
531 label_new = label
532 included_labels.add(label_new)
534 rename_errors = []
535 for label, label_new in renamed_tasks.items():
536 if label_new in tmp_pipeline.tasks:
537 rename_errors.append(f"Can't rename {label=} to existing {label_new=}")
538 else:
539 task = tmp_pipeline.tasks.pop(label)
540 task.label = label_new
541 tmp_pipeline.tasks[label_new] = task
543 if rename_errors:
544 raise ValueError("; ".join(rename_errors))
546 # Handle labeled subsets being specified in the include or exclude
547 # list, adding or removing labels.
548 if self.include is not None:
549 subsets_in_include = tmp_pipeline.labeled_subsets.keys() & self.include
550 for label in subsets_in_include:
551 included_labels.update(tmp_pipeline.labeled_subsets[label].subset)
553 elif self.exclude is not None:
554 subsets_in_exclude = tmp_pipeline.labeled_subsets.keys() & self.exclude
555 for label in subsets_in_exclude:
556 included_labels.difference_update(tmp_pipeline.labeled_subsets[label].subset)
558 if not self.importSteps:
559 tmp_pipeline.steps = []
561 tmp_pipeline = tmp_pipeline.subset_from_labels(included_labels, self.labeledSubsetModifyMode)
563 if not self.importContracts:
564 tmp_pipeline.contracts = []
566 return tmp_pipeline
568 def __eq__(self, other: object) -> bool:
569 if not isinstance(other, ImportIR):
570 return False
571 return all(
572 getattr(self, attr) == getattr(other, attr)
573 for attr in ("location", "include", "exclude", "importContracts")
574 )
577@dataclass
578class StepIR:
579 """Intermediate representation of a step definition."""
581 label: str
582 """The label associated with this step."""
584 dimensions: list[str]
585 """The dimensions to use when sharding this step."""
588class PipelineIR:
589 """Intermediate representation of a pipeline definition.
591 Parameters
592 ----------
593 loaded_yaml : `dict`
594 A dictionary which matches the structure that would be produced by a
595 yaml reader which parses a pipeline definition document.
597 Raises
598 ------
599 ValueError
600 Raised if:
602 - a pipeline is declared without a description;
603 - no tasks are declared in a pipeline, and no pipelines are to be
604 inherited;
605 - more than one instrument is specified;
606 - more than one inherited pipeline share a label.
607 """
609 def __init__(self, loaded_yaml: dict[str, Any]):
610 # Check required fields are present
611 if "description" not in loaded_yaml:
612 raise ValueError("A pipeline must be declared with a description")
613 if "tasks" not in loaded_yaml and len({"imports", "inherits"} - loaded_yaml.keys()) == 2:
614 raise ValueError("A pipeline must be declared with one or more tasks")
616 # These steps below must happen in this call order
618 # Process pipeline description
619 self.description = loaded_yaml.pop("description")
621 # Process tasks
622 self._read_tasks(loaded_yaml)
624 # Process instrument keys
625 inst = loaded_yaml.pop("instrument", None)
626 if isinstance(inst, list):
627 raise ValueError("Only one top level instrument can be defined in a pipeline")
628 self.instrument: str | None = inst
630 # Process any contracts
631 self._read_contracts(loaded_yaml)
633 # Process any defined parameters
634 self._read_parameters(loaded_yaml)
636 # Process any named label subsets
637 self._read_labeled_subsets(loaded_yaml)
639 # Process defined sets
640 self._read_step_declaration(loaded_yaml)
642 # Process any inherited pipelines
643 self._read_imports(loaded_yaml)
645 # verify named subsets, must be done after inheriting
646 self._verify_labeled_subsets()
648 # verify steps, must be done after inheriting
649 self._verify_steps()
651 def _read_contracts(self, loaded_yaml: dict[str, Any]) -> None:
652 """Process the contracts portion of the loaded yaml document
654 Parameters
655 ----------
656 loaded_yaml : `dict`
657 A dictionary which matches the structure that would be produced by
658 a yaml reader which parses a pipeline definition document
659 """
660 loaded_contracts = loaded_yaml.pop("contracts", [])
661 if isinstance(loaded_contracts, str):
662 loaded_contracts = [loaded_contracts]
663 self.contracts: list[ContractIR] = []
664 for contract in loaded_contracts:
665 if isinstance(contract, dict):
666 self.contracts.append(ContractIR(**contract))
667 if isinstance(contract, str):
668 self.contracts.append(ContractIR(contract=contract))
670 def _read_parameters(self, loaded_yaml: dict[str, Any]) -> None:
671 """Process the parameters portion of the loaded yaml document
673 Parameters
674 ----------
675 loaded_yaml : `dict`
676 A dictionary which matches the structure that would be produced by
677 a yaml reader which parses a pipeline definition document
678 """
679 loaded_parameters = loaded_yaml.pop("parameters", {})
680 if not isinstance(loaded_parameters, dict):
681 raise ValueError("The parameters section must be a yaml mapping")
682 self.parameters = ParametersIR(loaded_parameters)
684 def _read_labeled_subsets(self, loaded_yaml: dict[str, Any]) -> None:
685 """Process the subsets portion of the loaded yaml document
687 Parameters
688 ----------
689 loaded_yaml : `MutableMapping`
690 A dictionary which matches the structure that would be produced
691 by a yaml reader which parses a pipeline definition document
692 """
693 loaded_subsets = loaded_yaml.pop("subsets", {})
694 self.labeled_subsets: dict[str, LabeledSubset] = {}
695 if not loaded_subsets and "subset" in loaded_yaml:
696 raise ValueError("Top level key should be subsets and not subset, add an s")
697 for key, value in loaded_subsets.items():
698 self.labeled_subsets[key] = LabeledSubset.from_primitives(key, value)
700 def _read_step_declaration(self, loaded_yaml: dict[str, Any]) -> None:
701 """Process the steps portion of the loaded yaml document
703 Steps are subsets that are declared to be normal parts of the overall
704 processing of the pipeline. Not all subsets need to be a step, as they
705 can exist for certain targeted processing, such as debugging.
707 Parameters
708 ----------
709 loaded_yaml: `dict`
710 A dictionary which matches the structure that would be produced
711 by a yaml reader which parses a pipeline definition document
712 """
713 loaded_steps = loaded_yaml.pop("steps", [])
714 temp_steps: dict[str, StepIR] = {}
715 for declaration in loaded_steps:
716 new_step = StepIR(**declaration)
717 existing = temp_steps.setdefault(new_step.label, new_step)
718 if existing is not new_step:
719 raise ValueError(f"Step {existing.label} was declared twice.")
720 self.steps = [step for step in temp_steps.values()]
722 def _verify_labeled_subsets(self) -> None:
723 """Verify that all the labels in each named subset exist within the
724 pipeline.
725 """
726 # Verify that all labels defined in a labeled subset are in the
727 # Pipeline
728 for labeled_subset in self.labeled_subsets.values():
729 if not labeled_subset.subset.issubset(self.tasks.keys()):
730 raise ValueError(
731 f"Labels {labeled_subset.subset - self.tasks.keys()} were not found in the "
732 "declared pipeline"
733 )
734 # Verify subset labels are not already task labels
735 label_intersection = self.labeled_subsets.keys() & self.tasks.keys()
736 if label_intersection:
737 raise ValueError(f"Labeled subsets can not use the same label as a task: {label_intersection}")
739 def _verify_steps(self) -> None:
740 """Verify that all step definitions have a corresponding labeled
741 subset.
742 """
743 for step in self.steps:
744 if step.label not in self.labeled_subsets:
745 raise ValueError(
746 f"{step.label} was declared to be a step, but was not declared to be a labeled subset"
747 )
749 def _read_imports(self, loaded_yaml: dict[str, Any]) -> None:
750 """Process the inherits portion of the loaded yaml document
752 Parameters
753 ----------
754 loaded_yaml : `dict`
755 A dictionary which matches the structure that would be produced by
756 a yaml reader which parses a pipeline definition document
757 """
759 def process_args(argument: str | dict) -> dict:
760 if isinstance(argument, str):
761 return {"location": argument}
762 elif isinstance(argument, dict):
763 if "exclude" in argument and isinstance(argument["exclude"], str):
764 argument["exclude"] = [argument["exclude"]]
765 if "include" in argument and isinstance(argument["include"], str):
766 argument["include"] = [argument["include"]]
767 if "instrument" in argument and argument["instrument"] == "None":
768 argument["instrument"] = None
769 if "labeledSubsetModifyMode" in argument:
770 match argument["labeledSubsetModifyMode"]:
771 case "DROP":
772 argument["labeledSubsetModifyMode"] = PipelineSubsetCtrl.DROP
773 case "EDIT":
774 argument["labeledSubsetModifyMode"] = PipelineSubsetCtrl.EDIT
775 case unknown:
776 raise ValueError(f"{unknown} is not a valid mode for labeledSubsetModifyMode")
777 return argument
779 if not {"inherits", "imports"} - loaded_yaml.keys():
780 raise ValueError("Cannot define both inherits and imports sections, use imports")
781 tmp_import = loaded_yaml.pop("inherits", None)
782 if tmp_import is None:
783 tmp_import = loaded_yaml.pop("imports", None)
784 else:
785 raise ValueError("The 'inherits' key is not supported. Please use the key 'imports' instead")
786 if tmp_import is None:
787 self.imports: list[ImportIR] = []
788 elif isinstance(tmp_import, list):
789 self.imports = [ImportIR(**process_args(args)) for args in tmp_import]
790 else:
791 self.imports = [ImportIR(**process_args(tmp_import))]
793 self.merge_pipelines([fragment.toPipelineIR() for fragment in self.imports])
795 def merge_pipelines(self, pipelines: Iterable[PipelineIR]) -> None:
796 """Merge one or more other `PipelineIR` objects into this object.
798 Parameters
799 ----------
800 pipelines : `~collections.abc.Iterable` of `PipelineIR` objects
801 An `~collections.abc.Iterable` that contains one or more
802 `PipelineIR` objects to merge into this object.
804 Raises
805 ------
806 ValueError
807 Raised if there is a conflict in instrument specifications.
808 Raised if a task label appears in more than one of the input
809 `PipelineIR` objects which are to be merged.
810 Raised if a labeled subset appears in more than one of the input
811 `PipelineIR` objects which are to be merged, and with any subset
812 existing in this object.
813 """
814 # integrate any imported pipelines
815 accumulate_tasks: dict[str, TaskIR] = {}
816 accumulate_labeled_subsets: dict[str, LabeledSubset] = {}
817 accumulated_parameters = ParametersIR({})
818 accumulated_steps: dict[str, StepIR] = {}
820 for tmp_IR in pipelines:
821 if self.instrument is None:
822 self.instrument = tmp_IR.instrument
823 elif self.instrument != tmp_IR.instrument and tmp_IR.instrument is not None:
824 msg = (
825 "Only one instrument can be declared in a pipeline or its imports. "
826 f"Top level pipeline defines {self.instrument} but pipeline to merge "
827 f"defines {tmp_IR.instrument}."
828 )
829 raise ValueError(msg)
830 if duplicate_labels := accumulate_tasks.keys() & tmp_IR.tasks.keys():
831 msg = (
832 "Task labels in the imported pipelines must be unique. "
833 f"These labels appear multiple times: {duplicate_labels}"
834 )
835 raise ValueError(msg)
836 accumulate_tasks.update(tmp_IR.tasks)
837 self.contracts.extend(tmp_IR.contracts)
838 # verify that tmp_IR has unique labels for named subset among
839 # existing labeled subsets, and with existing task labels.
840 overlapping_subsets = accumulate_labeled_subsets.keys() & tmp_IR.labeled_subsets.keys()
841 task_subset_overlap = (
842 accumulate_labeled_subsets.keys() | tmp_IR.labeled_subsets.keys()
843 ) & accumulate_tasks.keys()
844 if overlapping_subsets or task_subset_overlap:
845 raise ValueError(
846 "Labeled subset names must be unique amongst imports in both labels and "
847 f" named Subsets. Duplicate: {overlapping_subsets | task_subset_overlap}"
848 )
849 accumulate_labeled_subsets.update(tmp_IR.labeled_subsets)
850 accumulated_parameters.update(tmp_IR.parameters)
851 for tmp_step in tmp_IR.steps:
852 existing = accumulated_steps.setdefault(tmp_step.label, tmp_step)
853 if existing != tmp_step:
854 raise ValueError(
855 f"There were conflicting step definitions in import {tmp_step}, {existing}"
856 )
858 for tmp_step in self.steps:
859 existing = accumulated_steps.setdefault(tmp_step.label, tmp_step)
860 if existing != tmp_step:
861 raise ValueError(f"There were conflicting step definitions in import {tmp_step}, {existing}")
863 # verify that any accumulated labeled subsets dont clash with a label
864 # from this pipeline
865 if accumulate_labeled_subsets.keys() & self.tasks.keys():
866 raise ValueError(
867 "Labeled subset names must be unique amongst imports in both labels and named Subsets"
868 )
869 # merge in the named subsets for self so this document can override any
870 # that have been declared
871 accumulate_labeled_subsets.update(self.labeled_subsets)
872 self.labeled_subsets = accumulate_labeled_subsets
874 # merge the dict of label:TaskIR objects, preserving any configs in the
875 # imported pipeline if the labels point to the same class
876 for label, task in self.tasks.items():
877 if label not in accumulate_tasks:
878 accumulate_tasks[label] = task
879 elif accumulate_tasks[label].klass == task.klass:
880 if task.config is not None:
881 for config in task.config:
882 accumulate_tasks[label].add_or_update_config(config)
883 else:
884 accumulate_tasks[label] = task
885 self.tasks: dict[str, TaskIR] = accumulate_tasks
886 accumulated_parameters.update(self.parameters)
887 self.parameters = accumulated_parameters
888 self.steps = list(accumulated_steps.values())
890 def _read_tasks(self, loaded_yaml: dict[str, Any]) -> None:
891 """Process the tasks portion of the loaded yaml document
893 Parameters
894 ----------
895 loaded_yaml : `dict`
896 A dictionary which matches the structure that would be produced by
897 a yaml reader which parses a pipeline definition document
898 """
899 self.tasks = {}
900 tmp_tasks = loaded_yaml.pop("tasks", None)
901 if tmp_tasks is None:
902 tmp_tasks = {}
904 if "parameters" in tmp_tasks:
905 raise ValueError("parameters is a reserved word and cannot be used as a task label")
907 for label, definition in tmp_tasks.items():
908 if isinstance(definition, str):
909 definition = {"class": definition}
910 config = definition.get("config", None)
911 if config is None:
912 task_config_ir = None
913 else:
914 if isinstance(config, dict):
915 config = [config]
916 task_config_ir = []
917 for c in config:
918 file = c.pop("file", None)
919 if file is None:
920 file = []
921 elif not isinstance(file, list):
922 file = [file]
923 task_config_ir.append(
924 ConfigIR(
925 python=c.pop("python", None), dataId=c.pop("dataId", None), file=file, rest=c
926 )
927 )
928 self.tasks[label] = TaskIR(label, definition["class"], task_config_ir)
930 def _remove_contracts(self, label: str) -> None:
931 """Remove any contracts that contain the given label
933 String comparison used in this way is not the most elegant and may
934 have issues, but it is the only feasible way when users can specify
935 contracts with generic strings.
936 """
937 new_contracts = []
938 for contract in self.contracts:
939 # match a label that is not preceded by an ASCII identifier, or
940 # is the start of a line and is followed by a dot
941 if re.match(f".*([^A-Za-z0-9_]|^){label}[.]", contract.contract):
942 continue
943 new_contracts.append(contract)
944 self.contracts = new_contracts
946 def subset_from_labels(
947 self, labelSpecifier: set[str], subsetCtrl: PipelineSubsetCtrl = PipelineSubsetCtrl.DROP
948 ) -> PipelineIR:
949 """Subset a pipelineIR to contain only labels specified in
950 labelSpecifier.
952 Parameters
953 ----------
954 labelSpecifier : `set` of `str`
955 Set containing labels that describes how to subset a pipeline.
956 subsetCtrl : `PipelineSubsetCtrl`
957 Control object which decides how subsets with missing labels are
958 handled. Setting to `PipelineSubsetCtrl.DROP` (the default) will
959 cause any subsets that have labels which are not in the set of all
960 task labels to be dropped. Setting to `PipelineSubsetCtrl.EDIT`
961 will cause the subset to instead be edited to remove the
962 nonexistent label.
964 Returns
965 -------
966 pipeline : `PipelineIR`
967 A new pipelineIR object that is a subset of the old pipelineIR.
969 Raises
970 ------
971 ValueError
972 Raised if there is an issue with specified labels.
974 Notes
975 -----
976 This method attempts to prune any contracts that contain labels which
977 are not in the declared subset of labels. This pruning is done using a
978 string based matching due to the nature of contracts and may prune more
979 than it should.
980 """
981 pipeline = copy.deepcopy(self)
983 # update the label specifier to expand any named subsets
984 toRemove = set()
985 toAdd = set()
986 for label in labelSpecifier:
987 if label in pipeline.labeled_subsets:
988 toRemove.add(label)
989 toAdd.update(pipeline.labeled_subsets[label].subset)
990 labelSpecifier.difference_update(toRemove)
991 labelSpecifier.update(toAdd)
992 # verify all the labels are in the pipeline
993 if not labelSpecifier.issubset(pipeline.tasks.keys() | pipeline.labeled_subsets):
994 difference = labelSpecifier.difference(pipeline.tasks.keys())
995 raise ValueError(
996 "Not all supplied labels (specified or named subsets) are in the pipeline "
997 f"definition, extra labels: {difference}"
998 )
999 # copy needed so as to not modify while iterating
1000 pipeline_labels = set(pipeline.tasks.keys())
1001 # Remove the labels from the pipelineIR, and any contracts that contain
1002 # those labels (see docstring on _remove_contracts for why this may
1003 # cause issues)
1004 for label in pipeline_labels:
1005 if label not in labelSpecifier:
1006 pipeline.tasks.pop(label)
1007 pipeline._remove_contracts(label)
1009 # create a copy of the object to iterate over
1010 labeled_subsets = copy.copy(pipeline.labeled_subsets)
1011 # remove or edit any labeled subsets that no longer have a complete set
1012 for label, labeled_subset in labeled_subsets.items():
1013 if extraTaskLabels := (labeled_subset.subset - pipeline.tasks.keys()):
1014 match subsetCtrl:
1015 case PipelineSubsetCtrl.DROP:
1016 del pipeline.labeled_subsets[label]
1017 case PipelineSubsetCtrl.EDIT:
1018 for extra in extraTaskLabels:
1019 labeled_subset.subset.discard(extra)
1020 elif subsetCtrl is PipelineSubsetCtrl.DROP and not labeled_subset.subset:
1021 # When mode is DROP, also drop any subsets that were already
1022 # empty. This ensures we drop steps that were emptied-out by
1023 # (earlier) imports with exclude in EDIT mode. Note that we
1024 # don't want to drop those steps when they're first excluded
1025 # down to nothing, because the pipeline might be about to add
1026 # new tasks back into them, and then we'd want to preserve the
1027 # step definitions.
1028 del pipeline.labeled_subsets[label]
1030 # remove any steps that correspond to removed subsets
1031 new_steps = []
1032 for step in pipeline.steps:
1033 if step.label not in pipeline.labeled_subsets:
1034 continue
1035 new_steps.append(step)
1036 pipeline.steps = new_steps
1038 return pipeline
1040 @classmethod
1041 def from_string(cls, pipeline_string: str) -> PipelineIR:
1042 """Create a `PipelineIR` object from a string formatted like a pipeline
1043 document.
1045 Parameters
1046 ----------
1047 pipeline_string : `str`
1048 A string that is formatted according like a pipeline document.
1049 """
1050 loaded_yaml = yaml.load(pipeline_string, Loader=PipelineYamlLoader)
1051 return cls(loaded_yaml)
1053 @classmethod
1054 def from_uri(cls, uri: ResourcePathExpression) -> PipelineIR:
1055 """Create a `PipelineIR` object from the document specified by the
1056 input uri.
1058 Parameters
1059 ----------
1060 uri : convertible to `~lsst.resources.ResourcePath`
1061 Location of document to use in creating a `PipelineIR` object.
1063 Returns
1064 -------
1065 pipelineIR : `PipelineIR`
1066 The loaded pipeline.
1067 """
1068 loaded_uri = ResourcePath(uri)
1069 with loaded_uri.open("r") as buffer:
1070 loaded_yaml = yaml.load(buffer, Loader=PipelineYamlLoader)
1071 return cls(loaded_yaml)
1073 def write_to_uri(self, uri: ResourcePathExpression) -> None:
1074 """Serialize this `PipelineIR` object into a yaml formatted string and
1075 write the output to a file at the specified uri.
1077 Parameters
1078 ----------
1079 uri : convertible to `~lsst.resources.ResourcePath`
1080 Location of document to write a `PipelineIR` object.
1081 """
1082 with ResourcePath(uri).open("w") as buffer:
1083 yaml.dump(self.to_primitives(), buffer, sort_keys=False, Dumper=MultilineStringDumper)
1085 def to_primitives(self) -> dict[str, Any]:
1086 """Convert to a representation used in yaml serialization.
1088 Returns
1089 -------
1090 primitives : `dict`
1091 Dictionary that maps directly to the serialized YAML form.
1092 """
1093 accumulate = {"description": self.description}
1094 if self.instrument is not None:
1095 accumulate["instrument"] = self.instrument
1096 if self.parameters:
1097 accumulate["parameters"] = self.parameters.to_primitives()
1098 accumulate["tasks"] = {m: t.to_primitives() for m, t in self.tasks.items()}
1099 if len(self.contracts) > 0:
1100 # sort contracts lexicographical order by the contract string in
1101 # absence of any other ordering principle
1102 contracts_list = [c.to_primitives() for c in self.contracts]
1103 contracts_list.sort(key=lambda x: x["contract"])
1104 accumulate["contracts"] = contracts_list
1105 if self.labeled_subsets:
1106 accumulate["subsets"] = {k: v.to_primitives() for k, v in self.labeled_subsets.items()}
1107 return accumulate
1109 def __str__(self) -> str:
1110 """Instance formatting as how it would look in yaml representation"""
1111 return yaml.dump(self.to_primitives(), sort_keys=False, Dumper=MultilineStringDumper)
1113 def __repr__(self) -> str:
1114 """Instance formatting as how it would look in yaml representation"""
1115 return str(self)
1117 def __eq__(self, other: object) -> bool:
1118 if not isinstance(other, PipelineIR):
1119 return False
1120 # special case contracts because it is a list, but order is not
1121 # important
1122 return (
1123 all(
1124 getattr(self, attr) == getattr(other, attr)
1125 for attr in ("tasks", "instrument", "labeled_subsets", "parameters")
1126 )
1127 and len(self.contracts) == len(other.contracts)
1128 and all(c in self.contracts for c in other.contracts)
1129 )