Coverage for python/lsst/source/injection/utils/_make_injection_pipeline.py: 65%
205 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-07 02:08 -0700
1# This file is part of source_injection.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://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 program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <https://www.gnu.org/licenses/>.
22from __future__ import annotations
24__all__ = ["make_injection_pipeline"]
26import logging
27import warnings
29from lsst.pipe.base import LabelSpecifier, Pipeline, PipelineGraph
30from lsst.pipe.base.pipelineIR import ContractError
33def _infer_injection_pipeline(dataset_type_name: str, logger: logging.Logger) -> str | None:
34 """Infer the injection pipeline from the dataset type name.
36 Parameters
37 ----------
38 dataset_type_name : `str`
39 Name of the dataset type being injected into.
40 logger : `~logging.Logger`
41 Logger for warning and info messages.
43 Returns
44 -------
45 injection_pipeline : `str` | `None`
46 Location of an injection pipeline definition YAML file stub, or None if
47 no suitable injection pipeline could be inferred.
48 """
49 injection_pipeline = None
50 match dataset_type_name:
51 case "postISRCCD" | "post_isr_image":
52 injection_pipeline = "$SOURCE_INJECTION_DIR/pipelines/inject_exposure.yaml"
53 case "icExp" | "calexp" | "initial_pvi" | "pvi" | "preliminary_visit_image" | "visit_image":
54 injection_pipeline = "$SOURCE_INJECTION_DIR/pipelines/inject_visit.yaml"
55 case (
56 "deepCoadd"
57 | "deepCoadd_calexp"
58 | "goodSeeingCoadd"
59 | "deep_coadd_predetection"
60 | "deep_coadd"
61 | "deep_coadd_cell_predetection"
62 | "template_coadd"
63 ):
64 injection_pipeline = "$SOURCE_INJECTION_DIR/pipelines/inject_coadd.yaml"
65 case _:
66 # Print a warning rather than a raise, as the user may wish to
67 # edit connection names without merging an injection pipeline.
68 logger.warning(
69 "Unable to infer injection pipeline stub from dataset type name '%s' and none was "
70 "provided. No injection pipeline will be merged into the output pipeline.",
71 dataset_type_name,
72 )
73 if injection_pipeline:
74 logger.info(
75 "Injected dataset type '%s' used to infer injection pipeline: %s",
76 dataset_type_name,
77 injection_pipeline,
78 )
79 return injection_pipeline
82def _merge_injection_pipeline(
83 pipeline: Pipeline,
84 injection_pipeline: Pipeline | str | None,
85 dataset_type_name: str,
86 prefix: str,
87) -> None | str:
88 """Merge an injection pipeline into an existing pipeline.
90 Parameters
91 ----------
92 pipeline : `~lsst.pipe.base.Pipeline`
93 Pipeline to merge the injection pipeline into.
94 injection_pipeline : `~lsst.pipe.base.Pipeline` | `str` | `None`
95 Injection pipeline to merge, or location of an injection pipeline
96 definition YAML file stub. If None, no injection pipeline is merged.
97 dataset_type_name : `str`
98 Name of the dataset type being injected into.
99 prefix : `str`
100 Prefix to prepend to each affected post-injection dataset type name.
102 Returns
103 -------
104 injection_task_label : `str` | `None`
105 Label of the injection task, if an injection pipeline was merged, or
106 None if no injection pipeline was merged.
108 Notes
109 -----
110 This function modifies the input pipeline in place.
111 """
112 if injection_pipeline is None: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 return None
114 if isinstance(injection_pipeline, str): 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 injection_pipeline = Pipeline.fromFile(injection_pipeline)
116 if len(injection_pipeline) != 1: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 raise RuntimeError(
118 f"The injection pipeline contains {len(injection_pipeline)} tasks; only 1 task is allowed."
119 )
120 pipeline.mergePipeline(injection_pipeline)
122 injection_task_label = next(iter(injection_pipeline.task_labels))
123 pipeline.addConfigOverride(injection_task_label, "connections.input_exposure", dataset_type_name)
124 pipeline.addConfigOverride(
125 injection_task_label, "connections.output_exposure", prefix + dataset_type_name
126 )
127 pipeline.addConfigOverride(
128 injection_task_label, "connections.output_catalog", prefix + dataset_type_name + "_catalog"
129 )
130 return injection_task_label
133def _parse_config_override(config_override: str) -> tuple[str, str, str]:
134 """Parse a config override string into a label, a key and a value.
136 Parameters
137 ----------
138 config_override : `str`
139 Config override string to parse.
141 Returns
142 -------
143 label : `str`
144 Label to override.
145 key : `str`
146 Key to override.
147 value : `str`
148 Value to override.
150 Raises
151 ------
152 TypeError
153 If the config override string cannot be parsed.
154 """
155 try:
156 label, keyvalue = config_override.split(":", 1)
157 except ValueError:
158 raise TypeError(
159 f"Unrecognized syntax for option 'config': '{config_override}' (does not match pattern "
160 "(?P<label>.+):(?P<value>.+=.+))"
161 ) from None
162 try:
163 key, value = keyvalue.split("=", 1)
164 except ValueError as e:
165 raise TypeError(
166 f"Could not parse key-value pair '{config_override}' using separator '=', with multiple values "
167 f"not allowed: {e}"
168 ) from None
169 return label, key, value
172def _configure_injection_pipeline(
173 pipeline: Pipeline,
174 config: str | list[str],
175 logger: logging.Logger,
176) -> None:
177 """Apply user-supplied config overrides to the pipeline.
179 Parameters
180 ----------
181 pipeline : `~lsst.pipe.base.Pipeline`
182 Pipeline to apply config overrides to. Pipeline is modified in place.
183 config : `str` | `list` [`str`]
184 Config override(s) to apply, in the format 'label:key=value'.
185 logger : `~logging.Logger`
186 Logger for warning and info messages.
188 Notes
189 -----
190 This function modifies the input pipeline in place.
191 """
192 if isinstance(config, str):
193 config = [config]
194 for conf in config:
195 config_label, config_key, config_value = _parse_config_override(conf)
196 try:
197 pipeline.addConfigOverride(config_label, config_key, config_value)
198 except LookupError:
199 logger.debug(
200 "Config override '%s' for label '%s' not found in the reference "
201 "pipeline, either due to a typo or the label not existing in "
202 "the reference pipeline.",
203 conf,
204 config_label,
205 )
208def _remove_excluded_tasks(
209 pipeline: Pipeline,
210 excluded_tasks: set[str] | str,
211 logger: logging.Logger,
212) -> Pipeline:
213 """Remove excluded tasks from the pipeline and any subsets,
214 and remove any empty subsets.
216 Parameters
217 ----------
218 pipeline : `~lsst.pipe.base.Pipeline`
219 Pipeline to remove tasks from. This pipeline is modified in place.
220 excluded_tasks : `set` [`str`] | `str`
221 Task labels to exclude from the injection pipeline.
222 logger : `~logging.Logger`
223 Logger for warning and info messages.
225 Returns
226 -------
227 pipeline : `~lsst.pipe.base.Pipeline`
228 The input pipeline with excluded tasks and empty subsets removed.
229 """
230 if isinstance(excluded_tasks, str): 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true
231 excluded_tasks = set(excluded_tasks.split(","))
232 all_tasks = set(pipeline.task_labels)
233 preserved_tasks = all_tasks - excluded_tasks
235 preserved_task_labels = LabelSpecifier(labels=preserved_tasks)
236 # EDIT mode removes tasks from parent subsets but keeps the subset itself.
237 pipeline = pipeline.subsetFromLabels(preserved_task_labels, pipeline.PipelineSubsetCtrl.EDIT)
239 if len(found_tasks := excluded_tasks & all_tasks) > 0: 239 ↛ 248line 239 didn't jump to line 248 because the condition on line 239 was always true
240 grammar = "task" if len(found_tasks) == 1 else "tasks"
241 logger.info(
242 "%d %s excluded from the output pipeline: %s",
243 len(found_tasks),
244 grammar,
245 ", ".join(sorted(found_tasks)),
246 )
248 removed_subsets = set()
249 for subset_label, subset_tasks in pipeline.subsets.items():
250 if not subset_tasks: 250 ↛ 251line 250 didn't jump to line 251 because the condition on line 250 was never true
251 removed_subsets.add(subset_label)
252 pipeline.removeLabeledSubset(subset_label)
253 if (removed_subsets_count := len(removed_subsets)) > 0: 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 grammar = "subset" if removed_subsets_count == 1 else "subsets"
255 logger.warning(
256 "Removed %d empty %s from the pipeline: %s.",
257 removed_subsets_count,
258 grammar,
259 ", ".join(sorted(removed_subsets)),
260 )
262 return pipeline
265def _get_pipeline_graph(pipeline: Pipeline, logger: logging.Logger) -> PipelineGraph:
266 """Get the pipeline graph, handling any contract errors.
268 Pipeline contracts that are violated by any modifications made to the
269 pipeline will be removed, with a warning logged for each contract that's
270 removed.
272 Parameters
273 ----------
274 pipeline : `~lsst.pipe.base.Pipeline`
275 Pipeline to validate contracts for.
276 logger : `~logging.Logger`
277 Logger for warning and info messages.
279 Returns
280 -------
281 pipeline_graph : `~lsst.pipe.base.PipelineGraph`
282 The pipeline graph for the input pipeline, with any violated contracts
283 removed from the input pipeline.
285 Notes
286 -----
287 This function modifies the input pipeline in place, removing any violated
288 contracts.
289 """
290 try:
291 with warnings.catch_warnings():
292 warnings.filterwarnings(
293 "ignore",
294 message=r".*formatted like a Pipeline parameter but was not found within the Pipeline.*",
295 category=UserWarning,
296 )
297 pipeline_graph = pipeline.to_graph()
298 except ContractError:
299 contracts_initial = pipeline._pipelineIR.contracts
300 pipeline._pipelineIR.contracts = []
301 contracts_passed = []
302 contracts_failed = []
304 for contract in contracts_initial:
305 pipeline._pipelineIR.contracts = [contract]
306 try:
307 _ = pipeline.to_graph()
308 except ContractError:
309 contracts_failed.append(contract)
310 continue
311 contracts_passed.append(contract)
313 pipeline._pipelineIR.contracts = contracts_passed
314 pipeline_graph = pipeline.to_graph()
316 if contracts_failed: 316 ↛ 321line 316 didn't jump to line 321 because the condition on line 316 was always true
317 logger.warning(
318 "The following contracts were violated and have been removed: \n%s",
319 "\n".join([str(contract) for contract in contracts_failed]),
320 )
321 return pipeline_graph
324def _collect_injected_task_labels(
325 pipeline_graph: PipelineGraph,
326 dataset_type_name: str,
327) -> set[str]:
328 """Collect tasks downstream of the injection point.
330 Parameters
331 ----------
332 pipeline_graph : `~lsst.pipe.base.PipelineGraph`
333 Pipeline graph to inspect.
334 dataset_type_name : `str`
335 Name of the dataset type being injected into.
337 Returns
338 -------
339 injected_task_labels : `set` [`str`]
340 Labels of all tasks that consume the injected dataset type directly or
341 indirectly, including the injection task itself if present.
342 """
343 injected_task_labels = set()
345 dataset_type_frontier = {dataset_type_name}
346 seen_dataset_types = set(dataset_type_frontier)
348 # Note: here we opt to walk the pipeline graph instead of using
349 # `pipeline_graph._xgraph.successors`. The `_xgraph` attribute is a private
350 # implementation detail and therefore not a guaranteed interface.
352 while dataset_type_frontier:
353 next_frontier = set()
354 for current_dataset_type in dataset_type_frontier:
355 for task_node in pipeline_graph.consumers_of(current_dataset_type):
356 if task_node.label in injected_task_labels: 356 ↛ 357line 356 didn't jump to line 357 because the condition on line 356 was never true
357 continue
359 injected_task_labels.add(task_node.label)
361 output_edges = task_node.iter_all_outputs()
363 for edge in output_edges:
364 output_dataset_type = edge.parent_dataset_type_name
365 if output_dataset_type not in seen_dataset_types: 365 ↛ 363line 365 didn't jump to line 363 because the condition on line 365 was always true
366 seen_dataset_types.add(output_dataset_type)
367 next_frontier.add(output_dataset_type)
368 dataset_type_frontier = next_frontier
370 return injected_task_labels
373def _add_injected_subsets(
374 pipeline: Pipeline,
375 injected_task_labels: set[str],
376 prefix: str,
377 logger: logging.Logger,
378) -> int:
379 """Create injected variants of existing subsets.
381 Parameters
382 ----------
383 pipeline : `~lsst.pipe.base.Pipeline`
384 Pipeline to modify in place.
385 injected_task_labels : `set` [`str`]
386 Labels of tasks downstream of the injection point.
387 prefix : `str`
388 Prefix to prepend to the subset names.
389 logger : `~logging.Logger`
390 Logger for warning and info messages.
392 Returns
393 -------
394 subset_count : `int`
395 Number of injected subsets created.
396 """
397 if not injected_task_labels: 397 ↛ 398line 397 didn't jump to line 398 because the condition on line 397 was never true
398 return 0
400 injected_label_specifier = LabelSpecifier(labels=injected_task_labels)
401 injected_pipeline = pipeline.subsetFromLabels(injected_label_specifier, pipeline.PipelineSubsetCtrl.EDIT)
403 injected_subset_labels = set()
404 for subset_label, subset_tasks in injected_pipeline.subsets.items():
405 if not subset_tasks: 405 ↛ 406line 405 didn't jump to line 406 because the condition on line 405 was never true
406 continue
407 injected_subset_label = prefix + subset_label
408 injected_subset_description = (
409 f"All tasks from the '{subset_label}' subset impacted by source injection."
410 )
412 # If the subset already exists, add any new tasks to the existing
413 # subset rather than creating a new subset with the same name.
414 # Used in pipelines with more than one injection tasks.
415 if injected_subset_label in pipeline.subsets.keys(): 415 ↛ 416line 415 didn't jump to line 416 because the condition on line 415 was never true
416 existing_subset_tasks = set(pipeline.subsets[injected_subset_label])
417 for injected_subset_task in set(subset_tasks) - existing_subset_tasks:
418 pipeline.addLabelToSubset(injected_subset_label, injected_subset_task)
419 else:
420 pipeline.addLabeledSubset(injected_subset_label, injected_subset_description, subset_tasks)
421 injected_subset_labels.add(injected_subset_label)
423 return len(injected_subset_labels)
426def _reconfigure_injection_pipeline(
427 pipeline: Pipeline,
428 dataset_type_name: str,
429 prefix: str,
430 injection_task_label: str | None,
431 update_subsets: bool,
432 logger: logging.Logger,
433) -> None:
434 """Reconfigure the injection pipeline by prefixing post-injection dataset
435 type names and updating subsets.
437 Parameters
438 ----------
439 pipeline : `~lsst.pipe.base.Pipeline`
440 Pipeline to configure. This pipeline is modified in place.
441 dataset_type_name : `str`
442 Name of the dataset type being injected into.
443 prefix : `str`
444 Prefix to prepend to each affected post-injection dataset type name.
445 injection_task_label : `str` | `None`
446 Label of the injection task.
447 update_subsets : `bool`
448 If True, update pipeline subsets to include the injection task.
449 logger : `~logging.Logger`
450 Logger for warning and info messages.
452 Notes
453 -----
454 This function modifies the input pipeline in place.
455 """
456 # Use pipeline graph to determine tasks with connections to be modified
457 pipeline_graph = _get_pipeline_graph(pipeline, logger)
458 injected_task_labels = _collect_injected_task_labels(pipeline_graph, dataset_type_name)
459 post_injection_tasks = pipeline_graph.consumers_of(dataset_type_name)
460 if len(post_injection_tasks) == 0: 460 ↛ 461line 460 didn't jump to line 461 because the condition on line 460 was never true
461 logger.warning(
462 "Dataset type '%s' not found in the reference pipeline; no input connection edits to be made.",
463 dataset_type_name,
464 )
465 if post_injection_tasks: 465 ↛ 468line 465 didn't jump to line 468 because the condition on line 465 was always true
466 post_injection_tasks = [task for task in post_injection_tasks if task.label != injection_task_label]
467 else:
468 post_injection_tasks = []
470 # Loop over each post injection task; prefix input connections only
471 for task_node in post_injection_tasks:
472 input_edges = task_node.iter_all_inputs()
474 for edge in input_edges:
475 if hasattr(task_node.config.connections.ConnectionsClass, edge.connection_name): 475 ↛ 474line 475 didn't jump to line 474 because the condition on line 475 was always true
476 if edge.parent_dataset_type_name == dataset_type_name: 476 ↛ 474line 476 didn't jump to line 474 because the condition on line 476 was always true
477 pipeline.addConfigOverride(
478 task_node.label,
479 "connections." + edge.connection_name,
480 prefix + edge.dataset_type_name,
481 )
483 # Update subsets to include the injection task
484 if ( 484 ↛ 493line 484 didn't jump to line 493 because the condition on line 484 was always true
485 update_subsets
486 and injection_task_label is not None
487 and (pre_injection_task := pipeline_graph.producer_of(dataset_type_name)) is not None
488 ):
489 precursor_subsets = pipeline.findSubsetsWithLabel(pre_injection_task.label)
490 for subset in precursor_subsets:
491 pipeline.addLabelToSubset(subset, injection_task_label)
493 injected_subset_count = 0
494 if update_subsets: 494 ↛ 497line 494 didn't jump to line 497 because the condition on line 494 was always true
495 injected_subset_count = _add_injected_subsets(pipeline, injected_task_labels, prefix, logger)
497 logger.info(
498 "Made an injection pipeline containing %d task%s and %d injected subset%s.",
499 len(pipeline),
500 "" if len(pipeline) == 1 else "s",
501 injected_subset_count,
502 "" if injected_subset_count == 1 else "s",
503 )
506def _add_additional_pipelines(
507 pipeline: Pipeline,
508 additional_pipelines: list[Pipeline] | list[str],
509 additional_subset: list[str] | str | None,
510 logger: logging.Logger,
511) -> None:
512 """Add additional pipelines to the injection pipeline, and optionally add
513 all additional tasks to existing or new subsets.
515 Parameters
516 ----------
517 pipeline : `~lsst.pipe.base.Pipeline`
518 Pipeline to add additional pipelines to. Pipeline is modified in place.
519 additional_pipelines : `list` [`~lsst.pipe.base.Pipeline`] | `list` [`str`]
520 Additional pipelines to merge, or locations of additional pipeline
521 definition YAML file stubs.
522 additional_subset : `list` [`str`] | `str` | `None`
523 A list of subset definitions in the form
524 "subset_name[:subset_description]".
525 These subsets will be created if they don't already exist. All tasks
526 from the additional pipelines will be added to these subsets.
527 If None, additional tasks will not be added to any subsets.
528 logger : `~logging.Logger`
529 Logger for warning and info messages.
531 Notes
532 -----
533 This function modifies the input pipeline in place.
534 """
535 # Merge all additional pipelines into the main pipeline
536 additional_tasks: set[str] = set()
537 for additional_pipeline in additional_pipelines:
538 if isinstance(additional_pipeline, str): 538 ↛ 539line 538 didn't jump to line 539 because the condition on line 538 was never true
539 additional_pipeline = Pipeline.fromFile(additional_pipeline)
540 additional_tasks.update(additional_pipeline.task_labels)
541 pipeline.mergePipeline(additional_pipeline)
543 # Add all tasks to subset_name; create the subset if it does not exist
544 subset_text = ""
545 if additional_subset is not None: 545 ↛ 573line 545 didn't jump to line 573 because the condition on line 545 was always true
546 if not isinstance(additional_subset, list): 546 ↛ 547line 546 didn't jump to line 547 because the condition on line 546 was never true
547 additional_subset = [additional_subset]
548 subset_names_old = []
549 subset_names_new = []
550 for subset in additional_subset:
551 # Parse the subset definition
552 if ":" in subset: 552 ↛ 555line 552 didn't jump to line 555 because the condition on line 552 was always true
553 subset_name, subset_description = subset.split(":", 1)
554 else:
555 subset_name = subset
556 subset_description = ""
557 # Add or create the subset with all additional tasks
558 if subset_name in pipeline.subsets: 558 ↛ 559line 558 didn't jump to line 559 because the condition on line 558 was never true
559 subset_names_old.append(subset_name)
560 for additional_task in additional_tasks:
561 pipeline.addLabelToSubset(subset_name, additional_task)
562 else:
563 subset_names_new.append(subset_name)
564 pipeline.addLabeledSubset(subset_name, subset_description, additional_tasks)
565 if subset_names_old: 565 ↛ 566line 565 didn't jump to line 566 because the condition on line 565 was never true
566 subset_text += f", and to existing subset{'s' if len(subset_names_old) > 1 else ''} "
567 subset_text += f"{', '.join(sorted(subset_names_old))}"
568 if subset_names_new: 568 ↛ 573line 568 didn't jump to line 573 because the condition on line 568 was always true
569 subset_text += f", and to new subset{'s' if len(subset_names_new) > 1 else ''} "
570 subset_text += f"{', '.join(sorted(subset_names_new))}"
572 # Revalidate the pipeline graph
573 _ = _get_pipeline_graph(pipeline, logger)
575 grammar = "task" if len(additional_tasks) == 1 else "tasks"
576 logger.info(
577 "Added %d %s to the pipeline%s: %s",
578 len(additional_tasks),
579 grammar,
580 subset_text,
581 ", ".join(sorted(additional_tasks)),
582 )
585def make_injection_pipeline(
586 dataset_type_name: str,
587 reference_pipeline: Pipeline | str,
588 injection_pipeline: Pipeline | str | None = None,
589 update_subsets: bool = True,
590 excluded_tasks: set[str] | str = {
591 "jointcal",
592 "gbdesAstrometricFit",
593 "fgcmBuildFromIsolatedStars",
594 "fgcmFitCycle",
595 "fgcmOutputProducts",
596 },
597 prefix: str = "injected_",
598 instrument: str | None = None,
599 config: str | list[str] | None = None,
600 additional_pipelines: list[Pipeline] | list[str] | None = None,
601 additional_subset: list[str] | str | None = None,
602 log_level: int = logging.INFO,
603) -> Pipeline:
604 """Make an expanded source injection pipeline.
606 This function takes a reference pipeline definition file and prefixes all
607 immediately post-injection dataset type names with the injected prefix. If
608 an optional injection pipeline definition YAML file is also provided, the
609 injection task will be merged into the pipeline.
611 Unless subset updates are explicitly disabled, all subsets from the
612 reference pipeline containing the task which generates the injection
613 dataset type will also be updated to include the injection task.
615 When the injection pipeline is constructed, a check on all existing
616 pipeline contracts is performed. If any contracts are violated, they're
617 removed from the pipeline. A warning is logged for each contract that is
618 removed.
620 Parameters
621 ----------
622 dataset_type_name : `str`
623 Name of the dataset type being injected into.
624 reference_pipeline : Pipeline | `str`
625 Location of a reference pipeline definition YAML file.
626 injection_pipeline : Pipeline | `str`, optional
627 Location of an injection pipeline definition YAML file stub. If not
628 provided, an attempt to infer the injection pipeline will be made based
629 on the injected dataset type name.
630 update_subsets : `bool`, optional
631 If True, update pipeline subsets to include the injection task.
632 excluded_tasks : `set` [`str`] | `str`
633 Set of task labels to exclude, or a comma-separated string of labels.
634 prefix : `str`, optional
635 Prefix to prepend to each affected post-injection dataset type name.
636 instrument : `str`, optional
637 Add instrument overrides. Must be a fully qualified class name.
638 config : `str` | `list` [`str`], optional
639 Config override for a task, in the format 'label:key=value'.
640 additional_pipelines: `list`[Pipeline] | `list`[`str`], optional
641 Additional pipelines to merge into the output pipeline, or their YAML
642 file locations. Tasks from these additional pipelines will be added to
643 the output injection pipeline.
644 additional_subset: `list`[`str`] | `str`, optional
645 A list of subset definitions in the form
646 "subset_name[:subset_description]".
647 These subsets will be created if they don't already exist.
648 All tasks from additional_pipelines will be added to these subsets.
649 log_level : `int`, optional
650 The log level to use for logging.
652 Returns
653 -------
654 pipeline : `lsst.pipe.base.Pipeline`
655 An expanded source injection pipeline.
656 """
657 logger = logging.getLogger(__name__)
658 logger.setLevel(log_level)
660 # Get the main reference pipeline
661 if isinstance(reference_pipeline, str): 661 ↛ 662line 661 didn't jump to line 662 because the condition on line 661 was never true
662 pipeline = Pipeline.fromFile(reference_pipeline)
663 else:
664 pipeline = reference_pipeline
666 # Add an instrument override
667 if instrument: 667 ↛ 671line 667 didn't jump to line 671 because the condition on line 667 was always true
668 pipeline.addInstrument(instrument)
670 # Infer the injection pipeline if not provided, and where possible
671 if not injection_pipeline: 671 ↛ 672line 671 didn't jump to line 672 because the condition on line 671 was never true
672 injection_pipeline = _infer_injection_pipeline(
673 dataset_type_name,
674 logger,
675 )
677 # Merge the injection pipeline into the main pipeline
678 injection_task_label = _merge_injection_pipeline(pipeline, injection_pipeline, dataset_type_name, prefix)
680 # Apply all user-supplied config overrides
681 if config is not None: 681 ↛ 682line 681 didn't jump to line 682 because the condition on line 681 was never true
682 _configure_injection_pipeline(pipeline, config, logger)
684 # Remove excluded tasks from the pipeline, and remove any empty subsets
685 pipeline = _remove_excluded_tasks(pipeline, excluded_tasks, logger)
687 # Prefix post-injection dataset type name connections and update subsets
688 _reconfigure_injection_pipeline(
689 pipeline, dataset_type_name, prefix, injection_task_label, update_subsets, logger
690 )
692 # Optionally include additional tasks in the injection pipeline.
693 if additional_pipelines: 693 ↛ 696line 693 didn't jump to line 696 because the condition on line 693 was always true
694 _add_additional_pipelines(pipeline, additional_pipelines, additional_subset, logger)
696 return pipeline