Move PipelineTask Config override handling

Move the handling of PipelineTaskConfig override handling into
the Config class and out of the Pipeline. This allows Config
classes more control over how overrides are applied, and a place
to add a hook to react to config values.

doc/changes/DM-39100.feature.md

@@ -0,0 +1 @@
+Added a method on `PipelineTaskConfig` objects named `applyConfigOverrides`. This method is called the system executing `PipelineTask`\ s within a pipeline, and is passed the instrument and config overrides defined within the pipeline for that task.

python/lsst/pipe/base/config.py

@@ -29,15 +29,20 @@
 # -------------------------------
 #  Imports of standard modules --
 # -------------------------------
+import os
 from numbers import Number
+from collections.abc import Iterable
 from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Type, TypeVar
 
 # -----------------------------
 #  Imports for other modules --
 # -----------------------------
 import lsst.pex.config as pexConfig
 
+from ._instrument import Instrument
+from .configOverrides import ConfigOverrides
 from .connections import PipelineTaskConnections
+from .pipelineIR import ConfigIR, ParametersIR
 
 if TYPE_CHECKING:
     from lsst.pex.config.callStack import StackFrame
@@ -199,6 +204,54 @@ class to allow configuration of the connections class. This dynamically
         doc="Flag to enable/disable saving of log output for a task, enabled by default.",
     )
 
+    def applyConfigOverrides(
+        self,
+        instrument: Instrument | None,
+        taskDefaultName: str,
+        pipelineConfigs: Iterable[ConfigIR] | None,
+        parameters: ParametersIR,
+        label: str,
+    ) -> None:
+        r"""Apply config overrides to this config instance.
+
+        Parameters
+        ---------
+        instrument : `Instrument` or `None`
+            An instance of the `Instrument` specified in a pipeline.
+            If `None` then the pipeline did not specify and instrument.
+        taskDefaultName : `str`
+            The default name associated with the `Task` class. This
+            may be used with instrumental overrides.
+        pipelineConfigs : `Iterable` of `ConfigIR`
+            An iterable of `ConfigIR` objects that contain overrides
+            to apply to this config instance.
+        parameters : `ParametersIR`
+            Parameters defined in a Pipeline which are used in formatting
+            of config values across multiple `Task`\ s in a pipeline.
+        label : `str`
+            The label associated with this class's Task in a pipeline.
+        """
+        overrides = ConfigOverrides()
+        if instrument is not None:
+            overrides.addInstrumentOverride(instrument, taskDefaultName)
+        if pipelineConfigs is not None:
+            for subConfig in (configIr.formatted(parameters) for configIr in pipelineConfigs):
+                if subConfig.dataId is not None:
+                    raise NotImplementedError(
+                        "Specializing a config on a partial data id is not yet "
+                        "supported in Pipeline definition"
+                    )
+                # only apply override if it applies to everything
+                if subConfig.dataId is None:
+                    if subConfig.file:
+                        for configFile in subConfig.file:
+                            overrides.addFileOverride(os.path.expandvars(configFile))
+                    if subConfig.python is not None:
+                        overrides.addPythonOverride(subConfig.python)
+                    for key, value in subConfig.rest.items():
+                        overrides.addValueOverride(key, value)
+        overrides.applyTo(self)
+
 
 class ResourceConfig(pexConfig.Config):
     """Configuration for resource requirements.

python/lsst/pipe/base/configOverrides.py

@@ -30,7 +30,8 @@
 from operator import attrgetter
 
 from lsst.resources import ResourcePath
-from lsst.utils import doImportType
+
+from ._instrument import Instrument
 
 OverrideTypes = Enum("OverrideTypes", "Value File Python Instrument")
 
@@ -183,21 +184,18 @@ def addPythonOverride(self, python_snippet: str) -> None:
         """
         self._overrides.append((OverrideTypes.Python, python_snippet))
 
-    def addInstrumentOverride(self, instrument: str, task_name: str) -> None:
+    def addInstrumentOverride(self, instrument: Instrument, task_name: str) -> None:
         """Apply any overrides that an instrument has for a task
 
         Parameters
         ----------
-        instrument: str
-            A string containing the fully qualified name of an instrument from
-            which configs should be loaded and applied
+        instrument: `Instrument`
+            An instrument instance which will apply configs
         task_name: str
             The _DefaultName of a task associated with a config, used to look
             up overrides from the instrument.
         """
-        instrument_cls: type = doImportType(instrument)
-        instrument_lib = instrument_cls()
-        self._overrides.append((OverrideTypes.Instrument, (instrument_lib, task_name)))
+        self._overrides.append((OverrideTypes.Instrument, (instrument, task_name)))
 
     def _parser(self, value, configParser):
         try:

python/lsst/pipe/base/pipeline.py

@@ -27,7 +27,6 @@
 
 import copy
 import logging
-import os
 import re
 import urllib.parse
 
@@ -62,9 +61,9 @@
 from lsst.utils.introspection import get_full_type_name
 
 from . import pipelineIR, pipeTools
+from ._instrument import Instrument as PipeBaseInstrument
 from ._task_metadata import TaskMetadata
 from .config import PipelineTaskConfig
-from .configOverrides import ConfigOverrides
 from .connections import iterConnections
 from .connectionTypes import Input
 from .pipelineTask import PipelineTask
@@ -773,26 +772,13 @@ def _buildTaskDef(self, label: str) -> TaskDef:
         taskClass: Type[PipelineTask] = doImportType(taskIR.klass)
         taskName = get_full_type_name(taskClass)
         config = taskClass.ConfigClass()
-        overrides = ConfigOverrides()
-        if self._pipelineIR.instrument is not None:
-            overrides.addInstrumentOverride(self._pipelineIR.instrument, taskClass._DefaultName)
-        if taskIR.config is not None:
-            for configIR in (configIr.formatted(self._pipelineIR.parameters) for configIr in taskIR.config):
-                if configIR.dataId is not None:
-                    raise NotImplementedError(
-                        "Specializing a config on a partial data id is not yet "
-                        "supported in Pipeline definition"
-                    )
-                # only apply override if it applies to everything
-                if configIR.dataId is None:
-                    if configIR.file:
-                        for configFile in configIR.file:
-                            overrides.addFileOverride(os.path.expandvars(configFile))
-                    if configIR.python is not None:
-                        overrides.addPythonOverride(configIR.python)
-                    for key, value in configIR.rest.items():
-                        overrides.addValueOverride(key, value)
-        overrides.applyTo(config)
+        instrument: PipeBaseInstrument | None = None
+        if (instrumentName := self._pipelineIR.instrument) is not None:
+            instrument_cls: type = doImportType(instrumentName)
+            instrument = instrument_cls()
+        config.applyConfigOverrides(
+            instrument, taskClass._DefaultName, taskIR.config, self._pipelineIR.parameters, label
+        )
         return TaskDef(taskName=taskName, config=config, taskClass=taskClass, label=label)
 
     def __iter__(self) -> Generator[TaskDef, None, None]:

tests/test_pipeTools.py

@@ -76,6 +76,7 @@ def _makeConfig(inputName, outputName, pipeline, label):
 
 class ExamplePipelineTask(PipelineTask):
     ConfigClass = ExamplePipelineTaskConfig
+    _DefaultName = "examplePipelineTask"
 
 
 def _makePipeline(tasks):